diff --git a/docs/deployed-diagnostics.md b/docs/deployed-diagnostics.md index 6d491a32..2ef2fa05 100644 --- a/docs/deployed-diagnostics.md +++ b/docs/deployed-diagnostics.md @@ -92,13 +92,26 @@ logic of its own by design: the boundary lives in one place, in this repo, with A cold container legitimately spends minutes in its first pass (#36 measured 61 minutes while the Relayfile mirror hydrated), so check `lastCompletedAtMs`: absent means "first pass since boot, still hydrating"; present and hours old means "was fine, then wedged". -- **How long a stall can last** — a sweep is bounded at `liveSubscription.reconcileTimeoutMs`, - 90 minutes by default (#296). On expiry the *wait* fails, so `consecutiveFailures` starts rising - and the loop schedules the next pass; the sweep itself is not cancelled, because it holds a durable - discovery lease, so `inFlightSinceMs` keeps ageing until it really finishes. A `stalled` state that - never turns into a rising `consecutiveFailures` therefore means the process is not running the loop - at all, which is a restart, not a wait. The deadline sits above #36's 61-minute measurement on +- **How long a stall can last** — two deadlines, at different scales. + + `liveSubscription.relayfileOperationTimeoutMs` bounds ONE relayfile call, five minutes by default + (#351). This is the one that catches a wedge. Expiry cancels the request, fails the pass with + `lastErrorClass: "RelayfileOperationTimeoutError"` and a `lastError` naming the call + (`relayfile listTree did not respond within 300000ms (GitHub issue ingestion)`), and unwinds the + sweep — which releases the discovery lease, so the next cycle starts clean. + + `liveSubscription.reconcileTimeoutMs` bounds the whole sweep, 90 minutes by default (#296). It is + the outer backstop only. On expiry the *wait* fails, so `consecutiveFailures` starts rising and the + loop schedules the next pass; the sweep itself is not cancelled, because it holds a durable + discovery lease, so `inFlightSinceMs` keeps ageing until it really finishes — and the next pass + coalesces onto that same running `runOnce()`. The deadline sits above #36's 61-minute measurement on purpose: setting it below realistic cold-mirror hydration would turn a slow boot into a crash loop. + Per-call bounds can be far tighter precisely because that cold-mirror cost is spread across + thousands of calls rather than concentrated in one. + + A `stalled` state that never turns into a rising `consecutiveFailures` means either the process is + not running the loop at all, or it predates #351 — on a current build a hung call fails within + `relayfileOperationTimeoutMs`. ### Why `ok` stays `true` while `status` goes amber diff --git a/src/cli/fleet.ts b/src/cli/fleet.ts index 679c544a..d242b0f0 100644 --- a/src/cli/fleet.ts +++ b/src/cli/fleet.ts @@ -2203,6 +2203,9 @@ async function buildMount( mount = await (deps.cloudMountFromConfig ?? RelayfileCloudMountClient.fromConfig)({ workspaceId: loaded.config.workspaceId, localMountRoot: loaded.config.localMountRoot, + // The transport half of #351: without a signal the SDK issues a bare + // fetch() with no deadline, which is what wedged the reconcile loop. + operationTimeoutMs: loaded.config.liveSubscription.relayfileOperationTimeoutMs, logger: observability.logger, onLocalMountHealth: observability.onLocalMountHealth, isAllowedDraft: (path, content, opts) => isAllowedFactoryDraft(path, content, opts, mount, loaded.config), diff --git a/src/config/schema.ts b/src/config/schema.ts index 8a01309d..890ebe1f 100644 --- a/src/config/schema.ts +++ b/src/config/schema.ts @@ -7,6 +7,7 @@ import { DEFAULT_FLEET_CONTROL_RESET_TIMEOUT_MS, DEFAULT_FLEET_ROSTER_TIMEOUT_MS, } from '../fleet/control-plane-circuit' +import { DEFAULT_RELAYFILE_OPERATION_TIMEOUT_MS } from '../mount/relayfile-operation-timeout' import { KubernetesEnvironmentConfigSchema } from '../environments/connection-registry.js' @@ -70,6 +71,19 @@ const liveSubscriptionSchema = z.object({ /** Bounds one sweep so a hung dependency call cannot stop the loop forever. */ reconcileTimeoutMs: z.number().int().min(50).max(6 * 60 * 60_000) .default(DEFAULT_READINESS_RECONCILE_TIMEOUT_MS), + /** + * Bounds ONE relayfile call (#351). + * + * Distinct from `reconcileTimeoutMs`, and deliberately far tighter. The + * cold-mirror cost that forces the sweep deadline up to 90 minutes is spread + * across thousands of calls; no single call has ever needed minutes. So a + * per-call bound catches a wedge in five minutes instead of ninety without + * re-creating the crash-loop risk described above — and unlike the sweep + * deadline it actually unwinds the pass, releasing the discovery lease so the + * next cycle starts clean rather than coalescing onto the hung one. + */ + relayfileOperationTimeoutMs: z.number().int().min(50).max(60 * 60_000) + .default(DEFAULT_RELAYFILE_OPERATION_TIMEOUT_MS), }).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. diff --git a/src/mount/relayfile-cloud-mount-client.test.ts b/src/mount/relayfile-cloud-mount-client.test.ts index e388b654..aa15968b 100644 --- a/src/mount/relayfile-cloud-mount-client.test.ts +++ b/src/mount/relayfile-cloud-mount-client.test.ts @@ -62,7 +62,7 @@ class FakeRelayFileClient implements RelayFileClientLike { path: string baseRevision: string }> = [] - readonly listTreeCalls: Array<{ workspaceId: string; options?: { path?: string; depth?: number; cursor?: string } }> = [] + readonly listTreeCalls: Array<{ workspaceId: string; options?: { path?: string; depth?: number; cursor?: string; signal?: AbortSignal } }> = [] readonly getEventsCalls: Array<{ workspaceId: string; opts?: { cursor?: string; limit?: number; provider?: string; last?: number } }> = [] readonly listLastNChangesCalls: Array<{ limit: number; context?: { workspaceId: string } }> = [] readonly getOpCalls: Array<{ workspaceId: string; opId: string }> = [] @@ -1456,8 +1456,11 @@ describe('RelayfileCloudMountClient', () => { expect(fake.readFileCalls[0]).toEqual({ workspaceId: 'rw_test', path: '/linear/issues/AR-1.json' }) expect(fake.listTreeCalls[0]).toEqual({ workspaceId: 'rw_test', - options: { path: '/linear/issues', cursor: undefined }, + // #351: reads carry the per-operation deadline's signal, so a relayfile + // call that stops answering is cancelled instead of waited on forever. + options: { path: '/linear/issues', cursor: undefined, signal: expect.any(AbortSignal) }, }) + expect(fake.listTreeCalls[0]?.options?.signal?.aborted).toBe(false) expect(fake.getEventsCalls[0]).toEqual({ workspaceId: 'rw_test', opts: { cursor: 'evt-0', limit: 10 } }) }) @@ -1486,6 +1489,104 @@ describe('RelayfileCloudMountClient', () => { expect(fake.listTreeCalls.map((call) => call.options?.cursor)).toEqual([undefined, '2', '4']) }) + // #351: the SDK attaches an AbortSignal to its fetch only when the caller + // supplies one, and nothing here did — so every relayfile read was a bare + // fetch() with no deadline. One of them stopped answering on 2026-08-23 and + // the readiness reconcile cycle that issued it never finished. + describe('bounded relayfile reads', () => { + class HangingListTreeClient extends FakeRelayFileClient { + seenSignal?: AbortSignal + + override async listTree( + workspaceId: string, + options?: { path?: string; depth?: number; cursor?: string; signal?: AbortSignal }, + ): Promise { + this.listTreeCalls.push({ workspaceId, options }) + this.seenSignal = options?.signal + return await new Promise((_resolve, reject) => { + options?.signal?.addEventListener('abort', () => { + reject((options.signal as AbortSignal & { reason?: unknown }).reason) + }) + }) + } + } + + it('cancels a read that stops answering and names the operation', async () => { + const client = new HangingListTreeClient() + const mount = new RelayfileCloudMountClient({ + workspaceId: 'rw_test', + client, + operationTimeoutMs: 25, + }) + + await expect(mount.listTree('/github/repos')).rejects.toMatchObject({ + name: 'RelayfileOperationTimeoutError', + operation: 'listTree', + }) + // Cancelled, not merely abandoned. An abandoned wait leaves the socket + // and the SDK's retry loop live, and hands the next caller the same + // wedged in-flight read. + expect(client.seenSignal?.aborted).toBe(true) + }) + + it('honours the ensureSubRoot timeout it used to accept and discard', async () => { + const client = new HangingListTreeClient() + const mount = new RelayfileCloudMountClient({ + workspaceId: 'rw_test', + client, + // Client-wide budget far past the caller's, so only the per-call + // argument can end this. It was silently dropped before #351, which is + // why `#ensureGithubIngestionReady` passing 90_000 bounded nothing. + operationTimeoutMs: 60_000, + }) + + await expect(mount.ensureSubRoot('/github/issues', { timeoutMs: 25 })).rejects.toMatchObject({ + name: 'RelayfileOperationTimeoutError', + operation: 'ensureSubRoot', + }) + }) + + it('caps an explicit ensureSubRoot timeout at the tighter client-wide budget', async () => { + const client = new HangingListTreeClient() + const mount = new RelayfileCloudMountClient({ + workspaceId: 'rw_test', + client, + operationTimeoutMs: 25, + }) + + // The caller's argument must cap, not replace: a client-wide budget + // tighter than the argument has to still cancel at the transport, or the + // orchestrator's backstop fires first and abandons the wait instead + // (codex on #354). + await expect(mount.ensureSubRoot('/github/issues', { timeoutMs: 60_000 })).rejects.toMatchObject({ + name: 'RelayfileOperationTimeoutError', + operation: 'ensureSubRoot', + timeoutMs: 25, + }) + expect(client.seenSignal?.aborted).toBe(true) + }) + + it('leaves the call unbounded when no budget is configured', async () => { + const client = new HangingListTreeClient() + const mount = new RelayfileCloudMountClient({ + workspaceId: 'rw_test', + client, + operationTimeoutMs: 0, + }) + + const pending = mount.listTree('/github/repos') + const settled = await Promise.race([ + pending.then(() => 'settled' as const, () => 'settled' as const), + new Promise<'pending'>((resolve) => { setTimeout(() => resolve('pending'), 50) }), + ]) + expect(settled).toBe('pending') + expect(client.seenSignal).toBeUndefined() + // Nothing will ever settle this; drop the reference explicitly so the + // rejection-free pending promise is not mistaken for a leak. + void pending.catch(() => undefined) + }) + }) + it('uses recent change-log events for provider-filtered getEvents tail reads', async () => { const fake = new FakeRelayFileClient() fake.events = [ diff --git a/src/mount/relayfile-cloud-mount-client.ts b/src/mount/relayfile-cloud-mount-client.ts index c9741a5e..f12c4e48 100644 --- a/src/mount/relayfile-cloud-mount-client.ts +++ b/src/mount/relayfile-cloud-mount-client.ts @@ -22,6 +22,15 @@ import { type WriteQueuedResponse, } from '@relayfile/sdk' import { RelayfileSetup } from '@relayfile/sdk/cli' +import { + DEFAULT_RELAYFILE_OPERATION_TIMEOUT_MS, + RelayfileOperationTimeoutError, + isRelayfileCallAbort, + relayfileCallBudgetMs, + relayfileCallDeadline, + tighterRelayfileBudgetMs, + withRelayfileCallDeadline, +} from './relayfile-operation-timeout' import { existsSync } from 'node:fs' import { isAbsolute, join, resolve } from 'node:path' @@ -260,10 +269,26 @@ export interface RelayfileCloudMountClientConfig { skipRegisteredMirrorLookup?: boolean isAllowedDraft?: (path: string, content: unknown, opts?: { guarded?: boolean }) => boolean | Promise isAllowedDelete?: (path: string, currentContent: unknown) => boolean | Promise + /** + * Deadline for one relayfile read, in milliseconds (#351). + * + * The SDK attaches an `AbortSignal` to its `fetch` only when the caller + * supplies one, so without this every read is a bare `fetch()` that can wait + * forever. Zero or a non-finite value restores that unbounded behaviour. + */ + operationTimeoutMs?: number } export type RelayFileClientLike = { - readFile(workspaceId: string, path: string): Promise + // `correlationId`/`signal` are the SDK's own trailing arguments. Declared + // here so a read can carry a deadline (#351); optional, so a narrower fake + // still satisfies the type. + readFile( + workspaceId: string, + path: string, + correlationId?: string, + signal?: AbortSignal, + ): Promise writeFile(input: WriteFileInput): Promise deleteFile(input: DeleteFileInput): Promise listTree(workspaceId: string, options?: ListTreeOptions): Promise @@ -335,6 +360,7 @@ export class RelayfileCloudMountClient implements MountClient { #disposed = false #isAllowedDraft?: (path: string, content: unknown, opts?: { guarded?: boolean }) => boolean | Promise readonly #isAllowedDelete?: (path: string, currentContent: unknown) => boolean | Promise + readonly #operationTimeoutMs: number readonly #lastOpByPath = new Map() readonly #confirmedExternalIdByPath = new Map() readonly #confirmedFailureReasonByPath = new Map() @@ -345,6 +371,7 @@ export class RelayfileCloudMountClient implements MountClient { } this.workspaceId = config.workspaceId ?? DEFAULT_WORKSPACE_ID + this.#operationTimeoutMs = config.operationTimeoutMs ?? DEFAULT_RELAYFILE_OPERATION_TIMEOUT_MS this.#client = config.client this.#tokenProvider = config.tokenProvider ?? (() => this.#client.getToken?.()) this.#baseUrl = config.baseUrl ?? this.#client.getBaseUrl?.() @@ -646,7 +673,8 @@ export class RelayfileCloudMountClient implements MountClient { } async readFile(path: string): Promise<{ content: unknown; revision?: string }> { - const response = await this.#client.readFile(this.workspaceId, path) + const response = await this.#bounded('readFile', this.#operationTimeoutMs, (signal) => + this.#client.readFile(this.workspaceId, path, undefined, signal)) return { content: parseRemoteContent(response), revision: response.revision, @@ -735,18 +763,88 @@ export class RelayfileCloudMountClient implements MountClient { } async listTree(prefix: string): Promise { - const paths: string[] = [] - let cursor: string | undefined - for (;;) { - const response = await this.#client.listTree(this.workspaceId, { - path: prefix, - cursor, + // ONE deadline for the whole walk, not one per page. The cursor loop is + // unbounded, so a per-page budget would let a server that keeps handing + // back a `nextCursor` run for as long as it likes — the loop would stay + // unbounded while every individual call looked bounded (#351). + const budgetMs = relayfileCallBudgetMs(this.#operationTimeoutMs) + const deadline = relayfileCallDeadline('listTree', budgetMs) + const deadlineAtMs = budgetMs === undefined ? undefined : Date.now() + budgetMs + try { + return await this.#named('listTree', budgetMs, async () => { + const paths: string[] = [] + let cursor: string | undefined + for (;;) { + const response = await withRelayfileCallDeadline( + 'listTree', + undefined, + this.#remainingMs('listTree', deadlineAtMs), + () => this.#client.listTree(this.workspaceId, { + path: prefix, + cursor, + ...(deadline.signal ? { signal: deadline.signal } : {}), + }), + ) + paths.push(...response.entries.map((entry) => entry.path)) + if (!response.nextCursor) break + cursor = response.nextCursor + } + return paths.sort() }) - paths.push(...response.entries.map((entry) => entry.path)) - if (!response.nextCursor) break - cursor = response.nextCursor + } finally { + deadline.dispose() + } + } + + /** What is left of a multi-call operation's budget, refusing at zero. */ + #remainingMs(operation: string, deadlineAtMs: number | undefined): number | undefined { + if (deadlineAtMs === undefined) return undefined + const remainingMs = deadlineAtMs - Date.now() + // Refuse rather than fall through: `withRelayfileCallDeadline` treats a + // non-positive budget as "no budget" and would run the call unbounded. + if (remainingMs <= 0) { + throw new RelayfileOperationTimeoutError(operation, this.#operationTimeoutMs) + } + return remainingMs + } + + /** + * One relayfile read under a deadline: cancelled at the transport where the + * SDK honours the signal, and abandoned by the race where it does not. + */ + async #bounded( + operation: string, + budgetMs: number | undefined, + start: (signal?: AbortSignal) => Promise, + ): Promise { + const usableMs = relayfileCallBudgetMs(budgetMs) + const deadline = relayfileCallDeadline(operation, usableMs) + try { + return await this.#named(operation, usableMs, () => + withRelayfileCallDeadline(operation, undefined, usableMs, () => start(deadline.signal))) + } finally { + deadline.dispose() + } + } + + /** + * Give the deadline a name an operator can act on. + * + * A signal expiry surfaces as a bare `TimeoutError` reading "the operation + * was aborted" — true and useless. `readinessReconcile.lastError` has to say + * which relayfile call it was waiting on, which is the whole point of #351. + * These paths pass no other signal, so an abort here can only be ours. + */ + async #named(operation: string, budgetMs: number | undefined, run: () => Promise): Promise { + try { + return await run() + } catch (error) { + if (budgetMs !== undefined && isRelayfileCallAbort(error)) { + if (error instanceof RelayfileOperationTimeoutError) throw error + throw new RelayfileOperationTimeoutError(operation, budgetMs) + } + throw error } - return paths.sort() } subscribe(globs: string[], onChange: (event: ChangeEvent) => void, opts?: SubscribeOptions): Subscription { @@ -970,9 +1068,21 @@ export class RelayfileCloudMountClient implements MountClient { return this.#confirmedExternalIdByPath.get(path) } - async ensureSubRoot(prefix: string, _opts?: { timeoutMs?: number }): Promise<'ready' | 'absent'> { + // `timeoutMs` used to be accepted and discarded, so `#ensureGithubIngestionReady` + // passed 90_000 and got no bound at all — the call site believed it was + // bounded and was not (#351). It is honoured now, and it *caps* rather than + // replaces the client-wide budget: a config that tightened + // `relayfileOperationTimeoutMs` below the caller's argument would otherwise + // let the transport run past the orchestrator's backstop, abandoning the wait + // instead of cancelling the call (codex on #354). + async ensureSubRoot(prefix: string, opts?: { timeoutMs?: number }): Promise<'ready' | 'absent'> { try { - await this.#client.listTree(this.workspaceId, { path: prefix, depth: 1 }) + await this.#bounded('ensureSubRoot', tighterRelayfileBudgetMs(opts?.timeoutMs, this.#operationTimeoutMs), (signal) => + this.#client.listTree(this.workspaceId, { + path: prefix, + depth: 1, + ...(signal ? { signal } : {}), + })) return 'ready' } catch (error) { if (isHttpStatus(error, 404)) return 'absent' diff --git a/src/mount/relayfile-operation-timeout.ts b/src/mount/relayfile-operation-timeout.ts new file mode 100644 index 00000000..f9b25dfe --- /dev/null +++ b/src/mount/relayfile-operation-timeout.ts @@ -0,0 +1,177 @@ +/** + * Per-call deadlines for relayfile operations (#351). + * + * `@relayfile/sdk`'s `performRequest` attaches an `AbortSignal` to its `fetch` + * only when the caller supplies one, and nothing in Factory did — so every + * relayfile read was a bare `fetch()` with no deadline. On 2026-08-23 one of + * them stopped returning and the readiness reconcile cycle that issued it never + * finished: 22 minutes with `consecutiveFailures: 0`, because a hang takes + * neither the success nor the failure path. + * + * The sweep-level deadline (#296) cannot cover this. It is 90 minutes by + * design — below realistic cold-mirror hydration a slow boot becomes a crash + * loop — and it rejects the *wait* while `runOnce()` keeps running, so the next + * cycle coalesces onto the same wedged promise and waits again. A per-call + * bound can be two orders of magnitude tighter, because a cold sweep's cost is + * spread across thousands of calls rather than concentrated in one, and its + * rejection unwinds the sweep — releasing the discovery lease, so the next + * cycle starts clean. + */ + +/** Fallback deadline for one relayfile call. See `config/schema.ts` for why. */ +export const DEFAULT_RELAYFILE_OPERATION_TIMEOUT_MS = 5 * 60_000 + +/** + * A relayfile call that did not answer inside its budget. + * + * The message is built only from code-controlled values — a closed set of + * operation and phase literals plus one integer — because it is persisted into + * the operator-facing `readinessReconcile.lastError`. The class name is what + * reaches the unauthenticated surface, through the `error-class` allowlist. + */ +export class RelayfileOperationTimeoutError extends Error { + readonly code = 'FACTORY_RELAYFILE_OPERATION_TIMEOUT' + + constructor( + readonly operation: string, + readonly timeoutMs: number, + readonly phase?: string, + ) { + super( + `relayfile ${operation} did not respond within ${timeoutMs}ms` + + `${phase === undefined ? '' : ` (${phase})`}`, + ) + this.name = 'RelayfileOperationTimeoutError' + } +} + +/** The budget to apply, or `undefined` when the caller configured none. */ +export const relayfileCallBudgetMs = (timeoutMs: number | undefined): number | undefined => + timeoutMs !== undefined && Number.isFinite(timeoutMs) && timeoutMs > 0 ? timeoutMs : undefined + +/** + * The tighter of several budgets, ignoring the ones that impose none. + * + * A caller's explicit per-call timeout must *cap* the client-wide budget rather + * than replace it: a config that tightened `relayfileOperationTimeoutMs` below + * an explicit argument would otherwise leave the transport running past the + * orchestrator's backstop, which abandons the wait instead of cancelling the + * call — the exact behaviour this module exists to avoid. + */ +export const tighterRelayfileBudgetMs = ( + ...timeoutsMs: ReadonlyArray +): number | undefined => { + const budgets = timeoutsMs + .map((timeoutMs) => relayfileCallBudgetMs(timeoutMs)) + .filter((budget): budget is number => budget !== undefined) + return budgets.length === 0 ? undefined : Math.min(...budgets) +} + +/** + * The same timeout, carrying the phase the orchestrator knows and the transport + * does not. + * + * The transport deadline is meant to win the race, so without this the + * persisted `lastError` would read `relayfile listTree did not respond within + * 300000ms` with no way to tell one list or read context from another — which + * is most of what naming the call was for. + */ +export function relayfileTimeoutWithPhase(error: unknown, phase: string | undefined): unknown { + if (phase === undefined) return error + if (!(error instanceof RelayfileOperationTimeoutError) || error.phase !== undefined) return error + const enriched = new RelayfileOperationTimeoutError(error.operation, error.timeoutMs, phase) + enriched.cause = error + return enriched +} + +/** True for the abort a `signal` deadline raises, in either transport's shape. */ +export function isRelayfileCallAbort(error: unknown): boolean { + if (error instanceof RelayfileOperationTimeoutError) return true + if (!(error instanceof Error)) return false + return error.name === 'TimeoutError' || error.name === 'AbortError' +} + +/** A cancellation signal plus the means to release its timer. */ +export interface RelayfileCallDeadline { + /** Undefined when no budget applies, so the call is made exactly as before. */ + readonly signal?: AbortSignal + /** Releases the timer. Always call it, or a settled call leaves one pending. */ + dispose(): void +} + +/** + * A deadline that cancels the request itself once the budget is spent. + * + * Preferred over racing the wait: an abandoned wait leaves the socket open and + * the SDK's own retry loop running, and a read the SDK has cached as in-flight + * would hand the *next* cycle the same wedged promise. + * + * Deliberately not `AbortSignal.timeout()`, whose timer cannot be cancelled: at + * a five-minute budget every completed read would leave a five-minute timer and + * a retained signal behind it, so a busy sweep accumulates thousands. This pair + * clears the timer the moment the call settles. + * + * The abort reason is the named error itself, so a transport that surfaces + * `signal.reason` (undici's `fetch` does) already reports which call it was. + */ +export function relayfileCallDeadline( + operation: string, + timeoutMs: number | undefined, +): RelayfileCallDeadline { + const budget = relayfileCallBudgetMs(timeoutMs) + if (budget === undefined) return { dispose: () => undefined } + + const controller = new AbortController() + const timer = setTimeout( + () => controller.abort(new RelayfileOperationTimeoutError(operation, budget)), + budget, + ) + timer.unref?.() + return { signal: controller.signal, dispose: () => clearTimeout(timer) } +} + +const CALL_TIMED_OUT = Symbol('relayfile-call-timed-out') + +type CallOutcome = { ok: true; value: T } | { ok: false; error: unknown } + +/** + * Await `start()`, or give up on it once `timeoutMs` is spent. + * + * The backstop behind `relayfileCallSignal`: it covers mount implementations + * that cannot honour a signal, and operations whose side effect makes real + * cancellation unsafe. Like `RelayFleetClient.#withinDeadline` (#306/#307) it + * abandons the *wait*, not the call — so the outcome is folded once, and a late + * rejection from the abandoned call cannot surface as an unhandled rejection. + * + * Takes a thunk rather than a promise so an unusable budget is decided before + * the request is made. + */ +export async function withRelayfileCallDeadline( + operation: string, + phase: string | undefined, + timeoutMs: number | undefined, + start: () => Promise, +): Promise { + const budget = relayfileCallBudgetMs(timeoutMs) + if (budget === undefined) return await start() + + let timer: ReturnType | undefined + try { + const inFlight: Promise> = start().then( + (value) => ({ ok: true, value }) as const, + (error: unknown) => ({ ok: false, error }) as const, + ) + const outcome = await Promise.race | typeof CALL_TIMED_OUT>([ + inFlight, + new Promise((resolve) => { + timer = setTimeout(() => resolve(CALL_TIMED_OUT), budget) + timer.unref?.() + }), + ]) + if (outcome === CALL_TIMED_OUT) throw new RelayfileOperationTimeoutError(operation, budget, phase) + if (outcome.ok) return outcome.value + throw outcome.error + } finally { + if (timer) clearTimeout(timer) + } +} diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 74987669..3d165078 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -30,6 +30,7 @@ import { } from '../index' import { LatePlacementReleasedError, changeEventPath } from './factory' import { RelaySpawnAckTimeoutError } from '../fleet/relay-fleet-client' +import { RelayfileOperationTimeoutError } from '../mount/relayfile-operation-timeout' import type { AgentWorktree, AgentWorktreeCleanupInspection, AgentWorktreeManager, AgentWorktreeRepository, ChangeEvent, EventPage, GithubConnectionRead, GithubConnectionWrite, GithubIssueStatus, GithubPublishPullRequestInput, GithubWriteback, LinearWriteback, PreviewReference, PreviewStartInput, ProviderSyncStatus, RosterEntry, SlackWriteback, SpawnInput, SpawnResult } from '../ports' import { FakeFleetClient, FakeMountClient, withDeadline } from '../testing' import type { CloseProbePrInput, GithubMergeGatePort, GithubMergeGateVerdict, GithubMergeInput, LinearIssue, VerificationGate, VerificationGateInput, VerificationVerdict } from '../index' @@ -13391,6 +13392,209 @@ describe('FactoryLoop', () => { } }) + // #351: the sweep deadline above cannot reach a call that never returns. + // It is 90 minutes in production — sized above #36's cold-mirror + // measurement on purpose — and it rejects only the WAIT, leaving + // `runOnce()` holding its discovery lease so the next cycle coalesces onto + // the same wedged promise. On 2026-08-23 one relayfile read stopped + // answering and the loop sat for 22 minutes with `consecutiveFailures: 0`. + // Only a bound on the CALL ends the pass, and ending it is what releases + // the lease and lets the next cycle actually run. + class HangingListTreeMount extends CountingEventsMount { + readonly hangStarted: Promise + hangListTree = false + #signalHangStarted: () => void = () => undefined + readonly #releases: Array<() => void> = [] + + constructor() { + super() + this.hangStarted = new Promise((resolve) => { this.#signalHangStarted = resolve }) + this.setSubRoot('/linear/issues', 'absent') + } + + /** Frees every parked read so teardown never inherits the wedge. */ + release(): void { + this.hangListTree = false + while (this.#releases.length > 0) this.#releases.pop()?.() + } + + override async listTree(prefix: string): Promise { + if (this.hangListTree) { + this.#signalHangStarted() + // Never settles: the shape of a relayfile fetch() issued with no + // AbortSignal, which is what the SDK does when no caller supplies one. + await new Promise((resolve) => { this.#releases.push(resolve) }) + } + return super.listTree(prefix) + } + } + + it('aborts a cycle whose relayfile read never returns, names it, and starts the next cycle', async () => { + const mount = new HangingListTreeMount() + const factory = createFactory(config({ issueSource: 'github' }), { + mount, + fleet: new FakeFleetClient(), + 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: 20, + // Far past this test's horizon, so the sweep deadline cannot be what + // ends the pass. Whatever recovers here is the per-call bound. + reconcileTimeoutMs: 60_000, + relayfileOperationTimeoutMs: 50, + }, + }) + try { + mount.hangListTree = true + await mount.hangStarted + const sweepsWhileHung = factory.status().counters.readinessReconcileSweeps ?? 0 + + await vi.waitFor(() => { + const status = factory.status() + // Loud: a counter an operator can alert on, not a silent `stalled`. + expect(status.readinessReconcile?.consecutiveFailures ?? 0).toBeGreaterThanOrEqual(1) + // Named: which call it was waiting on. During the outage `lastError` + // was absent entirely, which gave the operator nothing to act on. + expect(status.readinessReconcile?.lastError) + .toMatch(/relayfile listTree did not respond within \d+ms/u) + expect(status.readinessReconcile?.lastErrorClass).toBe('RelayfileOperationTimeoutError') + // Self-healing: a later cycle STARTED while the dependency is still + // hung. This is the number that stayed frozen for 22 minutes. + expect(status.counters.readinessReconcileSweeps ?? 0).toBeGreaterThan(sweepsWhileHung) + }, { timeout: 5_000 }) + + // And it recovers on its own once the dependency answers again — no + // restart, which is the only thing that cleared this in production. + mount.release() + await vi.waitFor(() => { + const readiness = factory.status().readinessReconcile + expect(readiness?.consecutiveFailures).toBe(0) + expect(readiness?.lastCompletedAtMs ?? 0) + .toBeGreaterThan(readiness?.lastFailureAtMs ?? Number.POSITIVE_INFINITY) + }, { timeout: 5_000 }) + } finally { + mount.release() + await factory.stop() + } + // Two sequential `vi.waitFor` windows do not fit the suite's 5s default. + }, 20_000) + + // With the real cloud mount the TRANSPORT deadline wins the race by design, + // and the mount does not know which phase it was serving — so without the + // orchestrator enriching it, `lastError` names the call but not the context + // and an operator cannot tell one of many list/read sites from another + // (codex on #354). + it('names the phase on a timeout the transport raised, not just its own', async () => { + class TransportTimeoutMount extends CountingEventsMount { + readonly hangStarted: Promise + failListTree = false + #signalHangStarted: () => void = () => undefined + + constructor() { + super() + this.hangStarted = new Promise((resolve) => { this.#signalHangStarted = resolve }) + this.setSubRoot('/linear/issues', 'absent') + } + + override async listTree(prefix: string): Promise { + if (this.failListTree) { + this.#signalHangStarted() + // Exactly what RelayfileCloudMountClient raises when its own signal + // fires: named operation, no phase. + throw new RelayfileOperationTimeoutError('listTree', 300_000) + } + return super.listTree(prefix) + } + } + + const mount = new TransportTimeoutMount() + const factory = createFactory(config({ issueSource: 'github' }), { + mount, + fleet: new FakeFleetClient(), + 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: 20, + reconcileTimeoutMs: 60_000, + // Out of reach: the orchestrator's own race must not be what produces + // the error, or this would not test the transport path at all. + relayfileOperationTimeoutMs: 60_000, + }, + }) + try { + mount.failListTree = true + await mount.hangStarted + + await vi.waitFor(() => { + const readiness = factory.status().readinessReconcile + expect(readiness?.consecutiveFailures ?? 0).toBeGreaterThanOrEqual(1) + expect(readiness?.lastErrorClass).toBe('RelayfileOperationTimeoutError') + // The parenthesised phase is the part the transport could not supply. + expect(readiness?.lastError) + .toMatch(/relayfile listTree did not respond within \d+ms \(.+\)/u) + }, { timeout: 5_000 }) + } finally { + mount.failListTree = false + await factory.stop() + } + }, 20_000) + + // The control for the test above. Same fixture, same hang, only the + // per-call bound moved out past the test horizon: the cycle must then NOT + // abort. Without this, a test that passed because of something else in the + // loop would look like proof of the abort. + it('control: with the per-call bound out of reach the same hang freezes the loop', async () => { + const mount = new HangingListTreeMount() + const factory = createFactory(config({ issueSource: 'github' }), { + mount, + fleet: new FakeFleetClient(), + 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: 20, + reconcileTimeoutMs: 60_000, + // The pre-#351 behaviour: no reachable bound on the call. + relayfileOperationTimeoutMs: 60_000, + }, + }) + try { + mount.hangListTree = true + await mount.hangStarted + const sweepsWhileHung = factory.status().counters.readinessReconcileSweeps ?? 0 + + // ~12 reconcile intervals. Long enough that a working abort would have + // fired many times over. + await new Promise((resolve) => { setTimeout(resolve, 250) }) + + const status = factory.status() + expect(status.counters.readinessReconcileSweeps ?? 0).toBe(sweepsWhileHung) + expect(status.readinessReconcile?.consecutiveFailures).toBe(0) + expect(status.readinessReconcile?.lastError).toBeUndefined() + } finally { + mount.release() + await factory.runOnce().catch(() => undefined) + await factory.stop() + } + }) + it('reports a pass still in flight past the stall threshold as stalled rather than healthy', async () => { const mount = new CountingEventsMount() mount.setSubRoot('/linear/issues', 'absent') diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index c54030a0..efa118fe 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -3,6 +3,12 @@ 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_RELAYFILE_OPERATION_TIMEOUT_MS, + RelayfileOperationTimeoutError, + relayfileTimeoutWithPhase, + withRelayfileCallDeadline, +} from '../mount/relayfile-operation-timeout' import { linearByStatePath, linearByIdPath, linearByUuidPath } from '../constants/linear' import { stateResolutionFromIds, type FactoryStateResolution } from '../linear/state-resolver' import { GithubMergeGate, closeProbePr, type GhRunner, type GithubMergeGate as GithubMergeGatePort } from '../github' @@ -480,9 +486,28 @@ const DISPATCH_FAILURE_HANDOFF_UNRESOLVED_TTL_MS = 5 * 60_000 const DEFAULT_LIVE_HEARTBEAT_INTERVAL_MS = 15_000 const REMOTE_OPERATION_PROGRESS_INTERVAL_MS = 15_000 const REMOTE_OPERATION_SLOW_WARN_MS = 30_000 +/** How far the cycle-level relayfile backstop sits above the transport deadline. */ +const RELAYFILE_OPERATION_BACKSTOP_RATIO = 1.25 const DISCOVERY_SWEEP_LEASE_MS = 5 * 60_000 const DISCOVERY_SWEEP_RENEW_MS = 30_000 const READINESS_RECONCILE_FAILURE_THRESHOLD = 3 + +/** + * A relayfile fault a swallowing catch must not turn into "no result". + * + * A 429 already escaped every one of these (#297): it is a fact about the + * dependency, not about the one item being read, so folding it into an empty + * result reports a clean pass over work that was never served. + * + * A per-call timeout (#351) is the same fact in a worse costume. The read that + * hung will hang for the next item too, and `#githubIssuePaths` swallowing it + * turns a wedged dependency into a *successful* sweep that discovered zero + * issues — `consecutiveFailures: 0`, no `lastError`, nothing dispatched. That + * is the exact silence this bound exists to remove, so it escapes here too. + */ +const isPassWideRelayfileFault = (error: unknown): boolean => + Boolean(relayfileOverload(error)) || error instanceof RelayfileOperationTimeoutError + 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`. */ @@ -778,6 +803,14 @@ export class FactoryLoop implements Factory { #readinessReconcileInFlight?: Promise #readinessReconcileIntervalMs = 60_000 #readinessReconcileTimeoutMs = DEFAULT_READINESS_RECONCILE_TIMEOUT_MS + /** + * Deadline for ONE relayfile call (#351). + * + * `#readinessReconcileTimeoutMs` bounds the sweep; this bounds the calls + * inside it. Only the second one can reach a dependency call that never + * returns: a deadline checked between awaits never regains control to check. + */ + #relayfileOperationTimeoutMs = DEFAULT_RELAYFILE_OPERATION_TIMEOUT_MS // 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 @@ -939,6 +972,9 @@ export class FactoryLoop implements Factory { constructor(config: FactoryConfig, ports: FactoryPorts) { this.#readOnly = ports.readOnly ?? false this.#config = config + // 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.#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 @@ -1522,6 +1558,7 @@ export class FactoryLoop implements Factory { // `start()` overrides skip the schema's cross-field check, so re-apply its // floor here: a deadline under one interval would kill every pass. this.#readinessReconcileTimeoutMs = Math.max(options.reconcileTimeoutMs, options.reconcileIntervalMs) + this.#relayfileOperationTimeoutMs = options.relayfileOperationTimeoutMs this.#liveConnectStartedAtMs = this.#clock.now() this.#liveReplaySkewMarginMs = options.replaySkewMarginMs const highWatermark = await this.#currentEventHighWatermark() @@ -1688,6 +1725,8 @@ export class FactoryLoop implements Factory { replaySkewMarginMs: overrides.replaySkewMarginMs ?? this.#config.liveSubscription.replaySkewMarginMs, reconcileIntervalMs: overrides.reconcileIntervalMs ?? this.#config.liveSubscription.reconcileIntervalMs, reconcileTimeoutMs: overrides.reconcileTimeoutMs ?? this.#config.liveSubscription.reconcileTimeoutMs, + relayfileOperationTimeoutMs: overrides.relayfileOperationTimeoutMs + ?? this.#config.liveSubscription.relayfileOperationTimeoutMs, } } @@ -3446,7 +3485,7 @@ export class FactoryLoop implements Factory { ? { available: false } : { available: true, highWatermark } } catch (error) { - if (relayfileOverload(error)) throw error + if (isPassWideRelayfileFault(error)) throw error this.#increment('discoveryHighWatermarkFailures') this.#logger.warn?.('[factory] discovery high-watermark unavailable; using a full sweep without caching', { error: describeError(error).errorMessage, @@ -3469,7 +3508,7 @@ export class FactoryLoop implements Factory { }) events = [...page.events].sort(compareDiscoveryEvents) } catch (error) { - if (relayfileOverload(error)) throw error + if (isPassWideRelayfileFault(error)) throw error this.#logger.warn?.('[factory] discovery change feed unavailable; refreshing tree prefixes once', { error: describeError(error).errorMessage, }) @@ -4200,7 +4239,22 @@ export class FactoryLoop implements Factory { } try { - const result = await fn() + // The bound the 2026-08-23 wedge needed (#351). `#readinessReconcileTimeoutMs` + // is 90 minutes and rejects only the WAIT — the sweep keeps its discovery + // lease and the next cycle coalesces onto it — so a call that never + // returns stopped dispatch for as long as the process lived. Bounding the + // call instead makes the pass unwind, which releases the lease and lets + // the next cycle actually start. + // + // Slightly above the transport's own deadline so the mount's cancellation + // wins the race and reports the more precise error; this is the backstop + // for mounts that cannot honour a signal. + const result = await withRelayfileCallDeadline( + operation, + details.phase, + this.#relayfileOperationBackstopMs(), + fn, + ) const elapsedMs = this.#elapsedSince(startedAtMs) const count = opts.count?.(result) if (opts.logComplete) { @@ -4259,7 +4313,11 @@ export class FactoryLoop implements Factory { error: describeError(error).errorMessage, }) } - throw error + // The transport deadline is meant to win the race above, and the mount + // does not know which phase it was serving. Without this, `lastError` + // would name the call but not the context — one of many list/read sites + // (codex on #354). + throw relayfileTimeoutWithPhase(error, details.phase) } finally { if (progressTimer) { clearInterval(progressTimer) @@ -4267,6 +4325,19 @@ export class FactoryLoop implements Factory { } } + /** + * The cycle-level backstop budget: the transport deadline plus a margin. + * + * Proportional rather than a flat grace so it holds at both ends of the + * range — a five-minute production budget and a millisecond test budget both + * leave the transport room to fire first. + */ + #relayfileOperationBackstopMs(): number | undefined { + const timeoutMs = this.#relayfileOperationTimeoutMs + if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) return undefined + return Math.ceil(timeoutMs * RELAYFILE_OPERATION_BACKSTOP_RATIO) + } + #elapsedSince(startedAtMs: number): number { return Math.max(0, this.#clock.now() - startedAtMs) } @@ -7299,7 +7370,7 @@ export class FactoryLoop implements Factory { this.#githubIssuePathIndexReady = true return [...issuePaths.values()].sort() } catch (error) { - if (relayfileOverload(error)) throw error + if (isPassWideRelayfileFault(error)) throw error this.#githubIssuePathIndexReady = false this.#increment('githubIssueListFailures') this.#logger.warn?.('[factory] failed to list GitHub issue source tree', error) @@ -7314,7 +7385,7 @@ export class FactoryLoop implements Factory { const { content } = await this.#readRelayfileFile(indexPath, 'GitHub issue index discovery') parsed = parseJsonContent(content) } catch (error) { - if (relayfileOverload(error)) throw error + if (isPassWideRelayfileFault(error)) throw error return undefined } if (!Array.isArray(parsed)) return undefined @@ -7424,7 +7495,7 @@ export class FactoryLoop implements Factory { } this.#increment('githubIssueMirrorsCreated') } catch (error) { - if (relayfileOverload(error)) throw error + if (isPassWideRelayfileFault(error)) throw error this.#logger.error?.('[factory] failed to ingest GitHub issue', error) } } @@ -8258,7 +8329,7 @@ export class FactoryLoop implements Factory { this.#indexDependencyIssue(resolvedIssue) return resolvedIssue } catch (error) { - if (relayfileOverload(error)) throw error + if (isPassWideRelayfileFault(error)) throw error if (isMissingIssueFileError(error) && isIssuePathUnderRoot(path)) { this.#increment('phantomSkipped') this.#logger.debug?.('[factory] skipped missing issue file discovered from issue tree', { path }) @@ -9415,7 +9486,7 @@ export class FactoryLoop implements Factory { try { paths = await this.#listRelayfileTree(root, 'exact-head PR confirmation') } catch (error) { - if (relayfileOverload(error)) throw error + if (isPassWideRelayfileFault(error)) throw error continue } for (const path of paths) { @@ -9686,7 +9757,7 @@ export class FactoryLoop implements Factory { try { return await this.#listRelayfileTree(root, 'published PR confirmation') } catch (error) { - if (relayfileOverload(error)) throw error + if (isPassWideRelayfileFault(error)) throw error return [] } }))).flat() @@ -11486,7 +11557,7 @@ export class FactoryLoop implements Factory { } } } catch (error) { - if (relayfileOverload(error)) throw error + if (isPassWideRelayfileFault(error)) throw error this.#logger.warn?.('[factory] unable to list GitHub issue comments for replay', { prefix, error }) } } @@ -14174,7 +14245,7 @@ export class FactoryLoop implements Factory { const tree = await this.#listRelayfileTree(root, 'PR meta path discovery') found.push(...tree.filter((path) => path.endsWith('.json') && numberSegment.test(path))) } catch (error) { - if (relayfileOverload(error)) throw error + if (isPassWideRelayfileFault(error)) throw error // try the next root } } @@ -15495,7 +15566,7 @@ export class FactoryLoop implements Factory { try { return await this.#listRelayfileTree(prefix, 'Slack identity lookup') } catch (error) { - if (relayfileOverload(error)) throw error + if (isPassWideRelayfileFault(error)) throw error return [] } } @@ -18847,7 +18918,7 @@ const resolveIssuePrFromMount = async ( try { for (const path of await listTree(root)) paths.add(path) } catch (error) { - if (relayfileOverload(error)) throw error + if (isPassWideRelayfileFault(error)) throw error listErrors.push(error) } } diff --git a/src/types.ts b/src/types.ts index 68c43fae..cdb05eb1 100644 --- a/src/types.ts +++ b/src/types.ts @@ -125,6 +125,12 @@ export interface FactoryLiveSubscriptionOptions { * sized above realistic worst-case mirror hydration, not to the interval. */ reconcileTimeoutMs: number + /** + * Deadline for one relayfile call inside a sweep (#351). Bounds what + * `reconcileTimeoutMs` cannot: a single dependency call that never returns, + * which no deadline checked *between* awaits can reach. + */ + relayfileOperationTimeoutMs: number } /**