mirror of
https://github.com/abhigyanpatwari/GitNexus.git
synced 2026-10-27 06:12:20 +00:00
feat: shared resilient-fetch (retries + circuit breaker) (#1448)
* feat: shared resilient-fetch (retries + circuit breaker)
Add a small, runtime-agnostic resilience layer in gitnexus-shared and
migrate every backend HTTP outbound call (CLI, MCP, wiki LLM, web → backend)
through it.
Helpers (gitnexus-shared/src/integrations/):
- retry.ts — withRetry(fn, opts) with caller-supplied
retryability classification and full-jitter
exponential backoff.
- circuit-breaker.ts — closed/open/half-open per-process breaker with
injectable clock, plus a keyed registry so
callers targeting the same endpoint share state.
- resilient-fetch.ts — composed wrapper: retries 5xx + 429 + retryable
network throws, treats AbortSignal.timeout()
and 4xx (other than 429) as terminal, honors
Retry-After (capped at 30s), throws
CircuitOpenError when the breaker opens.
Migrations (no behaviour regression — all existing tests pass):
- gitnexus/src/core/embeddings/http-client.ts (covers analyze + MCP
query path) — replaces inline linear-backoff retry.
- gitnexus/src/core/wiki/llm-client.ts — preserves Azure content-filter
branch; resilientFetch handles 5xx/429.
- gitnexus-web/src/services/backend-client.ts (fetchWithTimeout helper)
— small retry budget (2 attempts, 250–1500 ms) so a dead local
backend still fails fast for the user.
- gitnexus-web/src/core/llm/settings-service.ts (OpenRouter model list).
Deliberately not migrated:
- gitnexus-web/src/services/backend-client.ts streamJob() — Server-Sent
Events stream; the existing reconnect-with-Last-Event-ID logic is
not unary-fetch shaped.
- gitnexus-web/src/components/SettingsPanel.tsx checkOllamaStatus() —
one-shot health probe; retrying delays the "Ollama not running"
error rather than improving UX.
41 new helper tests cover backoff math, breaker state transitions,
Retry-After parsing (delta-seconds + HTTP-date), 401/422 terminal
classification, and breaker fail-fast on three exhausted retry batches.
* fix(review): apply autofix feedback
Address Claude's two MEDIUM blocking findings on PR #1448 plus the
CodeQL SSRF false-positive flag.
- backend-client `fetchWithTimeout` now uses `AbortSignal.timeout()`
merged with the caller's signal via `AbortSignal.any()`. Timer-fired
aborts surface as `DOMException(name='TimeoutError')` so
resilientFetch routes them through the terminal-network branch
(no retry, no breaker hit), instead of incrementing the breaker
for user-side network slowness.
- Method-aware retry budget in `fetchWithTimeout`: idempotent verbs
(GET/HEAD/OPTIONS) keep the 2-attempt budget; POST/PATCH/PUT/DELETE
default to single-attempt so a 5xx on `startAnalyze` cannot start
a duplicate job. New `forceRetry` parameter for callers that
know-idempotent mutations (e.g. DELETE of a known-deleted resource).
- `resilient-fetch.ts` carries a documented suppression for CodeQL
js/server-side-request-forgery on the inner fetch call. Every
concrete caller passes a hardcoded URL constant or a value from
configuration (env vars, saved settings); user request input never
flows into the URL parameter.
- New test file `backend-client-retry.test.ts` covers all three
paths: GET retries on 503, POST does not retry, timeout does not
increment the breaker.
* fix(resilient-fetch): address Codex adversarial findings
Closes the three blocking issues from Codex's review on PR #1448.
U1 — Add `recordNeutral()` to CircuitBreaker.
Third outcome path that's an explicit no-op for state and the
consecutive-failure counter. Distinct from `recordSuccess` (closes
the breaker) and `recordFailure` (may open it). Used for outcomes
that are neither evidence of backend health nor evidence of
backend failure.
U2 — Route terminal-client / terminal-network through `recordNeutral`.
Previously a 401 or local timeout called `recordSuccess`, which
reset `consecutiveFailures` to 0. A 5xx → 401 → 5xx → 401 → 5xx
sequence would NEVER trip the breaker because each 4xx in between
erased the running count. Also classify external `AbortError` as
terminal-network (was retryable-network), so caller-driven
cancellation no longer retries against an already-aborted signal
or counts toward breaker failures on exhaustion.
U3 — Per-origin breaker key in web `fetchWithTimeout`.
Was hardcoded to `'web-backend'` even though `_backendUrl` is
mutable via `setBackendUrl`. Switching backend URLs after a
circuit tripped on host-A would strand the user during the full
cooldown. Key is now `web-backend:<origin>`, so each backend URL
gets its own breaker state.
Tests: +5 recordNeutral, +4 resilient-fetch (interleaved 4xx/5xx,
external AbortError, prior-state preservation), +1 web switch-backend
regression. All 70 gitnexus integration tests + 15 web tests green.
* fix(resilient-fetch): tolerate header-less fetch mocks on 429
`classifyOutcome` called `resp.headers.get('Retry-After')` directly,
which crashed when a test stubs `fetch` with a plain object like
`{ ok: false, status: 429 }` (no `headers` field). Real `Response`
always has Headers, so this surfaces only in test setups, but the
helper has no business assuming caller-side correctness on this — the
defensive guard is cheap and a missing `Retry-After` falls through to
exponential-backoff retry like any 429 without the header.
Surfaced by `gitnexus/test/unit/http-embedder.test.ts > retries on
rate limit`, which the embeddings migration exercises against a
plain-object 429 stub. Locked in with a new
`classifies 429 from a header-less fetch mock without throwing` case.
* fix(review): apply autofix feedback
Closes findings from the third multi-agent review pass on PR #1448.
#1 (P1) callLLM had no per-attempt timeout
Wiki LLM calls passed no `signal` to resilientFetch; each of three
retry attempts could hang indefinitely on a frozen TCP connection.
Add `signal: AbortSignal.timeout(60_000)` so the per-attempt budget
matches what http-client.ts and backend-client.ts already provide.
#2 (P2) drop dead `lastRetryableResp` post-loop fallback
Variable was set in one switch arm but only read in unreachable code
after the loop. The retry loop always returns/throws on every
iteration. Keep only the defensive `throw` so TypeScript's
control-flow analysis still sees `Promise<Response>` as the return.
#5 (P2) gate test-only exports behind a subpath
`__resetBreakerRegistry__` and `classifyOutcome` were reachable from
the main `gitnexus-shared` barrel — production code calling
`__resetBreakerRegistry__` from a tool implementation would silently
nuke every circuit breaker process-wide. Move to a new
`gitnexus-shared/test-helpers` subpath export. Production callers
see the cleaner public API; tests import via the explicit
`gitnexus-shared/test-helpers` path.
#6 (P2) exhaustiveness guard on Outcome switch
Add a `default: const _: never = outcome` arm so a future sixth
`Outcome.kind` won't compile silently — it'll surface at the switch
site rather than fall through to a retry/no-retry default.
#9 (P3) document cumulative wall-clock budget
Add a "Cumulative wall-clock budget" paragraph to resilientFetch's
JSDoc explaining the worst-case total wait (`maxAttempts × (per-attempt
timeout + capDelayMs)` ≈ 60s with defaults) and pointing callers at
outer `AbortSignal.timeout()` when they want a tighter bound.
Deferred to follow-up PRs (per review's Auto-resolve recommendation):
- #3 idempotency knob to shared API (forceRetry into ResilientFetchOptions)
- #4 publish.ts migration to resilientFetch
- #7 parseRetryAfter past-HTTP-date / negative-seconds asymmetry
- #8 recordNeutral counter time-decay (documented breaker semantic)
* fix(circuit-breaker): gate half-open to a single in-flight probe
Closes the Codex adversarial-review finding on PR #1448 that flagged a
recovery-time thundering herd: when cooldown expired, every concurrent
caller transitioned the breaker to half-open and probed the still-
recovering dependency in lockstep, defeating the breaker's "fail fast"
promise.
U1 — probe-permit gate in CircuitBreaker.check()
Added a `probeInFlight: boolean` field. After cooldown expires, the
first `check()` admits the probe and consumes the permit; subsequent
callers throw `CircuitOpenError` with a configurable
`halfOpenRetryAfterMs` (default 1000ms) until the probe resolves.
Critical design point: `recordNeutral` now RELEASES the permit but
does NOT transition state. Without that split, a single `TimeoutError`
from per-attempt `AbortSignal.timeout` (which routes through neutral
classification) would permanently park the breaker in half-open. By
separating permit-release from state-resolution, we keep the
"neutral doesn't claim health" semantic without creating that wedge.
Other changes:
- `halfOpenRetryAfterMs` is now a constructor option for consumers
with long-running protected ops (LLM streaming, large uploads).
- `getState()` is documented as a pure read; the implicit
Open -> Half-Open transition lives in `check()` only, so tests
that inspect state never inadvertently consume a probe permit.
- `isProbeInFlight()` test-only accessor for assertion clarity.
- JSDoc on `check()` records the JS event-loop atomicity dependency
and the load-bearing `try/finally` pairing invariant.
U2 — End-to-end concurrency regression through resilientFetch
Three new scenarios in resilient-fetch.test.ts (26 -> 29):
- 3 concurrent calls + probe gets 200 -> 1 hits fetch, 2 throw
CircuitOpenError, breaker closes.
- 3 concurrent calls + probe gets 503 -> ResilientFetchExhaustedError
on probe; concurrent callers see halfOpenRetryAfterMs (1000ms);
fresh caller after probe resolves sees the FULL new cooldown
(10000ms), not the probe-in-flight default.
- Probe cancelled mid-flight via AbortError -> permit released,
state stays half-open, next caller becomes the new probe and
succeeds.
Plus 9 new circuit-breaker unit tests (16 -> 25) covering the permit
gate, recordNeutral-releases-permit semantic, fresh-cooldown distinction,
default vs configurable halfOpenRetryAfterMs, getState() purity, and
the three-probes-via-neutrals chain.
Total integration test count: 70 -> 82. All 106 gitnexus + 15 web
tests pass; both packages typecheck.
Maintainer decisions (deferred per plan 003 Open Questions):
- Plan 002's deferral judgement was reversed on Codex's argument
without new measurement / incident data. The reversal is defensible
on principle (Hystrix / Resilience4j alignment) but lacks workload-
driven evidence.
- Probe-blocked callers throw silently (no log / event hook). R4's
"no new public API" prevents adding observability; loosen if a
debug log on probe-blocked is wanted.
* refactor(embeddings): replace bespoke HF breaker with shared CircuitBreaker
Deleted the local `HfDownloadCircuitBreaker` class and the manual
retry loop in `withHfDownloadRetry`. Both are now backed by the
shared `gitnexus-shared` primitives:
- `hfDownloadCircuit` is `new CircuitBreaker({ failureThreshold,
cooldownMs, key: 'hf-download' })` — same state machine as before
PLUS the single-permit half-open gate that prevents recovery-time
stampedes when CLI + MCP embedders concurrently re-load the model.
- `withHfDownloadRetry` delegates the loop to `withRetry` from the
shared package. Per-attempt timeout (`withDownloadTimeout`),
network-vs-non-network classification, circuit recording, and the
`onRetry` callback wire through `withRetry`'s `isRetryable`
callback.
Behaviour preserved:
- Pre-flight `CIRCUIT_OPEN_TAG` rejection when the breaker is open.
- Mid-loop `CIRCUIT_OPEN_TAG` "opened after N consecutive failures"
when a network error trips the threshold.
- Non-network errors (e.g. CUDA unavailable) bypass retry and go
through `recordNeutral` instead of resetting the breaker's
failure-count progress.
- `onRetry(attempt+1, max, err)` fires only when there's a next
attempt, matching the prior semantic.
Generic CircuitBreaker gained two inspection accessors:
- `getOpenedAt(): number | null`
- `getCooldownMs(): number`
Used by `withHfDownloadRetry` to compute `secsUntilReset` without
consuming a probe permit (which `check()` would do).
Test consolidation: the 7 bespoke `HfDownloadCircuitBreaker`
state-machine tests in hf-env.test.ts were 1:1 duplicates of
existing tests in `circuit-breaker.test.ts` and were deleted.
Remaining 42 hf-env tests all pass; full integration sweep (148
gitnexus + 15 web) green.
This commit is contained in:
Gergő Magyar
GitHub
parent
75f2c9937d
commit
5b70af2f31
16 changed files with 2173 additions and 318 deletions
|
|
@ -10,6 +10,10 @@
|
|||
".": {
|
||||
"types": "./dist/index.d.ts",
|
||||
"default": "./dist/index.js"
|
||||
},
|
||||
"./test-helpers": {
|
||||
"types": "./dist/test-helpers.d.ts",
|
||||
"default": "./dist/test-helpers.js"
|
||||
}
|
||||
},
|
||||
"scripts": {
|
||||
|
|
|
|||
|
|
@ -143,6 +143,22 @@ export type { ScopeTree } from './scope-resolution/scope-tree.js';
|
|||
export { buildPositionIndex } from './scope-resolution/position-index.js';
|
||||
export type { PositionIndex } from './scope-resolution/position-index.js';
|
||||
|
||||
// Resilient fetch primitives — bounded retries + per-process circuit breaker.
|
||||
// Test-only helpers (`__resetBreakerRegistry__`, `classifyOutcome`) are
|
||||
// reachable via the separate `gitnexus-shared/test-helpers` subpath; do
|
||||
// NOT add them here. Production consumers must not call them.
|
||||
export { withRetry, computeBackoffMs } from './integrations/retry.js';
|
||||
export type { RetryOptions, RetryDecision } from './integrations/retry.js';
|
||||
export { CircuitBreaker, CircuitOpenError, getBreaker } from './integrations/circuit-breaker.js';
|
||||
export type { CircuitBreakerOptions } from './integrations/circuit-breaker.js';
|
||||
export {
|
||||
resilientFetch,
|
||||
ResilientFetchExhaustedError,
|
||||
RETRY_AFTER_CAP_MS,
|
||||
parseRetryAfter,
|
||||
} from './integrations/resilient-fetch.js';
|
||||
export type { ResilientFetchOptions } from './integrations/resilient-fetch.js';
|
||||
|
||||
// Understand-Quickly registry integration (opt-in)
|
||||
export {
|
||||
UNDERSTAND_QUICKLY_DISPATCH_URL,
|
||||
|
|
|
|||
273
gitnexus-shared/src/integrations/circuit-breaker.ts
Normal file
273
gitnexus-shared/src/integrations/circuit-breaker.ts
Normal file
|
|
@ -0,0 +1,273 @@
|
|||
/**
|
||||
* Per-process circuit breaker.
|
||||
*
|
||||
* Closed -> Open transition fires after `failureThreshold` consecutive
|
||||
* failures. While Open, `check` throws `CircuitOpenError` until
|
||||
* `cooldownMs` has elapsed since the breaker tripped. The first call
|
||||
* after the cooldown enters Half-Open and consumes the *probe permit*:
|
||||
* a recorded success returns to Closed; a recorded failure flips back
|
||||
* to Open with a fresh timestamp.
|
||||
*
|
||||
* Half-open admits exactly one in-flight probe at a time. Concurrent
|
||||
* callers attempting `check()` while a probe is outstanding receive
|
||||
* `CircuitOpenError` with `retryAfterMs = halfOpenRetryAfterMs` (default
|
||||
* 1000ms; configurable). This prevents the recovery-time thundering
|
||||
* herd that defeats the breaker's "fail fast" promise.
|
||||
*
|
||||
* Outcome reporting splits permit-release from state-resolution:
|
||||
* - `recordSuccess` — releases the probe permit, resets the failure
|
||||
* counter, transitions to Closed. Reserved for true 2xx/3xx outcomes.
|
||||
* - `recordFailure` — releases the probe permit, increments the
|
||||
* consecutive-failure counter, transitions to Open with a fresh
|
||||
* `openedAt` (when called from Half-Open or when the threshold
|
||||
* trips from Closed).
|
||||
* - `recordNeutral` — releases the probe permit, BUT leaves state and
|
||||
* counter untouched. Used for outcomes that are neither evidence of
|
||||
* backend health nor evidence of backend failure (caller-driven
|
||||
* cancellation, local timeout, terminal 4xx client errors). Critical
|
||||
* design point: if `recordNeutral` did not release the permit, a
|
||||
* single `TimeoutError` from per-attempt `AbortSignal.timeout` would
|
||||
* route through `recordNeutral` and permanently park the breaker in
|
||||
* half-open until process restart. Releasing the permit while leaving
|
||||
* state half-open keeps the "neutral doesn't claim health" semantic
|
||||
* without creating that wedge.
|
||||
*
|
||||
* Pairing invariant: every successful `check()` MUST be paired with
|
||||
* exactly one `record*()` on every code path including throws. Direct
|
||||
* consumers should wrap the protected operation in `try/finally`:
|
||||
*
|
||||
* breaker.check();
|
||||
* try {
|
||||
* const result = await operation();
|
||||
* breaker.recordSuccess();
|
||||
* return result;
|
||||
* } catch (err) {
|
||||
* // classify err and call recordFailure / recordNeutral / etc.
|
||||
* throw err;
|
||||
* }
|
||||
*
|
||||
* `resilientFetch`'s catch-all on `fetchImpl` already satisfies this
|
||||
* for that consumer.
|
||||
*
|
||||
* Atomicity model: the half-open gate relies on JavaScript event-loop
|
||||
* single-threadedness within a synchronous `check()` body. There is no
|
||||
* `await` inside `check()`; concurrent callers serialize on microtask
|
||||
* order, and exactly one observes `probeInFlight === false`. Do not
|
||||
* introduce `await` inside `check()` without revisiting the gate. If
|
||||
* this code is ever ported to a runtime with shared-memory threads
|
||||
* (Node `worker_threads` with `SharedArrayBuffer`, Web Workers with
|
||||
* shared registries), the boolean must become an atomic CAS — Resilience4j
|
||||
* and Hystrix use atomic permits *because* they run in JVM thread pools.
|
||||
*
|
||||
* Runtime-agnostic: depends only on a `now()` clock and standard JS —
|
||||
* no Node-only imports. Tests inject `now` to advance the clock
|
||||
* deterministically without `vi.useFakeTimers()`.
|
||||
*/
|
||||
|
||||
export class CircuitOpenError extends Error {
|
||||
override readonly name = 'CircuitOpenError';
|
||||
/** Approximate wait time before the breaker may transition to Half-Open
|
||||
* (or before the in-flight probe is expected to resolve). */
|
||||
readonly retryAfterMs: number;
|
||||
|
||||
constructor(retryAfterMs: number, key?: string) {
|
||||
super(
|
||||
key
|
||||
? `Circuit '${key}' is open; retry in ${Math.ceil(retryAfterMs / 1000)}s`
|
||||
: `Circuit is open; retry in ${Math.ceil(retryAfterMs / 1000)}s`,
|
||||
);
|
||||
this.retryAfterMs = retryAfterMs;
|
||||
}
|
||||
}
|
||||
|
||||
export interface CircuitBreakerOptions {
|
||||
/** Consecutive failures required to trip Closed -> Open. */
|
||||
failureThreshold?: number;
|
||||
/** Milliseconds Open before the next call may probe (Half-Open). */
|
||||
cooldownMs?: number;
|
||||
/**
|
||||
* Milliseconds to suggest in `CircuitOpenError.retryAfterMs` when the
|
||||
* breaker is Half-Open with the probe permit consumed. Default 1000ms.
|
||||
* Consumers with long-running protected ops (LLM streaming, large
|
||||
* uploads) should raise this — the cooldown clock is no longer the
|
||||
* right answer because cooldown has elapsed. Returning 0 invites
|
||||
* retry storms; returning the full cooldown misleads about wait.
|
||||
*/
|
||||
halfOpenRetryAfterMs?: number;
|
||||
/** Optional key for error messages and registry lookups. */
|
||||
key?: string;
|
||||
/** Clock override — defaults to `Date.now`. Tests inject deterministic time. */
|
||||
now?: () => number;
|
||||
}
|
||||
|
||||
type State = 'closed' | 'open' | 'half-open';
|
||||
|
||||
export class CircuitBreaker {
|
||||
private readonly failureThreshold: number;
|
||||
private readonly cooldownMs: number;
|
||||
private readonly halfOpenRetryAfterMs: number;
|
||||
private readonly key: string | undefined;
|
||||
private readonly now: () => number;
|
||||
|
||||
private state: State = 'closed';
|
||||
private consecutiveFailures = 0;
|
||||
private openedAt: number | null = null;
|
||||
/**
|
||||
* True between a successful `check()` and the next `record*()` call
|
||||
* during Half-Open. Gates concurrent callers from stampeding a still-
|
||||
* recovering dependency. Boolean rather than counter — single-permit
|
||||
* is the conservative end of the Hystrix/Resilience4j spectrum.
|
||||
*/
|
||||
private probeInFlight = false;
|
||||
|
||||
constructor(opts: CircuitBreakerOptions = {}) {
|
||||
this.failureThreshold = opts.failureThreshold ?? 3;
|
||||
this.cooldownMs = opts.cooldownMs ?? 30_000;
|
||||
this.halfOpenRetryAfterMs = opts.halfOpenRetryAfterMs ?? 1_000;
|
||||
this.key = opts.key;
|
||||
this.now = opts.now ?? (() => Date.now());
|
||||
}
|
||||
|
||||
/**
|
||||
* Throw `CircuitOpenError` if the breaker won't admit this call.
|
||||
* Otherwise consume the half-open probe permit (if applicable) and
|
||||
* return so the caller can attempt the protected work.
|
||||
*
|
||||
* Three rejection paths:
|
||||
* 1. Open and still in cooldown → throws with `retryAfterMs` =
|
||||
* remaining cooldown.
|
||||
* 2. Open with cooldown elapsed AND a probe is already in flight
|
||||
* (race: another caller transitioned to half-open and grabbed
|
||||
* the permit on a microtask before us) → throws with
|
||||
* `halfOpenRetryAfterMs`.
|
||||
* 3. Half-Open with probe in flight → throws with `halfOpenRetryAfterMs`.
|
||||
*
|
||||
* **Pairing invariant**: every successful return from `check()` MUST
|
||||
* be paired with exactly one `recordSuccess` / `recordFailure` /
|
||||
* `recordNeutral` on every code path including thrown exceptions.
|
||||
* Failing to pair leaves the probe permit consumed forever and
|
||||
* wedges the breaker. See file-header JSDoc for the canonical
|
||||
* try/finally pattern.
|
||||
*/
|
||||
check(): void {
|
||||
if (this.state === 'open' && this.openedAt !== null) {
|
||||
const elapsed = this.now() - this.openedAt;
|
||||
if (elapsed < this.cooldownMs) {
|
||||
throw new CircuitOpenError(this.cooldownMs - elapsed, this.key);
|
||||
}
|
||||
// Cooldown elapsed — transition to Half-Open. The very next
|
||||
// `probeInFlight` check below decides whether THIS caller gets
|
||||
// the permit or hits the gate.
|
||||
this.state = 'half-open';
|
||||
}
|
||||
|
||||
if (this.state === 'half-open') {
|
||||
if (this.probeInFlight) {
|
||||
throw new CircuitOpenError(this.halfOpenRetryAfterMs, this.key);
|
||||
}
|
||||
this.probeInFlight = true;
|
||||
}
|
||||
// Closed state falls through silently.
|
||||
}
|
||||
|
||||
recordSuccess(): void {
|
||||
this.probeInFlight = false;
|
||||
this.consecutiveFailures = 0;
|
||||
this.state = 'closed';
|
||||
this.openedAt = null;
|
||||
}
|
||||
|
||||
recordFailure(): void {
|
||||
this.probeInFlight = false;
|
||||
this.consecutiveFailures += 1;
|
||||
if (this.state === 'half-open' || this.consecutiveFailures >= this.failureThreshold) {
|
||||
this.state = 'open';
|
||||
this.openedAt = this.now();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Releases the probe permit BUT leaves state and counter untouched.
|
||||
* Use when an attempt produced a response or error that should not
|
||||
* influence breaker health in either direction — caller-driven aborts,
|
||||
* local AbortSignal timeouts, terminal 4xx client errors.
|
||||
*
|
||||
* Why permit-release-without-state-resolution: if `recordNeutral` did
|
||||
* not clear `probeInFlight`, a single `TimeoutError` from per-attempt
|
||||
* `AbortSignal.timeout` (which routes through neutral classification)
|
||||
* would permanently park the breaker in half-open. Since timeouts are
|
||||
* an *expected* outcome under flaky-dependency conditions, the cited
|
||||
* "per-attempt timeout bounds the stuck state" mitigation would itself
|
||||
* be the trigger for a permanent wedge. Releasing the permit closes
|
||||
* that loop while keeping the "neutral doesn't claim dependency
|
||||
* health" semantic.
|
||||
*
|
||||
* Calling `recordSuccess` for these would erase legitimate prior
|
||||
* failure signal; calling `recordFailure` would trip the breaker for
|
||||
* outcomes the backend isn't responsible for.
|
||||
*/
|
||||
recordNeutral(): void {
|
||||
this.probeInFlight = false;
|
||||
// State and consecutiveFailures are preserved by design.
|
||||
}
|
||||
|
||||
/**
|
||||
* Pure read — no state mutation, no permit accounting. Returns the
|
||||
* *would-be* state at the current instant: 'half-open' if the breaker
|
||||
* is open with cooldown elapsed (regardless of whether a probe is in
|
||||
* flight), 'open' if open and still in cooldown, 'closed' otherwise.
|
||||
*
|
||||
* Inspection-only; safe to call from tests without consuming a probe
|
||||
* permit. The implicit Open -> Half-Open transition that mutates
|
||||
* `state` lives in `check()` only.
|
||||
*/
|
||||
getState(): State {
|
||||
if (this.state === 'open' && this.openedAt !== null) {
|
||||
const elapsed = this.now() - this.openedAt;
|
||||
if (elapsed >= this.cooldownMs) return 'half-open';
|
||||
}
|
||||
return this.state;
|
||||
}
|
||||
getConsecutiveFailures(): number {
|
||||
return this.consecutiveFailures;
|
||||
}
|
||||
/** Inspection-only test accessor for the half-open probe permit. */
|
||||
isProbeInFlight(): boolean {
|
||||
return this.probeInFlight;
|
||||
}
|
||||
/** Timestamp (ms since epoch) when the breaker last transitioned to Open,
|
||||
* or `null` if it's currently Closed. Useful for computing remaining
|
||||
* cooldown without consuming a probe permit via `check()`. */
|
||||
getOpenedAt(): number | null {
|
||||
return this.openedAt;
|
||||
}
|
||||
/** Configured cooldown duration in milliseconds. */
|
||||
getCooldownMs(): number {
|
||||
return this.cooldownMs;
|
||||
}
|
||||
}
|
||||
|
||||
// ─── Per-process registry ────────────────────────────────────────────
|
||||
//
|
||||
// Single shared map keyed on caller-chosen strings. Used by
|
||||
// `resilient-fetch.ts` so multiple call sites targeting the same logical
|
||||
// endpoint share breaker state. Per-process only — not persisted.
|
||||
|
||||
const registry = new Map<string, CircuitBreaker>();
|
||||
|
||||
export function getBreaker(key: string, opts?: CircuitBreakerOptions): CircuitBreaker {
|
||||
let breaker = registry.get(key);
|
||||
if (!breaker) {
|
||||
breaker = new CircuitBreaker({ ...opts, key });
|
||||
registry.set(key, breaker);
|
||||
}
|
||||
return breaker;
|
||||
}
|
||||
|
||||
/**
|
||||
* Test-only: clear all registered breakers. Tests must call this in
|
||||
* `beforeEach` to prevent breaker state from leaking across test cases.
|
||||
*/
|
||||
export function __resetBreakerRegistry__(): void {
|
||||
registry.clear();
|
||||
}
|
||||
279
gitnexus-shared/src/integrations/resilient-fetch.ts
Normal file
279
gitnexus-shared/src/integrations/resilient-fetch.ts
Normal file
|
|
@ -0,0 +1,279 @@
|
|||
/**
|
||||
* `resilientFetch` — fetch wrapped in retry + circuit breaker, with
|
||||
* GitHub-flavoured retry classification baked in (Retry-After parsing,
|
||||
* 401/403/404/422 treated as terminal client errors).
|
||||
*
|
||||
* Designed for the `gitnexus publish` GitHub `repository_dispatch`
|
||||
* call, but the classification rules apply to any GitHub REST endpoint.
|
||||
* Runtime-agnostic — no Node-only imports.
|
||||
*/
|
||||
|
||||
import {
|
||||
CircuitBreaker,
|
||||
CircuitOpenError,
|
||||
getBreaker,
|
||||
type CircuitBreakerOptions,
|
||||
} from './circuit-breaker.js';
|
||||
import { computeBackoffMs, type RetryOptions } from './retry.js';
|
||||
|
||||
export { CircuitOpenError };
|
||||
|
||||
export interface ResilientFetchOptions {
|
||||
/** Optional fetch implementation override. Defaults to `globalThis.fetch`. */
|
||||
fetchImpl?: typeof fetch;
|
||||
/**
|
||||
* Logical key for the breaker. Defaults to `<host><pathname>` of the
|
||||
* request URL — call sites targeting the same endpoint share breaker
|
||||
* state regardless of query-string differences.
|
||||
*/
|
||||
breakerKey?: string;
|
||||
/** Per-call breaker override. Used for tests and one-off configuration. */
|
||||
breaker?: CircuitBreaker;
|
||||
/** Tuning knobs for the breaker registered under `breakerKey`. */
|
||||
breakerOptions?: CircuitBreakerOptions;
|
||||
/** Tuning knobs for the retry helper. */
|
||||
retry?: Partial<Pick<RetryOptions, 'maxAttempts' | 'baseDelayMs' | 'capDelayMs'>> & {
|
||||
sleep?: RetryOptions['sleep'];
|
||||
random?: RetryOptions['random'];
|
||||
};
|
||||
/** Clock override propagated into Retry-After HTTP-date math and breaker. */
|
||||
now?: () => number;
|
||||
}
|
||||
|
||||
/** Cap on any single Retry-After wait — protects CLI from a buggy registry. */
|
||||
export const RETRY_AFTER_CAP_MS = 30_000;
|
||||
|
||||
const DEFAULT_RETRY = {
|
||||
maxAttempts: 3,
|
||||
baseDelayMs: 500,
|
||||
capDelayMs: 5_000,
|
||||
};
|
||||
|
||||
/**
|
||||
* Parse a `Retry-After` header value into milliseconds.
|
||||
* Accepts either a delta-seconds integer (`"30"`) or an HTTP-date.
|
||||
* Returns null on parse failure or negative deltas.
|
||||
*/
|
||||
export function parseRetryAfter(value: string | null, now: () => number = Date.now): number | null {
|
||||
if (!value) return null;
|
||||
const trimmed = value.trim();
|
||||
if (trimmed === '') return null;
|
||||
|
||||
if (/^[0-9]+$/.test(trimmed)) {
|
||||
const seconds = parseInt(trimmed, 10);
|
||||
if (Number.isNaN(seconds) || seconds < 0) return null;
|
||||
return seconds * 1000;
|
||||
}
|
||||
|
||||
const target = Date.parse(trimmed);
|
||||
if (Number.isNaN(target)) return null;
|
||||
const delta = target - now();
|
||||
return delta >= 0 ? delta : 0;
|
||||
}
|
||||
|
||||
/** Internal: outcome classification used by the resilientFetch loop. */
|
||||
type Outcome =
|
||||
| { kind: 'success'; resp: Response }
|
||||
| { kind: 'terminal-client'; resp: Response } // 4xx other than 429: no retry, breaker neutral
|
||||
| { kind: 'retryable-status'; resp: Response; afterMs: number | undefined } // 5xx, 429
|
||||
| { kind: 'terminal-network'; err: unknown } // TimeoutError or AbortError: no retry, breaker neutral
|
||||
| { kind: 'retryable-network'; err: unknown }; // DNS, ECONNRESET, etc.
|
||||
|
||||
/** Exported for unit tests. */
|
||||
export function classifyOutcome(
|
||||
result: { kind: 'error'; err: unknown } | { kind: 'response'; resp: Response },
|
||||
now: () => number,
|
||||
): Outcome {
|
||||
if (result.kind === 'error') {
|
||||
// Both timer-fired aborts (`AbortSignal.timeout()` → `TimeoutError`)
|
||||
// and caller-driven aborts (`AbortController.abort()` → `AbortError`)
|
||||
// are terminal: retrying against an already-aborted signal would
|
||||
// fail again immediately, and neither outcome reflects backend
|
||||
// health. They route through the breaker's neutral path.
|
||||
if (
|
||||
result.err instanceof DOMException &&
|
||||
(result.err.name === 'TimeoutError' || result.err.name === 'AbortError')
|
||||
) {
|
||||
return { kind: 'terminal-network', err: result.err };
|
||||
}
|
||||
return { kind: 'retryable-network', err: result.err };
|
||||
}
|
||||
const resp = result.resp;
|
||||
if (resp.status >= 200 && resp.status < 400) return { kind: 'success', resp };
|
||||
if (resp.status === 429) {
|
||||
// `resp.headers` is always present on a real `Response`, but tests
|
||||
// sometimes stub `fetch` with a plain `{ ok, status }` object. Be
|
||||
// defensive — a missing `Retry-After` falls through to exponential
|
||||
// backoff, which is the correct behaviour anyway.
|
||||
const retryAfterHeader =
|
||||
typeof resp.headers?.get === 'function' ? resp.headers.get('Retry-After') : null;
|
||||
const parsed = parseRetryAfter(retryAfterHeader, now);
|
||||
return {
|
||||
kind: 'retryable-status',
|
||||
resp,
|
||||
afterMs: parsed !== null ? Math.min(parsed, RETRY_AFTER_CAP_MS) : undefined,
|
||||
};
|
||||
}
|
||||
if (resp.status >= 500) return { kind: 'retryable-status', resp, afterMs: undefined };
|
||||
return { kind: 'terminal-client', resp };
|
||||
}
|
||||
|
||||
const defaultSleep = (ms: number): Promise<void> =>
|
||||
new Promise((resolve) => setTimeout(resolve, ms));
|
||||
|
||||
function defaultBreakerKey(input: string | URL): string {
|
||||
try {
|
||||
const url = typeof input === 'string' ? new URL(input) : input;
|
||||
return `${url.host}${url.pathname}`;
|
||||
} catch {
|
||||
return String(input);
|
||||
}
|
||||
}
|
||||
|
||||
/** Final error thrown when retries are exhausted on a 5xx / 429. */
|
||||
export class ResilientFetchExhaustedError extends Error {
|
||||
override readonly name = 'ResilientFetchExhaustedError';
|
||||
constructor(public readonly response: Response) {
|
||||
super(`Request failed after retries (HTTP ${response.status})`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Wrap `fetch` with bounded retries and a per-process circuit breaker.
|
||||
*
|
||||
* Semantics:
|
||||
* - 5xx and 429 responses are retried; 429 honors `Retry-After` (capped).
|
||||
* - Network throws are retried unless they are `TimeoutError` DOMExceptions.
|
||||
* - Timeouts and 4xx (other than 429) are returned/thrown without retry
|
||||
* AND without incrementing the breaker — they reflect caller config
|
||||
* or local network state, not registry health.
|
||||
* - Each `fetch` call carries the caller-supplied `signal` (e.g. an
|
||||
* `AbortSignal.timeout()`) — that timeout bounds each individual
|
||||
* attempt, not the whole retry sequence.
|
||||
* - When the breaker is open, throws `CircuitOpenError` synchronously
|
||||
* without invoking `fetch`.
|
||||
* - When retries are exhausted on a 5xx / 429, throws
|
||||
* `ResilientFetchExhaustedError` carrying the last response.
|
||||
*
|
||||
* Cumulative wall-clock budget:
|
||||
* maxAttempts × (per-attempt-timeout + capDelayMs)
|
||||
* With defaults (3, 500ms base, 5000ms cap) and a typical 15s per-attempt
|
||||
* timeout from the caller's signal, worst case is ~3 × (15s + 5s) = 60s.
|
||||
* Callers that want a tighter total bound should reduce `maxAttempts` or
|
||||
* wrap `resilientFetch` in their own outer `AbortSignal.timeout()`.
|
||||
*/
|
||||
export async function resilientFetch(
|
||||
input: string | URL,
|
||||
init: RequestInit | undefined,
|
||||
opts: ResilientFetchOptions = {},
|
||||
): Promise<Response> {
|
||||
const fetchImpl = opts.fetchImpl ?? globalThis.fetch;
|
||||
const now = opts.now ?? (() => Date.now());
|
||||
const breaker =
|
||||
opts.breaker ?? getBreaker(opts.breakerKey ?? defaultBreakerKey(input), opts.breakerOptions);
|
||||
|
||||
const retryConfig = {
|
||||
maxAttempts: opts.retry?.maxAttempts ?? DEFAULT_RETRY.maxAttempts,
|
||||
baseDelayMs: opts.retry?.baseDelayMs ?? DEFAULT_RETRY.baseDelayMs,
|
||||
capDelayMs: opts.retry?.capDelayMs ?? DEFAULT_RETRY.capDelayMs,
|
||||
};
|
||||
const sleep = opts.retry?.sleep ?? defaultSleep;
|
||||
const random = opts.retry?.random ?? Math.random;
|
||||
|
||||
// Fail fast on an open breaker, before invoking fetch.
|
||||
breaker.check();
|
||||
|
||||
for (let attempt = 0; attempt < retryConfig.maxAttempts; attempt++) {
|
||||
let result: { kind: 'error'; err: unknown } | { kind: 'response'; resp: Response };
|
||||
try {
|
||||
// CodeQL js/server-side-request-forgery — flagged because `input`
|
||||
// is caller-supplied. Suppressed: every concrete caller passes
|
||||
// either a hardcoded URL constant (UNDERSTAND_QUICKLY_DISPATCH_URL,
|
||||
// OpenRouter base URL) or a value derived from configuration
|
||||
// (env vars, saved settings, the local backend URL). User-input
|
||||
// request fields (e.g. PR title, repo name) never flow into
|
||||
// `input`. Validating URL shape here would push false-positive
|
||||
// rejection onto every caller — wrong layer for the check.
|
||||
// lgtm[js/server-side-request-forgery]
|
||||
// codeql[js/server-side-request-forgery]
|
||||
const resp = await fetchImpl(input, init);
|
||||
result = { kind: 'response', resp };
|
||||
} catch (err) {
|
||||
result = { kind: 'error', err };
|
||||
}
|
||||
|
||||
const outcome = classifyOutcome(result, now);
|
||||
|
||||
switch (outcome.kind) {
|
||||
case 'success':
|
||||
breaker.recordSuccess();
|
||||
return outcome.resp;
|
||||
|
||||
case 'terminal-client':
|
||||
// 4xx: do not count as breaker failure (the server is healthy
|
||||
// and rejecting our request — auth, scope, or routing). But
|
||||
// also do NOT call recordSuccess: a 401 sandwiched between
|
||||
// 5xx responses would otherwise erase the running outage
|
||||
// signal. The breaker's neutral path leaves state untouched.
|
||||
breaker.recordNeutral();
|
||||
return outcome.resp;
|
||||
|
||||
case 'terminal-network':
|
||||
// Either `AbortSignal.timeout()` fired locally OR an external
|
||||
// caller cancelled the request via AbortController. The server
|
||||
// never had a chance to answer; this reflects the user's
|
||||
// network or an explicit cancel, not registry health. Don't
|
||||
// punish the breaker AND don't reset its outage signal.
|
||||
breaker.recordNeutral();
|
||||
throw outcome.err;
|
||||
|
||||
case 'retryable-status':
|
||||
if (attempt + 1 >= retryConfig.maxAttempts) {
|
||||
breaker.recordFailure();
|
||||
throw new ResilientFetchExhaustedError(outcome.resp);
|
||||
}
|
||||
await sleep(
|
||||
computeBackoffMs(
|
||||
attempt,
|
||||
retryConfig.baseDelayMs,
|
||||
retryConfig.capDelayMs,
|
||||
outcome.afterMs,
|
||||
random,
|
||||
),
|
||||
);
|
||||
break;
|
||||
|
||||
case 'retryable-network':
|
||||
if (attempt + 1 >= retryConfig.maxAttempts) {
|
||||
breaker.recordFailure();
|
||||
throw outcome.err;
|
||||
}
|
||||
await sleep(
|
||||
computeBackoffMs(
|
||||
attempt,
|
||||
retryConfig.baseDelayMs,
|
||||
retryConfig.capDelayMs,
|
||||
undefined,
|
||||
random,
|
||||
),
|
||||
);
|
||||
break;
|
||||
|
||||
default: {
|
||||
// Exhaustiveness guard. If a sixth `Outcome` kind is added in
|
||||
// future, TypeScript will refuse to assign it to `never` and
|
||||
// this line forces the maintainer to add an explicit arm
|
||||
// rather than silently fall through to retry/no-retry behaviour.
|
||||
const _exhaustive: never = outcome;
|
||||
throw new Error(`resilientFetch: unhandled outcome ${JSON.stringify(_exhaustive)}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Unreachable: every iteration of the loop either returns (success
|
||||
// / terminal-client) or throws (terminal-network / retry exhaustion).
|
||||
// The throw is here purely so TypeScript's control-flow analysis sees
|
||||
// the function never falls off the end without producing `Promise<Response>`.
|
||||
/* c8 ignore next 2 */
|
||||
throw new Error('resilientFetch: retry loop terminated unexpectedly');
|
||||
}
|
||||
105
gitnexus-shared/src/integrations/retry.ts
Normal file
105
gitnexus-shared/src/integrations/retry.ts
Normal file
|
|
@ -0,0 +1,105 @@
|
|||
/**
|
||||
* Bounded retry helper with full-jitter exponential backoff.
|
||||
*
|
||||
* Runtime-agnostic: depends only on `setTimeout`, `Math.random`, and the
|
||||
* Promise machinery — no Node-only imports. Safe to consume from CLI,
|
||||
* server, or browser callers.
|
||||
*
|
||||
* Pattern reference: gitnexus/src/core/embeddings/http-client.ts. This
|
||||
* helper is the upgraded form: classification is caller-supplied (so
|
||||
* 4xx-vs-5xx-vs-timeout decisions live with the protocol that knows
|
||||
* them), backoff is exponential with full jitter, and an optional
|
||||
* `afterMs` lets callers honor `Retry-After` headers.
|
||||
*/
|
||||
|
||||
export interface RetryOptions {
|
||||
/** Initial delay before the first retry attempt, in milliseconds. */
|
||||
baseDelayMs: number;
|
||||
/** Upper bound on any single delay, in milliseconds. */
|
||||
capDelayMs: number;
|
||||
/** Total attempts including the first call. Must be >= 1. */
|
||||
maxAttempts: number;
|
||||
/**
|
||||
* Decide whether to retry after a thrown error.
|
||||
* Return `{retry:false}` to terminate immediately and rethrow.
|
||||
* Return `{retry:true}` to retry with exponential-backoff jitter.
|
||||
* Return `{retry:true, afterMs}` to wait at least `afterMs` (still
|
||||
* subject to `capDelayMs`) — used by callers parsing `Retry-After`.
|
||||
*/
|
||||
isRetryable: (err: unknown, attempt: number) => RetryDecision;
|
||||
/** Sleep override — defaults to `setTimeout`. Tests inject fake timers. */
|
||||
sleep?: (ms: number) => Promise<void>;
|
||||
/** Random override — defaults to `Math.random`. Tests inject seeded values. */
|
||||
random?: () => number;
|
||||
}
|
||||
|
||||
export type RetryDecision = { retry: false } | { retry: true; afterMs?: number };
|
||||
|
||||
const defaultSleep = (ms: number): Promise<void> =>
|
||||
new Promise((resolve) => setTimeout(resolve, ms));
|
||||
|
||||
/**
|
||||
* Compute the delay before the next retry attempt.
|
||||
*
|
||||
* - When the caller specifies `afterMs` (e.g., from `Retry-After`), use
|
||||
* `min(afterMs, capDelayMs)` so a misbehaving server can't pin the
|
||||
* client for an arbitrarily long wait.
|
||||
* - Otherwise compute full-jitter exponential backoff:
|
||||
* `random() * min(cap, base * 2^attempt)`. Full jitter (rather than
|
||||
* "equal jitter") avoids retry-storm thundering herd, per AWS
|
||||
* guidance on backoff strategies.
|
||||
*/
|
||||
export function computeBackoffMs(
|
||||
attempt: number,
|
||||
baseDelayMs: number,
|
||||
capDelayMs: number,
|
||||
afterMs: number | undefined,
|
||||
random: () => number,
|
||||
): number {
|
||||
if (afterMs !== undefined) {
|
||||
return Math.min(Math.max(0, afterMs), capDelayMs);
|
||||
}
|
||||
const exponential = baseDelayMs * Math.pow(2, attempt);
|
||||
const upper = Math.min(capDelayMs, exponential);
|
||||
return Math.floor(random() * upper);
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute `fn` with bounded retries.
|
||||
*
|
||||
* The classification of "retryable" is the caller's responsibility — see
|
||||
* `resilient-fetch.ts` for the GitHub-dispatch-specific rules. This
|
||||
* helper is the mechanical retry loop only.
|
||||
*/
|
||||
export async function withRetry<T>(
|
||||
fn: (attempt: number) => Promise<T>,
|
||||
opts: RetryOptions,
|
||||
): Promise<T> {
|
||||
if (opts.maxAttempts < 1) {
|
||||
throw new Error(`withRetry: maxAttempts must be >= 1, got ${opts.maxAttempts}`);
|
||||
}
|
||||
const sleep = opts.sleep ?? defaultSleep;
|
||||
const random = opts.random ?? Math.random;
|
||||
|
||||
let lastError: unknown;
|
||||
for (let attempt = 0; attempt < opts.maxAttempts; attempt++) {
|
||||
try {
|
||||
return await fn(attempt);
|
||||
} catch (err) {
|
||||
lastError = err;
|
||||
const decision = opts.isRetryable(err, attempt);
|
||||
if (!decision.retry) throw err;
|
||||
// Don't sleep after the final attempt.
|
||||
if (attempt + 1 >= opts.maxAttempts) break;
|
||||
const delayMs = computeBackoffMs(
|
||||
attempt,
|
||||
opts.baseDelayMs,
|
||||
opts.capDelayMs,
|
||||
decision.afterMs,
|
||||
random,
|
||||
);
|
||||
if (delayMs > 0) await sleep(delayMs);
|
||||
}
|
||||
}
|
||||
throw lastError;
|
||||
}
|
||||
13
gitnexus-shared/src/test-helpers.ts
Normal file
13
gitnexus-shared/src/test-helpers.ts
Normal file
|
|
@ -0,0 +1,13 @@
|
|||
/**
|
||||
* Test-only helpers.
|
||||
*
|
||||
* Symbols here are reachable from `gitnexus-shared/test-helpers` so test
|
||||
* suites can reset shared registries or exercise internal classifiers,
|
||||
* but they are deliberately NOT re-exported from the main `gitnexus-shared`
|
||||
* barrel. Production consumers should never import this module — calling
|
||||
* `__resetBreakerRegistry__()` from a tool implementation would silently
|
||||
* nuke every circuit breaker process-wide.
|
||||
*/
|
||||
|
||||
export { __resetBreakerRegistry__ } from './integrations/circuit-breaker.js';
|
||||
export { classifyOutcome } from './integrations/resilient-fetch.js';
|
||||
|
|
@ -20,6 +20,7 @@ import {
|
|||
ProviderConfig,
|
||||
} from './types';
|
||||
import { DEFAULT_OPENROUTER_BASE_URL, DEFAULT_OLLAMA_BASE_URL } from '../../config/ui-constants';
|
||||
import { resilientFetch } from 'gitnexus-shared';
|
||||
|
||||
const STORAGE_KEY = 'gitnexus-llm-settings';
|
||||
|
||||
|
|
@ -407,7 +408,10 @@ export const getAvailableModels = (provider: LLMProvider): string[] => {
|
|||
*/
|
||||
export const fetchOpenRouterModels = async (): Promise<Array<{ id: string; name: string }>> => {
|
||||
try {
|
||||
const response = await fetch(`${DEFAULT_OPENROUTER_BASE_URL}/models`);
|
||||
const response = await resilientFetch(`${DEFAULT_OPENROUTER_BASE_URL}/models`, undefined, {
|
||||
breakerKey: 'openrouter-models',
|
||||
retry: { maxAttempts: 2, baseDelayMs: 500, capDelayMs: 2_000 },
|
||||
});
|
||||
if (!response.ok) throw new Error('Failed to fetch models');
|
||||
const data = await response.json();
|
||||
return data.data.map((model: any) => ({
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@
|
|||
*/
|
||||
|
||||
import type { GraphNode, GraphRelationship } from 'gitnexus-shared';
|
||||
import { CircuitOpenError, ResilientFetchExhaustedError, resilientFetch } from 'gitnexus-shared';
|
||||
|
||||
// ── Types ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
|
@ -237,29 +238,91 @@ export function normalizeServerUrl(input: string): string {
|
|||
const DEFAULT_TIMEOUT_MS = 30_000;
|
||||
const PROBE_TIMEOUT_MS = 2_000;
|
||||
|
||||
/** Idempotent HTTP methods. Other verbs (POST, PATCH, PUT, DELETE) get
|
||||
* a single-attempt retry budget by default to avoid duplicate side
|
||||
* effects on retry — a POST that 5xx'd may have already executed
|
||||
* server-side. Callers that have idempotency keys or otherwise know
|
||||
* their mutation is safe to retry can opt in via `forceRetry`. */
|
||||
const IDEMPOTENT_METHODS = new Set(['GET', 'HEAD', 'OPTIONS']);
|
||||
|
||||
const fetchWithTimeout = async (
|
||||
url: string,
|
||||
init: RequestInit = {},
|
||||
timeoutMs: number = DEFAULT_TIMEOUT_MS,
|
||||
/**
|
||||
* Force a retry budget on non-idempotent methods. Default false.
|
||||
* Pass true only when the endpoint is known-idempotent (e.g. DELETE
|
||||
* of a known-deleted resource — second call is a 404 / no-op) AND
|
||||
* the duplicate-side-effect window is acceptable.
|
||||
*/
|
||||
forceRetry = false,
|
||||
): Promise<Response> => {
|
||||
const controller = new AbortController();
|
||||
// Merge external signal if provided
|
||||
// Merge the external caller signal (if any) with an
|
||||
// `AbortSignal.timeout()` so a timer-fired abort produces a
|
||||
// `DOMException` with `name === 'TimeoutError'` — which
|
||||
// `resilientFetch` correctly classifies as terminal-network (no
|
||||
// retry, no breaker hit). A manual `AbortController.abort()` would
|
||||
// produce `name === 'AbortError'` and route through the
|
||||
// retryable-network branch, which mis-penalizes the breaker for
|
||||
// user-side network slowness.
|
||||
const timeoutSignal = AbortSignal.timeout(timeoutMs);
|
||||
const externalSignal = init.signal;
|
||||
if (externalSignal) {
|
||||
externalSignal.addEventListener('abort', () => controller.abort());
|
||||
const signal = externalSignal ? AbortSignal.any([timeoutSignal, externalSignal]) : timeoutSignal;
|
||||
|
||||
const method = (init.method ?? 'GET').toUpperCase();
|
||||
const isIdempotent = IDEMPOTENT_METHODS.has(method);
|
||||
const maxAttempts = isIdempotent || forceRetry ? 2 : 1;
|
||||
|
||||
// Key the breaker by the current backend origin so switching backend
|
||||
// URLs (e.g. recovering from a flapping local server by pointing at
|
||||
// a different host) gives the new origin a fresh breaker state. A
|
||||
// single shared `'web-backend'` key would otherwise leave a user
|
||||
// locked out for the full cooldown after one bad host trips the
|
||||
// circuit. The malformed-URL fallback is defensive — `setBackendUrl`
|
||||
// normalizes input, so this branch shouldn't fire in practice.
|
||||
let breakerKey: string;
|
||||
try {
|
||||
breakerKey = `web-backend:${new URL(_backendUrl).origin}`;
|
||||
} catch {
|
||||
breakerKey = 'web-backend:invalid';
|
||||
}
|
||||
const timer = setTimeout(() => controller.abort(), timeoutMs);
|
||||
|
||||
try {
|
||||
const response = await fetch(url, { ...init, signal: controller.signal });
|
||||
// Bounded retries + 5xx/429 handling are delegated to resilientFetch.
|
||||
// Method-aware budget: idempotent verbs retry once on transient
|
||||
// backend failures; mutations (POST/PATCH/PUT/DELETE) default to
|
||||
// single-attempt to avoid duplicate side effects.
|
||||
const response = await resilientFetch(
|
||||
url,
|
||||
{ ...init, signal },
|
||||
{
|
||||
breakerKey,
|
||||
retry: { maxAttempts, baseDelayMs: 250, capDelayMs: 1500 },
|
||||
},
|
||||
);
|
||||
return response;
|
||||
} catch (error: unknown) {
|
||||
if (error instanceof DOMException && error.name === 'AbortError') {
|
||||
if (externalSignal?.aborted) {
|
||||
throw new BackendError('Request aborted', 0, 'network');
|
||||
}
|
||||
if (error instanceof CircuitOpenError) {
|
||||
throw new BackendError(
|
||||
`GitNexus backend at ${_backendUrl} is unhealthy; retry in ${Math.ceil(error.retryAfterMs / 1000)}s`,
|
||||
0,
|
||||
'network',
|
||||
);
|
||||
}
|
||||
if (error instanceof ResilientFetchExhaustedError) {
|
||||
// Fall through to caller — surface the raw response so assertOk
|
||||
// can craft the BackendError with the right code.
|
||||
return error.response;
|
||||
}
|
||||
if (error instanceof DOMException && error.name === 'TimeoutError') {
|
||||
throw new BackendError(`Request to ${url} timed out after ${timeoutMs}ms`, 0, 'timeout');
|
||||
}
|
||||
if (error instanceof DOMException && error.name === 'AbortError') {
|
||||
// External caller-driven cancellation — `timeoutSignal` would
|
||||
// have surfaced as TimeoutError above, so this branch covers
|
||||
// only the externally-aborted case.
|
||||
throw new BackendError('Request aborted', 0, 'network');
|
||||
}
|
||||
if (error instanceof TypeError) {
|
||||
throw new BackendError(
|
||||
`Network error reaching GitNexus backend at ${_backendUrl}: ${error.message}`,
|
||||
|
|
@ -268,8 +331,6 @@ const fetchWithTimeout = async (
|
|||
);
|
||||
}
|
||||
throw error;
|
||||
} finally {
|
||||
clearTimeout(timer);
|
||||
}
|
||||
};
|
||||
|
||||
|
|
|
|||
110
gitnexus-web/test/unit/backend-client-retry.test.ts
Normal file
110
gitnexus-web/test/unit/backend-client-retry.test.ts
Normal file
|
|
@ -0,0 +1,110 @@
|
|||
/**
|
||||
* Method-aware retry budget + timeout-as-TimeoutError verification for
|
||||
* backend-client's `fetchWithTimeout`.
|
||||
*
|
||||
* Closes review findings on PR #1448:
|
||||
* - Non-idempotent POST/DELETE must NOT be retried by default —
|
||||
* a 5xx on `startAnalyze` could otherwise start a duplicate job.
|
||||
* - Timer-fired timeout must surface as `DOMException(name='TimeoutError')`,
|
||||
* not `AbortError`, so resilientFetch routes it through the
|
||||
* terminal-network branch (no retry, no breaker hit).
|
||||
*/
|
||||
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
import { getBreaker } from 'gitnexus-shared';
|
||||
import { __resetBreakerRegistry__ } from 'gitnexus-shared/test-helpers';
|
||||
import { fetchRepos, setBackendUrl, startAnalyze } from '../../src/services/backend-client';
|
||||
|
||||
const BASE = 'http://localhost:4747';
|
||||
|
||||
describe('backend-client retry budget (method-aware)', () => {
|
||||
beforeEach(() => {
|
||||
__resetBreakerRegistry__();
|
||||
setBackendUrl(BASE);
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.unstubAllGlobals();
|
||||
});
|
||||
|
||||
it('GET retries once on transient 503 (idempotent verb)', async () => {
|
||||
let n = 0;
|
||||
const fetchMock = vi.fn(async () => {
|
||||
n += 1;
|
||||
if (n === 1) return new Response('boom', { status: 503 });
|
||||
return new Response('[]', {
|
||||
status: 200,
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
});
|
||||
});
|
||||
vi.stubGlobal('fetch', fetchMock);
|
||||
|
||||
const repos = await fetchRepos();
|
||||
expect(repos).toEqual([]);
|
||||
// 1 retry budget on idempotent GET → 2 total fetch calls.
|
||||
expect(fetchMock).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it('POST does NOT retry on 503 by default (non-idempotent verb)', async () => {
|
||||
const fetchMock = vi.fn(async () => new Response('boom', { status: 503 }));
|
||||
vi.stubGlobal('fetch', fetchMock);
|
||||
|
||||
await expect(startAnalyze({ path: '/tmp/repo' })).rejects.toBeTruthy();
|
||||
// Single attempt — never duplicates a job-start POST.
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('switching backend URL after a circuit opens reaches a fresh breaker (U3)', async () => {
|
||||
// Pre-open the breaker for host-A by directly recording 3 failures.
|
||||
setBackendUrl('http://host-a.test:4747');
|
||||
const aKey = 'web-backend:http://host-a.test:4747';
|
||||
const breakerA = getBreaker(aKey);
|
||||
breakerA.recordFailure();
|
||||
breakerA.recordFailure();
|
||||
breakerA.recordFailure();
|
||||
expect(breakerA.getState()).toBe('open');
|
||||
|
||||
// Switch to host-B and make a request — must succeed against the
|
||||
// new origin without tripping the host-A circuit. Under the old
|
||||
// single-key behaviour the call would throw CircuitOpenError.
|
||||
setBackendUrl('http://host-b.test:4747');
|
||||
const fetchMock = vi.fn(
|
||||
async () =>
|
||||
new Response('[]', {
|
||||
status: 200,
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
}),
|
||||
);
|
||||
vi.stubGlobal('fetch', fetchMock);
|
||||
|
||||
const repos = await fetchRepos();
|
||||
expect(repos).toEqual([]);
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
|
||||
// Host-A's breaker is still open in cooldown.
|
||||
expect(breakerA.getState()).toBe('open');
|
||||
// Host-B has its own (fresh) breaker.
|
||||
const bKey = 'web-backend:http://host-b.test:4747';
|
||||
expect(getBreaker(bKey).getState()).toBe('closed');
|
||||
expect(getBreaker(bKey).getConsecutiveFailures()).toBe(0);
|
||||
});
|
||||
|
||||
it('breaker not incremented when timeout fires (TimeoutError, not AbortError)', async () => {
|
||||
// Reject directly with a TimeoutError DOMException, mimicking what
|
||||
// `fetch` produces when its `AbortSignal.timeout()`-wired signal
|
||||
// fires. The real-fetch path goes signal.reason → reject(reason);
|
||||
// we shortcut that here so the test doesn't have to wait the
|
||||
// 30-second default timeout.
|
||||
const fetchMock = vi.fn(async () => {
|
||||
throw new DOMException('aborted by timeout', 'TimeoutError');
|
||||
});
|
||||
vi.stubGlobal('fetch', fetchMock);
|
||||
|
||||
await expect(fetchRepos()).rejects.toMatchObject({ code: 'timeout' });
|
||||
|
||||
// The breaker must not have been penalized for a local timeout.
|
||||
expect(getBreaker(`web-backend:${BASE}`).getConsecutiveFailures()).toBe(0);
|
||||
// Timeout is terminal — no retry attempted.
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
|
|
@ -1,6 +1,8 @@
|
|||
import os from 'node:os';
|
||||
import { join } from 'node:path';
|
||||
|
||||
import { CircuitBreaker, withRetry } from 'gitnexus-shared';
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Download resilience defaults
|
||||
// ---------------------------------------------------------------------------
|
||||
|
|
@ -108,70 +110,19 @@ export function isNetworkFetchError(message: string): boolean {
|
|||
/** @internal Used by `withHfDownloadRetry` to mark a circuit-open rejection. */
|
||||
export const CIRCUIT_OPEN_TAG = 'hf-circuit-open';
|
||||
|
||||
/** Circuit-breaker states. */
|
||||
type CircuitState = 'closed' | 'open' | 'half-open';
|
||||
|
||||
/**
|
||||
* Circuit breaker for HuggingFace model downloads.
|
||||
*
|
||||
* After `failureThreshold` consecutive network failures the circuit opens and
|
||||
* all subsequent calls to `withHfDownloadRetry` fail immediately without
|
||||
* issuing any network requests. After `resetTimeoutMs` the circuit enters the
|
||||
* half-open state and the next call is attempted — if it succeeds the circuit
|
||||
* closes again; if it fails the circuit re-opens.
|
||||
*
|
||||
* Exported for unit-testing; production code should use the module-level
|
||||
* `hfDownloadCircuit` singleton.
|
||||
* Module-level singleton shared by both embedder entry points
|
||||
* (`core/embeddings/embedder.ts` + `mcp/core/embedder.ts`). Per-process
|
||||
* only — not persisted across restarts. Backed by the shared
|
||||
* `CircuitBreaker` from `gitnexus-shared` (same state machine, same
|
||||
* semantics, plus the single-permit half-open gate that prevents
|
||||
* recovery-time stampedes).
|
||||
*/
|
||||
export class HfDownloadCircuitBreaker {
|
||||
private _state: CircuitState = 'closed';
|
||||
private _failures = 0;
|
||||
/** Timestamp of the last recorded failure (ms since epoch). */
|
||||
lastFailureAt = 0;
|
||||
|
||||
constructor(
|
||||
readonly failureThreshold: number = CB_FAILURE_THRESHOLD,
|
||||
readonly resetTimeoutMs: number = CB_RESET_TIMEOUT_MS,
|
||||
) {}
|
||||
|
||||
/** Effective state, factoring in the reset-timeout transition. */
|
||||
get state(): CircuitState {
|
||||
if (this._state === 'open' && Date.now() - this.lastFailureAt > this.resetTimeoutMs) {
|
||||
this._state = 'half-open';
|
||||
}
|
||||
return this._state;
|
||||
}
|
||||
|
||||
/** Returns true when the circuit is open and calls should be rejected. */
|
||||
isOpen(): boolean {
|
||||
return this.state === 'open';
|
||||
}
|
||||
|
||||
/** Record a successful call — resets the failure counter and closes the circuit. */
|
||||
recordSuccess(): void {
|
||||
this._failures = 0;
|
||||
this._state = 'closed';
|
||||
}
|
||||
|
||||
/** Record a failed call — increments the counter and opens the circuit when the threshold is reached. */
|
||||
recordFailure(): void {
|
||||
this._failures++;
|
||||
this.lastFailureAt = Date.now();
|
||||
if (this._failures >= this.failureThreshold) {
|
||||
this._state = 'open';
|
||||
}
|
||||
}
|
||||
|
||||
/** @internal Reset to initial state (used in tests). */
|
||||
reset(): void {
|
||||
this._failures = 0;
|
||||
this._state = 'closed';
|
||||
this.lastFailureAt = 0;
|
||||
}
|
||||
}
|
||||
|
||||
/** Module-level singleton shared by both embedder entry points. */
|
||||
export const hfDownloadCircuit = new HfDownloadCircuitBreaker();
|
||||
export const hfDownloadCircuit = new CircuitBreaker({
|
||||
failureThreshold: CB_FAILURE_THRESHOLD,
|
||||
cooldownMs: CB_RESET_TIMEOUT_MS,
|
||||
key: 'hf-download',
|
||||
});
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Retry + timeout wrapper
|
||||
|
|
@ -219,11 +170,6 @@ export function withDownloadTimeout<T>(fn: () => Promise<T>, timeoutMs: number):
|
|||
});
|
||||
}
|
||||
|
||||
/** @internal Async sleep (exposed for testing). */
|
||||
export function sleep(ms: number): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
|
||||
export interface HfRetryOptions {
|
||||
/** Maximum total attempts including the initial one (default: `HF_MAX_ATTEMPTS`). */
|
||||
maxAttempts?: number;
|
||||
|
|
@ -235,7 +181,7 @@ export interface HfRetryOptions {
|
|||
* Circuit-breaker instance to use. Defaults to the module-level
|
||||
* `hfDownloadCircuit` singleton. Pass a fresh instance in tests.
|
||||
*/
|
||||
circuit?: HfDownloadCircuitBreaker;
|
||||
circuit?: CircuitBreaker;
|
||||
/**
|
||||
* Optional callback invoked before each retry (not the initial attempt).
|
||||
* @param attempt - 1-based retry number
|
||||
|
|
@ -295,49 +241,74 @@ export async function withHfDownloadRetry<T>(
|
|||
circuit = hfDownloadCircuit,
|
||||
onRetry,
|
||||
} = options;
|
||||
if (circuit.isOpen()) {
|
||||
const secsUntilReset = Math.ceil(
|
||||
(circuit.resetTimeoutMs - (Date.now() - circuit.lastFailureAt)) / 1000,
|
||||
);
|
||||
if (circuit.getState() === 'open') {
|
||||
// Compute remaining cooldown without consuming a probe permit.
|
||||
const openedAt = circuit.getOpenedAt();
|
||||
const secsUntilReset =
|
||||
openedAt !== null ? Math.ceil((circuit.getCooldownMs() - (Date.now() - openedAt)) / 1000) : 0;
|
||||
throw new Error(
|
||||
`${CIRCUIT_OPEN_TAG}: HuggingFace download circuit is open after repeated network failures` +
|
||||
(secsUntilReset > 0 ? ` — will reset in ~${secsUntilReset}s` : ''),
|
||||
);
|
||||
}
|
||||
|
||||
let lastError: Error = new Error('unknown error');
|
||||
// Retry budget delegated to `withRetry` from gitnexus-shared. The
|
||||
// HF-specific bits — per-attempt timeout, network-vs-non-network
|
||||
// classification, circuit-breaker recording, onRetry callback — wire
|
||||
// through the `isRetryable` callback. `circuitTripped` is the
|
||||
// sentinel that lets us replace the final thrown error with a
|
||||
// CIRCUIT_OPEN_TAG message when the breaker tripped mid-loop.
|
||||
let circuitTripped = false;
|
||||
|
||||
for (let attempt = 0; attempt < maxAttempts; attempt++) {
|
||||
try {
|
||||
const result = await withDownloadTimeout(fn, timeoutMs);
|
||||
circuit.recordSuccess();
|
||||
return result;
|
||||
} catch (err) {
|
||||
lastError = err instanceof Error ? err : new Error(String(err));
|
||||
|
||||
if (!isNetworkFetchError(lastError.message)) {
|
||||
// Non-network error (e.g. CUDA unavailable) — propagate without retry
|
||||
throw lastError;
|
||||
}
|
||||
|
||||
circuit.recordFailure();
|
||||
|
||||
if (circuit.isOpen()) {
|
||||
// Circuit just tripped — fail fast, no more retries
|
||||
throw new Error(
|
||||
`${CIRCUIT_OPEN_TAG}: HuggingFace download circuit opened after ${circuit.failureThreshold} consecutive failures`,
|
||||
);
|
||||
}
|
||||
|
||||
if (attempt < maxAttempts - 1) {
|
||||
const delay = baseDelayMs * Math.pow(2, attempt);
|
||||
onRetry?.(attempt + 1, maxAttempts, lastError);
|
||||
await sleep(delay);
|
||||
}
|
||||
try {
|
||||
return await withRetry(
|
||||
async () => {
|
||||
const result = await withDownloadTimeout(fn, timeoutMs);
|
||||
circuit.recordSuccess();
|
||||
return result;
|
||||
},
|
||||
{
|
||||
maxAttempts,
|
||||
baseDelayMs,
|
||||
// Disable the cap to match the bespoke pure-exponential
|
||||
// progression. With the default `HF_MAX_ATTEMPTS_CAP = 10` and
|
||||
// `baseDelayMs = 2000`, the largest possible delay is
|
||||
// `2000 * 2^9 = ~17 minutes` — bounded enough not to need a cap.
|
||||
capDelayMs: Number.MAX_SAFE_INTEGER,
|
||||
isRetryable: (err, attempt) => {
|
||||
const error = err instanceof Error ? err : new Error(String(err));
|
||||
if (!isNetworkFetchError(error.message)) {
|
||||
// Non-network error (e.g. CUDA unavailable) — propagate
|
||||
// without retry. Use recordNeutral so the breaker's existing
|
||||
// failure-count progress isn't reset by a non-network failure
|
||||
// that says nothing about the CDN's health.
|
||||
circuit.recordNeutral();
|
||||
return { retry: false };
|
||||
}
|
||||
circuit.recordFailure();
|
||||
if (circuit.getState() === 'open') {
|
||||
// Circuit just tripped — fail fast, no more retries.
|
||||
circuitTripped = true;
|
||||
return { retry: false };
|
||||
}
|
||||
// Mirror the bespoke onRetry contract: fire only when there's
|
||||
// actually a next attempt.
|
||||
if (attempt + 1 < maxAttempts) {
|
||||
onRetry?.(attempt + 1, maxAttempts, error);
|
||||
}
|
||||
return { retry: true };
|
||||
},
|
||||
},
|
||||
);
|
||||
} catch (err) {
|
||||
if (circuitTripped) {
|
||||
throw new Error(
|
||||
`${CIRCUIT_OPEN_TAG}: HuggingFace download circuit opened after ${CB_FAILURE_THRESHOLD} consecutive failures`,
|
||||
);
|
||||
}
|
||||
// All retries exhausted — rethrow the last network error so
|
||||
// isNetworkFetchError patterns in the calling code still match and
|
||||
// surface HF_ENDPOINT guidance.
|
||||
throw err;
|
||||
}
|
||||
|
||||
// All retries exhausted — throw the last network error so isNetworkFetchError
|
||||
// patterns in the calling code still match and surface HF_ENDPOINT guidance.
|
||||
throw lastError;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -3,13 +3,22 @@
|
|||
*
|
||||
* Shared fetch+retry logic for OpenAI-compatible /v1/embeddings endpoints.
|
||||
* Imported by both the core embedder (batch) and MCP embedder (query).
|
||||
*
|
||||
* Network resilience is delegated to `resilientFetch` from
|
||||
* `gitnexus-shared` — bounded retries with exponential-backoff jitter,
|
||||
* `Retry-After` honored on 429, and an in-process circuit breaker that
|
||||
* fails fast on a flapping endpoint. Per-attempt timeout is enforced
|
||||
* via `AbortSignal.timeout` on the underlying fetch.
|
||||
*/
|
||||
|
||||
import { CircuitOpenError, ResilientFetchExhaustedError, resilientFetch } from 'gitnexus-shared';
|
||||
|
||||
const HTTP_TIMEOUT_MS = 30_000;
|
||||
const HTTP_MAX_RETRIES = 2;
|
||||
const HTTP_RETRY_BACKOFF_MS = 1_000;
|
||||
const HTTP_BATCH_SIZE = 64;
|
||||
const DEFAULT_DIMS = 384;
|
||||
const HTTP_BREAKER_KEY = 'embeddings-http';
|
||||
|
||||
interface HttpConfig {
|
||||
baseUrl: string;
|
||||
|
|
@ -90,46 +99,51 @@ const httpEmbedBatch = async (
|
|||
model: string,
|
||||
apiKey: string,
|
||||
batchIndex = 0,
|
||||
attempt = 0,
|
||||
): Promise<EmbeddingItem[]> => {
|
||||
let resp: Response;
|
||||
try {
|
||||
resp = await fetch(url, {
|
||||
method: 'POST',
|
||||
signal: AbortSignal.timeout(HTTP_TIMEOUT_MS),
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
Authorization: `Bearer ${apiKey}`,
|
||||
resp = await resilientFetch(
|
||||
url,
|
||||
{
|
||||
method: 'POST',
|
||||
signal: AbortSignal.timeout(HTTP_TIMEOUT_MS),
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
Authorization: `Bearer ${apiKey}`,
|
||||
},
|
||||
body: JSON.stringify({ input: batch, model }),
|
||||
},
|
||||
body: JSON.stringify({ input: batch, model }),
|
||||
});
|
||||
{
|
||||
breakerKey: HTTP_BREAKER_KEY,
|
||||
retry: { maxAttempts: HTTP_MAX_RETRIES + 1, baseDelayMs: HTTP_RETRY_BACKOFF_MS },
|
||||
},
|
||||
);
|
||||
} catch (err) {
|
||||
// Timeouts should not be retried — the server is unresponsive.
|
||||
// AbortSignal.timeout() throws DOMException with name 'TimeoutError'.
|
||||
const isTimeout = err instanceof DOMException && err.name === 'TimeoutError';
|
||||
if (isTimeout) {
|
||||
if (err instanceof CircuitOpenError) {
|
||||
throw new Error(
|
||||
`Embedding endpoint circuit open (${safeUrl(url)}, batch ${batchIndex}): retry in ${Math.ceil(err.retryAfterMs / 1000)}s`,
|
||||
);
|
||||
}
|
||||
if (err instanceof DOMException && err.name === 'TimeoutError') {
|
||||
throw new Error(
|
||||
`Embedding request timed out after ${HTTP_TIMEOUT_MS}ms (${safeUrl(url)}, batch ${batchIndex})`,
|
||||
);
|
||||
}
|
||||
// DNS, connection errors — retry with backoff
|
||||
if (attempt < HTTP_MAX_RETRIES) {
|
||||
const delay = HTTP_RETRY_BACKOFF_MS * (attempt + 1);
|
||||
await new Promise((r) => setTimeout(r, delay));
|
||||
return httpEmbedBatch(url, batch, model, apiKey, batchIndex, attempt + 1);
|
||||
if (err instanceof ResilientFetchExhaustedError) {
|
||||
throw new Error(
|
||||
`Embedding endpoint returned ${err.response.status} (${safeUrl(url)}, batch ${batchIndex})`,
|
||||
);
|
||||
}
|
||||
const reason = err instanceof Error ? err.message : String(err);
|
||||
throw new Error(`Embedding request failed (${safeUrl(url)}, batch ${batchIndex}): ${reason}`);
|
||||
}
|
||||
|
||||
if (!resp.ok) {
|
||||
const status = resp.status;
|
||||
if ((status === 429 || status >= 500) && attempt < HTTP_MAX_RETRIES) {
|
||||
const delay = HTTP_RETRY_BACKOFF_MS * (attempt + 1);
|
||||
await new Promise((r) => setTimeout(r, delay));
|
||||
return httpEmbedBatch(url, batch, model, apiKey, batchIndex, attempt + 1);
|
||||
}
|
||||
throw new Error(`Embedding endpoint returned ${status} (${safeUrl(url)}, batch ${batchIndex})`);
|
||||
// resilientFetch already retried 5xx/429; any non-OK response here is
|
||||
// a terminal client error (4xx other than 429).
|
||||
throw new Error(
|
||||
`Embedding endpoint returned ${resp.status} (${safeUrl(url)}, batch ${batchIndex})`,
|
||||
);
|
||||
}
|
||||
|
||||
const data = (await resp.json()) as { data: EmbeddingItem[] };
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
import { logger } from '../logger.js';
|
||||
import { CircuitOpenError, ResilientFetchExhaustedError, resilientFetch } from 'gitnexus-shared';
|
||||
/**
|
||||
* LLM Client for Wiki Generation
|
||||
*
|
||||
|
|
@ -170,86 +171,85 @@ export async function callLLM(
|
|||
? { 'api-key': config.apiKey }
|
||||
: { Authorization: `Bearer ${config.apiKey}` };
|
||||
|
||||
const MAX_RETRIES = 3;
|
||||
let lastError: Error | null = null;
|
||||
|
||||
for (let attempt = 0; attempt < MAX_RETRIES; attempt++) {
|
||||
try {
|
||||
const response = await fetch(url, {
|
||||
// Network resilience (bounded retries with exponential-backoff jitter,
|
||||
// 5xx + 429 + Retry-After handling, in-process circuit breaker on the
|
||||
// LLM endpoint) is delegated to resilientFetch. Provider-specific
|
||||
// error parsing (Azure content filter, empty-content checks) stays
|
||||
// here since it requires response-body inspection.
|
||||
let response: Response;
|
||||
try {
|
||||
response = await resilientFetch(
|
||||
url,
|
||||
{
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
...authHeaders,
|
||||
},
|
||||
body: JSON.stringify(body),
|
||||
});
|
||||
|
||||
if (!response.ok) {
|
||||
const errorText = await response.text().catch(() => 'unknown error');
|
||||
|
||||
// Azure content filter — surface a clear message instead of a generic API error
|
||||
if (
|
||||
azure &&
|
||||
response.status === 400 &&
|
||||
(errorText.includes('content_filter') ||
|
||||
errorText.includes('ResponsibleAIPolicyViolation'))
|
||||
) {
|
||||
throw new Error(
|
||||
`Azure content filter blocked this request. The prompt triggered content policy. Details: ${errorText.slice(0, 300)}`,
|
||||
);
|
||||
}
|
||||
|
||||
// Rate limit — wait with exponential backoff and retry
|
||||
if (response.status === 429 && attempt < MAX_RETRIES - 1) {
|
||||
const retryAfter = parseInt(response.headers.get('retry-after') || '0', 10);
|
||||
const delay = retryAfter > 0 ? retryAfter * 1000 : 2 ** attempt * 3000;
|
||||
await sleep(delay);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Server error — retry with backoff
|
||||
if (response.status >= 500 && attempt < MAX_RETRIES - 1) {
|
||||
await sleep((attempt + 1) * 2000);
|
||||
continue;
|
||||
}
|
||||
|
||||
throw new Error(`LLM API error (${response.status}): ${errorText.slice(0, 500)}`);
|
||||
}
|
||||
|
||||
// Streaming path
|
||||
if (useStream && response.body) {
|
||||
return await readSSEStream(response.body, options!.onChunk!);
|
||||
}
|
||||
|
||||
// Non-streaming path
|
||||
const json = (await response.json()) as any;
|
||||
const choice = json.choices?.[0];
|
||||
if (!choice?.message?.content) {
|
||||
throw new Error('LLM returned empty response');
|
||||
}
|
||||
|
||||
return {
|
||||
content: choice.message.content,
|
||||
promptTokens: json.usage?.prompt_tokens,
|
||||
completionTokens: json.usage?.completion_tokens,
|
||||
};
|
||||
} catch (err: any) {
|
||||
lastError = err;
|
||||
|
||||
// Network error — retry with backoff
|
||||
if (
|
||||
attempt < MAX_RETRIES - 1 &&
|
||||
(err.code === 'ECONNREFUSED' || err.code === 'ETIMEDOUT' || err.message?.includes('fetch'))
|
||||
) {
|
||||
await sleep((attempt + 1) * 3000);
|
||||
continue;
|
||||
}
|
||||
|
||||
throw err;
|
||||
// Per-attempt timeout. Without this each retry can hang
|
||||
// indefinitely on a frozen TCP connection — the per-call
|
||||
// signal is the only timeout `resilientFetch` honors;
|
||||
// `capDelayMs` only bounds the *backoff* between attempts.
|
||||
// 60s matches typical LLM completion budgets.
|
||||
signal: AbortSignal.timeout(60_000),
|
||||
},
|
||||
{
|
||||
breakerKey: `wiki-llm-${new URL(url).host}`,
|
||||
retry: { maxAttempts: 3, baseDelayMs: 2_000, capDelayMs: 30_000 },
|
||||
},
|
||||
);
|
||||
} catch (err) {
|
||||
if (err instanceof CircuitOpenError) {
|
||||
throw new Error(
|
||||
`LLM endpoint circuit open: retry in ${Math.ceil(err.retryAfterMs / 1000)}s. ${err.message}`,
|
||||
);
|
||||
}
|
||||
if (err instanceof ResilientFetchExhaustedError) {
|
||||
const errorText = await err.response.text().catch(() => 'unknown error');
|
||||
throw new Error(
|
||||
`LLM API error (${err.response.status} after retries): ${errorText.slice(0, 500)}`,
|
||||
);
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
|
||||
throw lastError || new Error('LLM call failed after retries');
|
||||
if (!response.ok) {
|
||||
const errorText = await response.text().catch(() => 'unknown error');
|
||||
|
||||
// Azure content filter — surface a clear message instead of a generic API error.
|
||||
if (
|
||||
azure &&
|
||||
response.status === 400 &&
|
||||
(errorText.includes('content_filter') || errorText.includes('ResponsibleAIPolicyViolation'))
|
||||
) {
|
||||
throw new Error(
|
||||
`Azure content filter blocked this request. The prompt triggered content policy. Details: ${errorText.slice(0, 300)}`,
|
||||
);
|
||||
}
|
||||
|
||||
// Any other non-OK response here is a terminal 4xx — resilientFetch
|
||||
// already retried 5xx/429 to exhaustion and would have thrown above.
|
||||
throw new Error(`LLM API error (${response.status}): ${errorText.slice(0, 500)}`);
|
||||
}
|
||||
|
||||
// Streaming path
|
||||
if (useStream && response.body) {
|
||||
return await readSSEStream(response.body, options!.onChunk!);
|
||||
}
|
||||
|
||||
// Non-streaming path
|
||||
const json = (await response.json()) as any;
|
||||
const choice = json.choices?.[0];
|
||||
if (!choice?.message?.content) {
|
||||
throw new Error('LLM returned empty response');
|
||||
}
|
||||
|
||||
return {
|
||||
content: choice.message.content,
|
||||
promptTokens: json.usage?.prompt_tokens,
|
||||
completionTokens: json.usage?.completion_tokens,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -312,7 +312,3 @@ async function readSSEStream(
|
|||
|
||||
return { content };
|
||||
}
|
||||
|
||||
function sleep(ms: number): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,12 +1,12 @@
|
|||
import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest';
|
||||
import os from 'node:os';
|
||||
import { join } from 'node:path';
|
||||
import { CircuitBreaker } from 'gitnexus-shared';
|
||||
import {
|
||||
applyHfEnvOverrides,
|
||||
isNetworkFetchError,
|
||||
isHfDownloadFailure,
|
||||
isHfCircuitOpenError,
|
||||
HfDownloadCircuitBreaker,
|
||||
withDownloadTimeout,
|
||||
withHfDownloadRetry,
|
||||
CIRCUIT_OPEN_TAG,
|
||||
|
|
@ -154,85 +154,13 @@ describe('isHfDownloadFailure', () => {
|
|||
});
|
||||
});
|
||||
|
||||
describe('HfDownloadCircuitBreaker', () => {
|
||||
it('starts in closed state', () => {
|
||||
const cb = new HfDownloadCircuitBreaker();
|
||||
expect(cb.isOpen()).toBe(false);
|
||||
expect(cb.state).toBe('closed');
|
||||
});
|
||||
|
||||
it('opens after reaching the failure threshold', () => {
|
||||
const cb = new HfDownloadCircuitBreaker(3);
|
||||
cb.recordFailure();
|
||||
cb.recordFailure();
|
||||
expect(cb.isOpen()).toBe(false);
|
||||
cb.recordFailure(); // threshold reached
|
||||
expect(cb.isOpen()).toBe(true);
|
||||
expect(cb.state).toBe('open');
|
||||
});
|
||||
|
||||
it('closes on recordSuccess after being open', () => {
|
||||
const cb = new HfDownloadCircuitBreaker(1);
|
||||
cb.recordFailure();
|
||||
expect(cb.isOpen()).toBe(true);
|
||||
cb.recordSuccess();
|
||||
expect(cb.isOpen()).toBe(false);
|
||||
expect(cb.state).toBe('closed');
|
||||
});
|
||||
|
||||
it('transitions to half-open after the reset timeout', () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const cb = new HfDownloadCircuitBreaker(1, 100 /* 100ms */);
|
||||
cb.recordFailure();
|
||||
expect(cb.isOpen()).toBe(true);
|
||||
vi.advanceTimersByTime(200);
|
||||
expect(cb.isOpen()).toBe(false);
|
||||
expect(cb.state).toBe('half-open');
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it('reset() restores closed state', () => {
|
||||
const cb = new HfDownloadCircuitBreaker(1);
|
||||
cb.recordFailure();
|
||||
expect(cb.isOpen()).toBe(true);
|
||||
cb.reset();
|
||||
expect(cb.isOpen()).toBe(false);
|
||||
expect(cb.state).toBe('closed');
|
||||
});
|
||||
|
||||
it('re-opens when a failure is recorded in half-open state', () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const cb = new HfDownloadCircuitBreaker(1, 100 /* 100ms */);
|
||||
cb.recordFailure(); // opens the circuit
|
||||
vi.advanceTimersByTime(200); // advance past reset timeout
|
||||
expect(cb.state).toBe('half-open'); // getter transitions _state to half-open
|
||||
cb.recordFailure(); // failure in half-open → re-opens
|
||||
expect(cb.isOpen()).toBe(true);
|
||||
expect(cb.state).toBe('open');
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it('closes the circuit when success is recorded in half-open state', () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const cb = new HfDownloadCircuitBreaker(1, 100 /* 100ms */);
|
||||
cb.recordFailure(); // opens the circuit
|
||||
vi.advanceTimersByTime(200); // advance past reset timeout
|
||||
expect(cb.state).toBe('half-open');
|
||||
cb.recordSuccess(); // success in half-open → closes
|
||||
expect(cb.isOpen()).toBe(false);
|
||||
expect(cb.state).toBe('closed');
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
});
|
||||
// CircuitBreaker state-machine tests live in
|
||||
// `gitnexus/test/unit/integrations/circuit-breaker.test.ts` — that suite
|
||||
// already covers the closed/open/half-open transitions, recordSuccess/
|
||||
// recordFailure semantics, half-open probe gating, and configurable
|
||||
// thresholds. No need to duplicate here; this file's remaining tests
|
||||
// focus on HF-specific composition (withHfDownloadRetry, env-var
|
||||
// overrides, error classification).
|
||||
|
||||
describe('withDownloadTimeout', () => {
|
||||
it('resolves when fn completes before the timeout', async () => {
|
||||
|
|
@ -262,7 +190,7 @@ describe('withDownloadTimeout', () => {
|
|||
describe('withHfDownloadRetry', () => {
|
||||
it('returns the result on first success', async () => {
|
||||
const fn = vi.fn().mockResolvedValue('ok');
|
||||
const cb = new HfDownloadCircuitBreaker();
|
||||
const cb = new CircuitBreaker();
|
||||
const result = await withHfDownloadRetry(fn, { circuit: cb, baseDelayMs: 0 });
|
||||
expect(result).toBe('ok');
|
||||
expect(fn).toHaveBeenCalledTimes(1);
|
||||
|
|
@ -270,7 +198,7 @@ describe('withHfDownloadRetry', () => {
|
|||
|
||||
it('retries on network errors and succeeds on second attempt', async () => {
|
||||
const fn = vi.fn().mockRejectedValueOnce(new Error('fetch failed')).mockResolvedValue('ok');
|
||||
const cb = new HfDownloadCircuitBreaker();
|
||||
const cb = new CircuitBreaker();
|
||||
const result = await withHfDownloadRetry(fn, {
|
||||
circuit: cb,
|
||||
maxAttempts: 3,
|
||||
|
|
@ -282,7 +210,7 @@ describe('withHfDownloadRetry', () => {
|
|||
|
||||
it('throws the last network error after all attempts are exhausted', async () => {
|
||||
const fn = vi.fn().mockRejectedValue(new Error('ECONNREFUSED 127.0.0.1:443'));
|
||||
const cb = new HfDownloadCircuitBreaker(99 /* high threshold */);
|
||||
const cb = new CircuitBreaker({ failureThreshold: 99 });
|
||||
await expect(
|
||||
withHfDownloadRetry(fn, { circuit: cb, maxAttempts: 3, baseDelayMs: 0 }),
|
||||
).rejects.toThrow('ECONNREFUSED');
|
||||
|
|
@ -291,7 +219,7 @@ describe('withHfDownloadRetry', () => {
|
|||
|
||||
it('does not retry non-network errors', async () => {
|
||||
const fn = vi.fn().mockRejectedValue(new Error('Failed to initialize CUDA backend'));
|
||||
const cb = new HfDownloadCircuitBreaker();
|
||||
const cb = new CircuitBreaker();
|
||||
await expect(
|
||||
withHfDownloadRetry(fn, { circuit: cb, maxAttempts: 3, baseDelayMs: 0 }),
|
||||
).rejects.toThrow('Failed to initialize CUDA backend');
|
||||
|
|
@ -300,7 +228,7 @@ describe('withHfDownloadRetry', () => {
|
|||
|
||||
it('fails immediately when the circuit is already open', async () => {
|
||||
const fn = vi.fn().mockResolvedValue('ok');
|
||||
const cb = new HfDownloadCircuitBreaker(1);
|
||||
const cb = new CircuitBreaker({ failureThreshold: 1 });
|
||||
cb.recordFailure(); // open the circuit
|
||||
await expect(withHfDownloadRetry(fn, { circuit: cb })).rejects.toThrow(CIRCUIT_OPEN_TAG);
|
||||
expect(fn).not.toHaveBeenCalled();
|
||||
|
|
@ -308,12 +236,12 @@ describe('withHfDownloadRetry', () => {
|
|||
|
||||
it('opens the circuit after failureThreshold failures and throws a circuit-open error', async () => {
|
||||
const fn = vi.fn().mockRejectedValue(new Error('ENOTFOUND huggingface.co'));
|
||||
const cb = new HfDownloadCircuitBreaker(2 /* threshold */, 60_000);
|
||||
const cb = new CircuitBreaker({ failureThreshold: 2, cooldownMs: 60_000 });
|
||||
// First call: 2 attempts, threshold=2 → circuit opens on 2nd failure
|
||||
await expect(
|
||||
withHfDownloadRetry(fn, { circuit: cb, maxAttempts: 2, baseDelayMs: 0 }),
|
||||
).rejects.toThrow(CIRCUIT_OPEN_TAG);
|
||||
expect(cb.isOpen()).toBe(true);
|
||||
expect(cb.getState()).toBe('open');
|
||||
});
|
||||
|
||||
it('calls onRetry with correct arguments on each retry', async () => {
|
||||
|
|
@ -322,7 +250,7 @@ describe('withHfDownloadRetry', () => {
|
|||
.mockRejectedValueOnce(new Error('fetch failed'))
|
||||
.mockRejectedValueOnce(new Error('fetch failed'))
|
||||
.mockResolvedValue('ok');
|
||||
const cb = new HfDownloadCircuitBreaker(99);
|
||||
const cb = new CircuitBreaker({ failureThreshold: 99 });
|
||||
const onRetry = vi.fn();
|
||||
await withHfDownloadRetry(fn, { circuit: cb, maxAttempts: 3, baseDelayMs: 0, onRetry });
|
||||
expect(onRetry).toHaveBeenCalledTimes(2);
|
||||
|
|
@ -342,11 +270,11 @@ describe('withHfDownloadRetry', () => {
|
|||
|
||||
it('resets the circuit on success', async () => {
|
||||
const fn = vi.fn().mockResolvedValue('value');
|
||||
const cb = new HfDownloadCircuitBreaker(5);
|
||||
const cb = new CircuitBreaker({ failureThreshold: 5 });
|
||||
cb.recordFailure();
|
||||
cb.recordFailure(); // 2 failures, circuit still closed
|
||||
await withHfDownloadRetry(fn, { circuit: cb, baseDelayMs: 0 });
|
||||
expect(cb.state).toBe('closed');
|
||||
expect(cb.getState()).toBe('closed');
|
||||
});
|
||||
});
|
||||
|
||||
|
|
@ -371,7 +299,7 @@ describe('withHfDownloadRetry env overrides', () => {
|
|||
it('HF_MAX_ATTEMPTS=1 gives exactly 1 attempt', async () => {
|
||||
process.env.HF_MAX_ATTEMPTS = '1';
|
||||
const fn = vi.fn().mockRejectedValue(new Error('ECONNREFUSED 127.0.0.1:443'));
|
||||
const cb = new HfDownloadCircuitBreaker(99_999 /* high threshold */);
|
||||
const cb = new CircuitBreaker({ failureThreshold: 99_999 });
|
||||
await expect(withHfDownloadRetry(fn, { circuit: cb, baseDelayMs: 0 })).rejects.toThrow(
|
||||
'ECONNREFUSED',
|
||||
);
|
||||
|
|
@ -381,7 +309,7 @@ describe('withHfDownloadRetry env overrides', () => {
|
|||
it('HF_MAX_ATTEMPTS=2 gives exactly 2 attempts', async () => {
|
||||
process.env.HF_MAX_ATTEMPTS = '2';
|
||||
const fn = vi.fn().mockRejectedValue(new Error('ENOTFOUND huggingface.co'));
|
||||
const cb = new HfDownloadCircuitBreaker(99_999);
|
||||
const cb = new CircuitBreaker({ failureThreshold: 99_999 });
|
||||
await expect(withHfDownloadRetry(fn, { circuit: cb, baseDelayMs: 0 })).rejects.toThrow(
|
||||
'ENOTFOUND',
|
||||
);
|
||||
|
|
@ -391,7 +319,7 @@ describe('withHfDownloadRetry env overrides', () => {
|
|||
it('HF_MAX_ATTEMPTS=abc falls back to the built-in default', async () => {
|
||||
process.env.HF_MAX_ATTEMPTS = 'abc';
|
||||
const fn = vi.fn().mockRejectedValue(new Error('fetch failed'));
|
||||
const cb = new HfDownloadCircuitBreaker(99_999);
|
||||
const cb = new CircuitBreaker({ failureThreshold: 99_999 });
|
||||
await expect(withHfDownloadRetry(fn, { circuit: cb, baseDelayMs: 0 })).rejects.toThrow(
|
||||
'fetch failed',
|
||||
);
|
||||
|
|
@ -401,7 +329,7 @@ describe('withHfDownloadRetry env overrides', () => {
|
|||
it('HF_MAX_ATTEMPTS=0 falls back to the built-in default', async () => {
|
||||
process.env.HF_MAX_ATTEMPTS = '0';
|
||||
const fn = vi.fn().mockRejectedValue(new Error('fetch failed'));
|
||||
const cb = new HfDownloadCircuitBreaker(99_999);
|
||||
const cb = new CircuitBreaker({ failureThreshold: 99_999 });
|
||||
await expect(withHfDownloadRetry(fn, { circuit: cb, baseDelayMs: 0 })).rejects.toThrow(
|
||||
'fetch failed',
|
||||
);
|
||||
|
|
@ -411,7 +339,7 @@ describe('withHfDownloadRetry env overrides', () => {
|
|||
it('HF_MAX_ATTEMPTS=-1 falls back to the built-in default', async () => {
|
||||
process.env.HF_MAX_ATTEMPTS = '-1';
|
||||
const fn = vi.fn().mockRejectedValue(new Error('fetch failed'));
|
||||
const cb = new HfDownloadCircuitBreaker(99_999);
|
||||
const cb = new CircuitBreaker({ failureThreshold: 99_999 });
|
||||
await expect(withHfDownloadRetry(fn, { circuit: cb, baseDelayMs: 0 })).rejects.toThrow(
|
||||
'fetch failed',
|
||||
);
|
||||
|
|
@ -421,7 +349,7 @@ describe('withHfDownloadRetry env overrides', () => {
|
|||
it('HF_MAX_ATTEMPTS is clamped to HF_MAX_ATTEMPTS_CAP', async () => {
|
||||
process.env.HF_MAX_ATTEMPTS = '9999';
|
||||
const fn = vi.fn().mockRejectedValue(new Error('fetch failed'));
|
||||
const cb = new HfDownloadCircuitBreaker(99_999 /* very high threshold */);
|
||||
const cb = new CircuitBreaker({ failureThreshold: 99_999 });
|
||||
await expect(withHfDownloadRetry(fn, { circuit: cb, baseDelayMs: 0 })).rejects.toThrow(
|
||||
'fetch failed',
|
||||
);
|
||||
|
|
@ -431,7 +359,7 @@ describe('withHfDownloadRetry env overrides', () => {
|
|||
it('HF_MAX_ATTEMPTS=2.9 is floored to 2', async () => {
|
||||
process.env.HF_MAX_ATTEMPTS = '2.9';
|
||||
const fn = vi.fn().mockRejectedValue(new Error('fetch failed'));
|
||||
const cb = new HfDownloadCircuitBreaker(99_999);
|
||||
const cb = new CircuitBreaker({ failureThreshold: 99_999 });
|
||||
await expect(withHfDownloadRetry(fn, { circuit: cb, baseDelayMs: 0 })).rejects.toThrow(
|
||||
'fetch failed',
|
||||
);
|
||||
|
|
@ -443,7 +371,7 @@ describe('withHfDownloadRetry env overrides', () => {
|
|||
try {
|
||||
process.env.HF_DOWNLOAD_TIMEOUT_MS = '50';
|
||||
const neverResolves = () => new Promise<never>(() => {});
|
||||
const cb = new HfDownloadCircuitBreaker(99);
|
||||
const cb = new CircuitBreaker({ failureThreshold: 99 });
|
||||
const promise = withHfDownloadRetry(neverResolves, { circuit: cb, maxAttempts: 1 });
|
||||
vi.advanceTimersByTime(100);
|
||||
await expect(promise).rejects.toThrow('ETIMEDOUT');
|
||||
|
|
@ -458,7 +386,7 @@ describe('withHfDownloadRetry env overrides', () => {
|
|||
// we just verify that the env var rejection causes options.timeoutMs to be
|
||||
// the default constant (not -1) by confirming the resolved value is used.
|
||||
const fn = vi.fn().mockResolvedValue('ok');
|
||||
const cb = new HfDownloadCircuitBreaker(99);
|
||||
const cb = new CircuitBreaker({ failureThreshold: 99 });
|
||||
// Provide explicit timeoutMs to avoid the default 5-minute wait
|
||||
const result = await withHfDownloadRetry(fn, { circuit: cb, timeoutMs: 100 });
|
||||
expect(result).toBe('ok');
|
||||
|
|
@ -470,7 +398,7 @@ describe('withHfDownloadRetry env overrides', () => {
|
|||
// Set an env value exceeding the 30-minute cap
|
||||
process.env.HF_DOWNLOAD_TIMEOUT_MS = String(HF_MAX_TIMEOUT_MS + 60_000);
|
||||
const neverResolves = () => new Promise<never>(() => {});
|
||||
const cb = new HfDownloadCircuitBreaker(99);
|
||||
const cb = new CircuitBreaker({ failureThreshold: 99 });
|
||||
const promise = withHfDownloadRetry(neverResolves, { circuit: cb, maxAttempts: 1 });
|
||||
// Advance just past the 30-minute cap
|
||||
vi.advanceTimersByTime(HF_MAX_TIMEOUT_MS + 1);
|
||||
|
|
@ -483,7 +411,7 @@ describe('withHfDownloadRetry env overrides', () => {
|
|||
it('explicit options override env vars', async () => {
|
||||
process.env.HF_MAX_ATTEMPTS = '5';
|
||||
const fn = vi.fn().mockRejectedValue(new Error('fetch failed'));
|
||||
const cb = new HfDownloadCircuitBreaker(99);
|
||||
const cb = new CircuitBreaker({ failureThreshold: 99 });
|
||||
// explicit maxAttempts: 2 must win over HF_MAX_ATTEMPTS=5
|
||||
await expect(
|
||||
withHfDownloadRetry(fn, { circuit: cb, maxAttempts: 2, baseDelayMs: 0 }),
|
||||
|
|
|
|||
395
gitnexus/test/unit/integrations/circuit-breaker.test.ts
Normal file
395
gitnexus/test/unit/integrations/circuit-breaker.test.ts
Normal file
|
|
@ -0,0 +1,395 @@
|
|||
import { describe, it, expect, beforeEach } from 'vitest';
|
||||
import { CircuitBreaker, CircuitOpenError, getBreaker } from 'gitnexus-shared';
|
||||
import { __resetBreakerRegistry__ } from 'gitnexus-shared/test-helpers';
|
||||
|
||||
describe('CircuitBreaker', () => {
|
||||
beforeEach(() => __resetBreakerRegistry__());
|
||||
|
||||
function makeClock(start = 1_700_000_000_000) {
|
||||
let t = start;
|
||||
return {
|
||||
now: () => t,
|
||||
advance: (ms: number) => {
|
||||
t += ms;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
it('runs through check/recordSuccess in closed state', () => {
|
||||
const clock = makeClock();
|
||||
const b = new CircuitBreaker({ failureThreshold: 3, cooldownMs: 30_000, now: clock.now });
|
||||
expect(b.getState()).toBe('closed');
|
||||
b.check(); // does not throw
|
||||
b.recordSuccess();
|
||||
expect(b.getState()).toBe('closed');
|
||||
expect(b.getConsecutiveFailures()).toBe(0);
|
||||
});
|
||||
|
||||
it('stays closed below the failure threshold', () => {
|
||||
const clock = makeClock();
|
||||
const b = new CircuitBreaker({ failureThreshold: 3, cooldownMs: 30_000, now: clock.now });
|
||||
b.recordFailure();
|
||||
b.recordFailure();
|
||||
expect(b.getState()).toBe('closed');
|
||||
expect(b.getConsecutiveFailures()).toBe(2);
|
||||
});
|
||||
|
||||
it('opens after failureThreshold consecutive failures and check throws', () => {
|
||||
const clock = makeClock();
|
||||
const b = new CircuitBreaker({ failureThreshold: 3, cooldownMs: 30_000, now: clock.now });
|
||||
b.recordFailure();
|
||||
b.recordFailure();
|
||||
b.recordFailure();
|
||||
expect(b.getState()).toBe('open');
|
||||
expect(() => b.check()).toThrow(CircuitOpenError);
|
||||
});
|
||||
|
||||
it('CircuitOpenError.retryAfterMs decreases as time advances', () => {
|
||||
const clock = makeClock();
|
||||
const b = new CircuitBreaker({ failureThreshold: 1, cooldownMs: 30_000, now: clock.now });
|
||||
b.recordFailure();
|
||||
let caught: CircuitOpenError | null = null;
|
||||
try {
|
||||
b.check();
|
||||
} catch (err) {
|
||||
caught = err as CircuitOpenError;
|
||||
}
|
||||
expect(caught?.retryAfterMs).toBe(30_000);
|
||||
|
||||
clock.advance(10_000);
|
||||
try {
|
||||
b.check();
|
||||
} catch (err) {
|
||||
caught = err as CircuitOpenError;
|
||||
}
|
||||
expect(caught?.retryAfterMs).toBe(20_000);
|
||||
});
|
||||
|
||||
it('transitions Open -> Half-Open after cooldown elapses (via check)', () => {
|
||||
const clock = makeClock();
|
||||
const b = new CircuitBreaker({ failureThreshold: 1, cooldownMs: 30_000, now: clock.now });
|
||||
b.recordFailure();
|
||||
expect(b.getState()).toBe('open');
|
||||
clock.advance(31_000);
|
||||
b.check(); // should not throw
|
||||
// After check, internal state is half-open (next call probes).
|
||||
expect(b.getConsecutiveFailures()).toBe(1); // unchanged until next outcome
|
||||
});
|
||||
|
||||
it('half-open + recordSuccess -> closed and counter reset', () => {
|
||||
const clock = makeClock();
|
||||
const b = new CircuitBreaker({ failureThreshold: 1, cooldownMs: 30_000, now: clock.now });
|
||||
b.recordFailure();
|
||||
clock.advance(31_000);
|
||||
b.check();
|
||||
b.recordSuccess();
|
||||
expect(b.getState()).toBe('closed');
|
||||
expect(b.getConsecutiveFailures()).toBe(0);
|
||||
});
|
||||
|
||||
it('half-open + recordFailure -> open with fresh openedAt', () => {
|
||||
const clock = makeClock();
|
||||
const b = new CircuitBreaker({ failureThreshold: 1, cooldownMs: 30_000, now: clock.now });
|
||||
b.recordFailure();
|
||||
const firstOpen = clock.now();
|
||||
clock.advance(31_000); // cooldown expired
|
||||
b.check(); // half-open
|
||||
b.recordFailure();
|
||||
// Open with fresh timestamp — full cooldown again.
|
||||
let caught: CircuitOpenError | null = null;
|
||||
try {
|
||||
b.check();
|
||||
} catch (err) {
|
||||
caught = err as CircuitOpenError;
|
||||
}
|
||||
expect(caught).toBeInstanceOf(CircuitOpenError);
|
||||
expect(caught?.retryAfterMs).toBe(30_000);
|
||||
// Sanity: not the original openedAt (would be negative remaining).
|
||||
expect(clock.now()).toBeGreaterThan(firstOpen);
|
||||
});
|
||||
|
||||
it('recordSuccess from closed state with prior partial failures resets counter', () => {
|
||||
const b = new CircuitBreaker({ failureThreshold: 5 });
|
||||
b.recordFailure();
|
||||
b.recordFailure();
|
||||
expect(b.getConsecutiveFailures()).toBe(2);
|
||||
b.recordSuccess();
|
||||
expect(b.getConsecutiveFailures()).toBe(0);
|
||||
expect(b.getState()).toBe('closed');
|
||||
});
|
||||
|
||||
describe('recordNeutral (U1)', () => {
|
||||
it('is a no-op from closed state with zero prior failures', () => {
|
||||
const b = new CircuitBreaker({ failureThreshold: 3 });
|
||||
b.recordNeutral();
|
||||
expect(b.getState()).toBe('closed');
|
||||
expect(b.getConsecutiveFailures()).toBe(0);
|
||||
});
|
||||
|
||||
it('preserves partial-failure progress (does not reset counter)', () => {
|
||||
const b = new CircuitBreaker({ failureThreshold: 3 });
|
||||
b.recordFailure();
|
||||
b.recordFailure();
|
||||
b.recordNeutral();
|
||||
expect(b.getConsecutiveFailures()).toBe(2);
|
||||
expect(b.getState()).toBe('closed');
|
||||
// Real third failure still trips the breaker — neutrals didn't
|
||||
// erase the running count toward the threshold.
|
||||
b.recordFailure();
|
||||
expect(b.getState()).toBe('open');
|
||||
});
|
||||
|
||||
it('does not reset openedAt or transition out of open state', () => {
|
||||
const clock = makeClock();
|
||||
const b = new CircuitBreaker({ failureThreshold: 1, cooldownMs: 30_000, now: clock.now });
|
||||
b.recordFailure();
|
||||
expect(b.getState()).toBe('open');
|
||||
b.recordNeutral();
|
||||
// Still open; cooldown clock unchanged.
|
||||
expect(() => b.check()).toThrow(CircuitOpenError);
|
||||
});
|
||||
|
||||
it('leaves half-open state alone (next true outcome decides)', () => {
|
||||
const clock = makeClock();
|
||||
const b = new CircuitBreaker({ failureThreshold: 1, cooldownMs: 30_000, now: clock.now });
|
||||
b.recordFailure();
|
||||
clock.advance(31_000);
|
||||
b.check(); // half-open
|
||||
b.recordNeutral();
|
||||
// Still half-open; a subsequent recordFailure flips to open.
|
||||
b.recordFailure();
|
||||
let caught: CircuitOpenError | null = null;
|
||||
try {
|
||||
b.check();
|
||||
} catch (err) {
|
||||
caught = err as CircuitOpenError;
|
||||
}
|
||||
expect(caught).toBeInstanceOf(CircuitOpenError);
|
||||
});
|
||||
|
||||
it('integration: 2 failures + 5 neutrals + 1 failure → opens on third real failure', () => {
|
||||
const b = new CircuitBreaker({ failureThreshold: 3 });
|
||||
b.recordFailure();
|
||||
b.recordFailure();
|
||||
for (let i = 0; i < 5; i++) b.recordNeutral();
|
||||
expect(b.getConsecutiveFailures()).toBe(2);
|
||||
expect(b.getState()).toBe('closed');
|
||||
b.recordFailure();
|
||||
expect(b.getState()).toBe('open');
|
||||
});
|
||||
});
|
||||
|
||||
describe('half-open probe permit gate (U1)', () => {
|
||||
it('admits exactly one caller after cooldown; subsequent check() throws halfOpenRetryAfterMs', () => {
|
||||
const clock = makeClock();
|
||||
const b = new CircuitBreaker({
|
||||
failureThreshold: 1,
|
||||
cooldownMs: 30_000,
|
||||
halfOpenRetryAfterMs: 1_000,
|
||||
now: clock.now,
|
||||
});
|
||||
b.recordFailure();
|
||||
clock.advance(31_000);
|
||||
|
||||
// Caller A: gets the probe permit.
|
||||
b.check();
|
||||
expect(b.isProbeInFlight()).toBe(true);
|
||||
|
||||
// Caller B: blocked.
|
||||
let caught: CircuitOpenError | null = null;
|
||||
try {
|
||||
b.check();
|
||||
} catch (err) {
|
||||
caught = err as CircuitOpenError;
|
||||
}
|
||||
expect(caught).toBeInstanceOf(CircuitOpenError);
|
||||
expect(caught?.retryAfterMs).toBe(1_000);
|
||||
|
||||
// Caller A's recordSuccess clears the breaker.
|
||||
b.recordSuccess();
|
||||
expect(b.isProbeInFlight()).toBe(false);
|
||||
expect(b.getState()).toBe('closed');
|
||||
|
||||
// Caller C: succeeds in closed state.
|
||||
b.check();
|
||||
expect(b.getState()).toBe('closed');
|
||||
});
|
||||
|
||||
it('recordFailure on probe re-opens with fresh cooldown (NOT halfOpenRetryAfterMs)', () => {
|
||||
const clock = makeClock();
|
||||
const b = new CircuitBreaker({
|
||||
failureThreshold: 1,
|
||||
cooldownMs: 30_000,
|
||||
halfOpenRetryAfterMs: 1_000,
|
||||
now: clock.now,
|
||||
});
|
||||
b.recordFailure();
|
||||
clock.advance(31_000);
|
||||
|
||||
b.check(); // A: probe
|
||||
expect(() => b.check()).toThrow(CircuitOpenError); // B: blocked
|
||||
|
||||
b.recordFailure(); // A reports failure → reopens with fresh openedAt
|
||||
|
||||
// C: should see the fresh cooldown remaining, not the probe-in-flight 1s default.
|
||||
let caught: CircuitOpenError | null = null;
|
||||
try {
|
||||
b.check();
|
||||
} catch (err) {
|
||||
caught = err as CircuitOpenError;
|
||||
}
|
||||
expect(caught).toBeInstanceOf(CircuitOpenError);
|
||||
// Fresh openedAt = current clock; cooldown is 30s; retryAfter ≈ 30s.
|
||||
expect(caught?.retryAfterMs).toBe(30_000);
|
||||
});
|
||||
|
||||
it('recordNeutral releases the probe permit but leaves state half-open', () => {
|
||||
const clock = makeClock();
|
||||
const b = new CircuitBreaker({ failureThreshold: 1, cooldownMs: 30_000, now: clock.now });
|
||||
b.recordFailure();
|
||||
clock.advance(31_000);
|
||||
|
||||
b.check(); // A: probe
|
||||
expect(b.isProbeInFlight()).toBe(true);
|
||||
|
||||
b.recordNeutral(); // A: neutral — permit released, state untouched
|
||||
expect(b.isProbeInFlight()).toBe(false);
|
||||
expect(b.getState()).toBe('half-open');
|
||||
|
||||
// B: succeeds (becomes the new probe), no longer blocked.
|
||||
b.check();
|
||||
expect(b.isProbeInFlight()).toBe(true);
|
||||
|
||||
// B's recordSuccess clears the breaker.
|
||||
b.recordSuccess();
|
||||
expect(b.getState()).toBe('closed');
|
||||
});
|
||||
|
||||
it('three sequential probes via neutrals: A → A.neutral → B → B.neutral → C', () => {
|
||||
const clock = makeClock();
|
||||
const b = new CircuitBreaker({ failureThreshold: 1, cooldownMs: 30_000, now: clock.now });
|
||||
b.recordFailure();
|
||||
const initialFailures = b.getConsecutiveFailures();
|
||||
clock.advance(31_000);
|
||||
|
||||
for (let i = 0; i < 3; i++) {
|
||||
b.check();
|
||||
b.recordNeutral();
|
||||
}
|
||||
// Counter unchanged; state still half-open; permit released.
|
||||
expect(b.getConsecutiveFailures()).toBe(initialFailures);
|
||||
expect(b.getState()).toBe('half-open');
|
||||
expect(b.isProbeInFlight()).toBe(false);
|
||||
});
|
||||
|
||||
it('5 same-tick sequential callers: exactly one passes, the other 4 throw', () => {
|
||||
// `check()` is synchronous — these calls execute on a single
|
||||
// microtask in declaration order. The first mutates probeInFlight
|
||||
// = true; the next four observe the mutation and throw. This
|
||||
// tests mutation ordering, not true concurrency (the actual
|
||||
// interleaved-async-microtask scenario lives in U2).
|
||||
const clock = makeClock();
|
||||
const b = new CircuitBreaker({ failureThreshold: 1, cooldownMs: 30_000, now: clock.now });
|
||||
b.recordFailure();
|
||||
clock.advance(31_000);
|
||||
|
||||
const results: Array<'pass' | 'throw'> = [];
|
||||
for (let i = 0; i < 5; i++) {
|
||||
try {
|
||||
b.check();
|
||||
results.push('pass');
|
||||
} catch {
|
||||
results.push('throw');
|
||||
}
|
||||
}
|
||||
expect(results.filter((r) => r === 'pass').length).toBe(1);
|
||||
expect(results.filter((r) => r === 'throw').length).toBe(4);
|
||||
});
|
||||
|
||||
it('probe permit consumed; clock advances another full cooldown without record*; still throws', () => {
|
||||
const clock = makeClock();
|
||||
const b = new CircuitBreaker({ failureThreshold: 1, cooldownMs: 30_000, now: clock.now });
|
||||
b.recordFailure();
|
||||
clock.advance(31_000);
|
||||
|
||||
b.check(); // probe permit consumed
|
||||
clock.advance(60_000); // another full cooldown elapses, no record*
|
||||
|
||||
// Half-open semantics: wait for an outcome, not a timer. The
|
||||
// permit-consumed state doesn't auto-resolve on time.
|
||||
expect(() => b.check()).toThrow(CircuitOpenError);
|
||||
});
|
||||
|
||||
it('halfOpenRetryAfterMs default is 1000 when not configured', () => {
|
||||
const clock = makeClock();
|
||||
const b = new CircuitBreaker({ failureThreshold: 1, cooldownMs: 30_000, now: clock.now });
|
||||
b.recordFailure();
|
||||
clock.advance(31_000);
|
||||
b.check();
|
||||
|
||||
let caught: CircuitOpenError | null = null;
|
||||
try {
|
||||
b.check();
|
||||
} catch (err) {
|
||||
caught = err as CircuitOpenError;
|
||||
}
|
||||
expect(caught?.retryAfterMs).toBe(1_000);
|
||||
});
|
||||
|
||||
it('halfOpenRetryAfterMs is configurable for long-running protected ops', () => {
|
||||
const clock = makeClock();
|
||||
const b = new CircuitBreaker({
|
||||
failureThreshold: 1,
|
||||
cooldownMs: 30_000,
|
||||
halfOpenRetryAfterMs: 10_000, // LLM-streaming-friendly
|
||||
now: clock.now,
|
||||
});
|
||||
b.recordFailure();
|
||||
clock.advance(31_000);
|
||||
b.check();
|
||||
|
||||
let caught: CircuitOpenError | null = null;
|
||||
try {
|
||||
b.check();
|
||||
} catch (err) {
|
||||
caught = err as CircuitOpenError;
|
||||
}
|
||||
expect(caught?.retryAfterMs).toBe(10_000);
|
||||
});
|
||||
|
||||
it('getState() is a pure read — does not consume the probe permit', () => {
|
||||
const clock = makeClock();
|
||||
const b = new CircuitBreaker({ failureThreshold: 1, cooldownMs: 30_000, now: clock.now });
|
||||
b.recordFailure();
|
||||
clock.advance(31_000);
|
||||
|
||||
// Test calls getState() to inspect — must not consume the permit.
|
||||
expect(b.getState()).toBe('half-open');
|
||||
expect(b.isProbeInFlight()).toBe(false);
|
||||
// First check() still gets the permit.
|
||||
b.check();
|
||||
expect(b.isProbeInFlight()).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe('getBreaker registry', () => {
|
||||
it('returns the same instance for the same key', () => {
|
||||
const a = getBreaker('endpoint-a');
|
||||
const b = getBreaker('endpoint-a');
|
||||
expect(a).toBe(b);
|
||||
});
|
||||
|
||||
it('returns different instances for different keys', () => {
|
||||
const a = getBreaker('endpoint-a');
|
||||
const b = getBreaker('endpoint-b');
|
||||
expect(a).not.toBe(b);
|
||||
});
|
||||
|
||||
it('__resetBreakerRegistry__ clears all instances', () => {
|
||||
const a = getBreaker('endpoint-a');
|
||||
__resetBreakerRegistry__();
|
||||
const a2 = getBreaker('endpoint-a');
|
||||
expect(a2).not.toBe(a);
|
||||
});
|
||||
});
|
||||
});
|
||||
558
gitnexus/test/unit/integrations/resilient-fetch.test.ts
Normal file
558
gitnexus/test/unit/integrations/resilient-fetch.test.ts
Normal file
|
|
@ -0,0 +1,558 @@
|
|||
import { describe, it, expect, beforeEach, vi } from 'vitest';
|
||||
import {
|
||||
CircuitBreaker,
|
||||
CircuitOpenError,
|
||||
parseRetryAfter,
|
||||
resilientFetch,
|
||||
ResilientFetchExhaustedError,
|
||||
RETRY_AFTER_CAP_MS,
|
||||
} from 'gitnexus-shared';
|
||||
import { __resetBreakerRegistry__, classifyOutcome } from 'gitnexus-shared/test-helpers';
|
||||
|
||||
describe('parseRetryAfter', () => {
|
||||
it('parses delta-seconds form', () => {
|
||||
expect(parseRetryAfter('30')).toBe(30_000);
|
||||
expect(parseRetryAfter('0')).toBe(0);
|
||||
});
|
||||
it('returns null on negative or non-numeric garbage', () => {
|
||||
expect(parseRetryAfter(null)).toBeNull();
|
||||
expect(parseRetryAfter('')).toBeNull();
|
||||
expect(parseRetryAfter(' ')).toBeNull();
|
||||
expect(parseRetryAfter('not-a-number')).toBeNull();
|
||||
});
|
||||
it('parses HTTP-date form against an injected clock', () => {
|
||||
const now = () => Date.parse('Wed, 21 Oct 2025 07:28:00 GMT');
|
||||
expect(parseRetryAfter('Wed, 21 Oct 2025 07:28:30 GMT', now)).toBe(30_000);
|
||||
});
|
||||
it('returns 0 (not negative) on past HTTP-date', () => {
|
||||
const now = () => Date.parse('Wed, 21 Oct 2025 08:00:00 GMT');
|
||||
expect(parseRetryAfter('Wed, 21 Oct 2025 07:28:00 GMT', now)).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe('classifyOutcome', () => {
|
||||
const now = () => 1_700_000_000_000;
|
||||
|
||||
it('classifies 2xx as success', () => {
|
||||
const resp = new Response(null, { status: 204 });
|
||||
const out = classifyOutcome({ kind: 'response', resp }, now);
|
||||
expect(out.kind).toBe('success');
|
||||
});
|
||||
it('classifies 5xx as retryable-status without afterMs', () => {
|
||||
const resp = new Response(null, { status: 503 });
|
||||
const out = classifyOutcome({ kind: 'response', resp }, now);
|
||||
expect(out.kind).toBe('retryable-status');
|
||||
if (out.kind === 'retryable-status') expect(out.afterMs).toBeUndefined();
|
||||
});
|
||||
it('classifies 429 with Retry-After (capped) as retryable-status', () => {
|
||||
const resp = new Response(null, { status: 429, headers: { 'Retry-After': '99999' } });
|
||||
const out = classifyOutcome({ kind: 'response', resp }, now);
|
||||
expect(out.kind).toBe('retryable-status');
|
||||
if (out.kind === 'retryable-status') expect(out.afterMs).toBe(RETRY_AFTER_CAP_MS);
|
||||
});
|
||||
it('classifies 429 from a header-less fetch mock without throwing', () => {
|
||||
// Tests sometimes stub `fetch` with a plain `{ ok, status }` object
|
||||
// (e.g. http-embedder.test.ts). Real `Response` always carries
|
||||
// `Headers`, but the helper must not crash when the stub does not.
|
||||
// Falls through to exponential-backoff retry like a 429 with no
|
||||
// Retry-After header.
|
||||
const resp = { ok: false, status: 429 } as unknown as Response;
|
||||
const out = classifyOutcome({ kind: 'response', resp }, now);
|
||||
expect(out.kind).toBe('retryable-status');
|
||||
if (out.kind === 'retryable-status') expect(out.afterMs).toBeUndefined();
|
||||
});
|
||||
it('classifies 401/403/404/422 as terminal-client', () => {
|
||||
for (const status of [401, 403, 404, 422, 400]) {
|
||||
const resp = new Response(null, { status });
|
||||
const out = classifyOutcome({ kind: 'response', resp }, now);
|
||||
expect(out.kind).toBe('terminal-client');
|
||||
}
|
||||
});
|
||||
it('classifies TimeoutError as terminal-network', () => {
|
||||
const err = new DOMException('aborted', 'TimeoutError');
|
||||
const out = classifyOutcome({ kind: 'error', err }, now);
|
||||
expect(out.kind).toBe('terminal-network');
|
||||
});
|
||||
it('classifies generic network throw as retryable-network', () => {
|
||||
const err = new TypeError('fetch failed');
|
||||
const out = classifyOutcome({ kind: 'error', err }, now);
|
||||
expect(out.kind).toBe('retryable-network');
|
||||
});
|
||||
});
|
||||
|
||||
describe('resilientFetch', () => {
|
||||
const URL_STR = 'https://example.test/api/dispatch';
|
||||
|
||||
beforeEach(() => __resetBreakerRegistry__());
|
||||
|
||||
function jsonResp(status: number, headers?: Record<string, string>): Response {
|
||||
return new Response(null, { status, headers });
|
||||
}
|
||||
|
||||
function makeBreaker(opts: Partial<ConstructorParameters<typeof CircuitBreaker>[0]> = {}) {
|
||||
let t = 1_700_000_000_000;
|
||||
const breaker = new CircuitBreaker({
|
||||
failureThreshold: 3,
|
||||
cooldownMs: 30_000,
|
||||
key: 'test',
|
||||
now: () => t,
|
||||
...opts,
|
||||
});
|
||||
return { breaker, advance: (ms: number) => (t += ms) };
|
||||
}
|
||||
|
||||
it('204 returns immediately, no retries, breaker stays closed', async () => {
|
||||
const fetchImpl = vi.fn(async () => jsonResp(204));
|
||||
const sleep = vi.fn(async () => {});
|
||||
const { breaker } = makeBreaker();
|
||||
const resp = await resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep },
|
||||
});
|
||||
expect(resp.status).toBe(204);
|
||||
expect(fetchImpl).toHaveBeenCalledTimes(1);
|
||||
expect(sleep).not.toHaveBeenCalled();
|
||||
expect(breaker.getState()).toBe('closed');
|
||||
expect(breaker.getConsecutiveFailures()).toBe(0);
|
||||
});
|
||||
|
||||
it('one 503 then 204 → retried once, returns 204, breaker stays closed', async () => {
|
||||
let n = 0;
|
||||
const fetchImpl = vi.fn(async () => {
|
||||
n += 1;
|
||||
return n === 1 ? jsonResp(503) : jsonResp(204);
|
||||
});
|
||||
const sleep = vi.fn(async () => {});
|
||||
const { breaker } = makeBreaker();
|
||||
const resp = await resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep, random: () => 0.5, baseDelayMs: 100, capDelayMs: 1000 },
|
||||
});
|
||||
expect(resp.status).toBe(204);
|
||||
expect(fetchImpl).toHaveBeenCalledTimes(2);
|
||||
expect(sleep).toHaveBeenCalledTimes(1);
|
||||
expect(breaker.getConsecutiveFailures()).toBe(0);
|
||||
});
|
||||
|
||||
it('429 with Retry-After honored (capped at RETRY_AFTER_CAP_MS)', async () => {
|
||||
let n = 0;
|
||||
const fetchImpl = vi.fn(async () => {
|
||||
n += 1;
|
||||
return n === 1 ? jsonResp(429, { 'Retry-After': '1' }) : jsonResp(204);
|
||||
});
|
||||
const sleep = vi.fn(async () => {});
|
||||
const { breaker } = makeBreaker();
|
||||
await resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep },
|
||||
});
|
||||
expect(sleep).toHaveBeenCalledWith(1000); // 1s
|
||||
});
|
||||
|
||||
it('429 with absurd Retry-After is capped to RETRY_AFTER_CAP_MS', async () => {
|
||||
let n = 0;
|
||||
const fetchImpl = vi.fn(async () => {
|
||||
n += 1;
|
||||
return n === 1 ? jsonResp(429, { 'Retry-After': '99999' }) : jsonResp(204);
|
||||
});
|
||||
const sleep = vi.fn(async () => {});
|
||||
const { breaker } = makeBreaker();
|
||||
await resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep, capDelayMs: 999_999 }, // ensure cap comes from RETRY_AFTER_CAP_MS, not retry config
|
||||
});
|
||||
expect(sleep).toHaveBeenCalledWith(RETRY_AFTER_CAP_MS);
|
||||
});
|
||||
|
||||
it('429 without Retry-After falls back to exponential-backoff delay', async () => {
|
||||
let n = 0;
|
||||
const fetchImpl = vi.fn(async () => {
|
||||
n += 1;
|
||||
return n === 1 ? jsonResp(429) : jsonResp(204);
|
||||
});
|
||||
const sleep = vi.fn(async () => {});
|
||||
const { breaker } = makeBreaker();
|
||||
await resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep, baseDelayMs: 100, capDelayMs: 1000, random: () => 0.5 },
|
||||
});
|
||||
// attempt 0: full-jitter upper = min(1000, 100*1) = 100; floor(0.5*100) = 50
|
||||
expect(sleep).toHaveBeenCalledWith(50);
|
||||
});
|
||||
|
||||
it('401 returned as Response, no retry, breaker not incremented', async () => {
|
||||
const fetchImpl = vi.fn(async () => jsonResp(401));
|
||||
const sleep = vi.fn(async () => {});
|
||||
const { breaker } = makeBreaker();
|
||||
const resp = await resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep },
|
||||
});
|
||||
expect(resp.status).toBe(401);
|
||||
expect(fetchImpl).toHaveBeenCalledTimes(1);
|
||||
expect(breaker.getConsecutiveFailures()).toBe(0);
|
||||
});
|
||||
|
||||
it('422 returned as Response, no retry', async () => {
|
||||
const fetchImpl = vi.fn(async () => jsonResp(422));
|
||||
const { breaker } = makeBreaker();
|
||||
const resp = await resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep: async () => {} },
|
||||
});
|
||||
expect(resp.status).toBe(422);
|
||||
expect(fetchImpl).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('TimeoutError rethrown immediately, no retry, breaker not incremented', async () => {
|
||||
const fetchImpl = vi.fn(async () => {
|
||||
throw new DOMException('aborted', 'TimeoutError');
|
||||
});
|
||||
const sleep = vi.fn(async () => {});
|
||||
const { breaker } = makeBreaker();
|
||||
await expect(
|
||||
resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep },
|
||||
}),
|
||||
).rejects.toThrow(DOMException);
|
||||
expect(fetchImpl).toHaveBeenCalledTimes(1);
|
||||
expect(sleep).not.toHaveBeenCalled();
|
||||
expect(breaker.getConsecutiveFailures()).toBe(0);
|
||||
});
|
||||
|
||||
it('three consecutive 503 throws ResilientFetchExhaustedError; breaker increments by 1', async () => {
|
||||
const fetchImpl = vi.fn(async () => jsonResp(503));
|
||||
const sleep = vi.fn(async () => {});
|
||||
const { breaker } = makeBreaker();
|
||||
await expect(
|
||||
resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep, maxAttempts: 3 },
|
||||
}),
|
||||
).rejects.toBeInstanceOf(ResilientFetchExhaustedError);
|
||||
expect(fetchImpl).toHaveBeenCalledTimes(3);
|
||||
expect(breaker.getConsecutiveFailures()).toBe(1);
|
||||
});
|
||||
|
||||
it('after three exhausted 503 batches, breaker opens and fails fast', async () => {
|
||||
const fetchImpl = vi.fn(async () => jsonResp(503));
|
||||
const { breaker } = makeBreaker({ failureThreshold: 3, cooldownMs: 60_000 });
|
||||
for (let i = 0; i < 3; i++) {
|
||||
await expect(
|
||||
resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep: async () => {}, maxAttempts: 3 },
|
||||
}),
|
||||
).rejects.toBeInstanceOf(ResilientFetchExhaustedError);
|
||||
}
|
||||
expect(breaker.getState()).toBe('open');
|
||||
// 4th call: breaker open, no fetch invoked.
|
||||
const fetchCallsBefore = fetchImpl.mock.calls.length;
|
||||
await expect(
|
||||
resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep: async () => {}, maxAttempts: 3 },
|
||||
}),
|
||||
).rejects.toBeInstanceOf(CircuitOpenError);
|
||||
expect(fetchImpl.mock.calls.length).toBe(fetchCallsBefore);
|
||||
});
|
||||
|
||||
it('retryable-network error retries, breaker counts only on exhaustion', async () => {
|
||||
const fetchImpl = vi.fn(async () => {
|
||||
throw new TypeError('fetch failed');
|
||||
});
|
||||
const { breaker } = makeBreaker();
|
||||
await expect(
|
||||
resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep: async () => {}, maxAttempts: 3 },
|
||||
}),
|
||||
).rejects.toBeInstanceOf(TypeError);
|
||||
expect(fetchImpl).toHaveBeenCalledTimes(3);
|
||||
expect(breaker.getConsecutiveFailures()).toBe(1);
|
||||
});
|
||||
|
||||
describe('U2: terminal outcomes route through recordNeutral', () => {
|
||||
it('401 does not erase prior partial-failure progress on the breaker', async () => {
|
||||
const { breaker } = makeBreaker();
|
||||
// Pre-seed the breaker with 2 failures (still closed; threshold 3).
|
||||
breaker.recordFailure();
|
||||
breaker.recordFailure();
|
||||
expect(breaker.getConsecutiveFailures()).toBe(2);
|
||||
|
||||
const fetchImpl = vi.fn(async () => jsonResp(401));
|
||||
await resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep: async () => {} },
|
||||
});
|
||||
|
||||
// Counter MUST stay at 2 — under the old behaviour recordSuccess
|
||||
// would have reset to 0 and the next 5xx batch would have started
|
||||
// from scratch instead of tipping over the threshold.
|
||||
expect(breaker.getConsecutiveFailures()).toBe(2);
|
||||
expect(breaker.getState()).toBe('closed');
|
||||
});
|
||||
|
||||
it('TimeoutError does not erase prior partial-failure progress', async () => {
|
||||
const { breaker } = makeBreaker();
|
||||
breaker.recordFailure();
|
||||
breaker.recordFailure();
|
||||
|
||||
const fetchImpl = vi.fn(async () => {
|
||||
throw new DOMException('aborted by timeout', 'TimeoutError');
|
||||
});
|
||||
await expect(
|
||||
resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep: async () => {} },
|
||||
}),
|
||||
).rejects.toBeInstanceOf(DOMException);
|
||||
|
||||
expect(breaker.getConsecutiveFailures()).toBe(2);
|
||||
});
|
||||
|
||||
it('external AbortError is terminal: no retry, breaker untouched', async () => {
|
||||
const { breaker } = makeBreaker();
|
||||
breaker.recordFailure();
|
||||
|
||||
const fetchImpl = vi.fn(async () => {
|
||||
throw new DOMException('aborted by caller', 'AbortError');
|
||||
});
|
||||
const sleep = vi.fn(async () => {});
|
||||
|
||||
await expect(
|
||||
resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep, maxAttempts: 3 },
|
||||
}),
|
||||
).rejects.toMatchObject({ name: 'AbortError' });
|
||||
|
||||
expect(fetchImpl).toHaveBeenCalledTimes(1);
|
||||
expect(sleep).not.toHaveBeenCalled();
|
||||
// Counter unchanged — neither incremented (no failure) nor reset
|
||||
// (no synthetic success).
|
||||
expect(breaker.getConsecutiveFailures()).toBe(1);
|
||||
});
|
||||
|
||||
it('interleaved 5xx + 401 + 5xx + 401 + 5xx opens breaker on third real failure', async () => {
|
||||
const { breaker } = makeBreaker({ failureThreshold: 3 });
|
||||
const sequence = [503, 401, 503, 401, 503];
|
||||
let i = 0;
|
||||
const fetchImpl = vi.fn(async () => jsonResp(sequence[i++]));
|
||||
|
||||
// Each call uses maxAttempts:1 so each surfaces a single response
|
||||
// (5xx → ResilientFetchExhaustedError; 4xx → returned Response).
|
||||
const driveOne = () =>
|
||||
resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep: async () => {}, maxAttempts: 1 },
|
||||
});
|
||||
|
||||
await expect(driveOne()).rejects.toBeInstanceOf(ResilientFetchExhaustedError); // 5xx fail #1
|
||||
await driveOne(); // 401 neutral
|
||||
await expect(driveOne()).rejects.toBeInstanceOf(ResilientFetchExhaustedError); // 5xx fail #2
|
||||
await driveOne(); // 401 neutral
|
||||
await expect(driveOne()).rejects.toBeInstanceOf(ResilientFetchExhaustedError); // 5xx fail #3 → opens
|
||||
|
||||
expect(breaker.getState()).toBe('open');
|
||||
expect(fetchImpl).toHaveBeenCalledTimes(5);
|
||||
});
|
||||
});
|
||||
|
||||
describe('half-open single-probe gating (U2)', () => {
|
||||
/** Test helper: a fetch mock whose Response is controlled by the test. */
|
||||
function deferredFetch(): {
|
||||
promise: Promise<Response>;
|
||||
resolve: (resp: Response) => void;
|
||||
reject: (err: unknown) => void;
|
||||
} {
|
||||
let resolve!: (resp: Response) => void;
|
||||
let reject!: (err: unknown) => void;
|
||||
const promise = new Promise<Response>((res, rej) => {
|
||||
resolve = res;
|
||||
reject = rej;
|
||||
});
|
||||
return { promise, resolve, reject };
|
||||
}
|
||||
|
||||
/** Builds a clock-injected breaker pre-opened with cooldown elapsed. */
|
||||
function preOpenedBreaker(opts: { cooldownMs: number; halfOpenRetryAfterMs?: number }): {
|
||||
breaker: CircuitBreaker;
|
||||
advance: (ms: number) => void;
|
||||
} {
|
||||
let t = 1_700_000_000_000;
|
||||
const breaker = new CircuitBreaker({
|
||||
failureThreshold: 1,
|
||||
cooldownMs: opts.cooldownMs,
|
||||
halfOpenRetryAfterMs: opts.halfOpenRetryAfterMs ?? 1_000,
|
||||
key: 'test',
|
||||
now: () => t,
|
||||
});
|
||||
breaker.recordFailure();
|
||||
t += opts.cooldownMs + 1; // cooldown elapsed
|
||||
return { breaker, advance: (ms) => (t += ms) };
|
||||
}
|
||||
|
||||
it('happy: 3 concurrent calls — exactly 1 hits fetch, others throw CircuitOpenError', async () => {
|
||||
const { breaker } = preOpenedBreaker({ cooldownMs: 10 });
|
||||
const deferred = deferredFetch();
|
||||
const fetchImpl = vi.fn(() => deferred.promise);
|
||||
|
||||
// Synchronous portion of each `resilientFetch` runs eagerly up to
|
||||
// the first await, so by the time r2/r3 are constructed the probe
|
||||
// permit is already consumed by r1 and they reject synchronously.
|
||||
const r1 = resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep: async () => {}, maxAttempts: 1 },
|
||||
});
|
||||
const r2 = resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep: async () => {}, maxAttempts: 1 },
|
||||
});
|
||||
const r3 = resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep: async () => {}, maxAttempts: 1 },
|
||||
});
|
||||
|
||||
// Resolve the probe with 200; r1 should now settle.
|
||||
deferred.resolve(new Response(null, { status: 200 }));
|
||||
|
||||
const results = await Promise.allSettled([r1, r2, r3]);
|
||||
|
||||
expect(results[0].status).toBe('fulfilled');
|
||||
if (results[0].status === 'fulfilled') {
|
||||
expect(results[0].value.status).toBe(200);
|
||||
}
|
||||
expect(results[1].status).toBe('rejected');
|
||||
if (results[1].status === 'rejected') {
|
||||
expect(results[1].reason).toBeInstanceOf(CircuitOpenError);
|
||||
}
|
||||
expect(results[2].status).toBe('rejected');
|
||||
if (results[2].status === 'rejected') {
|
||||
expect(results[2].reason).toBeInstanceOf(CircuitOpenError);
|
||||
}
|
||||
|
||||
// Only ONE underlying fetch was invoked.
|
||||
expect(fetchImpl).toHaveBeenCalledTimes(1);
|
||||
// Breaker closed after the probe's success.
|
||||
expect(breaker.getState()).toBe('closed');
|
||||
});
|
||||
|
||||
it('error: probe gets 503 — exhausted error; subsequent caller sees fresh full cooldown', async () => {
|
||||
const { breaker } = preOpenedBreaker({ cooldownMs: 10_000, halfOpenRetryAfterMs: 1_000 });
|
||||
const deferred = deferredFetch();
|
||||
const fetchImpl = vi.fn(() => deferred.promise);
|
||||
|
||||
const r1 = resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep: async () => {}, maxAttempts: 1 },
|
||||
});
|
||||
const r2 = resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep: async () => {}, maxAttempts: 1 },
|
||||
});
|
||||
const r3 = resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep: async () => {}, maxAttempts: 1 },
|
||||
});
|
||||
|
||||
// Probe fails with 503 → exhausted (maxAttempts: 1) → recordFailure → reopen.
|
||||
deferred.resolve(new Response(null, { status: 503 }));
|
||||
|
||||
const results = await Promise.allSettled([r1, r2, r3]);
|
||||
|
||||
expect(results[0].status).toBe('rejected');
|
||||
if (results[0].status === 'rejected') {
|
||||
expect(results[0].reason).toBeInstanceOf(ResilientFetchExhaustedError);
|
||||
}
|
||||
expect(results[1].status).toBe('rejected');
|
||||
if (results[1].status === 'rejected') {
|
||||
expect(results[1].reason).toBeInstanceOf(CircuitOpenError);
|
||||
// Blocked-while-half-open used the halfOpenRetryAfterMs default.
|
||||
expect((results[1].reason as CircuitOpenError).retryAfterMs).toBe(1_000);
|
||||
}
|
||||
|
||||
// Breaker has re-opened with a fresh openedAt.
|
||||
expect(breaker.getState()).toBe('open');
|
||||
|
||||
// r4: should see the fresh full cooldown, NOT the probe-in-flight 1000ms.
|
||||
let r4Caught: CircuitOpenError | null = null;
|
||||
try {
|
||||
await resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep: async () => {}, maxAttempts: 1 },
|
||||
});
|
||||
} catch (err) {
|
||||
r4Caught = err as CircuitOpenError;
|
||||
}
|
||||
expect(r4Caught).toBeInstanceOf(CircuitOpenError);
|
||||
expect(r4Caught?.retryAfterMs).toBe(10_000);
|
||||
});
|
||||
|
||||
it('cancellation: probe AbortError releases permit; next caller becomes new probe', async () => {
|
||||
const { breaker } = preOpenedBreaker({ cooldownMs: 10_000 });
|
||||
const deferred1 = deferredFetch();
|
||||
const deferred2 = deferredFetch();
|
||||
let callIdx = 0;
|
||||
const fetchImpl = vi.fn(() => (callIdx++ === 0 ? deferred1.promise : deferred2.promise));
|
||||
|
||||
// r1 admitted as the probe; r2 blocked while r1 still in flight.
|
||||
const r1 = resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep: async () => {}, maxAttempts: 1 },
|
||||
});
|
||||
const r2 = resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep: async () => {}, maxAttempts: 1 },
|
||||
});
|
||||
await expect(r2).rejects.toBeInstanceOf(CircuitOpenError);
|
||||
|
||||
// Cancel the probe — `AbortError` routes through terminal-network →
|
||||
// `recordNeutral` → permit released, state stays half-open.
|
||||
deferred1.reject(new DOMException('aborted by caller', 'AbortError'));
|
||||
await expect(r1).rejects.toMatchObject({ name: 'AbortError' });
|
||||
|
||||
expect(breaker.isProbeInFlight()).toBe(false);
|
||||
expect(breaker.getState()).toBe('half-open');
|
||||
|
||||
// r3: now succeeds and becomes the new probe.
|
||||
const r3 = resilientFetch(URL_STR, undefined, {
|
||||
fetchImpl: fetchImpl as unknown as typeof fetch,
|
||||
breaker,
|
||||
retry: { sleep: async () => {}, maxAttempts: 1 },
|
||||
});
|
||||
deferred2.resolve(new Response(null, { status: 200 }));
|
||||
const r3Resp = await r3;
|
||||
expect(r3Resp.status).toBe(200);
|
||||
expect(breaker.getState()).toBe('closed');
|
||||
// Two fetches total: the cancelled probe + the recovery probe.
|
||||
expect(fetchImpl).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
});
|
||||
});
|
||||
128
gitnexus/test/unit/integrations/retry.test.ts
Normal file
128
gitnexus/test/unit/integrations/retry.test.ts
Normal file
|
|
@ -0,0 +1,128 @@
|
|||
import { describe, it, expect, vi } from 'vitest';
|
||||
import { computeBackoffMs, withRetry, type RetryOptions } from 'gitnexus-shared';
|
||||
|
||||
describe('computeBackoffMs', () => {
|
||||
it('returns afterMs (capped) when caller supplies it', () => {
|
||||
expect(computeBackoffMs(0, 500, 5000, 1500, () => 0.5)).toBe(1500);
|
||||
expect(computeBackoffMs(0, 500, 5000, 99_999, () => 0.5)).toBe(5000);
|
||||
expect(computeBackoffMs(0, 500, 5000, 0, () => 0.5)).toBe(0);
|
||||
expect(computeBackoffMs(0, 500, 5000, -1, () => 0.5)).toBe(0);
|
||||
});
|
||||
|
||||
it('full-jitter delay falls within [0, min(cap, base * 2^attempt)]', () => {
|
||||
// attempt 0: upper = min(5000, 500 * 1) = 500
|
||||
expect(computeBackoffMs(0, 500, 5000, undefined, () => 0)).toBe(0);
|
||||
expect(computeBackoffMs(0, 500, 5000, undefined, () => 0.999)).toBeLessThan(500);
|
||||
// attempt 1: upper = min(5000, 500 * 2) = 1000
|
||||
expect(computeBackoffMs(1, 500, 5000, undefined, () => 0.5)).toBe(500);
|
||||
// attempt 4: 500 * 16 = 8000, capped at 5000
|
||||
expect(computeBackoffMs(4, 500, 5000, undefined, () => 0.5)).toBe(2500);
|
||||
expect(computeBackoffMs(4, 500, 5000, undefined, () => 0.999)).toBeLessThan(5000);
|
||||
});
|
||||
});
|
||||
|
||||
describe('withRetry', () => {
|
||||
function makeOpts(overrides: Partial<RetryOptions> = {}): RetryOptions {
|
||||
return {
|
||||
maxAttempts: 3,
|
||||
baseDelayMs: 10,
|
||||
capDelayMs: 100,
|
||||
isRetryable: () => ({ retry: true }),
|
||||
sleep: vi.fn(async () => {}),
|
||||
random: () => 0.5,
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
it('returns immediately when fn succeeds first try', async () => {
|
||||
const sleep = vi.fn(async () => {});
|
||||
const fn = vi.fn(async () => 'ok');
|
||||
const result = await withRetry(fn, makeOpts({ sleep }));
|
||||
expect(result).toBe('ok');
|
||||
expect(fn).toHaveBeenCalledTimes(1);
|
||||
expect(sleep).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('retries when isRetryable returns retry:true and second call succeeds', async () => {
|
||||
const sleep = vi.fn(async () => {});
|
||||
let calls = 0;
|
||||
const fn = async () => {
|
||||
calls += 1;
|
||||
if (calls === 1) throw new Error('boom');
|
||||
return 'ok';
|
||||
};
|
||||
const result = await withRetry(fn, makeOpts({ sleep }));
|
||||
expect(result).toBe('ok');
|
||||
expect(calls).toBe(2);
|
||||
expect(sleep).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('honors afterMs returned by isRetryable', async () => {
|
||||
const sleep = vi.fn(async () => {});
|
||||
let calls = 0;
|
||||
const fn = async () => {
|
||||
calls += 1;
|
||||
if (calls === 1) throw new Error('throttle');
|
||||
return 'ok';
|
||||
};
|
||||
await withRetry(
|
||||
fn,
|
||||
makeOpts({
|
||||
sleep,
|
||||
isRetryable: () => ({ retry: true, afterMs: 1500 }),
|
||||
capDelayMs: 5000,
|
||||
}),
|
||||
);
|
||||
expect(sleep).toHaveBeenCalledWith(1500);
|
||||
});
|
||||
|
||||
it('caps afterMs at capDelayMs', async () => {
|
||||
const sleep = vi.fn(async () => {});
|
||||
let calls = 0;
|
||||
const fn = async () => {
|
||||
calls += 1;
|
||||
if (calls === 1) throw new Error('throttle');
|
||||
return 'ok';
|
||||
};
|
||||
await withRetry(
|
||||
fn,
|
||||
makeOpts({
|
||||
sleep,
|
||||
isRetryable: () => ({ retry: true, afterMs: 10_000 }),
|
||||
capDelayMs: 3000,
|
||||
}),
|
||||
);
|
||||
expect(sleep).toHaveBeenCalledWith(3000);
|
||||
});
|
||||
|
||||
it('rethrows immediately when isRetryable returns retry:false', async () => {
|
||||
const sleep = vi.fn(async () => {});
|
||||
const fn = vi.fn(async () => {
|
||||
throw new Error('terminal');
|
||||
});
|
||||
await expect(
|
||||
withRetry(fn, makeOpts({ sleep, isRetryable: () => ({ retry: false }) })),
|
||||
).rejects.toThrow('terminal');
|
||||
expect(fn).toHaveBeenCalledTimes(1);
|
||||
expect(sleep).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('throws the last error when maxAttempts exhausted', async () => {
|
||||
const sleep = vi.fn(async () => {});
|
||||
let calls = 0;
|
||||
const fn = async () => {
|
||||
calls += 1;
|
||||
throw new Error(`boom-${calls}`);
|
||||
};
|
||||
await expect(withRetry(fn, makeOpts({ sleep, maxAttempts: 3 }))).rejects.toThrow('boom-3');
|
||||
expect(calls).toBe(3);
|
||||
// 3 attempts → 2 sleeps between them; final attempt does not sleep.
|
||||
expect(sleep).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it('rejects maxAttempts < 1', async () => {
|
||||
await expect(withRetry(async () => 'ok', makeOpts({ maxAttempts: 0 }))).rejects.toThrow(
|
||||
/maxAttempts must be >= 1/,
|
||||
);
|
||||
});
|
||||
});
|
||||
Loading…
Add table
Reference in a new issue