From cea7db086c3c9535b10c3299cc9d531cd9851f75 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Thu, 20 Aug 2026 16:19:23 +0200 Subject: [PATCH 1/5] fix(orchestrator): respect the advertised Retry-After and stop discarding a sweep on one 429 (#297) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit relayfile answers an overloaded workspace DO with a 429 in MILLISECONDS carrying `Retry-After: 5`. Factory answered by sleeping up to 300 seconds, latching on the first 429 raised anywhere in the sweep, discarding the entire sweep including work that had already succeeded, and clearing the ratchet only after a fully clean sweep. Root-caused during the 2026-08-20 cloud outage, where `/healthz` still read degraded seven minutes after the upstream fix went live purely because of this. Respect the advertised delay. `discoveryOverloadBackoffMs` treated `Retry-After` as a floor under an independent 5/10/20/40/80/160/300s ladder. It is now authoritative in both directions: the first rung, so we never retry sooner than the dependency allows, and the ceiling (DISCOVERY_OVERLOAD_ADVERTISED_BACKOFF_MAX_MS, or the advertised value itself when that is larger), so we never sleep for minutes over a request to wait seconds. With no advertised delay there is nothing to respect and the original five-minute ladder governs unchanged. `Retry-After: 0` is floored, or `0 * 2**n` would pin the ladder at zero. Stop discarding a whole sweep for one latched 429. This is the same shape #292/#293 fixed for dispatch errors, and the per-item catch is inverted the same way rather than by inventing a second pattern: a 429 on one work unit — its ready-issue read or its dispatch — skips that unit and the sweep continues. `#isPassFatalFailure` no longer treats the sweep-wide latch as fatal; what is fatal now is DISCOVERY_OVERLOAD_PER_SWEEP_LIMIT shed operations in one sweep, so skipping cannot degenerate into grinding a shedding dependency through a whole backlog one 429 at a time. A sweep that was shed and got NO unit through still fails: there is no progress to preserve, and committing it would leave readinessReconcile green over a dependency that served none of it. Decay the ratchet on partial progress. `consecutiveOverloads` reset only in completeDiscoverySweep, so requiring a perfect sweep meant the ratchet rarely cleared under sustained mild load. The signal is now whether relayfile served any of the sweep's work units — a shedding DO rejects all background traffic, so a genuinely overloaded workspace still escalates to the cap, while a sweep that got units through decays one rung. Decay, not reset: surviving shedding is not evidence the overload is over. `completeDiscoverySweep` takes an optional residual so a committed-but-shed sweep keeps a decayed ratchet and a backoff. Log the reason, not just the message. relayfile's four reason codes (inflight_limit, oldest_inflight_age, router_inflight_limit, durable_object_overloaded) share ONE message string and mean four different things; `relayfileOverload()` always parsed the reason and nothing printed it. Every shed operation now warns with its reason, per-reason counters are exposed on status(), and readinessReconcile.lastError carries it — during the incident that ambiguity was the single biggest obstacle to diagnosis. The reason is allowlisted where it reaches stdout or a counter key, the same public-surface rule #293 applied to error class names. Tests: five must-fire regressions, each verified failing first against unmodified source, plus three must-not-fire controls — the five-minute ladder still governs without an advertised Retry-After, repeated shedding still escalates the durable ratchet from the decayed counter, and a sweep relayfile shed entirely still fails. Co-Authored-By: Claude Opus 5 Session-Id: 964fa2f4-17fc-4afc-93e3-0f7ee72f1316 --- src/orchestrator/factory.test.ts | 372 +++++++++++++++++++++++++++++ src/orchestrator/factory.ts | 352 ++++++++++++++++++++++++--- src/ports/state.ts | 6 + src/state/file-state-store.ts | 7 +- src/state/in-memory-state-store.ts | 7 +- 5 files changed, 710 insertions(+), 34 deletions(-) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 2ebc419..7986b3f 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -4096,6 +4096,378 @@ describe('FactoryLoop', () => { })) }) + // #297: relayfile answers an overloaded workspace DO with a 429 in + // milliseconds carrying `Retry-After: 5`. Factory answered by sleeping up to + // 300 SECONDS, latching on the first 429 raised anywhere in the sweep, and + // discarding everything the sweep had already accomplished. Root-caused + // during the 2026-08-20 cloud outage. + describe('Relayfile discovery overload (#297)', () => { + // All four relayfile reason codes share this one message string, which is + // why the reason has to be logged separately — during the incident that + // ambiguity was the single biggest obstacle to diagnosis. + const OVERLOAD_MESSAGE = 'workspace durable object is busy; retry after the advertised delay' + + const overloadError = ( + opts: { retryAfterSeconds?: number; reason?: string } = {}, + ): Error => + Object.assign(new Error(OVERLOAD_MESSAGE), { + status: 429, + details: { + reason: opts.reason ?? 'oldest_inflight_age', + ...(opts.retryAfterSeconds === undefined + ? {} + : { retryAfterSeconds: opts.retryAfterSeconds }), + }, + }) + + type Warning = { message: string; details?: unknown } + + const backoffs = (warnings: Warning[]) => + warnings + .filter((entry) => + entry.message === '[factory] Relayfile discovery overloaded; backing off before another sweep') + .map((entry) => entry.details as { delayMs: number; consecutiveOverloads: number }) + + /** + * The durable ratchet after each overloaded sweep, whichever way the sweep + * ended. A sweep that skipped every shed unit and ran to the end still + * moves the ratchet — it just does so from the commit path rather than the + * deferral path. + */ + const ratchet = (warnings: Warning[]) => + warnings + .filter((entry) => + entry.message === '[factory] Relayfile discovery overloaded; backing off before another sweep' || + entry.message === '[factory] discovery sweep committed despite Relayfile overload') + .map((entry) => entry.details as { + delayMs: number + consecutiveOverloads: number + previousOverloads: number + reason: string + }) + + /** + * `oldest_inflight_age` rejects ALL background traffic regardless of how + * few requests are in flight — one stuck op poisons everyone — so a sweep + * against a shedding object gets nothing done at all. This is the shape the + * ratchet exists for, and it must keep escalating. + */ + class ShedEveryTreeReadMount extends FakeMountClient { + constructor(readonly makeOverload: () => Error) { + super() + } + + override async listTree(): Promise { + throw this.makeOverload() + } + } + + const shedFactory = (mount: FakeMountClient, clock: ManualClock, warnings: Warning[]) => + createFactory(config({ issueSource: 'github' }), { + mount, + fleet: new FakeFleetClient(), + stateStore: new InMemoryStateStore({ batchSize: 2 }), + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + clock, + logger: { warn: (message, details) => warnings.push({ message, details }) }, + }) + + const drainLadder = async (mount: FakeMountClient, rungs: number) => { + const clock = new ManualClock() + const warnings: Warning[] = [] + const factory = shedFactory(mount, clock, warnings) + for (let rung = 0; rung < rungs; rung += 1) { + await expect(factory.runOnce()).rejects.toThrow(OVERLOAD_MESSAGE) + } + return { factory, warnings, delays: backoffs(warnings) } + } + + // MUST FIRE. The dependency asked for 5 seconds and Factory slept for up to + // 300 — a self-imposed outage, and the reason /healthz still read degraded + // seven minutes after the upstream fix went live. + it('caps the ladder near the advertised Retry-After instead of climbing to five minutes', async () => { + const mount = new ShedEveryTreeReadMount(() => overloadError({ retryAfterSeconds: 5 })) + + const { delays } = await drainLadder(mount, 7) + + // Never below the advertised delay, never wildly above it, and still + // rising: the ratchet is intact, it is just bounded by what the + // dependency actually asked for. + expect(delays.map((entry) => entry.delayMs)).toEqual([ + 5_000, + 10_000, + 20_000, + 30_000, + 30_000, + 30_000, + 30_000, + ]) + // MUST NOT FIRE: repeated genuine overload still escalates the durable + // ratchet. Capping the sleep must not make the counter meaningless. + expect(delays.map((entry) => entry.consecutiveOverloads)).toEqual([1, 2, 3, 4, 5, 6, 7]) + }) + + // MUST NOT FIRE. With no advertised delay there is nothing to respect, so + // the original five-minute ceiling still governs. Without this the fix + // could quietly become "hammer the dependency every five seconds forever". + it('still climbs the five-minute ladder when the 429 advertises no Retry-After', async () => { + const mount = new ShedEveryTreeReadMount(() => overloadError()) + + const { delays } = await drainLadder(mount, 8) + + expect(delays.map((entry) => entry.delayMs)).toEqual([ + 5_000, + 10_000, + 20_000, + 40_000, + 80_000, + 160_000, + 300_000, + 300_000, + ]) + expect(delays.map((entry) => entry.consecutiveOverloads)).toEqual([1, 2, 3, 4, 5, 6, 7, 8]) + }) + + const shedPath = githubIssuePath('AgentWorkforce', 'pear', 59) + const freshPath = githubIssuePath('AgentWorkforce', 'pear', 60) + + /** Sheds the read of specific issue files, leaving the rest of the sweep intact. */ + class ShedSomeIssueReadsMount extends FakeMountClient { + shedPaths = new Set() + + constructor(files: Record, readonly makeOverload: () => Error) { + super(files) + } + + override async readFile(path: string): Promise<{ content: unknown; revision?: string }> { + if (this.shedPaths.has(path)) throw this.makeOverload() + return await super.readFile(path) + } + } + + const twoReadyIssues = () => + new ShedSomeIssueReadsMount( + { + [shedPath]: githubIssueFile(59, { labels: ['factory', 'pear'] }), + [freshPath]: githubIssueFile(60, { labels: ['factory', 'pear'] }), + }, + () => overloadError({ retryAfterSeconds: 5, reason: 'inflight_limit' }), + ) + + // MUST FIRE. Same principle as #292/#293: one work unit must not decide the + // fate of the others. A sweep that pulled 40 issues and hit one 429 on the + // 39th must not throw away the other 39. + it('skips the work unit relayfile shed and dispatches the rest of the sweep', async () => { + const mount = twoReadyIssues() + mount.shedPaths.add(shedPath) + const fleet = new LocalLifecycleFleetClient() + const warnings: Warning[] = [] + const factory = createFactory(config({ issueSource: 'github', batchSize: 4 }), { + mount, + fleet, + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + logger: { warn: (message, details) => warnings.push({ message, details }) }, + }) + + const report = await factory.runOnce() + + expect(report.dispatched.map((result) => result.issue.key)).toEqual(['60']) + expect(report.skipped).toContainEqual(expect.objectContaining({ + // The issue body was never readable, so the ref is reconstructed from + // the path — the key is what an operator needs to correlate it. + issue: expect.objectContaining({ key: '59' }), + // Sanitized the way #293 sanitized per-item skip reasons: a fixed + // classification plus an allowlisted relayfile reason code. + reason: 'relayfile overloaded (inflight_limit)', + })) + expect(fleet.spawns.map((spawn) => spawn.name)).toEqual([ + 'ar-60-impl-pear', + 'ar-60-review-pear', + ]) + expect(factory.status().counters.discoveryOverloadItemsSkipped).toBe(1) + }) + + // MUST FIRE. Deliverable 4: four distinct reason codes share one message + // string, and only the reason separates a DO-local limiter from a runtime + // shed from a Worker-global one. + it('logs the 429 reason code, not just its message', async () => { + const mount = twoReadyIssues() + mount.shedPaths.add(shedPath) + const warnings: Warning[] = [] + const factory = createFactory(config({ issueSource: 'github', batchSize: 4 }), { + mount, + fleet: new LocalLifecycleFleetClient(), + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + logger: { warn: (message, details) => warnings.push({ message, details }) }, + }) + + await factory.runOnce() + + expect(warnings).toContainEqual(expect.objectContaining({ + message: '[factory] relayfile shed a discovery operation', + details: expect.objectContaining({ + operation: 'readFile', + status: 429, + reason: 'inflight_limit', + retryAfterSeconds: 5, + }), + })) + expect(factory.status().counters['discoveryOverloadReason:inflight_limit']).toBe(1) + }) + + // MUST FIRE. Deliverable 3: requiring a perfect sweep to clear the ratchet + // means the ratchet rarely clears under sustained mild load, so recovery + // stays pinned at the cap long after the dependency recovered. + it('decays the ratchet when a sweep makes progress despite an overload', async () => { + const mount = twoReadyIssues() + const clock = new ManualClock() + const warnings: Warning[] = [] + const stateStore = new InMemoryStateStore({ batchSize: 4 }) + const factory = createFactory(config({ issueSource: 'github', batchSize: 4 }), { + mount, + fleet: new LocalLifecycleFleetClient(), + stateStore, + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + clock, + logger: { warn: (message, details) => warnings.push({ message, details }) }, + }) + + // Every work unit is shed, so three sweeps in a row get nothing through. + // Skipping a shed unit is no longer fatal, but a sweep that served NO + // unit accomplished nothing, so it still fails and still ratchets. + mount.shedPaths.add(shedPath) + mount.shedPaths.add(freshPath) + for (let rung = 0; rung < 3; rung += 1) { + await expect(factory.runOnce()).rejects.toThrow(OVERLOAD_MESSAGE) + } + expect(ratchet(warnings).map((entry) => entry.consecutiveOverloads)).toEqual([1, 2, 3]) + + // The dependency partially recovers: issue 60 now reads and dispatches, + // issue 59 is still shed. That is progress, so the ratchet must come + // down rather than stay pinned or climb. + mount.shedPaths.delete(freshPath) + const report = await factory.runOnce() + + expect(report.dispatched.map((result) => result.issue.key)).toEqual(['60']) + expect(warnings).toContainEqual(expect.objectContaining({ + message: '[factory] discovery sweep committed despite Relayfile overload', + details: expect.objectContaining({ + previousOverloads: 3, + consecutiveOverloads: 2, + reason: 'inflight_limit', + // The decayed rung, still bounded by the advertised ceiling rather + // than by DISCOVERY_OVERLOAD_BACKOFF_MAX_MS. + delayMs: 10_000, + }), + })) + }) + + // MUST NOT FIRE. Partial-progress decay must not make the ratchet + // meaningless: a sweep that gets nothing done still escalates, and the + // durable counter it escalates from is the one the decayed sweep left. + it('still escalates from the decayed counter when the next sweep gets nothing done', async () => { + const mount = twoReadyIssues() + const clock = new ManualClock() + const warnings: Warning[] = [] + const factory = createFactory(config({ issueSource: 'github', batchSize: 4 }), { + mount, + fleet: new LocalLifecycleFleetClient(), + stateStore: new InMemoryStateStore({ batchSize: 4 }), + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + clock, + logger: { warn: (message, details) => warnings.push({ message, details }) }, + }) + + mount.shedPaths.add(shedPath) + mount.shedPaths.add(freshPath) + await expect(factory.runOnce()).rejects.toThrow(OVERLOAD_MESSAGE) + await expect(factory.runOnce()).rejects.toThrow(OVERLOAD_MESSAGE) + expect(ratchet(warnings).map((entry) => entry.consecutiveOverloads)).toEqual([1, 2]) + + // One sweep gets work done and decays the counter to 1... + mount.shedPaths.delete(freshPath) + expect((await factory.runOnce()).dispatched.map((result) => result.issue.key)).toEqual(['60']) + expect(ratchet(warnings).at(-1)?.consecutiveOverloads).toBe(1) + + // ...and the next sweep, which gets nothing through, climbs again from + // there rather than restarting at zero. + mount.shedPaths.add(freshPath) + await expect(factory.runOnce()).rejects.toThrow(OVERLOAD_MESSAGE) + expect(ratchet(warnings).at(-1)).toMatchObject({ + previousOverloads: 1, + consecutiveOverloads: 2, + }) + }) + + // MUST NOT FIRE. Skipping shed work units must never hand back a green + // sweep over a dependency that served none of it: `readinessReconcile` is + // the signal operators and monitors read, and a sweep that dispatched + // nothing because every unit was shed is an outage, not a clean pass. + it('still fails a sweep in which relayfile shed every work unit', async () => { + const mount = twoReadyIssues() + mount.shedPaths.add(shedPath) + mount.shedPaths.add(freshPath) + const fleet = new LocalLifecycleFleetClient() + const warnings: Warning[] = [] + const factory = createFactory(config({ issueSource: 'github', batchSize: 4 }), { + mount, + fleet, + stateStore: new InMemoryStateStore({ batchSize: 4 }), + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + clock: new ManualClock(), + logger: { warn: (message, details) => warnings.push({ message, details }) }, + }) + + // Two shed units is below DISCOVERY_OVERLOAD_PER_SWEEP_LIMIT, so the fuse + // is not what saves this — "no unit got through" is. + await expect(factory.runOnce()).rejects.toThrow(OVERLOAD_MESSAGE) + expect(fleet.spawns).toEqual([]) + expect(backoffs(warnings)).toEqual([ + expect.objectContaining({ delayMs: 5_000, consecutiveOverloads: 1 }), + ]) + }) + + // MUST NOT FIRE. Skipping shed work units must not degenerate into grinding + // a shedding dependency through a whole backlog one 429 at a time. Once the + // sweep has been shed repeatedly, the pass aborts and backs off. + it('still aborts the sweep once relayfile sheds it repeatedly', async () => { + const numbers = [71, 72, 73, 74, 75, 76, 77] + const paths = numbers.map((number) => githubIssuePath('AgentWorkforce', 'pear', number)) + const mount = new ShedSomeIssueReadsMount( + Object.fromEntries(paths.map((path, index) => [ + path, + githubIssueFile(numbers[index]!, { labels: ['factory', 'pear'] }), + ])), + () => overloadError({ retryAfterSeconds: 5, reason: 'durable_object_overloaded' }), + ) + for (const path of paths) mount.shedPaths.add(path) + const fleet = new LocalLifecycleFleetClient() + const warnings: Warning[] = [] + const factory = createFactory(config({ issueSource: 'github', batchSize: 5 }), { + mount, + fleet, + stateStore: new InMemoryStateStore({ batchSize: 5 }), + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + clock: new ManualClock(), + logger: { warn: (message, details) => warnings.push({ message, details }) }, + }) + + await expect(factory.runOnce()).rejects.toThrow(OVERLOAD_MESSAGE) + + expect(fleet.spawns).toEqual([]) + expect(backoffs(warnings)).toEqual([ + expect.objectContaining({ delayMs: 5_000, consecutiveOverloads: 1 }), + ]) + }) + }) + it('filters GitHub startup discovery through the Relayfile issue index', async () => { const root = await mkdtemp(join(tmpdir(), 'factory-github-index-discovery-')) const readyPath = githubIssueCompactPath('AgentWorkforce', 'pear', 70) diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 6f0fe2a..07e35e0 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -448,6 +448,38 @@ const DISCOVERY_SWEEP_RENEW_MS = 30_000 const READINESS_RECONCILE_FAILURE_THRESHOLD = 3 const DISCOVERY_CHANGE_EVENT_LIMIT = 1_000 const DISCOVERY_OVERLOAD_BACKOFF_MAX_MS = 5 * 60_000 +/** First rung of the ladder when the 429 advertises no `Retry-After`. */ +const DISCOVERY_OVERLOAD_BACKOFF_BASE_MS = 5_000 +/** + * Floor for the advertised delay. `Retry-After: 0` would otherwise pin the + * whole ladder at zero (`0 * 2 ** n` is still zero) and turn respecting the + * dependency into hammering it. + */ +const DISCOVERY_OVERLOAD_BACKOFF_MIN_MS = 1_000 +/** + * Ceiling for the ladder once the dependency has told us how long to wait. + * + * relayfile sheds an overloaded workspace DO with a 429 in milliseconds + * carrying `Retry-After: 5`, and #297 is what happened when the ladder ignored + * that and climbed to `DISCOVERY_OVERLOAD_BACKOFF_MAX_MS` anyway: the + * dependency asked for five seconds, Factory slept for five minutes, probed + * for recovery once per cap-length window, and presented a transient upstream + * blip as a sustained outage. The ratchet still escalates — it is just bounded + * by roughly what was actually asked for, and never *below* it, so this is a + * ceiling and not a licence to retry sooner than the dependency allows. + */ +const DISCOVERY_OVERLOAD_ADVERTISED_BACKOFF_MAX_MS = 30_000 +/** + * How many relayfile operations one sweep may have shed before the sweep is + * abandoned rather than continued. + * + * Per-item overload skips that item and keeps going (#297, the same principle + * as #292), but skipping must not degenerate into grinding a shedding + * dependency through an entire backlog one 429 at a time. Past this many, the + * dependency is not serving this sweep at all: abort, back off, and let the + * ratchet do its job. + */ +const DISCOVERY_OVERLOAD_PER_SWEEP_LIMIT = 5 const GITHUB_FACTORY_LABEL = 'factory' const GITHUB_LIFECYCLE_LABELS = new Set(['factory:in-progress', 'factory:human-review']) const GITHUB_MIRROR_TITLE_PREFIX = '[factory]' @@ -768,7 +800,25 @@ export class FactoryLoop implements Factory { // so the loop's catch no longer runs the failure-handoff reaper for it; the // pass reaps inline and must write to the same paths runLoop would. #loopReapPaths?: { heartbeatPath: string; registryPath: string } + /** + * The first 429 relayfile raised during this sweep, kept for its + * `Retry-After` and reason when the sweep decides how long to back off. + * + * Before #297 this doubled as a sweep-wide kill switch: any 429 from any + * relayfile call latched here and `#runOnceWithDiscoveryFence` then threw it + * away along with everything the sweep had already accomplished. It is now + * only evidence, never a verdict — see `#discoverySweepOverloads` for the + * fuse that still ends a sweep the dependency is genuinely refusing to serve. + */ #discoveryOverloadError?: unknown + /** Relayfile operations this sweep has been shed on. */ + #discoverySweepOverloads = 0 + /** + * Whether relayfile served at least one ready work unit end to end during + * this sweep — its issue read, or a dispatch built on it. This is what + * decays the durable overload ratchet; see `#discoveryOverloadOutcome`. + */ + #discoverySweepProgress = false #resolvedIssueSource?: IssueSource #integrationInstructions?: string #integrationInstructionsRefresh?: Promise @@ -1547,7 +1597,15 @@ export class FactoryLoop implements Factory { skipped: report.skipped.length, }) } catch (error) { - const errorMessage = describeError(error).errorMessage + // #297: all four relayfile overload reason codes share one message, and + // `lastError` is what an operator reads from /evidence. Without the + // reason, "workspace durable object is busy" cannot be told apart from + // three other conditions with three different remedies. + const overload = relayfileOverload(error) + const errorMessage = overload + ? `${describeError(error).errorMessage} [relayfile ${overload.status} ${overload.reason}` + + `${overload.retryAfterSeconds === undefined ? '' : `; retry-after=${overload.retryAfterSeconds}s`}]` + : describeError(error).errorMessage this.#readinessReconcileConsecutiveFailures += 1 this.#readinessReconcileLastDurationMs = this.#elapsedSince(startedAtMs) this.#readinessReconcileLastFailureAtMs = this.#clock.now() @@ -2328,12 +2386,28 @@ export class FactoryLoop implements Factory { this.#discoverySweepStartedAtMs = sweepStartedAtMs this.#discoverySweepLeaseLost = false this.#discoveryOverloadError = undefined + this.#discoverySweepOverloads = 0 + this.#discoverySweepProgress = false this.#startDiscoverySweepRenewal(claim.lease.epoch) let leaseReleased = false try { this.#discoverySession = await this.#prepareDiscoverySession(claim) + // #297: a 429 raised anywhere in the sweep used to latch and be rethrown + // here, discarding a completed pass — every issue read, every dispatch — + // because of one transient shed operation. The work this sweep did is + // now kept instead, and the ratchet below records that the dependency is + // shedding but still serving. const report = await this.#performRunOnce(opts) - if (this.#discoveryOverloadError) throw this.#discoveryOverloadError + // The exception, and the reason skipping shed units cannot make a sweep + // unconditionally green: a sweep that was shed AND got no work unit + // through accomplished nothing. There is no progress to preserve, and + // committing it would report a clean sweep over a dependency that served + // none of it — leaving `readinessReconcile` healthy while Factory + // dispatches nothing, which is the #292 wedge wearing the other costume. + // Fail it so the ratchet escalates and readiness reflects reality. + if (this.#discoveryOverloadError !== undefined && !this.#discoverySweepProgress) { + throw this.#discoveryOverloadError + } const checkpoint = await this.#finalizeDiscoveryCheckpoint() // Do not clear the durable lease while a renewal can still be waiting on // the same state-file lock. A late renewal that observes the completed @@ -2348,6 +2422,7 @@ export class FactoryLoop implements Factory { this.#discoverySweepOwner, claim.lease.epoch, checkpoint, + this.#discoveryOverloadOutcome(claim.state.consecutiveOverloads, 'committed'), ) leaseReleased = completed if (!completed) throw new Error('discovery sweep lease was lost before completion') @@ -2362,24 +2437,24 @@ export class FactoryLoop implements Factory { await this.#stopDiscoverySweepRenewal() const overload = relayfileOverload(error) if (overload) { - const consecutiveOverloads = claim.state.consecutiveOverloads + 1 - const delayMs = discoveryOverloadBackoffMs(overload.retryAfterSeconds, consecutiveOverloads) - const backoffUntilMs = this.#clock.now() + delayMs + const outcome = this.#discoveryOverloadOutcome(claim.state.consecutiveOverloads, 'aborted', error)! leaseReleased = await this.#state.deferDiscoverySweep( this.#workspaceId, this.#discoverySweepOwner, claim.lease.epoch, - backoffUntilMs, - consecutiveOverloads, + outcome.backoffUntilMs, + outcome.consecutiveOverloads, ) - this.#increment('discoveryOverloadBackoffs') this.#logger.warn?.('[factory] Relayfile discovery overloaded; backing off before another sweep', { status: overload.status, reason: overload.reason, retryAfterSeconds: overload.retryAfterSeconds, - delayMs, - backoffUntilMs, - consecutiveOverloads, + delayMs: outcome.delayMs, + backoffUntilMs: outcome.backoffUntilMs, + consecutiveOverloads: outcome.consecutiveOverloads, + previousOverloads: claim.state.consecutiveOverloads, + sweepOverloads: this.#discoverySweepOverloads, + sweepProgress: this.#discoverySweepProgress, }) // backoffUntilMs is already durable via deferDiscoverySweep, and the // next runOnce() honors it at the pre-claim wait above — sleeping @@ -2394,6 +2469,8 @@ export class FactoryLoop implements Factory { this.#discoverySweepEpoch = undefined this.#discoverySweepStartedAtMs = undefined this.#discoveryOverloadError = undefined + this.#discoverySweepOverloads = 0 + this.#discoverySweepProgress = false // This sweep is over either way (committed, deferred, or lease lost) — // a stale `true` here would otherwise make every #listRelayfileTree // call outside a fresh claim (Slack lookups, PR confirmation, the @@ -2410,6 +2487,63 @@ export class FactoryLoop implements Factory { } } + /** + * How the durable overload ratchet should read after this sweep, or + * `undefined` if relayfile never shed anything and the ordinary reset + * applies. + * + * #297, deliverable 3. `consecutiveOverloads` used to clear only on a + * *fully* clean sweep, so under sustained mild load the ratchet essentially + * never cleared: it climbed to the cap on the first bad sweep and stayed + * there, probing for recovery once per cap-length window, long after the + * dependency had recovered. Requiring perfection to clear a ratchet means + * the ratchet does not clear. + * + * The signal that replaces "was this sweep perfect" is "did relayfile serve + * any of this sweep's work units", because that is what the ratchet is + * actually for. A shedding DO rejects ALL background traffic — + * `oldest_inflight_age` does this regardless of how few requests are in + * flight — so a sweep against a genuinely overloaded workspace gets not one + * unit through and still escalates, all the way to the cap. A sweep that did + * get units through proves the dependency is serving us, so it decays by one + * rung. Decay, not reset: getting work done while being shed is not evidence + * that the overload is over, only that it is survivable. + */ + #discoveryOverloadOutcome( + previousOverloads: number, + sweepOutcome: 'committed' | 'aborted', + error?: unknown, + ): { consecutiveOverloads: number; backoffUntilMs: number; delayMs: number } | undefined { + const overload = relayfileOverload(error) ?? relayfileOverload(this.#discoveryOverloadError) + if (!overload) return undefined + // Deliberately NOT "the sweep reached its end". A sweep in which relayfile + // shed EVERY unit still runs the loop to the end, and it got nothing done; + // treating that as progress would decay the ratchet in exactly the case it + // exists for. (`#runOnceWithDiscoveryFence` turns that sweep into an + // aborted one before it can commit, so the escalate branch here is reached + // only through `sweepOutcome === 'aborted'` — but the rule is a property of + // progress, not of which caller asked, and is written that way.) + const consecutiveOverloads = this.#discoverySweepProgress + ? Math.max(0, previousOverloads - 1) + : previousOverloads + 1 + const delayMs = discoveryOverloadBackoffMs(overload.retryAfterSeconds, consecutiveOverloads) + const backoffUntilMs = this.#clock.now() + delayMs + this.#increment('discoveryOverloadBackoffs') + if (sweepOutcome === 'committed') { + this.#logger.warn?.('[factory] discovery sweep committed despite Relayfile overload', { + status: overload.status, + reason: overload.reason, + retryAfterSeconds: overload.retryAfterSeconds, + sweepOverloads: this.#discoverySweepOverloads, + previousOverloads, + consecutiveOverloads, + delayMs, + backoffUntilMs, + }) + } + return { consecutiveOverloads, backoffUntilMs, delayMs } + } + async #performRunOnce(opts: { dryRun?: boolean } = {}): Promise { const dryRun = opts.dryRun ?? this.#config.dryRun const startedAtMs = this.#clock.now() @@ -2457,8 +2591,35 @@ export class FactoryLoop implements Factory { const issueEntries: Array<{ path: string; issue?: LinearIssue }> = [] for (const path of paths) { - const issue = await this.#readIssue(path) + let issue: LinearIssue | undefined + let shed = false + try { + issue = await this.#readIssue(path) + } catch (error) { + // #297: `#readIssue` rethrows relayfile overload and swallows every + // other read fault, so this catch only ever sees the backend + // shedding THIS issue's read. That is a fact about one work unit: + // a sweep that pulled 40 issues and was shed on the 39th must keep + // the other 39, exactly as #292 argued for dispatch failures. The + // fuse below is what still ends a sweep the backend is refusing. + const overload = relayfileOverload(error) + const fuse = this.#discoveryOverloadFuseError() + if (!overload || fuse) throw fuse ?? error + shed = true + this.#increment('discoveryOverloadItemsSkipped') + this.#logger.warn?.('[factory] relayfile shed a ready-issue read; skipping it and continuing the sweep', { + path, + status: overload.status, + reason: overload.reason, + retryAfterSeconds: overload.retryAfterSeconds, + sweepOverloads: this.#discoverySweepOverloads, + }) + recordSkip({ issue: issueRefFromPath(path), reason: perItemDispatchSkipReason(error) }) + } readyIssueReads += 1 + // Relayfile served this work unit's read: the dependency is shedding + // but not refusing, which is what decays the ratchet (#297). + if (!shed) this.#discoverySweepProgress = true lastReadyReadProgressAtMs = this.#logTimedProgress( this.#config.issueSource === 'github' ? '[factory] GitHub ready issue read progress' @@ -2467,10 +2628,12 @@ export class FactoryLoop implements Factory { lastReadyReadProgressAtMs, { read: readyIssueReads, total: paths.length, path }, ) - if (issue && issueSource === 'linear') { - await this.#recordCanonicalIssueState(issue) + if (!shed) { + if (issue && issueSource === 'linear') { + await this.#recordCanonicalIssueState(issue) + } + issueEntries.push({ path, issue }) } - issueEntries.push({ path, issue }) await this.#refreshLiveHeartbeatIfDue() } if (issueSource === 'github') { @@ -2564,6 +2727,9 @@ export class FactoryLoop implements Factory { // A completed dispatch — even one that parks or escalates the issue — // proves the pipeline still works, so the fuse below starts over. unclassifiedFailuresSinceDispatch = 0 + // ...and proves relayfile is still serving this sweep, which is what + // decays the durable overload ratchet (#297). + this.#discoverySweepProgress = true if (result.agents.length === 0 && !dryRun) { const reason = result.hold?.kind === 'dependency-cycle' ? `dependency cycle detected: ${result.hold.cycle?.join(' -> ') ?? 'unknown cycle'}` @@ -2579,8 +2745,27 @@ export class FactoryLoop implements Factory { // that is about ONE unit costs that unit and nothing else. Only the // conditions named in `#isPassFatalFailure` — the ones where // continuing the pass is meaningless — abort the whole sweep. - if (this.#isPassFatalFailure(error, dryRun)) throw error - if (!isClassifiedPerItemDispatchFailure(error)) { + if (this.#isPassFatalFailure(error, dryRun)) { + // The overload fuse may have been tripped by a 429 that a caller + // swallowed, leaving an unrelated error in hand. The fence keys + // the durable backoff off `relayfileOverload(error)`, so hand it + // the 429 rather than whatever surfaced last. + throw this.#discoveryOverloadFuseError() ?? error + } + const overload = relayfileOverload(error) + if (overload) { + // #297: shedding is a state of the dependency, not a fault of this + // work unit, so it stays out of `counters.errors` and gets its own + // counter — the same split #293 made for undispatchable units. + this.#increment('discoveryOverloadItemsSkipped') + this.#logger.warn?.('[factory] relayfile shed this work unit; skipping it and continuing the sweep', { + issue: issueRef(issue).key, + status: overload.status, + reason: overload.reason, + retryAfterSeconds: overload.retryAfterSeconds, + sweepOverloads: this.#discoverySweepOverloads, + }) + } else if (!isClassifiedPerItemDispatchFailure(error)) { unclassifiedFailuresSinceDispatch += 1 // A pass-wide fault can arrive disguised as a run of per-item // faults. Skipping every unit would then hand back a green report @@ -2652,6 +2837,21 @@ export class FactoryLoop implements Factory { } } + /** + * The 429 that ended this sweep, when relayfile has shed + * `DISCOVERY_OVERLOAD_PER_SWEEP_LIMIT` operations and skipping the next work + * unit would just be grinding a shedding dependency. Undefined below that. + * + * Returns the *latched* 429 rather than whatever error is in hand, because + * `#runOnceWithDiscoveryFence` keys the durable backoff off + * `relayfileOverload(error)` and a caller may have swallowed the 429 that + * tripped the fuse. + */ + #discoveryOverloadFuseError(): unknown { + if (this.#discoverySweepOverloads < DISCOVERY_OVERLOAD_PER_SWEEP_LIMIT) return undefined + return this.#discoveryOverloadError + } + /** * Whether a failure raised while processing ONE work unit must abort the * whole readiness pass instead of skipping that unit. @@ -2672,10 +2872,11 @@ export class FactoryLoop implements Factory { * one would be recorded as an ordinary per-issue skip. The run report * would then claim a clean pass over work this process no longer has the * right to touch. - * - Relayfile signalled overload for this sweep. The backend is shedding - * load; grinding through the remaining units makes it worse, and - * `#runOnceWithDiscoveryFence` is going to rethrow this at the fence - * anyway. + * - Relayfile has shed `DISCOVERY_OVERLOAD_PER_SWEEP_LIMIT` operations in + * this sweep. A single shed operation is per-item and skippable (#297), + * but past this the backend is not serving this sweep at all: grinding + * through the remaining units makes it worse, and the fence needs the 429 + * to set the durable backoff. * - The factory is stopping. Teardown is in progress and dispatching more * agents now leaks them past the shutdown deadline. * - The fleet control-plane circuit is no longer closed, **on a live pass**. @@ -2704,7 +2905,15 @@ export class FactoryLoop implements Factory { #isPassFatalFailure(error: unknown, dryRun: boolean): boolean { // Sweep-scoped: these are about this process's right or ability to run the // pass at all, so they hold for a dry run exactly as for a live one. - if (this.#discoverySweepLeaseLost || this.#discoveryOverloadError !== undefined || this.#stopping) { + if (this.#discoverySweepLeaseLost || this.#stopping) { + return true + } + // #297: relayfile overload used to sit alongside those two, and it did not + // belong there. A 429 on ONE work unit is a fact about that unit's read or + // write, not about this process's right to run the pass — and because the + // flag latched for the whole sweep, the first transient shed also made + // every later unit fatal. Only sustained shedding is now pass-fatal. + if (this.#discoverySweepOverloads >= DISCOVERY_OVERLOAD_PER_SWEEP_LIMIT) { return true } // Fleet-scoped, and therefore live-only. See the doc comment above. @@ -3651,14 +3860,37 @@ export class FactoryLoop implements Factory { } return result } catch (error) { - if (relayfileOverload(error) && this.#discoverySweepEpoch !== undefined) { + const overload = relayfileOverload(error) + if (overload && this.#discoverySweepEpoch !== undefined) { this.#discoveryOverloadError ??= error + this.#discoverySweepOverloads += 1 + this.#increment('discoveryOverloadOperations') + // #297: relayfile's four overload reason codes — inflight_limit, + // oldest_inflight_age, router_inflight_limit, durable_object_overloaded + // — all share ONE message string, and they mean four different things: + // a DO-local admission cap, one stuck op poisoning every background + // caller, a Worker-global isolate cap, and the Cloudflare runtime + // shedding the object outright. `relayfileOverload()` has always parsed + // the reason; nothing on this path ever printed it, and during the + // 2026-08-20 outage that ambiguity was the single biggest obstacle to + // diagnosis. Unconditional, unlike the failure warn below: a 429 is + // always worth one line. + this.#increment(`discoveryOverloadReason:${relayfileOverloadReasonLabel(overload.reason)}`) + this.#logger.warn?.('[factory] relayfile shed a discovery operation', { + ...metadata, + status: overload.status, + reason: overload.reason, + retryAfterSeconds: overload.retryAfterSeconds, + sweepOverloads: this.#discoverySweepOverloads, + elapsedMs: this.#elapsedSince(startedAtMs), + }) } if (opts.logFailure || waitWarnings > 0) { this.#increment('relayfileOperationFailures') this.#logger.warn?.('[factory] relayfile operation failed', { ...metadata, elapsedMs: this.#elapsedSince(startedAtMs), + ...(overload ? { status: overload.status, reason: overload.reason } : {}), error: describeError(error).errorMessage, }) } @@ -16376,6 +16608,16 @@ const githubIssueAuthor = (issue: LinearIssue): string | undefined => { return source ? undefined : githubAuthorLogin(payload)?.trim() || undefined } +/** + * An `IssueRef` for a path whose issue body could not be read at all — the + * shape a relayfile-shed ready-issue read leaves behind (#297). The key is + * what an operator needs to correlate the skip; the uuid falls back to it. + */ +const issueRefFromPath = (path: string): IssueRef => { + const key = keyFromPath(path) + return { uuid: uuidFromPath(path) ?? key, key, path } +} + const issueRef = (issue: LinearIssue): IssueRef => ({ uuid: issue.uuid, key: issue.key, path: issue.path }) // Preserve the historical Linear state namespace while keeping GitHub-native @@ -18665,16 +18907,60 @@ const relayfileOverload = (error: unknown): RelayfileOverload | undefined => { return { status, reason, ...(retryAfterSeconds === undefined ? {} : { retryAfterSeconds }) } } +/** + * relayfile's overload reason codes, allowlisted. + * + * `IterationReport.skipped[].reason` is serialized to stdout by `factory + * run-once`, so it stays a fixed classification plus a known code — the same + * public-surface rule #293 applied to error class names — rather than + * whatever string the dependency happened to send. + */ +const RELAYFILE_OVERLOAD_REASONS = new Set([ + // Admission gate inside the workspace durable object. + 'inflight_limit', + 'oldest_inflight_age', + 'write_admission_limit', + // Worker-side backpressure, per isolate rather than per workspace. + 'router_inflight_limit', + // The Cloudflare runtime shed the object; relayfile only relabels it. + 'durable_object_overloaded', + // relayfileOverload()'s fallback when the body carried no reason at all. + 'rate_limited', +]) + +/** + * Both the run-report reason and the per-reason counter key are built from + * this, so an unknown code from the dependency can neither leak into stdout + * nor open an unbounded counter namespace. + */ +const relayfileOverloadReasonLabel = (reason: string): string => + RELAYFILE_OVERLOAD_REASONS.has(reason) ? reason : 'unrecognized' + +/** + * How long to wait before the next discovery sweep after relayfile shed this + * one. + * + * The advertised `Retry-After` is authoritative in BOTH directions (#297): + * it is the first rung, so we never retry sooner than the dependency allows, + * and it bounds the ceiling, so we never sleep for minutes because of a + * request to wait seconds. Without an advertised delay there is nothing to + * respect and the original five-minute ladder governs unchanged. + */ const discoveryOverloadBackoffMs = ( retryAfterSeconds: number | undefined, consecutiveOverloads: number, ): number => { - const retryAfterMs = Math.max(0, Math.ceil((retryAfterSeconds ?? 0) * 1_000)) - const exponentialMs = Math.min( - DISCOVERY_OVERLOAD_BACKOFF_MAX_MS, - 5_000 * (2 ** Math.min(10, Math.max(0, consecutiveOverloads - 1))), - ) - return Math.max(retryAfterMs, exponentialMs) + const advertisedMs = retryAfterSeconds === undefined + ? undefined + : Math.max(DISCOVERY_OVERLOAD_BACKOFF_MIN_MS, Math.ceil(retryAfterSeconds * 1_000)) + const baseMs = advertisedMs ?? DISCOVERY_OVERLOAD_BACKOFF_BASE_MS + // A dependency that asks for longer than the advertised ceiling still gets + // what it asked for; the ceiling only stops the ladder from overshooting it. + const ceilingMs = advertisedMs === undefined + ? DISCOVERY_OVERLOAD_BACKOFF_MAX_MS + : Math.max(advertisedMs, DISCOVERY_OVERLOAD_ADVERTISED_BACKOFF_MAX_MS) + const steps = Math.min(10, Math.max(0, consecutiveOverloads - 1)) + return Math.min(ceilingMs, baseMs * (2 ** steps)) } const eventSequenceNumber = (eventId: string): number | undefined => { @@ -19237,7 +19523,11 @@ const UNCLASSIFIED_DISPATCH_FAILURE_LIMIT = 5 */ const isClassifiedPerItemDispatchFailure = (error: unknown): boolean => error instanceof LiveDispatchStateChangedError || - error instanceof DispatchLifecycleClaimRefusedError + error instanceof DispatchLifecycleClaimRefusedError || + // Relayfile shedding one operation is a state of the dependency, not an + // unexplained fault, and it has its own fuse — see #297 and + // DISCOVERY_OVERLOAD_PER_SWEEP_LIMIT. + relayfileOverload(error) !== undefined /** * The run-report reason recorded for a work unit the pass could not dispatch. @@ -19249,6 +19539,8 @@ const isClassifiedPerItemDispatchFailure = (error: unknown): boolean => * `describeControlPlaneError` makes for circuit state. */ const perItemDispatchSkipReason = (error: unknown): string => { + const overload = relayfileOverload(error) + if (overload) return `relayfile overloaded (${relayfileOverloadReasonLabel(overload.reason)})` if (error instanceof LiveDispatchStateChangedError) return 'live state changed during dispatch' if (error instanceof DispatchLifecycleClaimRefusedError) { return error.refusal === 'terminal' diff --git a/src/ports/state.ts b/src/ports/state.ts index de79841..f94a16e 100644 --- a/src/ports/state.ts +++ b/src/ports/state.ts @@ -406,6 +406,12 @@ export interface StateStore { owner: string, epoch: number, checkpoint?: DiscoveryCheckpoint, + /** + * Residual overload state for a sweep that committed its work while + * relayfile was shedding some of it (#297). Omitted for a clean sweep, + * which clears the ratchet and the backoff as it always has. + */ + overload?: { consecutiveOverloads: number; backoffUntilMs: number }, ): Promise deferDiscoverySweep( workspaceId: string, diff --git a/src/state/file-state-store.ts b/src/state/file-state-store.ts index 325eaed..7e53c77 100644 --- a/src/state/file-state-store.ts +++ b/src/state/file-state-store.ts @@ -154,6 +154,7 @@ export class DocumentStateStore extends InMemoryStateStore { owner: string, epoch: number, checkpoint?: DiscoveryCheckpoint, + overload?: { consecutiveOverloads: number; backoffUntilMs: number }, ): Promise { return await this.#exclusive(async () => this.#withMutationLock(async () => { const document = await this.#loadFromDisk() @@ -164,8 +165,10 @@ export class DocumentStateStore extends InMemoryStateStore { // now empty") — keep the last good checkpoint so the next sweep can // still diff from it instead of falling back to a full walk. if (checkpoint) state.checkpoint = cloneDiscoveryCheckpoint(checkpoint) - state.consecutiveOverloads = 0 - state.backoffUntilMs = 0 + // A sweep that committed while relayfile was shedding it keeps a decayed + // ratchet and a backoff instead of clearing both outright (#297). + state.consecutiveOverloads = overload?.consecutiveOverloads ?? 0 + state.backoffUntilMs = overload?.backoffUntilMs ?? 0 delete state.lease await this.#persist(document) return true diff --git a/src/state/in-memory-state-store.ts b/src/state/in-memory-state-store.ts index 47d2f9e..9fc9398 100644 --- a/src/state/in-memory-state-store.ts +++ b/src/state/in-memory-state-store.ts @@ -137,6 +137,7 @@ export class InMemoryStateStore implements StateStore { owner: string, epoch: number, checkpoint?: DiscoveryCheckpoint, + overload?: { consecutiveOverloads: number; backoffUntilMs: number }, ): Promise { const state = this.#workspace(workspaceId).discoverySweep if (!discoveryLeaseMatches(state, owner, epoch)) return false @@ -145,8 +146,10 @@ export class InMemoryStateStore implements StateStore { // now empty") — keep the last good checkpoint so the next sweep can // still diff from it instead of falling back to a full walk. if (checkpoint) state.checkpoint = cloneDiscoveryCheckpoint(checkpoint) - state.consecutiveOverloads = 0 - state.backoffUntilMs = 0 + // A sweep that committed while relayfile was shedding it keeps a decayed + // ratchet and a backoff instead of clearing both outright (#297). + state.consecutiveOverloads = overload?.consecutiveOverloads ?? 0 + state.backoffUntilMs = overload?.backoffUntilMs ?? 0 delete state.lease return true } From cb1f34f31ca985125e58da4abcae7e002884b972 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Thu, 20 Aug 2026 16:24:18 +0200 Subject: [PATCH 2/5] fix(orchestrator): surface a non-overload ready-issue read fault as itself MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up on the #297 read-loop skip. `#readIssue` only ever rethrows a 429 — every other read fault is swallowed and returns undefined — so the non-overload branch is defensive. It was folded in with the fuse check, which meant a fault arriving while the fuse was already tripped would have surfaced as the latched 429 instead of itself. Split them: anything that is not a 429 is not ours to reclassify. Co-Authored-By: Claude Opus 5 Session-Id: 964fa2f4-17fc-4afc-93e3-0f7ee72f1316 --- src/orchestrator/factory.ts | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 07e35e0..ac87a8f 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -2601,10 +2601,13 @@ export class FactoryLoop implements Factory { // shedding THIS issue's read. That is a fact about one work unit: // a sweep that pulled 40 issues and was shed on the 39th must keep // the other 39, exactly as #292 argued for dispatch failures. The - // fuse below is what still ends a sweep the backend is refusing. + // fuse is what still ends a sweep the backend is refusing. const overload = relayfileOverload(error) + // Defensive: anything `#readIssue` did not swallow and is not a 429 + // is not ours to reclassify, and must surface as itself. + if (!overload) throw error const fuse = this.#discoveryOverloadFuseError() - if (!overload || fuse) throw fuse ?? error + if (fuse) throw fuse shed = true this.#increment('discoveryOverloadItemsSkipped') this.#logger.warn?.('[factory] relayfile shed a ready-issue read; skipping it and continuing the sweep', { From 430f2c1e41963c298806fd1ab58f67df92ed6790 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Thu, 20 Aug 2026 16:52:50 +0200 Subject: [PATCH 3/5] fix(orchestrator,state): address the #298 review findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five findings, each with a test proven to fail first. P1 (codex + cubic, independently) — reap before skipping a shed dispatch. A 429 raised after `#dispatchUnlocked` has spawned leaves half-started agents behind, persisted as failure handoffs. runLoop's catch used to reap them because the error aborted the pass; skipping the unit meant nobody did, so the fix leaked spawned agents and a later retry would duplicate them — worse than the abort it replaced. The reap now runs for every per-item skip except the two lifecycle refusals, which are decided before anything is spawned. Deliberately a denylist: an unclassified new failure mode should default to reaping, since a needless reap is one no-op pass over an empty handoff list. P2 (cubic + coderabbit) — allowlist the reason on the readiness surface. `readinessReconcile.lastError` embedded `overload.reason` raw while every other surface in the change routed it through `relayfileOverloadReasonLabel`. It is returned from `status()` AND written into the loop heartbeat file, so an unbounded dependency-controlled string reached an operator-facing artifact on disk; the regression test drives a reason containing markup and asserts it cannot land there. P2 (codex) — honour the longest Retry-After the sweep saw. `#discoveryOverloadError` latches the FIRST 429, so both the committed outcome and the fuse derived their delay from whichever arrived first. An early 1s ask followed by a later 30s ask produced a 1s backoff, breaking the guarantee this change is built on. The sweep now tracks the maximum advertised delay. P2 (cubic) — a read that produced nothing is not progress. `#readIssue` returns `undefined` for a body it could not read, and that was credited as a served work unit, decaying the ratchet on a sweep that served nothing. cubic reached this via a "concurrent 429" that cannot occur — `#readIssue` rethrows overloads rather than swallowing them — but the conclusion holds by a reachable route: the known phantom condition, where the tree lists issue paths whose bodies are absent, makes every read return `undefined`. That is what the test uses. P2 (cubic) — do not lose the overload residual silently. Carrying it as an optional fifth argument to `completeDiscoverySweep` meant any store compiled against the old signature dropped it without a word, losing the backoff so the next sweep retried immediately. Replaced with an optional `completeDiscoverySweepWithOverload`, following this port's own `renewDiscoverySweepWithDetails` precedent: a legacy store still works, and its inability to carry the state is detectable. When one is injected and a residual exists, the sweep warns and increments `discoveryOverloadResidualUnsupported` instead of degrading in silence. Co-Authored-By: Claude Opus 5 Session-Id: 964fa2f4-17fc-4afc-93e3-0f7ee72f1316 --- src/orchestrator/factory.test.ts | 245 ++++++++++++++++++++++++++++- src/orchestrator/factory.ts | 141 ++++++++++++++--- src/ports/state.ts | 25 ++- src/state/file-state-store.ts | 25 ++- src/state/in-memory-state-store.ts | 27 +++- 5 files changed, 432 insertions(+), 31 deletions(-) diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 7986b3f..4379d29 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -45,7 +45,7 @@ import { type ResourceSubscriptionsClient, } from '../subscriptions' import { InternalFleetClient, type HarnessDriverClientLike } from '../fleet/internal-fleet-client' -import type { ConversationMessage, ConversationSessionState, DiscoverySweepClaim, DiscoverySweepRenewal, DispatchLifecycle } from '../ports/state' +import type { ConversationMessage, ConversationSessionState, DiscoverySweepClaim, DiscoverySweepRenewal, DispatchLifecycle, StateStore } from '../ports/state' import { DEFAULT_FLEET_CONTROL_FAILURE_THRESHOLD, DEFAULT_FLEET_ROSTER_TIMEOUT_MS, @@ -4466,6 +4466,249 @@ describe('FactoryLoop', () => { expect.objectContaining({ delayMs: 5_000, consecutiveOverloads: 1 }), ]) }) + + // Review follow-up on #298 (P2). `#discoveryOverloadError` latches the + // FIRST 429 of the sweep, so deriving the delay from it alone can let the + // durable backoff expire before a LATER, longer Retry-After permits — + // contradicting the guarantee this PR is built on. + it('backs off by the longest Retry-After the sweep saw, not the first', async () => { + class ShedWithPerPathDelayMount extends FakeMountClient { + constructor(files: Record, readonly delaysByPath: Map) { + super(files) + } + + override async readFile(path: string): Promise<{ content: unknown; revision?: string }> { + const retryAfterSeconds = this.delaysByPath.get(path) + if (retryAfterSeconds !== undefined) throw overloadError({ retryAfterSeconds }) + return await super.readFile(path) + } + } + + // Issue 59 is read first and asks for one second; issue 60 then asks for + // thirty. Honouring only the first would re-sweep 29 seconds early. + const mount = new ShedWithPerPathDelayMount( + { + [shedPath]: githubIssueFile(59, { labels: ['factory', 'pear'] }), + [freshPath]: githubIssueFile(60, { labels: ['factory', 'pear'] }), + }, + new Map([[shedPath, 1], [freshPath, 30]]), + ) + const warnings: Warning[] = [] + const factory = createFactory(config({ issueSource: 'github', batchSize: 4 }), { + mount, + fleet: new LocalLifecycleFleetClient(), + stateStore: new InMemoryStateStore({ batchSize: 4 }), + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + clock: new ManualClock(), + logger: { warn: (message, details) => warnings.push({ message, details }) }, + }) + + await expect(factory.runOnce()).rejects.toThrow(OVERLOAD_MESSAGE) + + expect(backoffs(warnings)).toEqual([ + expect.objectContaining({ delayMs: 30_000, retryAfterSeconds: 30 }), + ]) + }) + + // Review follow-up on #298 (P1). A 429 raised after `#dispatchUnlocked` + // has spawned leaves half-started agents behind, persisted as failure + // handoffs. runLoop's catch used to reap them because the error aborted + // the pass; now that an overloaded unit is merely skipped, the reap has to + // happen inside the pass or those agents leak and a later retry doubles + // them up. + it('reaps the agents a shed dispatch left behind instead of leaking them', async () => { + const mount = new FakeMountClient({ [issuePath(72)]: issueFile(72) }) + const fleet = new CapturedPidFleetClient([ + { name: 'ar-72-impl-pear', sessionRef: 'session-ar-72-impl-pear', pid: 7_201 }, + { name: 'ar-72-review', sessionRef: 'session-ar-72-review', pid: 7_203 }, + ]) + // Sheds only after the team is spawned, which is the window the leak + // lives in: the agents exist, the dispatch does not complete. + const linear: LinearWriteback = { + async postComment() {}, + async setState() { + throw overloadError({ retryAfterSeconds: 5, reason: 'oldest_inflight_age' }) + }, + async createIssue() { + throw new Error('not used') + }, + async verify() { + return true + }, + } + const factory = createFactory(config({}), { + mount, + fleet, + triage: new StaticTriage(), + linear, + readChildPids: async () => [], + kill: () => { + throw Object.assign(new Error('not running'), { code: 'ESRCH' }) + }, + terminationGraceMs: 0, + }) + + const report = await factory.runOnce() + + expect(report.skipped).toContainEqual(expect.objectContaining({ + issue: expect.objectContaining({ key: 'AR-72' }), + reason: 'relayfile overloaded (oldest_inflight_age)', + })) + // The dispatch got far enough to spawn and to persist the handoffs... + expect(factory.status().counters.dispatchFailureReaperHandoffs).toBe(1) + // ...so the pass must have released them before returning. + expect(fleet.releases.map((release) => release.name).sort()).toEqual([ + 'ar-72-impl-pear', + 'ar-72-review', + ]) + expect(fleet.releases.every((release) => release.reason === 'dispatch failed')).toBe(true) + }) + + // Review follow-up on #298 (P2, found by cubic and coderabbit). Every other + // surface in this PR routes the reason through the allowlist; + // `readinessReconcile.lastError` did not — and it is returned from + // `status()` AND written to the loop heartbeat file on disk, so a raw + // dependency-controlled string reaches an operator-facing artifact. + it('allowlists the 429 reason before it reaches readinessReconcile.lastError', async () => { + const hostileReason = 'not_a_reason\n' + + class OverloadedClaimStateStore extends InMemoryStateStore { + override async claimDiscoverySweep(): Promise { + throw Object.assign(new Error(OVERLOAD_MESSAGE), { + status: 429, + details: { reason: hostileReason, retryAfterSeconds: 5 }, + }) + } + } + + const mount = new CountingEventsMount() + mount.setSubRoot('/linear/issues', 'absent') + const root = await mkdtemp(join(tmpdir(), 'factory-overload-reason-allowlist-')) + const heartbeatPath = join(root, 'heartbeat.json') + const factory = createFactory(config({ + issueSource: 'github', + loop: { heartbeatPath, registryPath: join(root, 'registry.json') }, + }), { + mount, + fleet: new FakeFleetClient(), + stateStore: new OverloadedClaimStateStore({ batchSize: 2 }), + triage: new StaticTriage(), + githubWriteback: new RecordingGithubWriteback(), + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + }) + + await factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 50 }, + }) + try { + await vi.waitFor(() => { + const lastError = factory.status().readinessReconcile?.lastError + expect(lastError).toBeDefined() + // The operator still learns it was a 429 and how long to wait... + expect(lastError).toContain('[relayfile 429 unrecognized; retry-after=5s]') + // ...without the dependency choosing what lands in status(). + expect(lastError).not.toContain('