diff --git a/review-enrichment/src/analyzer-circuit-breaker.ts b/review-enrichment/src/analyzer-circuit-breaker.ts new file mode 100644 index 0000000000..faff772f57 --- /dev/null +++ b/review-enrichment/src/analyzer-circuit-breaker.ts @@ -0,0 +1,83 @@ +// Per-analyzer circuit breaker (#2541). Analyzers that depend on a third-party HTTP API (registry lookups, +// GitHub API calls, endoflife.date, etc) have no memory of recent failures by default -- every incoming +// enrichment request re-attempts a currently-unhealthy dependency from a cold state, even seconds after an +// identical call just timed out or errored. Trip a short, in-process cooldown after a run of CONSECUTIVE +// thrown failures (a timeout counts -- runWithTimeout's rejection is a thrown failure) and skip that analyzer +// entirely -- no network/CLI call at all -- for the cooldown window, falling through the SAME plan.skipped +// path any other skip reason already uses. In-process only (no persistence layer): review-enrichment is a +// single long-lived process (Railway), matching the main app's equivalent per-provider AI circuit breaker +// (src/selfhost/ai.ts's createChainAi). +// +// Half-open probing: review-enrichment serves CONCURRENT requests (one per in-flight PR review), so once the +// cooldown expires, a burst of near-simultaneous requests would all see the circuit as "not open" and all +// retry the same still-unhealthy dependency at once. `isAnalyzerCircuitOpen` claims a single probe slot the +// first time it observes an expired cooldown; every other caller sees the circuit as still open until that +// one probe resolves (recordAnalyzerCircuitSuccess/Failure). `releaseAnalyzerCircuitProbe` frees a claimed +// slot without recording an outcome, for the case where the analyzer never actually ran (budget/timeout +// capped in brief.ts before reaching the real call) -- otherwise a stuck claim would block re-probing forever. +import type { AnalyzerName } from "./analyzers/types.js"; + +const ANALYZER_CIRCUIT_FAILURE_STREAK = 3; +const ANALYZER_CIRCUIT_COOLDOWN_MS = 5 * 60_000; + +interface AnalyzerCircuitState { + consecutiveFailures: number; + cooldownUntilMs: number; + probeClaimed: boolean; +} + +const analyzerCircuits = new Map(); + +/** True while `name`'s breaker should skip the caller: either still within the full cooldown window, or past + * it but another caller already claimed this cycle's single half-open probe. The FIRST caller to observe an + * expired cooldown claims the probe as a side effect and gets `false` (proceed) -- this is the one function + * planning calls to decide runnable vs skipped, so the claim has to happen here, not at execution time. + * + * `cooldownUntilMs === 0` means the circuit has NEVER actually tripped (below the streak threshold) -- + * recordAnalyzerCircuitFailure only sets a non-zero cooldownUntilMs once consecutiveFailures reaches the + * threshold, so this is a reliable "never opened" check. Without it, a caller after just 1-2 failures would + * claim the half-open probe slot too, spuriously skipping a concurrent second caller as circuit_open even + * though the breaker was never actually open. */ +export function isAnalyzerCircuitOpen(name: AnalyzerName, nowMs = Date.now()): boolean { + const state = analyzerCircuits.get(name); + if (state === undefined || state.cooldownUntilMs === 0) return false; + if (state.cooldownUntilMs > nowMs) return true; + if (state.probeClaimed) return true; + state.probeClaimed = true; + return false; +} + +/** A completed run (whether a clean "ok" or a non-throwing "degraded"/"capped" partial result) resets the + * streak -- the dependency responded, so it is not the failure mode this breaker guards against. */ +export function recordAnalyzerCircuitSuccess(name: AnalyzerName): void { + analyzerCircuits.delete(name); +} + +/** A THROWN failure (including the analyzer_timeout rejection) is the signal this breaker tracks. Trips the + * cooldown once the consecutive count reaches the streak threshold; stays open (extends nothing further -- + * the analyzer is simply skipped while open, so no additional failures accrue until it is tried again). A + * half-open probe's failure re-extends the cooldown via this same threshold check, since consecutiveFailures + * is already at/above it by the time a probe can be claimed -- no separate re-trip path needed. */ +export function recordAnalyzerCircuitFailure(name: AnalyzerName, nowMs = Date.now()): void { + const state = analyzerCircuits.get(name) ?? { consecutiveFailures: 0, cooldownUntilMs: 0, probeClaimed: false }; + state.consecutiveFailures += 1; + state.probeClaimed = false; + if (state.consecutiveFailures >= ANALYZER_CIRCUIT_FAILURE_STREAK) { + state.cooldownUntilMs = nowMs + ANALYZER_CIRCUIT_COOLDOWN_MS; + } + analyzerCircuits.set(name, state); +} + +/** Frees a claimed half-open probe WITHOUT recording success or failure -- for when the probing attempt never + * actually reached the analyzer call (capped by budget/timeout in brief.ts first). Safe no-op when `name` has + * no circuit state or no claimed probe, so callers can call this unconditionally on every capped early-return + * without needing to know whether this particular call was the one that claimed the probe. */ +export function releaseAnalyzerCircuitProbe(name: AnalyzerName): void { + const state = analyzerCircuits.get(name); + if (state !== undefined) state.probeClaimed = false; +} + +/** Test-only reset so circuit-breaker state from one test can't leak into the next (module-level Map). */ +export function resetAnalyzerCircuitsForTest(): void { + analyzerCircuits.clear(); +} diff --git a/review-enrichment/src/brief.ts b/review-enrichment/src/brief.ts index 0982d4f52d..3543f3c0ee 100644 --- a/review-enrichment/src/brief.ts +++ b/review-enrichment/src/brief.ts @@ -14,6 +14,11 @@ import type { AnalyzerRunContext, AnalyzerCostClass, } from "./analyzers/types.js"; +import { + recordAnalyzerCircuitFailure, + recordAnalyzerCircuitSuccess, + releaseAnalyzerCircuitProbe, +} from "./analyzer-circuit-breaker.js"; import { createAnalysisContext, type AnalysisContext, @@ -258,6 +263,10 @@ export async function buildBrief( }; partial = true; analysis.metrics.recordCappedWork("analyzer_budget", 1); + // #2541: budget exhaustion, not a dependency-health signal -- if this call had claimed the circuit + // breaker's half-open probe (isAnalyzerCircuitOpen), free it so a later request can still probe rather + // than leaving the slot claimed forever with no outcome ever recorded. + releaseAnalyzerCircuitProbe(name); return; } const timeoutMs = analyzerTimeoutMs( @@ -279,6 +288,8 @@ export async function buildBrief( }; partial = true; analysis.metrics.recordCappedWork(`analyzer_${item.descriptor.cost}`, 1); + // #2541: same as above -- release a claimed half-open probe without recording an outcome. + releaseAnalyzerCircuitProbe(name); return; } try { @@ -296,6 +307,11 @@ export async function buildBrief( }, ); findings[name] = result as never; + // #2541: the analyzer completed WITHOUT throwing -- whether "ok" or a non-throwing partial/degraded + // result (its own internal cap, not a dependency failure) -- so the dependency responded. Reset the + // circuit rather than only resetting on a clean "ok"; a benign internal partial must not itself count + // toward tripping the breaker. + recordAnalyzerCircuitSuccess(name); if (resultIsPartial(result) || diagnostics.partialStatus === "partial") { const status = statusFromDiagnostics(diagnostics, "degraded"); const partialReason = publicPartialReason( @@ -342,6 +358,10 @@ export async function buildBrief( }; } } catch (error) { + // #2541: a THROWN failure (including the analyzer_timeout rejection from runWithTimeout) is the signal + // the circuit breaker tracks -- the dependency did not respond at all, unlike a non-throwing partial + // result above. + recordAnalyzerCircuitFailure(name); const status = timeoutStatus(error, diagnostics); const partialReason = publicPartialReason(diagnostics.partialReason, "analyzer_error"); analyzerStatus[name] = status; @@ -374,6 +394,18 @@ export async function buildBrief( } } + // #2541 (cost-class parallelization evaluated, NOT implemented): cost classes run strictly sequentially -- + // this loop awaits each class's bounded worker pool (runWithConcurrency) before starting the next -- with + // cheaper/more-certain classes (local, then registry) always draining before expensive/less-essential ones + // (github-heavy, tooling). This is deliberate prioritization, not an oversight: it guarantees the cheap, + // always-safe signals are collected first, and a shrinking remainingMs budget (see analyzerTimeoutMs above) + // correctly starves LATER, less-essential classes first when time runs short -- never the reverse. Running + // every class's worker pool concurrently would sum EVERY class's concurrency limit at once (8+3+2+1+1 = 15 + // simultaneous external calls on the "deep" profile instead of at most 8), spiking the third-party burst + // rate exactly when third-party health is already the concern this issue is about, and would let an + // expensive/uncertain "tooling" call start competing for budget with a cheap "local" one instead of only + // running once local has had its turn. That risk is not "low", so this stays sequential; the per-analyzer + // circuit breaker above is the intended fix for a specific unhealthy dependency, not a scheduling change. for (const cost of COST_ORDER) { const items = plan.runnable.filter((item) => item.descriptor.cost === cost); if (!items.length) continue; diff --git a/review-enrichment/src/scheduler.ts b/review-enrichment/src/scheduler.ts index 401d777cc3..aa7a9d8238 100644 --- a/review-enrichment/src/scheduler.ts +++ b/review-enrichment/src/scheduler.ts @@ -1,4 +1,5 @@ import type { AnalysisContext } from "./analysis-context.js"; +import { isAnalyzerCircuitOpen } from "./analyzer-circuit-breaker.js"; import { ANALYZER_NAMES, getAnalyzerDescriptor, @@ -321,7 +322,19 @@ function skipReasonForAnalyzer( return "missing_github_token"; } - return inputSkipReason(descriptor.name, analysis, req); + const inputSkip = inputSkipReason(descriptor.name, analysis, req); + if (inputSkip) return inputSkip; + + // #2541: checked LAST, only once every other skip reason has cleared -- isAnalyzerCircuitOpen claims a + // single half-open probe as a side effect when the cooldown has expired, and that claim is only ever + // released inside runAnalyzer (brief.ts), which never runs for a plan.skipped item. Checking this any + // earlier could claim the probe for an analyzer that's about to be skipped for an UNRELATED reason (a + // missing head SHA, no dependency manifest, etc.), leaking the claim forever with no outcome ever recorded. + // An EXPLICIT request (req.analyzers) does not bypass this -- the circuit is about the dependency being + // down right now, which an explicit request can't fix. + if (isAnalyzerCircuitOpen(descriptor.name)) return "circuit_open"; + + return null; } function inputSkipReason( diff --git a/review-enrichment/test/analyzer-circuit-breaker.test.ts b/review-enrichment/test/analyzer-circuit-breaker.test.ts new file mode 100644 index 0000000000..544cb4bdee --- /dev/null +++ b/review-enrichment/test/analyzer-circuit-breaker.test.ts @@ -0,0 +1,309 @@ +import assert from "node:assert/strict"; +import test, { afterEach } from "node:test"; + +import { buildBrief } from "../dist/brief.js"; +import { + isAnalyzerCircuitOpen, + recordAnalyzerCircuitFailure, + recordAnalyzerCircuitSuccess, + releaseAnalyzerCircuitProbe, + resetAnalyzerCircuitsForTest, +} from "../dist/analyzer-circuit-breaker.js"; + +afterEach(() => { + resetAnalyzerCircuitsForTest(); +}); + +const baseReq = { + repoFullName: "JSONbored/gittensory", + prNumber: 1811, + analyzers: ["history"], + githubToken: "token", + author: "jsonbored", + headSha: "abcdef1234567890", + files: [{ path: "src/a.ts", patch: "@@ -1,0 +1,1 @@\n+export const a = 1;" }], + budget: { timeoutMs: 2000 }, +}; + +test("does not open the circuit before the failure streak threshold — every request still calls the analyzer", async () => { + let calls = 0; + const failing = { history: async () => { calls += 1; throw new Error("boom"); } }; + for (let i = 0; i < 2; i += 1) { + const brief = await buildBrief(baseReq, failing); + assert.equal(brief.analyzerStatus.history, "degraded"); + } + assert.equal(calls, 2); + assert.equal(isAnalyzerCircuitOpen("history"), false); +}); + +test("opens the circuit after 3 consecutive failures and SKIPS the analyzer on the next request — zero calls to the broken dependency", async () => { + let calls = 0; + const failing = { history: async () => { calls += 1; throw new Error("boom"); } }; + for (let i = 0; i < 3; i += 1) { + await buildBrief(baseReq, failing); + } + assert.equal(calls, 3); + assert.equal(isAnalyzerCircuitOpen("history"), true); + + const brief = await buildBrief(baseReq, failing); + + assert.equal(calls, 3); // UNCHANGED — the 4th "attempt" never happened, it was skipped at planning time + assert.equal(brief.analyzerStatus.history, "skipped"); + assert.equal(brief.telemetry.analyzers.history.skipReason, "circuit_open"); +}); + +test("a timeout counts as a circuit-breaker failure, same as a thrown error", async () => { + // 300ms matches scheduler.test.ts's own proven-stable timeout budget: tight enough to time out reliably, + // but not so tight it races into "capped" (the reserved-response-budget pre-check) instead of "timeout". + const hanging = { history: async () => new Promise(() => undefined) }; + const timeoutReq = { ...baseReq, budget: { timeoutMs: 300 } }; + for (let i = 0; i < 3; i += 1) { + const brief = await buildBrief(timeoutReq, hanging); + assert.equal(brief.analyzerStatus.history, "timeout"); + } + assert.equal(isAnalyzerCircuitOpen("history"), true); +}); + +test("a non-throwing DEGRADED/partial result does NOT count as a circuit-breaker failure (the dependency responded)", async () => { + // resultIsPartial (brief.ts) checks per-entry `.partial === true`, matching the real analyzer-result shape. + // Uses "secret" (a flat SecretFinding[] result, unlike history's nested similarPastPrs render requirement). + const secretReq = { ...baseReq, analyzers: ["secret"] }; + const partiallyOk = { secret: async () => [{ file: "a.ts", line: 1, kind: "test", confidence: "high", partial: true }] }; + for (let i = 0; i < 5; i += 1) { + const brief = await buildBrief(secretReq, partiallyOk); + assert.equal(brief.analyzerStatus.secret, "degraded"); + assert.notEqual(brief.analyzerStatus.secret, "skipped"); + } + assert.equal(isAnalyzerCircuitOpen("secret"), false); +}); + +test("a success resets the streak so it does not carry over into a LATER, separate run of failures", async () => { + recordAnalyzerCircuitFailure("history"); + recordAnalyzerCircuitFailure("history"); + recordAnalyzerCircuitSuccess("history"); + let calls = 0; + const failing = { history: async () => { calls += 1; throw new Error("boom"); } }; + // Two MORE failures after the reset — still below the streak threshold on their own. + await buildBrief(baseReq, failing); + await buildBrief(baseReq, failing); + assert.equal(calls, 2); + assert.equal(isAnalyzerCircuitOpen("history"), false); +}); + +test("REGRESSION: a circuit-expired analyzer is tried again rather than staying open forever", async () => { + const realNow = Date.now(); + let fakeNow = realNow; + const originalNow = Date.now; + try { + Date.now = () => fakeNow; + recordAnalyzerCircuitFailure("history"); + recordAnalyzerCircuitFailure("history"); + recordAnalyzerCircuitFailure("history"); + assert.equal(isAnalyzerCircuitOpen("history"), true); + + fakeNow = realNow + 5 * 60_000 + 1; // past the cooldown window + + assert.equal(isAnalyzerCircuitOpen("history"), false); + } finally { + Date.now = originalNow; + } +}); + +test("recordAnalyzerCircuitSuccess on an analyzer with no prior failures is a safe no-op", () => { + assert.doesNotThrow(() => recordAnalyzerCircuitSuccess("secret")); + assert.equal(isAnalyzerCircuitOpen("secret"), false); +}); + +test("an EXPLICITLY requested analyzer (req.analyzers) is still skipped while its circuit is open — the explicit request can't fix a down dependency", async () => { + let calls = 0; + const failing = { history: async () => { calls += 1; throw new Error("boom"); } }; + for (let i = 0; i < 3; i += 1) { + await buildBrief(baseReq, failing); + } + assert.equal(calls, 3); + + const explicitReq = { ...baseReq, analyzers: ["history"] }; + const brief = await buildBrief(explicitReq, failing); + + assert.equal(calls, 3); + assert.equal(brief.analyzerStatus.history, "skipped"); +}); + +// Half-open probing (#2624 review follow-up): once the cooldown expires, only ONE caller should get to +// re-try the analyzer at a time — a burst of concurrent requests must not all hit the same still-unhealthy +// dependency simultaneously just because the cooldown clock happened to expire. +test("REGRESSION: below the failure-streak threshold, isAnalyzerCircuitOpen never claims a probe — a second concurrent caller is NOT skipped as circuit_open", async () => { + // Before the fix, isAnalyzerCircuitOpen claimed probeClaimed for ANY existing state (cooldownUntilMs <= + // nowMs is true even at cooldownUntilMs === 0, i.e. never-tripped), so a circuit with only 1-2 recorded + // failures would spuriously block a second concurrent caller — even though the breaker never actually opened. + recordAnalyzerCircuitFailure("history"); + recordAnalyzerCircuitFailure("history"); // 2 failures — still below the 3-failure trip threshold + + assert.equal(isAnalyzerCircuitOpen("history"), false); // first caller — not open + assert.equal(isAnalyzerCircuitOpen("history"), false); // second, concurrent caller — also not open + + let calls = 0; + const failing = { history: async () => { calls += 1; throw new Error("boom"); } }; + const [first, second] = await Promise.all([buildBrief(baseReq, failing), buildBrief(baseReq, failing)]); + + assert.equal(calls, 2); // both concurrent calls actually invoked the analyzer + assert.notEqual(first.analyzerStatus.history, "skipped"); + assert.notEqual(second.analyzerStatus.history, "skipped"); +}); + +test("half-open: only the FIRST caller after cooldown expiry gets to probe — a second caller in the same instant is still blocked", async () => { + const realNow = Date.now(); + let fakeNow = realNow; + const originalNow = Date.now; + try { + Date.now = () => fakeNow; + recordAnalyzerCircuitFailure("history"); + recordAnalyzerCircuitFailure("history"); + recordAnalyzerCircuitFailure("history"); + fakeNow = realNow + 5 * 60_000 + 1; // past the cooldown window + + assert.equal(isAnalyzerCircuitOpen("history"), false); // first caller claims the probe + assert.equal(isAnalyzerCircuitOpen("history"), true); // second caller, same instant — still blocked + } finally { + Date.now = originalNow; + } +}); + +test("half-open: a successful probe fully closes the circuit — a later caller is not treated as another probe", async () => { + const realNow = Date.now(); + let fakeNow = realNow; + const originalNow = Date.now; + try { + Date.now = () => fakeNow; + recordAnalyzerCircuitFailure("history"); + recordAnalyzerCircuitFailure("history"); + recordAnalyzerCircuitFailure("history"); + fakeNow = realNow + 5 * 60_000 + 1; + + assert.equal(isAnalyzerCircuitOpen("history"), false); // claims the probe + recordAnalyzerCircuitSuccess("history"); + + assert.equal(isAnalyzerCircuitOpen("history"), false); // fully closed, not "another probe" + } finally { + Date.now = originalNow; + } +}); + +test("half-open: a failed probe re-extends the cooldown and immediately blocks new callers again", async () => { + const realNow = Date.now(); + let fakeNow = realNow; + const originalNow = Date.now; + try { + Date.now = () => fakeNow; + recordAnalyzerCircuitFailure("history"); + recordAnalyzerCircuitFailure("history"); + recordAnalyzerCircuitFailure("history"); + fakeNow = realNow + 5 * 60_000 + 1; + + assert.equal(isAnalyzerCircuitOpen("history"), false); // claims the probe + recordAnalyzerCircuitFailure("history", fakeNow); + + assert.equal(isAnalyzerCircuitOpen("history"), true); // re-tripped, new cooldown active + } finally { + Date.now = originalNow; + } +}); + +test("releaseAnalyzerCircuitProbe frees a claimed slot without recording an outcome, so a later caller can still probe immediately", async () => { + const realNow = Date.now(); + let fakeNow = realNow; + const originalNow = Date.now; + try { + Date.now = () => fakeNow; + recordAnalyzerCircuitFailure("history"); + recordAnalyzerCircuitFailure("history"); + recordAnalyzerCircuitFailure("history"); + fakeNow = realNow + 5 * 60_000 + 1; + + assert.equal(isAnalyzerCircuitOpen("history"), false); // claims the probe + assert.equal(isAnalyzerCircuitOpen("history"), true); // second caller blocked + + releaseAnalyzerCircuitProbe("history"); // e.g. the probing analyzer never ran (budget-capped) + + assert.equal(isAnalyzerCircuitOpen("history"), false); // released — a fresh probe can be claimed + } finally { + Date.now = originalNow; + } +}); + +test("releaseAnalyzerCircuitProbe on an analyzer with no circuit state, or no claimed probe, is a safe no-op", () => { + assert.doesNotThrow(() => releaseAnalyzerCircuitProbe("secret")); + recordAnalyzerCircuitFailure("secret"); + assert.doesNotThrow(() => releaseAnalyzerCircuitProbe("secret")); // tripped but cooling down, no probe claimed + assert.equal(isAnalyzerCircuitOpen("secret"), false); // below the streak threshold — unaffected either way +}); + +test("end-to-end: two concurrent buildBrief calls right after cooldown expiry — only the FIRST invokes the analyzer, the second is skipped as circuit_open", async () => { + const realNow = Date.now(); + let fakeNow = realNow; + const originalNow = Date.now; + try { + Date.now = () => fakeNow; + recordAnalyzerCircuitFailure("secret"); + recordAnalyzerCircuitFailure("secret"); + recordAnalyzerCircuitFailure("secret"); + fakeNow = realNow + 5 * 60_000 + 1; + + let calls = 0; + const secretReq = { ...baseReq, analyzers: ["secret"] }; + const ok = { secret: async () => { calls += 1; return []; } }; + // Async functions run synchronously up to their first `await`, so both buildBrief() calls' planning + // phases (fully synchronous, including isAnalyzerCircuitOpen) resolve in call order BEFORE either + // promise is awaited — this deterministically reproduces the "burst right after cooldown expiry" race. + const [first, second] = await Promise.all([buildBrief(secretReq, ok), buildBrief(secretReq, ok)]); + + assert.equal(calls, 1); + assert.notEqual(first.analyzerStatus.secret, "skipped"); + assert.equal(second.analyzerStatus.secret, "skipped"); + assert.equal(second.telemetry.analyzers.secret.skipReason, "circuit_open"); + } finally { + Date.now = originalNow; + } +}); + +test("REGRESSION: a half-open probe claim is not leaked when the SAME planning pass skips the analyzer for an UNRELATED reason", async () => { + // Before the fix, isAnalyzerCircuitOpen was checked FIRST in skipReasonForAnalyzer, so it could claim the + // half-open probe even for a request that's about to be skipped for a totally unrelated reason (e.g. no + // added lines for "secret"). Since a plan.skipped item never reaches runAnalyzer -- the only place a + // claimed probe is released -- that claim would leak forever, permanently blocking every later request + // from ever probing the analyzer again even once it's actually healthy. + const realNow = Date.now(); + let fakeNow = realNow; + const originalNow = Date.now; + try { + Date.now = () => fakeNow; + recordAnalyzerCircuitFailure("secret"); + recordAnalyzerCircuitFailure("secret"); + recordAnalyzerCircuitFailure("secret"); + fakeNow = realNow + 5 * 60_000 + 1; // past the cooldown window + + // A pure deletion (no `+` line) — "secret" requires added lines, so this is skipped as "no_added_lines", + // unrelated to the circuit breaker. + const noAddedLinesReq = { + ...baseReq, + analyzers: ["secret"], + files: [{ path: "src/a.ts", patch: "@@ -1,1 +1,0 @@\n-export const a = 1;" }], + }; + const noop = { secret: async () => [] }; + const unrelatedSkip = await buildBrief(noAddedLinesReq, noop); + assert.equal(unrelatedSkip.analyzerStatus.secret, "skipped"); + assert.equal(unrelatedSkip.telemetry.analyzers.secret.skipReason, "no_added_lines"); + + // A LATER, normal request must still be able to claim a fresh probe — not spuriously blocked as + // circuit_open by a claim the unrelated skip above should never have made in the first place. + let calls = 0; + const secretReq = { ...baseReq, analyzers: ["secret"] }; + const ok = { secret: async () => { calls += 1; return []; } }; + const probe = await buildBrief(secretReq, ok); + + assert.equal(calls, 1); + assert.notEqual(probe.analyzerStatus.secret, "skipped"); + } finally { + Date.now = originalNow; + } +});