From 06637a6371dbcd92aacc7f3e377b78aef1d848c9 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 23 Aug 2026 23:55:09 +0200 Subject: [PATCH 1/2] fix(orchestrator): bound each relayfile call so a hung read cannot wedge reconcile (#351) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `readinessReconcile` started a cycle at 20:29:40Z on the live container and had not finished it 22 minutes later. Heartbeat still ticking, fleet agent online, `fleetControlPlane: closed`, and `consecutiveFailures: 0` — it was not erroring, it was blocked inside a call, and a failure counter cannot see a hang. Every other dependency boundary in the cycle is already bounded: the fleet roster probe at 5s through `FleetControlPlaneCircuit`, GitHub REST reads at 30s via `AbortSignal.timeout`, spawn and resume by the spawn-ack deadline. Relayfile was not. `@relayfile/sdk`'s `performRequest` attaches a signal to its `fetch` only when the caller supplies one, and `RelayfileCloudMountClient` never did — so every `readFile`, `listTree` and `ensureSubRoot` was a bare `fetch()` that could wait forever. `listTree` also walked an unbounded cursor loop, and `ensureSubRoot` accepted a `timeoutMs` and discarded it, so `#ensureGithubIngestionReady` passing 90_000 bounded nothing. The #296 sweep deadline could not cover this. It is 90 minutes by design — below realistic cold-mirror hydration a slow boot becomes a crash loop — and it rejects only the *wait*: `runOnce()` keeps its discovery lease, so the next cycle coalesces onto the same wedged promise. Nothing but a restart recovered. This bounds the call instead: - `RelayfileCloudMountClient` derives a deadline per operation and passes its signal to the SDK, so the request is cancelled rather than abandoned — an abandoned wait leaves the socket and the SDK's retry loop live and hands the next cycle the same in-flight read. One deadline covers the whole `listTree` walk, and `ensureSubRoot` honours the `timeoutMs` it used to drop. - `#withRelayfileOperation` races the same budget as a backstop for mounts that cannot honour a signal, and throws `RelayfileOperationTimeoutError` naming the operation and phase. That reaches the existing failure path: `consecutiveFailures` rises, `lastError` says what it was waiting on, and `lastErrorClass` publishes through the existing allowlist. - A timeout now escapes the swallowing catches alongside a relayfile 429 (#297). Without that, `#githubIssuePaths` folded it into an empty result and a wedged dependency became a *successful* sweep that discovered zero issues — the same silence wearing a different costume. - Failing the call unwinds the pass, which releases the discovery lease, so the next cycle starts clean instead of joining the hang. `liveSubscription.relayfileOperationTimeoutMs` defaults to 5 minutes, two orders of magnitude below the sweep deadline. That is safe precisely where the sweep deadline is not: #36's 61-minute cold-mirror reconcile is spread across thousands of calls, so a per-call bound does not re-create the crash-loop risk. Tests: a reconcile whose `listTree` never resolves aborts, counts, names the call, starts the next cycle, and recovers with no restart — with a control that moves the bound out of reach and asserts the loop then freezes, so the test cannot pass for another reason. Verified by ablation: removing the bound fails the test at the counter assertion. Co-Authored-By: Claude Opus 5 --- docs/deployed-diagnostics.md | 25 ++- src/cli/fleet.ts | 3 + src/config/schema.ts | 14 ++ .../relayfile-cloud-mount-client.test.ts | 85 ++++++++++- src/mount/relayfile-cloud-mount-client.ts | 134 +++++++++++++++-- src/mount/relayfile-operation-timeout.ts | 142 ++++++++++++++++++ src/orchestrator/factory.test.ts | 137 +++++++++++++++++ src/orchestrator/factory.ts | 92 ++++++++++-- src/types.ts | 6 + 9 files changed, 603 insertions(+), 35 deletions(-) create mode 100644 src/mount/relayfile-operation-timeout.ts 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..d1317d42 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,84 @@ 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('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..ae8f00fc 100644 --- a/src/mount/relayfile-cloud-mount-client.ts +++ b/src/mount/relayfile-cloud-mount-client.ts @@ -22,6 +22,14 @@ import { type WriteQueuedResponse, } from '@relayfile/sdk' import { RelayfileSetup } from '@relayfile/sdk/cli' +import { + DEFAULT_RELAYFILE_OPERATION_TIMEOUT_MS, + RelayfileOperationTimeoutError, + isRelayfileCallAbort, + relayfileCallBudgetMs, + relayfileCallDeadline, + withRelayfileCallDeadline, +} from './relayfile-operation-timeout' import { existsSync } from 'node:fs' import { isAbsolute, join, resolve } from 'node:path' @@ -260,10 +268,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 +359,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 +370,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 +672,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 +762,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 +1067,18 @@ 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, falling back to the + // client-wide operation budget. + 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', 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..acabe413 --- /dev/null +++ b/src/mount/relayfile-operation-timeout.ts @@ -0,0 +1,142 @@ +/** + * 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' + } +} + +/** 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' +} + +/** 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 + +/** 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..03391329 100644 --- a/src/orchestrator/factory.test.ts +++ b/src/orchestrator/factory.test.ts @@ -13391,6 +13391,143 @@ 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) + + // 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..e59e9efb 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -3,6 +3,11 @@ 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, + 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 +485,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 +802,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 +971,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 +1557,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 +1724,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 +3484,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 +3507,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 +4238,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) { @@ -4267,6 +4320,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 +7365,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 +7380,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 +7490,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 +8324,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 +9481,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 +9752,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 +11552,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 +14240,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 +15561,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 +18913,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 } /** From a6e024cdcab8de53cffcdbe5509b8c2e1b4ea13a Mon Sep 17 00:00:00 2001 From: Khaliq Date: Mon, 24 Aug 2026 00:07:15 +0200 Subject: [PATCH 2/2] fix(mount): cap ensureSubRoot at the tighter budget and name the phase on transport timeouts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two P2s from codex on #354, both real. `ensureSubRoot`'s explicit `timeoutMs` replaced the client-wide budget instead of capping it. With `relayfileOperationTimeoutMs` configured below 72s, the reconcile caller's hard-coded 90_000 left the transport running past the orchestrator's 1.25x backstop — so the backstop abandoned the wait rather than the transport cancelling the call, which is precisely the behaviour this change exists to avoid. It now takes the tighter of the two. The transport deadline is designed to win the race, and the mount does not know which phase it was serving, so `lastError` read `relayfile listTree did not respond within 300000ms` with no way to tell one of many list/read contexts from another. The orchestrator now enriches a phase-less transport timeout with the phase it knows as the error crosses its boundary, keeping the original as `cause`. That restores the message the PR and the diagnostics doc describe. Both covered, and both verified by ablation: reverting the cap times the ensureSubRoot test out, and reverting the enrichment fails the new orchestrator test on the phase assertion specifically. Co-Authored-By: Claude Opus 5 --- .../relayfile-cloud-mount-client.test.ts | 20 ++++++ src/mount/relayfile-cloud-mount-client.ts | 10 ++- src/mount/relayfile-operation-timeout.ts | 43 ++++++++++-- src/orchestrator/factory.test.ts | 67 +++++++++++++++++++ src/orchestrator/factory.ts | 7 +- 5 files changed, 139 insertions(+), 8 deletions(-) diff --git a/src/mount/relayfile-cloud-mount-client.test.ts b/src/mount/relayfile-cloud-mount-client.test.ts index d1317d42..aa15968b 100644 --- a/src/mount/relayfile-cloud-mount-client.test.ts +++ b/src/mount/relayfile-cloud-mount-client.test.ts @@ -1546,6 +1546,26 @@ describe('RelayfileCloudMountClient', () => { }) }) + 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({ diff --git a/src/mount/relayfile-cloud-mount-client.ts b/src/mount/relayfile-cloud-mount-client.ts index ae8f00fc..f12c4e48 100644 --- a/src/mount/relayfile-cloud-mount-client.ts +++ b/src/mount/relayfile-cloud-mount-client.ts @@ -28,6 +28,7 @@ import { isRelayfileCallAbort, relayfileCallBudgetMs, relayfileCallDeadline, + tighterRelayfileBudgetMs, withRelayfileCallDeadline, } from './relayfile-operation-timeout' import { existsSync } from 'node:fs' @@ -1069,11 +1070,14 @@ export class RelayfileCloudMountClient implements MountClient { // `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, falling back to the - // client-wide operation budget. + // 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.#bounded('ensureSubRoot', opts?.timeoutMs ?? this.#operationTimeoutMs, (signal) => + await this.#bounded('ensureSubRoot', tighterRelayfileBudgetMs(opts?.timeoutMs, this.#operationTimeoutMs), (signal) => this.#client.listTree(this.workspaceId, { path: prefix, depth: 1, diff --git a/src/mount/relayfile-operation-timeout.ts b/src/mount/relayfile-operation-timeout.ts index acabe413..f9b25dfe 100644 --- a/src/mount/relayfile-operation-timeout.ts +++ b/src/mount/relayfile-operation-timeout.ts @@ -45,6 +45,45 @@ export class RelayfileOperationTimeoutError extends Error { } } +/** 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 @@ -52,10 +91,6 @@ export function isRelayfileCallAbort(error: unknown): boolean { return error.name === 'TimeoutError' || error.name === 'AbortError' } -/** 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 - /** 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. */ diff --git a/src/orchestrator/factory.test.ts b/src/orchestrator/factory.test.ts index 03391329..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' @@ -13484,6 +13485,72 @@ describe('FactoryLoop', () => { // 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 diff --git a/src/orchestrator/factory.ts b/src/orchestrator/factory.ts index e59e9efb..efa118fe 100644 --- a/src/orchestrator/factory.ts +++ b/src/orchestrator/factory.ts @@ -6,6 +6,7 @@ import { DEFAULT_READINESS_RECONCILE_TIMEOUT_MS, FactoryConfigSchema, type Facto import { DEFAULT_RELAYFILE_OPERATION_TIMEOUT_MS, RelayfileOperationTimeoutError, + relayfileTimeoutWithPhase, withRelayfileCallDeadline, } from '../mount/relayfile-operation-timeout' import { linearByStatePath, linearByIdPath, linearByUuidPath } from '../constants/linear' @@ -4312,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)