diff --git a/src/config/schema.ts b/src/config/schema.ts index 890ebe1f..6e3370d9 100644 --- a/src/config/schema.ts +++ b/src/config/schema.ts @@ -61,6 +61,39 @@ const subscriptionSchema = z.object({ */ export const DEFAULT_READINESS_RECONCILE_TIMEOUT_MS = 90 * 60_000 +/** + * Aggregate budget for one discovery sweep (#372), when nothing narrows it. + * + * Equal to `DEFAULT_READINESS_RECONCILE_TIMEOUT_MS` on purpose — and a config + * that omits `sweepBudgetMs` tracks whatever `reconcileTimeoutMs` it set, not + * this constant (see `resolvedSweepBudgetMs`). The two numbers describe the + * same envelope; what changes is the MECHANISM, and the mechanism is the + * deliverable. `reconcileTimeoutMs` rejects the caller's wait + * and leaves `runOnce()` running, so every later cycle coalesces onto the + * wedged pass and the daemon never recovers. The sweep budget rejects from + * inside the fence, so the lease is released and the next cycle starts clean. + * + * Shipping it equal keeps this change free of new timing risk: no sweep that + * survives today is killed by it. Tightening it is a separate, evidence-driven + * decision with a real cost — the sweep commits its checkpoint only at the end, + * so a budget below realistic cold-mirror hydration (#36 measured 61 minutes in + * production) converts a slow boot into a loop that never makes progress, + * which is the same trap `reconcileTimeoutMs` documents above. That is why this + * is a config dial and not a constant. + */ +export const DEFAULT_DISCOVERY_SWEEP_BUDGET_MS = DEFAULT_READINESS_RECONCILE_TIMEOUT_MS + +/** + * The effective budget for a config, given what it did or did not set. + * + * Exported so the orchestrator's field initializer and the schema agree on one + * rule rather than two that happen to match today. + */ +export const resolvedSweepBudgetMs = ( + sweepBudgetMs: number | undefined, + reconcileTimeoutMs: number, +): number => Math.min(sweepBudgetMs ?? reconcileTimeoutMs, reconcileTimeoutMs) + const liveSubscriptionSchema = z.object({ transport: z.enum(['subscribe-and-poll', 'subscribe', 'poll']).default('subscribe-and-poll'), pollIntervalMs: z.number().int().min(50).default(5_000), @@ -84,6 +117,18 @@ const liveSubscriptionSchema = z.object({ */ relayfileOperationTimeoutMs: z.number().int().min(50).max(60 * 60_000) .default(DEFAULT_RELAYFILE_OPERATION_TIMEOUT_MS), + /** + * Bounds the WHOLE sweep (#372). + * + * Distinct from both neighbours above, and the only one of the three that is + * agnostic to which dependency hangs. `relayfileOperationTimeoutMs` bounds one + * relayfile call and cannot see a retry loop around it or a call on another + * transport; `reconcileTimeoutMs` bounds the caller's wait and leaves the + * sweep running underneath it. This one is charged against a single timer for + * the entire pass, so a sweep cannot outlive it however many calls, retries + * or transports it is spread across. + */ + sweepBudgetMs: z.number().int().min(50).max(6 * 60 * 60_000).optional(), }).superRefine((value, ctx) => { // A deadline below the interval kills every pass that takes longer than one // tick, which is most of them on a cold mirror. @@ -94,7 +139,27 @@ const liveSubscriptionSchema = z.object({ message: `reconcileTimeoutMs (${value.reconcileTimeoutMs}) must be at least reconcileIntervalMs (${value.reconcileIntervalMs})`, }) } -}).default({}) + // The sweep budget has to be the tighter of the two, or the wait gives up + // first and the sweep it abandoned keeps running for the next cycle to + // coalesce onto — the exact behaviour the budget exists to remove. Checked + // only when it was set explicitly: an omitted one is derived below and + // cannot violate this. + if (value.sweepBudgetMs !== undefined && value.sweepBudgetMs > value.reconcileTimeoutMs) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + path: ['sweepBudgetMs'], + message: `sweepBudgetMs (${value.sweepBudgetMs}) must not exceed reconcileTimeoutMs (${value.reconcileTimeoutMs})`, + }) + } +}).transform((value) => ({ + ...value, + // Derived from the SIBLING, never from a constant. A fixed 90-minute default + // would reject every config that already tightened `reconcileTimeoutMs` + // below it — the schema throws, so Factory would not start — and would + // silently cap every config that loosened it above. "Omitted" means "the + // same envelope as the wait", whatever that wait is configured to be. + sweepBudgetMs: value.sweepBudgetMs ?? value.reconcileTimeoutMs, +})).default({}) export const DEFAULT_AGENT_HOLD_TIMEOUT_MS = 4 * 60 * 60_000 diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 97d3c411..f54c7113 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -15523,6 +15523,20 @@ describe('FactoryLoop', () => { // took neither: the subsystem reported `healthy` with zero failures while // dispatching nothing, and only a process restart recovered it. describe('bounded readiness reconciliation', () => { + /** + * Selects the pre-#372 backstop these assertions are about. + * + * The aggregate sweep budget aborts a wedged sweep at or before the + * readiness deadline, so at its default there IS no abandoned-but-still- + * running sweep left for this block to observe — which is the fix, covered + * in `sweep-budget.test.ts`. What survives underneath it is what this block + * has always tested: the #296/#301 abandoned-wait accounting, which is + * still the behaviour when the budget is disabled, and still the shape a + * sweep degrades to if a future teardown path cannot be abandoned. `0` + * disables the budget, the same idiom #368 used for its own control. + */ + const NO_SWEEP_BUDGET = { sweepBudgetMs: 0 } as const + class HangingDiscoveryStateStore extends InMemoryStateStore { hangClaims = false readonly hangStarted: Promise @@ -15576,7 +15590,7 @@ describe('FactoryLoop', () => { await factory.start({ mode: 'live', - liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 50, reconcileTimeoutMs: 300 }, + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 50, reconcileTimeoutMs: 300, ...NO_SWEEP_BUDGET }, }) try { stateStore.hangClaims = true @@ -16134,7 +16148,7 @@ describe('FactoryLoop', () => { await factory.start({ mode: 'live', - liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 300, reconcileTimeoutMs: 300 }, + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 300, reconcileTimeoutMs: 300, ...NO_SWEEP_BUDGET }, }) let stopped = false let stopping: Promise | undefined @@ -16182,7 +16196,7 @@ describe('FactoryLoop', () => { // possible by carrying the abandoned sweep's own start time. await factory.start({ mode: 'live', - liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 400, reconcileTimeoutMs: 1_000 }, + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 400, reconcileTimeoutMs: 1_000, ...NO_SWEEP_BUDGET }, }) try { stateStore.hangClaims = true @@ -16221,7 +16235,7 @@ describe('FactoryLoop', () => { await factory.start({ mode: 'live', - liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 400, reconcileTimeoutMs: 400 }, + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 400, reconcileTimeoutMs: 400, ...NO_SWEEP_BUDGET }, }) try { stateStore.hangClaims = true @@ -16262,7 +16276,7 @@ describe('FactoryLoop', () => { await factory.start({ mode: 'live', - liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 300, reconcileTimeoutMs: 300 }, + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 300, reconcileTimeoutMs: 300, ...NO_SWEEP_BUDGET }, }) try { stateStore.hangClaims = true @@ -16307,7 +16321,7 @@ describe('FactoryLoop', () => { await factory.start({ mode: 'live', - liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 300, reconcileTimeoutMs: 300 }, + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 300, reconcileTimeoutMs: 300, ...NO_SWEEP_BUDGET }, }) let stopped = false let stopping: Promise | undefined @@ -16370,7 +16384,7 @@ describe('FactoryLoop', () => { await factory.start({ mode: 'live', - liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 100, reconcileTimeoutMs: 200 }, + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 100, reconcileTimeoutMs: 200, ...NO_SWEEP_BUDGET }, }) try { stateStore.hangClaims = true diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index 73150a74..10f205b6 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -3,7 +3,13 @@ import { randomUUID } from 'node:crypto' import { readFile } from 'node:fs/promises' import { dirname, isAbsolute, resolve } from 'node:path' -import { DEFAULT_READINESS_RECONCILE_TIMEOUT_MS, FactoryConfigSchema, type FactoryConfig } from '../config/schema' +import { + DEFAULT_DISCOVERY_SWEEP_BUDGET_MS, + DEFAULT_READINESS_RECONCILE_TIMEOUT_MS, + FactoryConfigSchema, + resolvedSweepBudgetMs, + type FactoryConfig, +} from '../config/schema' import { DEFAULT_RELAYFILE_OPERATION_TIMEOUT_MS, RelayfileOperationTimeoutError, @@ -65,6 +71,13 @@ import type { Clock, Logger } from '../ports/system' import type { AgentWorktree, AgentWorktreeManager, AgentWorktreeRepository } from '../ports/worktree' import { factoryWorktreeIssueSlug, factoryWorktreePath } from '../git/agent-worktree' import { InMemoryStateStore } from '../state/in-memory-state-store' +import { + DISCOVERY_SWEEP_TEARDOWN_TIMEOUT_MS, + DiscoverySweepBudgetExceededError, + startDiscoverySweepBudget, + withSweepTeardownDeadline, + type DiscoverySweepBudget, +} from './sweep-budget' import { dispatchHandedOffToBabysitters, dispatchLifecycleOccupiesSlot, @@ -877,6 +890,31 @@ export class FactoryLoop implements Factory { * returns: a deadline checked between awaits never regains control to check. */ #relayfileOperationTimeoutMs = DEFAULT_RELAYFILE_OPERATION_TIMEOUT_MS + /** + * Aggregate budget for ONE sweep (#372). + * + * The third bound on this path and the only transport-agnostic one. + * `#relayfileOperationTimeoutMs` bounds one relayfile call and cannot see the + * retry loop around it; `#readinessReconcileTimeoutMs` bounds the *wait* and + * leaves `runOnce()` running for the next cycle to coalesce onto. This is + * charged against one timer for the whole pass, so however many calls, + * retries and transports the sweep is spread across, it cannot outlive this. + * + * Held at `min(configured, #readinessReconcileTimeoutMs)`: a budget looser + * than the wait would let the wait give up first, which is the behaviour it + * exists to remove. + */ + #discoverySweepBudgetMs = DEFAULT_DISCOVERY_SWEEP_BUDGET_MS + /** + * The budgets of every sweep currently in flight. + * + * `stop()` drains the sweep it started (see `#readinessReconcileAbandonedWait` + * below), so a wedged sweep holds shutdown open for the whole budget — 90 + * minutes at the default, and unbounded before this budget existed. A set + * rather than one field because a live sweep and a mismatched dry-run one can + * be in flight at the same time. + */ + readonly #discoverySweepBudgets = new Set() // Set for exactly as long as a sweep is running. `state` is derived from // this, so an in-flight pass can no longer masquerade as the last settled one. #readinessReconcileInFlightSinceMs?: number @@ -1121,6 +1159,10 @@ export class FactoryLoop implements Factory { // Also read here, not only in `#startLiveSubscription`: a standalone // `runOnce()` never starts the live subscription and must still be bounded. this.#relayfileOperationTimeoutMs = config.liveSubscription.relayfileOperationTimeoutMs + this.#discoverySweepBudgetMs = resolvedSweepBudgetMs( + config.liveSubscription.sweepBudgetMs, + config.liveSubscription.reconcileTimeoutMs, + ) this.#mount = ports.mount // Resolved role<->state mapping. The CLI injects a name-resolved, per-team // resolution via ports; fall back to one built from explicit stateIds plus @@ -1541,27 +1583,45 @@ export class FactoryLoop implements Factory { if (this.#heldAgentDeadlineTimer) clearTimeout(this.#heldAgentDeadlineTimer) this.#heldAgentDeadlineTimer = undefined this.#heldAgentDeadlineDueAtMs = undefined - await this.#heldAgentDeadlineSweepInFlight - for (const timer of this.#dispatchLifecycleRetryTimers.values()) clearTimeout(timer) - this.#dispatchLifecycleRetryTimers.clear() - this.#abandonedDispatchReasons.clear() - this.#dispatchLifecycleCapacityWaits.clear() - this.#dispatchLifecycleOwnershipWaitLogged.clear() - if (this.#completionSweepTimer) clearTimeout(this.#completionSweepTimer) - this.#completionSweepTimer = undefined - if (this.#readinessReconcileTimer) clearTimeout(this.#readinessReconcileTimer) - this.#readinessReconcileTimer = undefined - if (this.#previewSweepTimer) clearTimeout(this.#previewSweepTimer) - this.#previewSweepTimer = undefined - await this.#readinessReconcileInFlight - // #301 review: the deadline ends the *wait*, so `#readinessReconcileInFlight` - // can settle with its `runOnce()` still live. Shutdown releases dispatch - // lifecycle leases and disposes ports below, and `#isPassFatalFailure` only - // fences a stopping sweep once something in it throws — so a sweep whose - // dependency recovers cleanly would otherwise dispatch through torn-down - // state. Draining here restores exactly the pre-deadline shutdown contract: - // stop() outlives the sweep it started. - await this.#readinessReconcileAbandonedWait + // #372: the drain below is unbounded in the one case that matters — a + // wedged sweep — so shutdown inherits the sweep budget, 90 minutes at the + // default. Spending it after one teardown window bounds shutdown without + // discarding a sweep that was about to finish. + // + // Armed BEFORE the first shutdown await, not just before the sweep drain + // (cubic-dev-ai, #374 review). The grace is a timer, so arming it costs + // nothing and starts the clock at the moment shutdown starts; arming it + // after `#heldAgentDeadlineSweepInFlight` made the lever's start depend on + // an UNRELATED in-flight sweep finishing first, so a slow held-agent pass + // simply added its own latency to the wedged discovery sweep's reprieve — + // in the limit the lever never arms at all, which is the bound this whole + // change exists to provide. + const releaseSweepBudgetGrace = this.#cutSweepBudgetsShortForStop() + try { + await this.#heldAgentDeadlineSweepInFlight + for (const timer of this.#dispatchLifecycleRetryTimers.values()) clearTimeout(timer) + this.#dispatchLifecycleRetryTimers.clear() + this.#abandonedDispatchReasons.clear() + this.#dispatchLifecycleCapacityWaits.clear() + this.#dispatchLifecycleOwnershipWaitLogged.clear() + if (this.#completionSweepTimer) clearTimeout(this.#completionSweepTimer) + this.#completionSweepTimer = undefined + if (this.#readinessReconcileTimer) clearTimeout(this.#readinessReconcileTimer) + this.#readinessReconcileTimer = undefined + if (this.#previewSweepTimer) clearTimeout(this.#previewSweepTimer) + this.#previewSweepTimer = undefined + await this.#readinessReconcileInFlight + // #301 review: the deadline ends the *wait*, so `#readinessReconcileInFlight` + // can settle with its `runOnce()` still live. Shutdown releases dispatch + // lifecycle leases and disposes ports below, and `#isPassFatalFailure` only + // fences a stopping sweep once something in it throws — so a sweep whose + // dependency recovers cleanly would otherwise dispatch through torn-down + // state. Draining here restores exactly the pre-deadline shutdown contract: + // stop() outlives the sweep it started. + await this.#readinessReconcileAbandonedWait + } finally { + releaseSweepBudgetGrace() + } await this.#previewSweepInFlight this.#stoppingHeartbeatRefreshActive = await this.#stopLiveHeartbeat('stopping') try { @@ -1963,6 +2023,13 @@ export class FactoryLoop implements Factory { // floor here: a deadline under one interval would kill every pass. this.#readinessReconcileTimeoutMs = Math.max(options.reconcileTimeoutMs, options.reconcileIntervalMs) this.#relayfileOperationTimeoutMs = options.relayfileOperationTimeoutMs + // Same re-application as the deadline above: `start()` overrides bypass the + // schema's cross-field check, and a budget looser than the wait is not a + // budget. + this.#discoverySweepBudgetMs = resolvedSweepBudgetMs( + options.sweepBudgetMs, + this.#readinessReconcileTimeoutMs, + ) this.#liveConnectStartedAtMs = this.#clock.now() this.#liveReplaySkewMarginMs = options.replaySkewMarginMs const highWatermark = await this.#currentEventHighWatermark() @@ -2137,6 +2204,7 @@ export class FactoryLoop implements Factory { reconcileTimeoutMs: overrides.reconcileTimeoutMs ?? this.#config.liveSubscription.reconcileTimeoutMs, relayfileOperationTimeoutMs: overrides.relayfileOperationTimeoutMs ?? this.#config.liveSubscription.relayfileOperationTimeoutMs, + sweepBudgetMs: overrides.sweepBudgetMs ?? this.#config.liveSubscription.sweepBudgetMs, } } @@ -3028,17 +3096,141 @@ export class FactoryLoop implements Factory { } } + /** + * One sweep, under one aggregate budget (#372). + * + * The budget wraps the fence rather than the fence's caller, and that is the + * whole point. #296's deadline lives in `#runOnceWithReadinessDeadline`, + * outside `runOnce()`: expiry rejects the wait and leaves the sweep running, + * so the next cycle coalesces onto the same wedged promise (`runOnce()`, the + * `#runOnceInFlight` branch) and the daemon never recovers. Expiring in HERE + * unwinds the body below, which releases the discovery lease on its way out + * and lets `runOnce()` clear `#runOnceInFlight` — so the next cycle claims a + * fresh lease and runs clean. + * + * See `sweep-budget.ts` for what the mechanism can and cannot interrupt. In + * short: it abandons the in-flight await, it does not cancel the call. + */ async #runOnceWithDiscoveryFence(opts: { dryRun?: boolean }): Promise { + const budget = startDiscoverySweepBudget(this.#discoverySweepBudgetMs) + this.#discoverySweepBudgets.add(budget) + const sweepStartedAtMs = this.#clock.now() + // A sweep that starts while shutdown is already draining would otherwise + // get a full fresh budget to hold `stop()` open with. + if (this.#stopping) budget.expire() + try { + return await this.#runDiscoverySweep(opts, budget) + } catch (error) { + if (error instanceof DiscoverySweepBudgetExceededError) { + this.#increment('discoverySweepBudgetExceeded') + this.#logger.error?.('[factory] discovery sweep aborted at its aggregate budget', { + budgetMs: error.budgetMs, + // The await the sweep was abandoned on. The one diagnostic no + // per-call bound can produce once the sweep is already wedged: it + // says WHICH transport this wedge is on without anyone having to + // guess which layer to bound next. + phase: error.phase, + elapsedMs: this.#elapsedSince(sweepStartedAtMs), + }) + } + throw error + } finally { + this.#discoverySweepBudgets.delete(budget) + budget.dispose() + } + } + + /** + * Cut every in-flight sweep's budget short once shutdown starts draining. + * + * `stop()` deliberately outlives the sweep it started (#301), so a wedged + * sweep makes shutdown as long as the budget. The grace window is what keeps + * an ordinary restart from throwing away a sweep that was about to commit; + * after it, expiry routes the sweep into exactly the abort path a real expiry + * takes — lease released, teardown bounded — instead of holding the process. + * + * Returns the cancel for the grace timer. Always call it: on a prompt + * shutdown there is nothing to cut short. + */ + #cutSweepBudgetsShortForStop(): () => void { + if (this.#discoverySweepBudgets.size === 0) return () => undefined + const timer = setTimeout(() => { + for (const budget of this.#discoverySweepBudgets) { + if (budget.expired()) continue + this.#increment('discoverySweepBudgetsCutShortForStop') + this.#logger.warn?.('[factory] shutdown is draining a sweep; spending its budget now', { + graceMs: STOP_TEARDOWN_TIMEOUT_MS, + budgetMs: budget.budgetMs, + }) + budget.expire() + } + }, STOP_TEARDOWN_TIMEOUT_MS) + timer.unref?.() + return () => clearTimeout(timer) + } + + /** + * Claim the sweep lease under the budget, compensating for a claim that + * lands after we stopped waiting for it. + * + * The budget abandons the wait, not the call, so the store can still persist + * a lease for a claim this sweep has already given up on — and that lease + * would be held by an owner that will never renew, commit or release it, so + * every later sweep defers until it expires. It is self-expiring and a later + * sweep reclaims it as an orphan, which makes this a latency fix rather than + * a correctness one; the latency is one whole lease window with no discovery, + * which is the thing this PR exists to stop paying. + */ + async #claimDiscoverySweepUnderBudget(budget: DiscoverySweepBudget): Promise { + // Issued INSIDE the budget callback, so a budget that is already spent + // rejects the phase without opening a lease it could only hand straight + // back. The handle is kept out here because the compensation below needs + // the promise the wait was abandoned on. + let claim: Promise | undefined + try { + return await budget.run('discovery-lease-claim', () => { + claim = this.#state.claimDiscoverySweep( + this.#workspaceId, + this.#discoverySweepOwner, + this.#clock.now(), + DISCOVERY_SWEEP_LEASE_MS, + ) + return claim + }) + } catch (error) { + if (!(error instanceof DiscoverySweepBudgetExceededError) || claim === undefined) throw error + void claim.then( + async (late) => { + if (!late.acquired || !late.lease) return + this.#increment('discoverySweepStrandedClaimsReleased') + this.#logger.warn?.('[factory] releasing a discovery lease that was claimed after the sweep budget expired', { + epoch: late.lease.epoch, + budgetMs: error.budgetMs, + }) + await this.#sweepTeardownStep('stranded discovery lease release', () => this.#state.releaseDiscoverySweep( + this.#workspaceId, + this.#discoverySweepOwner, + late.lease!.epoch, + )) + }, + () => undefined, + ).catch(() => undefined) + throw error + } + } + + async #runDiscoverySweep( + opts: { dryRun?: boolean }, + budget: DiscoverySweepBudget, + ): Promise { const sweepStartedAtMs = this.#clock.now() if (!(opts.dryRun ?? this.#config.dryRun)) { - await this.#assertFleetControlPlaneAvailable() + // Under the budget, and first, because this is where the 2026-08-25 + // 07:52:59Z wedge sat: a pre-claim probe on a transport neither #351 nor + // #368 covers. The budget does not care which one it is. + await budget.run('fleet-control-plane-probe', () => this.#assertFleetControlPlaneAvailable()) } - let claim = await this.#state.claimDiscoverySweep( - this.#workspaceId, - this.#discoverySweepOwner, - this.#clock.now(), - DISCOVERY_SWEEP_LEASE_MS, - ) + let claim = await this.#claimDiscoverySweepUnderBudget(budget) if (!claim.acquired && claim.reason === 'backoff') { const delayMs = Math.max(0, claim.state.backoffUntilMs - this.#clock.now()) this.#increment('discoveryBackoffWaits') @@ -3047,13 +3239,8 @@ export class FactoryLoop implements Factory { backoffUntilMs: claim.state.backoffUntilMs, consecutiveOverloads: claim.state.consecutiveOverloads, }) - await this.#clock.sleep(delayMs) - claim = await this.#state.claimDiscoverySweep( - this.#workspaceId, - this.#discoverySweepOwner, - this.#clock.now(), - DISCOVERY_SWEEP_LEASE_MS, - ) + await budget.run('discovery-backoff-wait', () => this.#clock.sleep(delayMs)) + claim = await this.#claimDiscoverySweepUnderBudget(budget) } if (!claim.acquired || !claim.lease) { this.#increment('discoverySweepsSkippedInFlight') @@ -3099,13 +3286,16 @@ export class FactoryLoop implements Factory { this.#startDiscoverySweepRenewal(claim.lease.epoch) let leaseReleased = false try { - this.#discoverySession = await this.#prepareDiscoverySession(claim) + this.#discoverySession = await budget.run( + 'discovery-session', + () => 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) + const report = await budget.run('run-once', () => this.#performRunOnce(opts, budget)) // 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 @@ -3116,17 +3306,23 @@ export class FactoryLoop implements Factory { if (this.#discoveryOverloadError !== undefined && !this.#discoverySweepProgress) { throw this.#discoveryOverloadError } - const checkpoint = await this.#finalizeDiscoveryCheckpoint() + const checkpoint = await budget.run('discovery-checkpoint', () => 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 // (lease-less) checkpoint is a false lease-loss signal and can poison an // otherwise successful reconcile cycle. - await this.#stopDiscoverySweepRenewal() + await budget.run('discovery-renewal-stop', () => this.#stopDiscoverySweepRenewal()) if (this.#discoverySweepLeaseLost) { throw new Error('discovery sweep lease was lost before checkpoint commit') } const residual = this.#discoveryOverloadOutcome(claim.state.consecutiveOverloads, 'committed') - const completed = await this.#commitDiscoverySweep(claim.lease.epoch, checkpoint, residual) + // Under the budget too. A hung commit is the same class of wedge as a + // hung read, and the epoch guard in the store makes a late one a no-op: + // the teardown below has already released this epoch's lease. + const completed = await budget.run( + 'discovery-commit', + () => this.#commitDiscoverySweep(claim.lease!.epoch, checkpoint, residual), + ) leaseReleased = completed if (!completed) throw new Error('discovery sweep lease was lost before completion') // This is the progress boundary consumed by deployment health. A timer @@ -3141,17 +3337,23 @@ export class FactoryLoop implements Factory { }) return report } catch (error) { - await this.#stopDiscoverySweepRenewal() + await this.#sweepTeardownStep('discovery sweep renewal stop', () => this.#stopDiscoverySweepRenewal()) const overload = relayfileOverload(error) if (overload) { const outcome = this.#discoveryOverloadOutcome(claim.state.consecutiveOverloads, 'aborted', error)! - leaseReleased = await this.#state.deferDiscoverySweep( - this.#workspaceId, - this.#discoverySweepOwner, - claim.lease.epoch, - outcome.backoffUntilMs, - outcome.consecutiveOverloads, - ) + // Bounded for the same reason the release below is: this is the other + // path that hands the lease back, and an unbounded one would hold + // `#runOnceInFlight` open past the budget that just expired. + leaseReleased = await withSweepTeardownDeadline( + DISCOVERY_SWEEP_TEARDOWN_TIMEOUT_MS, + () => this.#state.deferDiscoverySweep( + this.#workspaceId, + this.#discoverySweepOwner, + claim.lease!.epoch, + outcome.backoffUntilMs, + outcome.consecutiveOverloads, + ), + ) ?? false this.#logger.warn?.('[factory] Relayfile discovery overloaded; backing off before another sweep', { status: overload.status, reason: overload.reason, @@ -3171,7 +3373,7 @@ export class FactoryLoop implements Factory { } throw error } finally { - await this.#stopDiscoverySweepRenewal() + await this.#sweepTeardownStep('discovery sweep renewal stop', () => this.#stopDiscoverySweepRenewal()) this.#discoverySession = undefined this.#discoverySweepEpoch = undefined this.#discoverySweepStartedAtMs = undefined @@ -3188,15 +3390,48 @@ export class FactoryLoop implements Factory { // happens to run and reset it at the top of this method. this.#discoverySweepLeaseLost = false if (!leaseReleased) { - await this.#state.releaseDiscoverySweep( + // The half of the budget that makes the NEXT cycle clean, so it gets + // its own deadline rather than the spent one: an unbounded release + // would re-create the wedge one layer down, which is the pattern this + // change exists to end. An abandoned release is survivable — the + // durable lease carries its own expiry and a later sweep reclaims it + // as an orphan (`claim.reclaimedLease` above). + await this.#sweepTeardownStep('discovery sweep lease release', () => this.#state.releaseDiscoverySweep( this.#workspaceId, this.#discoverySweepOwner, - claim.lease.epoch, - ) + claim.lease!.epoch, + )) } } } + /** + * One sweep-teardown step, under its own deadline. + * + * Teardown cannot run under the sweep's aggregate budget: on the path that + * matters the budget is already spent, so every step would reject and the + * lease would never be released. It gets a short independent deadline + * instead. Abandoning it costs an orphaned lease for one expiry window; + * NOT bounding it costs the whole invariant, because a hung release holds + * `#runOnceInFlight` open and every later cycle coalesces onto it. + */ + async #sweepTeardownStep(label: string, step: () => Promise): Promise { + const outcome = await withSweepTeardownDeadline( + DISCOVERY_SWEEP_TEARDOWN_TIMEOUT_MS, + async () => { + await step() + return true as const + }, + ) + if (outcome === undefined) { + this.#increment('discoverySweepTeardownDeadlineExceeded') + this.#logger.warn?.('[factory] discovery sweep teardown step abandoned at its deadline', { + step: label, + timeoutMs: DISCOVERY_SWEEP_TEARDOWN_TIMEOUT_MS, + }) + } + } + /** * Commit the sweep, carrying any residual overload backoff into the store. * @@ -3310,7 +3545,10 @@ export class FactoryLoop implements Factory { } } - async #performRunOnce(opts: { dryRun?: boolean } = {}): Promise { + async #performRunOnce( + opts: { dryRun?: boolean } = {}, + budget?: DiscoverySweepBudget, + ): Promise { const dryRun = opts.dryRun ?? this.#config.dryRun const startedAtMs = this.#clock.now() const relayfileWaitWarningsAtStart = this.#counters.relayfileOperationWaitWarnings ?? 0 @@ -3371,6 +3609,12 @@ export class FactoryLoop implements Factory { const issueEntries: Array<{ path: string; issue?: LinearIssue }> = [] for (const path of paths) { + // A between-await check, worth exactly what #368 said such a check is + // worth against a call that never returns: nothing. What it buys is + // the other half — a pass already abandoned at the budget unwinds at + // its next iteration if it ever regains control, instead of running to + // completion beside the sweep that replaced it. + budget?.assertNotExpired('run-once') let issue: LinearIssue | undefined let shed = false try { @@ -3445,6 +3689,10 @@ export class FactoryLoop implements Factory { } for (const { issue } of issueEntries) { + // The dispatch half of the same fence as the read loop above. Without + // it a pass abandoned during enumeration would go on to dispatch after + // its lease had been handed back, racing the sweep that replaced it. + budget?.assertNotExpired('run-once') await this.#refreshLiveHeartbeatIfDue() if (!issue) { continue @@ -4702,9 +4950,35 @@ export class FactoryLoop implements Factory { return paths ? [...paths] : undefined } + /** + * True when this call is a continuation of a sweep that has already ended. + * + * #372: the aggregate budget abandons the WAIT, not the call, so a read + * issued by an aborted sweep can still resolve — by which time the shared + * `#discoverySession` and `#discoverySweepEpoch` may belong to the sweep + * that replaced it. `discoveryEnumerationPass` is an `AsyncLocalStorage`, so + * its store follows the async continuation and still carries the epoch that + * ISSUED the read; comparing the two is what tells a live write from a + * late one. + * + * Only when a store exists: a caller outside a discovery pass legitimately + * has none, and treating that as stale would silence the live event drain. + */ + #isStaleDiscoveryContinuation(): boolean { + const issuingPass = discoveryEnumerationPass.getStore() + return issuingPass !== undefined && issuingPass.epoch !== this.#discoverySweepEpoch + } + async #rememberDiscoveryTree(prefix: string, paths: string[]): Promise { const session = this.#discoverySession if (!session) return + // Committing this listing would put a tree from an abandoned pass into the + // replacement sweep's checkpoint, under a watermark that claims to describe + // the replacement. That is checkpoint corruption, and it outlives the sweep. + if (this.#isStaleDiscoveryContinuation()) { + this.#increment('discoveryStaleTreeWritesDropped') + return + } const uniquePaths = new Set() for (let index = 0; index < paths.length; index += 1) { uniquePaths.add(paths[index]!) @@ -4800,7 +5074,10 @@ export class FactoryLoop implements Factory { return result } catch (error) { const overload = relayfileOverload(error) - if (overload && this.#discoverySweepEpoch !== undefined) { + // The stale check for the same reason as the tree write above: a 429 that + // arrives after its sweep was abandoned is not this sweep's evidence, and + // attributing it here would drive the replacement's overload ratchet. + if (overload && this.#discoverySweepEpoch !== undefined && !this.#isStaleDiscoveryContinuation()) { this.#discoveryOverloadError ??= error this.#discoverySweepOverloads += 1 if (overload.retryAfterSeconds !== undefined) { diff --git a/src/orchestrator/sweep-budget.test.ts b/src/orchestrator/sweep-budget.test.ts new file mode 100644 index 00000000..71c24da3 --- /dev/null +++ b/src/orchestrator/sweep-budget.test.ts @@ -0,0 +1,810 @@ +import { describe, expect, it } from 'vitest' +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { + DEFAULT_DISCOVERY_SWEEP_BUDGET_MS, + DEFAULT_READINESS_RECONCILE_TIMEOUT_MS, + resolvedSweepBudgetMs, +} from '../config/schema' +import { + FactoryConfigSchema, + createFactory, + type FactoryConfig, + type LinearIssue, + type TriageDecision, + type TriageEngine, +} from '../index' +import { FakeFleetClient, FakeMountClient } from '../testing' +import { withDeadline } from '../testing/deadline' +import { InMemoryStateStore } from '../state/in-memory-state-store' +import type { DiscoverySweepClaim } from '../ports/state' +import { + DISCOVERY_SWEEP_TEARDOWN_TIMEOUT_MS, + DiscoverySweepBudgetExceededError, + startDiscoverySweepBudget, + withSweepTeardownDeadline, +} from './sweep-budget' + +/** + * The aggregate sweep budget (#372). + * + * Three unbounded calls have wedged this sweep in one day, on three different + * transports, each one bounded afterwards and each time the wedge returned one + * layer down. So the subject here is not another call. It is the property that + * a sweep cannot be in flight for longer than its budget REGARDLESS of what it + * is waiting on — which is what turns the next unbounded call into a degraded + * sweep instead of the end of dispatch. + * + * Every must-fire below hangs a call that never returns. Before this change + * each one hangs the test itself, at vitest's 5 s default: that is the defect, + * in the same shape production had it. + */ + +const NEVER = (): Promise => new Promise(() => undefined) + +describe('sweep budget primitive', () => { + it('must-fire: abandons an in-flight await that never returns, naming the phase', async () => { + const budget = startDiscoverySweepBudget(30) + try { + await expect(budget.run('discovery-session', NEVER)).rejects.toThrow(DiscoverySweepBudgetExceededError) + } finally { + budget.dispose() + } + }) + + it('must-fire: charges every phase against ONE clock, so the sum cannot exceed the budget', async () => { + // The property a per-call deadline cannot have. Each call here is well + // inside the budget on its own; only their sum is not. + const budget = startDiscoverySweepBudget(120) + const sleep = (ms: number) => new Promise((resolve) => setTimeout(() => resolve('served'), ms)) + try { + expect(await budget.run('discovery-session', () => sleep(40))).toBe('served') + expect(await budget.run('discovery-checkpoint', () => sleep(40))).toBe('served') + // The third 40 ms call is identical to the two that were served. It is + // rejected because the SWEEP is out of time, not because it is slow. + await expect(budget.run('discovery-commit', () => sleep(80))).rejects.toMatchObject({ + name: 'DiscoverySweepBudgetExceededError', + phase: 'discovery-commit', + budgetMs: 120, + }) + } finally { + budget.dispose() + } + }) + + it('must-fire: a bounded call inside an unbounded retry loop still ends at the budget (L3)', async () => { + // The deployed 0.1.75 shape: #368's per-call deadline fires every time, and + // the retry around it makes the total unbounded anyway. `attempts` is + // deliberately unbounded — the budget is the only thing that stops it. + const budget = startDiscoverySweepBudget(80) + let attempts = 0 + const alwaysRejects = async (): Promise => { + attempts += 1 + await new Promise((resolve) => setTimeout(resolve, 5)) + throw new Error('relayfile getEvents did not respond within 5ms') + } + const retryForever = async (): Promise => { + for (;;) { + try { + return await budget.run('run-once', alwaysRejects) + } catch (error) { + if (error instanceof DiscoverySweepBudgetExceededError) throw error + // Exactly what L3 does: swallow the per-call deadline and go again. + } + } + } + try { + await expect(withDeadline(retryForever(), 4_000, 'retry loop never ended')).rejects + .toThrow(DiscoverySweepBudgetExceededError) + expect(attempts).toBeGreaterThan(1) + } finally { + budget.dispose() + } + }) + + it('must-fire: aborts its signal at expiry, so anything that honours one is really cancelled', async () => { + const budget = startDiscoverySweepBudget(25) + try { + expect(budget.signal.aborted).toBe(false) + await expect(budget.run('run-once', NEVER)).rejects.toThrow(DiscoverySweepBudgetExceededError) + expect(budget.signal.aborted).toBe(true) + expect(budget.signal.reason).toBeInstanceOf(DiscoverySweepBudgetExceededError) + } finally { + budget.dispose() + } + }) + + it('must-fire: refuses to start new work once the budget is spent', async () => { + const budget = startDiscoverySweepBudget(25) + let started = false + try { + await expect(budget.run('run-once', NEVER)).rejects.toThrow(DiscoverySweepBudgetExceededError) + await expect(budget.run('discovery-commit', async () => { + started = true + return 'served' + })).rejects.toThrow(DiscoverySweepBudgetExceededError) + // A spent budget must not issue another request against the dependency + // it just gave up on. + expect(started).toBe(false) + expect(() => budget.assertNotExpired('run-once')).toThrow(DiscoverySweepBudgetExceededError) + } finally { + budget.dispose() + } + }) + + it('must-not-fire: work that finishes inside the budget returns unchanged', async () => { + const budget = startDiscoverySweepBudget(2_000) + try { + const report = { dispatched: ['ar-1'], candidates: 2 } + expect(await budget.run('run-once', async () => report)).toBe(report) + expect(budget.expired()).toBe(false) + expect(budget.signal.aborted).toBe(false) + expect(() => budget.assertNotExpired('run-once')).not.toThrow() + // The caller's own failure still surfaces as itself, never re-clothed as + // a budget expiry — the two have completely different remedies. + const failure = new Error('dispatch failed') + await expect(budget.run('run-once', async () => { throw failure })).rejects.toBe(failure) + } finally { + budget.dispose() + } + }) + + it('must-not-fire: the budget timer holds the event loop, and stops holding it the moment the sweep settles', async () => { + // The pair for the shutdown lever below. The trivially wrong way to stop a + // deadline timer holding the process open is to `unref()` it — which also + // lets Node exit before the budget fires, so a one-shot `runOnce()` returns + // having neither reported the wedge nor released the lease. This assertion + // fails for that version: `getActiveResourcesInfo()` lists only resources + // that are KEEPING THE EVENT LOOP ALIVE, so an unref'd timer never appears. + const timers = () => process.getActiveResourcesInfo().filter((resource) => resource === 'Timeout').length + const before = timers() + const budget = startDiscoverySweepBudget(90 * 60_000) + // `dispose()` has to run on the failing path too, not just the passing one. + // This is the ONE test in the file that deliberately arms the real 90-minute + // timer, so an assertion that throws before an unguarded `dispose()` would + // leave a *referenced* Timeout in the worker — the exact resource the next + // line proves keeps the event loop alive. The failure would then present as + // a hung suite instead of a named assertion, which is the worst way to + // report it (cubic-dev-ai, #374 review). + try { + expect(timers()).toBe(before + 1) + expect(await budget.run('run-once', async () => 'served')).toBe('served') + } finally { + budget.dispose() + } + // ...and the other half: a settled sweep leaves nothing behind, which is + // what makes keeping it referenced affordable at a 90-minute budget. + expect(timers()).toBe(before) + }) + + it('must-not-fire: with no budget the same hung call stays pending, so the rejections above are the budget', async () => { + // The control for every must-fire above. Without it a wrapper that + // rejected unconditionally would pass all of them. + const budget = startDiscoverySweepBudget(0) + let settled = false + try { + const pending = budget.run('run-once', NEVER).then( + () => { settled = true }, + () => { settled = true }, + ) + await Promise.race([pending, new Promise((resolve) => setTimeout(resolve, 200))]) + expect(settled).toBe(false) + expect(budget.expired()).toBe(false) + expect(budget.budgetMs).toBeUndefined() + } finally { + budget.dispose() + } + }) +}) + +describe('sweep teardown deadline', () => { + it('must-fire: abandons a teardown step that never returns, rather than holding the sweep open', async () => { + expect(await withDeadline( + withSweepTeardownDeadline(25, NEVER), + 4_000, + 'teardown deadline never fired', + )).toBeUndefined() + }) + + it('must-not-fire: a served teardown returns its value, and a failing one still throws', async () => { + expect(await withSweepTeardownDeadline(2_000, async () => true)).toBe(true) + const failure = new Error('release rejected') + await expect(withSweepTeardownDeadline(2_000, async () => { throw failure })).rejects.toBe(failure) + // A teardown deadline is not a licence to swallow failures that used to + // surface, and it is short enough that it cannot be the thing that wedges. + expect(DISCOVERY_SWEEP_TEARDOWN_TIMEOUT_MS).toBeLessThan(60_000) + }) +}) + +describe('the configured budget', () => { + const live = (liveSubscription: Record) => FactoryConfigSchema.parse({ + workspaceId: 'factory-sweep-budget-config', + repos: { + byLabel: { pear: 'AgentWorkforce/pear' }, + clonePaths: { 'AgentWorkforce/pear': '/work/pear' }, + default: 'AgentWorkforce/pear', + }, + stateIds: { readyForAgent: ready, agentImplementing: implementing, done, inPlanning: planning }, + liveSubscription, + }).liveSubscription + + it('must-not-fire: a config that already tightened reconcileTimeoutMs still parses', () => { + // The regression that would have taken Factory down on deploy rather than + // fixing it: a fixed 90-minute default is ABOVE such a config's timeout, so + // the cross-field check rejects it and `FactoryConfigSchema.parse` throws + // before the daemon can start. + expect(live({ reconcileTimeoutMs: 5 * 60_000 }).sweepBudgetMs).toBe(5 * 60_000) + }) + + it('must-not-fire: a config that loosened reconcileTimeoutMs is not silently capped at 90 minutes', () => { + expect(live({ reconcileTimeoutMs: 3 * 60 * 60_000 }).sweepBudgetMs).toBe(3 * 60 * 60_000) + }) + + it('must-not-fire: an omitted budget tracks the default timeout', () => { + expect(live({}).sweepBudgetMs).toBe(DEFAULT_DISCOVERY_SWEEP_BUDGET_MS) + expect(DEFAULT_DISCOVERY_SWEEP_BUDGET_MS).toBe(DEFAULT_READINESS_RECONCILE_TIMEOUT_MS) + }) + + it('must-fire: an explicit budget looser than the wait is still rejected', () => { + // A budget the wait gives up on first is not a budget: the sweep it + // abandoned keeps running and the next cycle coalesces onto it. + expect(() => live({ reconcileTimeoutMs: 60_000, sweepBudgetMs: 120_000 })).toThrow(/sweepBudgetMs/) + expect(live({ reconcileTimeoutMs: 120_000, sweepBudgetMs: 60_000 }).sweepBudgetMs).toBe(60_000) + }) + + it('must-not-fire: resolving is the same rule on both sides of the schema', () => { + expect(resolvedSweepBudgetMs(undefined, 5 * 60_000)).toBe(5 * 60_000) + expect(resolvedSweepBudgetMs(60_000, 5 * 60_000)).toBe(60_000) + // `start()` overrides bypass the schema, so the clamp has to hold here too. + expect(resolvedSweepBudgetMs(10 * 60_000, 5 * 60_000)).toBe(5 * 60_000) + }) +}) + +/* ------------------------------------------------------------------------- */ +/* The same invariant, driven through a real sweep. */ +/* ------------------------------------------------------------------------- */ + +const ready = '11111111-1111-4111-8111-111111111111' +const implementing = '22222222-2222-4222-8222-222222222222' +const done = '33333333-3333-4333-8333-333333333333' +const planning = '44444444-4444-4444-8444-444444444444' + +const config = ( + sweepBudgetMs: number, + registryRoot: string, + dispatch?: { agentHoldTimeoutMs?: number; agentlessHoldTimeoutMs?: number }, +): FactoryConfig => FactoryConfigSchema.parse({ + workspaceId: 'factory-sweep-budget', + repos: { + byLabel: { pear: 'AgentWorkforce/pear' }, + clonePaths: { 'AgentWorkforce/pear': '/work/pear' }, + default: 'AgentWorkforce/pear', + }, + triage: { maxImplementers: 4 }, + batchSize: 4, + stateIds: { readyForAgent: ready, agentImplementing: implementing, done, inPlanning: planning }, + verification: { enabled: false }, + loop: { + registryPath: join(registryRoot, 'registry.json'), + heartbeatPath: join(registryRoot, 'heartbeat.json'), + }, + liveSubscription: { sweepBudgetMs }, + ...(dispatch ? { dispatch } : {}), +}) + +const issuePath = (n: number) => `/linear/issues/AR-${n}__uuid-${n}.json` + +const issueFile = (n: number) => ({ + provider: 'linear', + objectType: 'issue', + objectId: `uuid-${n}`, + payload: { + id: `uuid-${n}`, + identifier: `AR-${n}`, + title: `[factory-e2e] Fix factory issue ${n}`, + description: 'Implement the requested fix in src/orchestrator/factory.ts and verify it with tests.', + stateId: ready, + url: `https://linear.app/agent-relay/issue/AR-${n}/factory-issue-${n}`, + labels: [{ name: 'pear' }], + labelIds: ['label-id-not-used-by-parser'], + team: { key: 'AR', name: 'Agent Relay' }, + project: { name: 'Factory' }, + state: { id: ready, name: 'Ready for Agent' }, + }, +}) + +class StaticTriage implements TriageEngine { + async triage(issue: LinearIssue): Promise { + const number = issue.key.match(/\d+/)?.[0] ?? '0' + return { + issue: { uuid: issue.uuid, key: issue.key, path: issue.path }, + routes: [{ repo: 'AgentWorkforce/pear', clonePath: '/work/pear', rationale: 'test route' }], + scope: 'single', + implementers: [{ + name: `ar-${number}-impl`, + role: 'implementer', + capability: 'spawn:codex', + model: 'codex', + task: `Implement ${issue.key}`, + repo: 'AgentWorkforce/pear', + clonePath: '/work/pear', + node: 'self', + }], + reviewer: { + name: `ar-${number}-review`, + role: 'reviewer', + capability: 'spawn:claude', + model: 'claude', + task: `Review ${issue.key}`, + repo: 'AgentWorkforce/pear', + clonePath: '/work/pear', + node: 'self', + }, + thin: false, + confidence: 'high', + rationale: 'static test decision', + } + } +} + +/** + * A mount whose first watermark read never answers. + * + * `#prepareDiscoverySession` makes this call immediately after the discovery + * lease is claimed, so a sweep that hangs here holds the lease — the position + * the 0.1.74 wedge occupied, and the position every later layer has reoccupied + * on a different transport. + */ +class HangingWatermarkMount extends FakeMountClient { + hang = true + /** + * How many watermark reads to serve before hanging. + * + * `#startLiveSubscription` reads the watermark once itself, BEFORE the + * startup backfill and outside the sweep, so no sweep budget covers it. + * Nothing in the orchestrator bounds it either: `#currentEventHighWatermark` + * (factory.ts:2192) awaits `mount.getEventHighWatermark()` under a bare + * try/catch, not through `#withRelayfileOperation`. What bounds it in + * production is one layer lower — the deployed client's own `#bounded()` + * (relayfile-cloud-mount-client.ts:1057, #368), fed from + * `relayfileOperationTimeoutMs` at cli/fleet.ts:2215 — so a `MountClient` + * without that deadline has none here at all. Serving the read is what puts + * the hang inside the sweep, which is the subject of this file. + */ + serveFirst = 0 + served = 0 + hungCalls = 0 + + override async getEventHighWatermark(opts: { provider?: string } = {}): Promise { + if (this.hang && this.served >= this.serveFirst) { + this.hungCalls += 1 + return await NEVER() + } + this.served += 1 + return await super.getEventHighWatermark(opts) + } +} + +/** A fleet whose control-plane probe is slow enough to spend a tight budget. */ +class SlowRosterFleet extends FakeFleetClient { + rosterDelayMs = 0 + + override async roster(): ReturnType { + if (this.rosterDelayMs > 0) { + await new Promise((resolve) => setTimeout(resolve, this.rosterDelayMs)) + } + return await super.roster() + } +} + +/** + * A fleet that parks the held-agent deadline sweep partway through, so a sweep + * UNRELATED to discovery is genuinely in flight when `stop()` is called. + * + * `#sweepHeldAgentDeadlines` releases the agents it reaps through + * `fleet.release(name, reason)`, so blocking that one reason is enough to hold + * `#heldAgentDeadlineSweepInFlight` open for as long as the test wants, + * without a timer or a sleep deciding the outcome. + */ +class HeldSweepParkingFleet extends FakeFleetClient { + static readonly HELD_PAST_DEADLINE = 'held-past-deadline' + + #open?: () => void + // Once unparked, STAY unparked: a held dispatch releases every one of its + // agents, so a one-shot gate would park the second release forever and the + // test would fail on shutdown rather than on the assertion it exists for. + #opened = false + #entered!: () => void + /** Resolves once the held-agent sweep is parked inside its release. */ + readonly parked: Promise = new Promise((resolve) => { this.#entered = resolve }) + + override async release(name: string, reason?: string): Promise { + if (reason === HeldSweepParkingFleet.HELD_PAST_DEADLINE && !this.#opened) { + this.#entered() + await new Promise((resolve) => { this.#open = resolve }) + } + await super.release(name, reason) + } + + /** Lets the parked held-agent sweep finish, and keeps it free thereafter. */ + unpark(): void { + this.#opened = true + this.#open?.() + } +} + +/** A store that records the lease handbacks, so "released" is observed, not inferred. */ +class LeaseWatchingStateStore extends InMemoryStateStore { + readonly released: number[] = [] + claims = 0 + /** Milliseconds a claim takes to answer. Long enough and the budget gives up first. */ + claimDelayMs = 0 + + override async claimDiscoverySweep( + workspaceId: string, + owner: string, + nowMs: number, + leaseMs: number, + ): Promise { + this.claims += 1 + if (this.claimDelayMs > 0) { + await new Promise((resolve) => setTimeout(resolve, this.claimDelayMs)) + } + return await super.claimDiscoverySweep(workspaceId, owner, nowMs, leaseMs) + } + + override async releaseDiscoverySweep(workspaceId: string, owner: string, epoch: number): Promise { + this.released.push(epoch) + await super.releaseDiscoverySweep(workspaceId, owner, epoch) + } +} + +describe('a wedged sweep is bounded end to end (#372)', () => { + it('must-fire: aborts at the budget, releases the lease, and the NEXT cycle runs clean', async () => { + const root = await mkdtemp(join(tmpdir(), 'factory-sweep-budget-')) + const mount = new HangingWatermarkMount({ + [issuePath(901)]: issueFile(901), + }) + const fleet = new FakeFleetClient() + const stateStore = new LeaseWatchingStateStore({ batchSize: 4 }) + const factory = createFactory(config(400, root), { + mount, + fleet, + stateStore, + triage: new StaticTriage(), + logger: {}, + }) + + try { + // BEFORE this change this call never settles and the test dies at + // vitest's 5 s default — the production defect, not an assertion detail. + // The 4 s guard is inside that default so the failure names itself. + const aborted = await withDeadline( + factory.runOnce().then(() => undefined, (error: unknown) => error), + 4_000, + 'sweep never settled', + ) + expect(aborted).toMatchObject({ name: 'DiscoverySweepBudgetExceededError', budgetMs: 400 }) + // Asserted BEFORE the phase, so a mis-timed run says which sweep phase + // actually ran out rather than blaming a string. One hung call is the + // proof that the budget reached `#prepareDiscoverySession`; the two + // phases ahead of it here are a fake fleet roster and an in-memory lease + // claim, and 400 ms is roughly three orders of magnitude of headroom + // over both. + expect(mount.hungCalls).toBe(1) + // The abandoned await is named, which is the diagnostic no per-call + // bound can produce once the sweep is already wedged. + expect((aborted as { phase?: string }).phase).toBe('discovery-session') + // The lease is handed back on the way out. Without this the next cycle + // has nothing to claim and the wedge simply moves. + expect(stateStore.released).toHaveLength(1) + + // The next cycle. It must start a NEW sweep rather than coalesce onto + // the abandoned one — the failure mode #296's deadline left behind. + mount.hang = false + const report = await withDeadline(factory.runOnce(), 4_000, 'next cycle never ran') + expect(report.pulled).toHaveLength(1) + expect(report.dispatched).toHaveLength(1) + expect(report.discoveryDeferred).toBeUndefined() + expect(fleet.spawns.map((spawn) => spawn.name)).toEqual(['ar-901-impl-pear', 'ar-901-review']) + } finally { + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }) + + it('must-fire: a live daemon whose sweep is wedged still shuts down, because nothing is left in flight', async () => { + // The counterpart to the #296/#301 abandoned-wait tests, which assert that + // `stop()` must NOT complete while a sweep abandoned by the readiness + // deadline is still running. That state is real, and it is exactly what + // held the deployed daemon: the sweep outlives every wait on it. With the + // budget it cannot happen — the sweep is aborted, not abandoned — so + // shutdown has nothing to drain. + const root = await mkdtemp(join(tmpdir(), 'factory-sweep-budget-stop-')) + const mount = new HangingWatermarkMount({ [issuePath(901)]: issueFile(901) }) + mount.serveFirst = 1 + const factory = createFactory(config(200, root), { + mount, + fleet: new FakeFleetClient(), + stateStore: new LeaseWatchingStateStore({ batchSize: 4 }), + triage: new StaticTriage(), + logger: {}, + }) + try { + // The startup backfill is a discovery pass like any other and wedges on + // the same call. Before this change `start()` itself never returns. + await withDeadline(factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 50, reconcileTimeoutMs: 60_000 }, + }), 4_000, 'start never returned') + expect(mount.hungCalls).toBeGreaterThan(0) + await withDeadline(factory.stop(), 4_000, 'stop never completed') + } finally { + await factory.stop().catch(() => undefined) + await rm(root, { recursive: true, force: true }) + } + }) + + it('must-fire: shutdown does not inherit the sweep budget when a sweep is wedged', async () => { + // `stop()` deliberately outlives the sweep it started (#301), so without a + // shutdown lever a wedged sweep makes shutdown as long as the budget — 90 + // minutes at the default. A 60 s budget here stands in for that: if + // shutdown waited for it, the 4 s guard below fires. + const root = await mkdtemp(join(tmpdir(), 'factory-sweep-budget-drain-')) + const mount = new HangingWatermarkMount({ [issuePath(901)]: issueFile(901) }) + // Serve the pre-backfill read AND the startup backfill's own, so the + // daemon comes up healthy and it is a LATER, periodic sweep that wedges. + mount.serveFirst = 2 + const factory = createFactory(config(60_000, root), { + mount, + fleet: new FakeFleetClient(), + stateStore: new LeaseWatchingStateStore({ batchSize: 4 }), + triage: new StaticTriage(), + logger: {}, + }) + try { + await withDeadline(factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 50, reconcileTimeoutMs: 60_000 }, + }), 4_000, 'start never returned') + await withDeadline( + (async () => { + while (mount.hungCalls === 0) await new Promise((resolve) => setTimeout(resolve, 25)) + })(), + 4_000, + 'no periodic sweep ever wedged', + ) + await withDeadline(factory.stop(), 4_000, 'stop waited for the whole sweep budget') + } finally { + await factory.stop().catch(() => undefined) + await rm(root, { recursive: true, force: true }) + } + }) + + it( + 'must-fire: the shutdown lever arms even while an unrelated held-agent sweep is still in flight', + // Generous on purpose: every bound inside is a failure deadline, so this is + // only ever consumed by a run that is already failing. + { timeout: 60_000 }, + async () => { + // The pair for the test above. That one proves the lever exists; this one + // proves WHEN it is armed, which is the half a passing shutdown cannot + // distinguish (cubic-dev-ai, #374 review). + // + // `stop()` arms the grace timer and then drains. Arming it after + // `#heldAgentDeadlineSweepInFlight` made the wedged discovery sweep's + // reprieve equal to `held-agent sweep duration + grace` instead of just + // `grace` — an unrelated subsystem silently extending the one bound this + // change exists to provide, without limit if that sweep never returns. + // + // The grace itself is `STOP_TEARDOWN_TIMEOUT_MS` (factory.ts), 2.5 s, and + // the budget here is 60 s — so a counter that is set can only have been + // set by the lever, never by a budget that expired on its own. + const root = await mkdtemp(join(tmpdir(), 'factory-sweep-budget-held-')) + const mount = new HangingWatermarkMount({ [issuePath(901)]: issueFile(901) }) + // As above: serve the pre-backfill read and the backfill's own, so the + // daemon comes up healthy and a LATER periodic sweep is the one that wedges. + mount.serveFirst = 2 + const fleet = new HeldSweepParkingFleet() + // 50 ms puts the dispatched agents past their hold deadline almost + // immediately, so the held-agent sweep is running well before shutdown. + const factory = createFactory(config(60_000, root, { agentHoldTimeoutMs: 50 }), { + mount, + fleet, + stateStore: new LeaseWatchingStateStore({ batchSize: 4 }), + triage: new StaticTriage(), + logger: {}, + }) + try { + await withDeadline(factory.start({ + mode: 'live', + liveSubscription: { transport: 'subscribe', reconcileIntervalMs: 50, reconcileTimeoutMs: 60_000 }, + }), 8_000, 'start never returned') + // Both preconditions are OBSERVED, not slept for: a discovery sweep is + // wedged, and a held-agent sweep is parked mid-release. + await withDeadline( + (async () => { + while (mount.hungCalls === 0) await new Promise((resolve) => setTimeout(resolve, 25)) + })(), + 8_000, + 'no periodic sweep ever wedged', + ) + await withDeadline(fleet.parked, 8_000, 'the held-agent deadline sweep never started') + + const stopping = factory.stop() + // POLLED, not slept (cubic-dev-ai, #374 review). A fixed wait just past + // the 2.5 s grace leaves a few hundred ms of headroom, which on a loaded + // worker is a new flake — in a suite that already carries two (#342, + // #373) and in a PR whose whole subject is a wedge. Polling costs the + // discrimination nothing: the held-agent sweep stays parked until + // `unpark()` below, so against the previous ordering `stop()` never + // reaches the arming call at all and this can only end in its own + // deadline. The margin disappears; the must-fire does not. + await withDeadline( + (async () => { + while (factory.status().counters.discoverySweepBudgetsCutShortForStop === undefined) { + await new Promise((resolve) => setTimeout(resolve, 25)) + } + })(), + 12_000, + 'the shutdown lever never armed while an unrelated held-agent sweep was in flight', + ) + expect(factory.status().counters.discoverySweepBudgetsCutShortForStop).toBe(1) + + fleet.unpark() + await withDeadline(stopping, 8_000, 'stop never completed') + } finally { + fleet.unpark() + await factory.stop().catch(() => undefined) + await rm(root, { recursive: true, force: true }) + } + }, + ) + + it('must-fire: a sweep aborted before the claim phase never opens a lease', async () => { + // A lease taken after the budget expired makes every later sweep defer — + // the same "later cycles wait on a pass that is already over" failure the + // PR's table pins on `reconcileTimeoutMs`. + // + // Scope, stated because it is narrower than it looks: this drives the + // ordinary path, where the budget is spent in an EARLIER phase and that + // phase's own `budget.run` throws first. Every entry into + // `#claimDiscoverySweepUnderBudget` is preceded by such a `budget.run`, so + // "already spent on entry" is not reachable by construction — it is the + // microtask gap between that check and the claim, which `stop()` made + // reachable when it gained the ability to spend a budget asynchronously. + // Issuing the claim inside the budget callback closes it by construction; + // the guarantee that makes that work is asserted directly on the primitive + // above ("refuses to start new work once the budget is spent", which + // asserts the thunk is never invoked). + const root = await mkdtemp(join(tmpdir(), 'factory-sweep-budget-spent-')) + const stateStore = new LeaseWatchingStateStore({ batchSize: 4 }) + const fleet = new SlowRosterFleet() + // Longer than the budget, so the budget is already spent by the time the + // claim phase is entered. + fleet.rosterDelayMs = 500 + const factory = createFactory(config(150, root), { + mount: new FakeMountClient({ [issuePath(901)]: issueFile(901) }), + fleet, + stateStore, + triage: new StaticTriage(), + logger: {}, + }) + try { + await expect(withDeadline(factory.runOnce(), 4_000, 'sweep never settled')) + .rejects.toMatchObject({ + name: 'DiscoverySweepBudgetExceededError', + phase: 'fleet-control-plane-probe', + }) + expect(stateStore.claims).toBe(0) + expect(stateStore.released).toHaveLength(0) + } finally { + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }) + + it('must-not-fire: a healthy sweep still claims exactly once and hands the lease back', async () => { + // The pair for the test above: the trivially wrong way to stop a spent + // budget from claiming is to stop claiming. + const root = await mkdtemp(join(tmpdir(), 'factory-sweep-budget-claims-')) + const stateStore = new LeaseWatchingStateStore({ batchSize: 4 }) + const factory = createFactory(config(30_000, root), { + mount: new FakeMountClient({ [issuePath(901)]: issueFile(901) }), + fleet: new SlowRosterFleet(), + stateStore, + triage: new StaticTriage(), + logger: {}, + }) + try { + const report = await withDeadline(factory.runOnce(), 4_000, 'healthy sweep never settled') + expect(report.dispatched).toHaveLength(1) + expect(stateStore.claims).toBe(1) + } finally { + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }) + + it('must-fire: a lease that lands after the budget gave up on the claim is handed straight back', async () => { + // The budget abandons the WAIT, not the call, so the store can persist a + // lease for a claim this sweep has already left. Nobody would renew, + // commit or release it, so every later sweep would defer for a full lease + // window — the wedge this PR exists to stop, one layer down. + const root = await mkdtemp(join(tmpdir(), 'factory-sweep-budget-claim-')) + const stateStore = new LeaseWatchingStateStore({ batchSize: 4 }) + stateStore.claimDelayMs = 600 + const factory = createFactory(config(150, root), { + mount: new FakeMountClient({ [issuePath(901)]: issueFile(901) }), + fleet: new FakeFleetClient(), + stateStore, + triage: new StaticTriage(), + logger: {}, + }) + try { + await expect(withDeadline(factory.runOnce(), 4_000, 'sweep never settled')) + .rejects.toMatchObject({ + name: 'DiscoverySweepBudgetExceededError', + phase: 'discovery-lease-claim', + }) + // The claim was still in flight when the budget expired; the compensation + // fires when it lands. + await withDeadline( + (async () => { + while (stateStore.released.length === 0) { + await new Promise((resolve) => setTimeout(resolve, 25)) + } + })(), + 4_000, + 'stranded lease was never released', + ) + expect(stateStore.released).toHaveLength(1) + } finally { + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + }) + + it('must-not-fire: a sweep that completes inside its budget is not aborted and its result is unchanged', async () => { + // Without this the trivial wrong fix — abort every sweep immediately — + // passes the must-fire above. + const run = async (sweepBudgetMs: number) => { + const root = await mkdtemp(join(tmpdir(), 'factory-sweep-budget-ok-')) + const fleet = new FakeFleetClient() + const factory = createFactory(config(sweepBudgetMs, root), { + mount: new FakeMountClient({ [issuePath(901)]: issueFile(901), [issuePath(902)]: issueFile(902) }), + fleet, + stateStore: new InMemoryStateStore({ batchSize: 4 }), + triage: new StaticTriage(), + logger: {}, + }) + try { + const report = await withDeadline(factory.runOnce(), 4_000, 'healthy sweep never settled') + return { + pulled: report.pulled.map((issue) => issue.key).sort(), + dispatched: report.dispatched.length, + skipped: report.skipped.length, + spawns: fleet.spawns.map((spawn) => spawn.name), + // The risk the stale-continuation fence introduces is the opposite of + // the one it removes: a fence that mistook a LIVE write for a late + // one would silently drop this sweep's own tree listings and the + // checkpoint would go empty. Zero, on a sweep that enumerated. + staleTreeWritesDropped: factory.status().counters.discoveryStaleTreeWritesDropped ?? 0, + } + } finally { + await factory.stop() + await rm(root, { recursive: true, force: true }) + } + } + + // A snug budget, and an effectively unbounded control. Identical results + // are what "the budget did not change this sweep" means. + const budgeted = await run(30_000) + const control = await run(90 * 60_000) + expect(budgeted.dispatched).toBeGreaterThan(0) + expect(budgeted.staleTreeWritesDropped).toBe(0) + expect(budgeted).toEqual(control) + }) +}) diff --git a/src/orchestrator/sweep-budget.ts b/src/orchestrator/sweep-budget.ts new file mode 100644 index 00000000..f5d0f744 --- /dev/null +++ b/src/orchestrator/sweep-budget.ts @@ -0,0 +1,254 @@ +/** + * One aggregate deadline for one discovery sweep. + * + * ## Why a per-call bound is not enough + * + * Three separate unbounded dependency calls have wedged this sweep in a single + * day, each one real, each one bounded, and each time the wedge came back one + * layer down: + * + * - the FACTORY_STATE Durable Object calls (factory-cloud#78), + * - the relayfile change-feed tail reads (#368, shipped in 0.1.75), + * - the retry of the now-bounded call, which is itself unbounded. + * + * The pattern is structural, not incidental. A per-call deadline bounds ONE + * await; a sweep is thousands of awaits plus every retry loop between them, so + * "every call is bounded" never adds up to "the sweep is bounded". The + * deployed 0.1.75 proves it: `readinessReconcile.state` reached `retrying`, + * which means #368's deadline genuinely fires, and `inFlightMs` still climbed + * with wall clock to 31 minutes on a 60-second interval. + * + * This module bounds the total instead. Time is charged against ONE timer for + * the whole sweep, so it does not matter which await is slow, how many there + * are, or how many times the sweep retries one of them: the sum cannot exceed + * the budget. The next unbounded call is then a degraded sweep rather than the + * end of dispatch. + * + * ## What it can and cannot interrupt + * + * `run()` is a race, not a cancellation — the same honest limitation #368 + * documented for `withRelayfileCallDeadline`, and stated here for the same + * reason: the next lane over has to be able to reason about what is left. + * + * It CAN abandon an in-flight await. Expiry rejects the sweep from inside + * `#runOnceWithDiscoveryFence`, so the sweep's own teardown runs: the lease is + * released and `#runOnceInFlight` is cleared, which is what lets the next cycle + * start a fresh sweep instead of coalescing onto the wedged one. That is the + * whole difference from the #296 sweep deadline, which rejected the *caller's + * wait* and left `runOnce()` running for every later cycle to coalesce onto. + * + * It CANNOT stop the abandoned work. The socket stays open, the SDK's own + * retry loop keeps running, and any side effect already in flight still lands. + * Two things narrow that: + * + * - `signal` aborts at expiry, so anything downstream that honours an + * `AbortSignal` is really cancelled. Nothing in the sweep consumes it yet + * (the relayfile client mints its own per-call signal, and that file is + * owned by another lane this week) — it is exported so the wiring is a + * one-line change rather than a redesign. + * - `assertNotExpired()` is a between-await check, which is worth exactly + * what #368 said it was worth — nothing against a call that never returns + * — but it does make an abandoned pass unwind at its next loop iteration + * if it ever regains control, instead of running to completion beside the + * sweep that replaced it. + */ + +/** The phases a sweep can be abandoned in. A closed set: see the error below. */ +export type DiscoverySweepPhase = + | 'fleet-control-plane-probe' + | 'discovery-lease-claim' + | 'discovery-backoff-wait' + | 'discovery-session' + | 'run-once' + | 'discovery-checkpoint' + | 'discovery-renewal-stop' + | 'discovery-commit' + +/** + * A sweep that did not finish inside its aggregate budget. + * + * Built only from code-controlled values — one integer and one literal from the + * closed set above — because it is persisted verbatim into the operator-facing + * `readinessReconcile.lastError`, and the class name reaches the + * unauthenticated health surface through the `error-class` allowlist. + * + * The phase is the diagnostic that matters: it names the await the sweep was + * abandoned on, which is the one thing no per-call bound could report once the + * sweep was already wedged. + */ +export class DiscoverySweepBudgetExceededError extends Error { + readonly code = 'FACTORY_DISCOVERY_SWEEP_BUDGET_EXCEEDED' + + constructor( + readonly budgetMs: number, + readonly phase: DiscoverySweepPhase, + ) { + super(`discovery sweep exceeded its ${budgetMs}ms budget while waiting on ${phase}`) + this.name = 'DiscoverySweepBudgetExceededError' + } +} + +/** The budget to apply, or `undefined` when the caller configured none. */ +export const discoverySweepBudgetMs = (timeoutMs: number | undefined): number | undefined => + timeoutMs !== undefined && Number.isFinite(timeoutMs) && timeoutMs > 0 ? timeoutMs : undefined + +export interface DiscoverySweepBudget { + /** The applied budget, or `undefined` when the sweep is deliberately unbounded. */ + readonly budgetMs?: number + /** Aborted the moment the budget is spent. Reason is the error below. */ + readonly signal: AbortSignal + /** True once the budget is spent, whether or not anything is awaiting. */ + expired(): boolean + /** Await `start()`, or abandon that wait once the sweep's budget is spent. */ + run(phase: DiscoverySweepPhase, start: () => Promise): Promise + /** Throw if the budget is already spent. A between-await check; see above. */ + assertNotExpired(phase: DiscoverySweepPhase): void + /** + * Spend the budget now. + * + * Shutdown's lever. `stop()` drains the sweep it started, so a wedged sweep + * would otherwise hold shutdown open for the whole budget — 90 minutes by + * default. Expiring it turns that into the sweep's ordinary abort path: + * lease released, teardown bounded, `stop()` free to continue. + */ + expire(): void + /** Releases the timer. Always call it, or a settled sweep leaves one pending. */ + dispose(): void +} + +const BUDGET_EXPIRED = Symbol('discovery-sweep-budget-expired') + +type SweepOutcome = { ok: true; value: T } | { ok: false; error: unknown } + +/** + * Start the clock for one sweep. + * + * Deliberately not `AbortSignal.timeout()`: its timer cannot be cancelled, so a + * sweep that finishes in four seconds would leave a ninety-minute timer and a + * retained signal behind it. `dispose()` clears the timer the moment the sweep + * settles. + * + * One timer and one expiry promise are shared by every `run()` on this budget. + * That sharing is the aggregate property: a sweep does not get a fresh budget + * per await, so elapsed time accumulates across phases and across retries. + */ +export function startDiscoverySweepBudget(timeoutMs: number | undefined): DiscoverySweepBudget { + const budgetMs = discoverySweepBudgetMs(timeoutMs) + const controller = new AbortController() + let expired = false + let timer: ReturnType | undefined + let expire: () => void = () => undefined + // Never rejects, so an unawaited race arm cannot surface as an unhandled + // rejection when the sweep finishes first. + const expiry = new Promise((resolve) => { + expire = () => resolve(BUDGET_EXPIRED) + }) + + const spend = (): void => { + if (expired || budgetMs === undefined) return + expired = true + if (timer) clearTimeout(timer) + // `run()` supplies the phase on the throw; the abort reason cannot know + // which await was in flight, so it names the sweep as a whole. + controller.abort(new DiscoverySweepBudgetExceededError(budgetMs, 'run-once')) + expire() + } + + if (budgetMs !== undefined) { + timer = setTimeout(spend, budgetMs) + // Deliberately NOT unref'd, unlike the per-call deadline in + // `relayfile-operation-timeout.ts`. That one is created thousands of times + // per sweep and can afford to lose a race with process exit; this one is + // the guarantee. Under a one-shot `runOnce()` whose only pending work is a + // promise nothing else references, an unref'd timer lets Node exit before + // the budget fires — the command would return without ever reporting the + // wedge or releasing the lease. It lives for at most one budget and + // `dispose()` clears it from a `finally` on every path. + } + + const assertNotExpired = (phase: DiscoverySweepPhase): void => { + if (expired && budgetMs !== undefined) throw new DiscoverySweepBudgetExceededError(budgetMs, phase) + } + + return { + ...(budgetMs === undefined ? {} : { budgetMs }), + signal: controller.signal, + expired: () => expired, + assertNotExpired, + expire: spend, + async run(phase: DiscoverySweepPhase, start: () => Promise): Promise { + if (budgetMs === undefined) return await start() + // Decided before the call is made, so an already-spent budget cannot + // start new work against the dependency it is abandoning. + assertNotExpired(phase) + // The outcome is folded once. A late rejection from the abandoned call + // then has a handler attached and cannot crash the process. + const inFlight: Promise> = start().then( + (value) => ({ ok: true, value }) as const, + (error: unknown) => ({ ok: false, error }) as const, + ) + const outcome = await Promise.race | typeof BUDGET_EXPIRED>([inFlight, expiry]) + if (outcome === BUDGET_EXPIRED) throw new DiscoverySweepBudgetExceededError(budgetMs, phase) + if (outcome.ok) return outcome.value + throw outcome.error + }, + dispose: () => { + if (timer) clearTimeout(timer) + }, + } +} + +/** + * How long one sweep-teardown step may take. + * + * Teardown cannot run under the aggregate budget: by the time it runs the + * budget is spent by construction, so every step would reject and the lease + * would never be released — which is the half of the fix that makes the next + * cycle clean. It gets its own, much shorter, deadline instead, because an + * unbounded release call would re-create the very wedge this module exists to + * end, one layer further down. + */ +export const DISCOVERY_SWEEP_TEARDOWN_TIMEOUT_MS = 30_000 + +const TEARDOWN_TIMED_OUT = Symbol('discovery-sweep-teardown-timed-out') + +/** + * Run one teardown step under its own deadline. + * + * Returns `undefined` if the step did not finish in time; the caller decides + * what an abandoned teardown means. Errors propagate exactly as they did + * before this wrapper existed — a teardown deadline is not a licence to start + * swallowing failures that used to surface. + * + * An abandoned lease release is survivable and is why this returns rather than + * throws: the durable lease carries its own expiry, so it is reclaimed as an + * orphan by a later sweep instead of being held forever. + */ +export async function withSweepTeardownDeadline( + timeoutMs: number, + step: () => Promise, +): Promise { + const budget = discoverySweepBudgetMs(timeoutMs) + if (budget === undefined) return await step() + + let timer: ReturnType | undefined + try { + const inFlight: Promise> = step().then( + (value) => ({ ok: true, value }) as const, + (error: unknown) => ({ ok: false, error }) as const, + ) + const outcome = await Promise.race | typeof TEARDOWN_TIMED_OUT>([ + inFlight, + new Promise((resolve) => { + // Referenced, for the same reason as the budget timer above: this is + // the deadline that guarantees the lease is handed back. + timer = setTimeout(() => resolve(TEARDOWN_TIMED_OUT), budget) + }), + ]) + if (outcome === TEARDOWN_TIMED_OUT) return undefined + if (outcome.ok) return outcome.value + throw outcome.error + } finally { + if (timer) clearTimeout(timer) + } +} diff --git a/src/types.ts b/src/types.ts index f7ef0659..b36b7e7d 100644 --- a/src/types.ts +++ b/src/types.ts @@ -134,6 +134,13 @@ export interface FactoryLiveSubscriptionOptions { * which no deadline checked *between* awaits can reach. */ relayfileOperationTimeoutMs: number + /** + * Aggregate budget for one whole sweep (#372). Bounds what neither neighbour + * can: a bounded call wrapped in an unbounded retry, or a hang on a transport + * nobody has bounded yet. Expiry aborts the sweep from inside its own fence, + * so the discovery lease is released and the next cycle starts clean. + */ + sweepBudgetMs: number } /**