From 71e5c87718f295afc639b29bdc6876384bc54141 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Tue, 25 Aug 2026 08:54:25 +0200 Subject: [PATCH] fix(mount): bound the relayfile change feed so a hung tail read cannot wedge the sweep MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #354 bounded every relayfile read except the two the discovery sweep makes first. `getEventHighWatermark()` and `getEvents()` went straight to the SDK with no deadline and no signal, and per `relayfile-operation-timeout.ts` the SDK attaches an `AbortSignal` to its `fetch` only when the caller supplies one — so both were bare `fetch()` calls that can wait forever. That is the deployed wedge. `#prepareDiscoverySession` calls `#discoveryHighWatermark()` -> `mount.getEventHighWatermark()` as the FIRST network read after `#runOnceWithDiscoveryFence` claims the discovery lease, before `#performRunOnce` logs anything. On the live instance (0.1.74, boot 2026-08-25T06:13:07Z) the sweep that started 06:17:21.603Z was still in flight 24 minutes later with `consecutiveFailures: 0`, and the daemon's own output ring over that window contains no `run-once started`, no ready-issue read progress, and no discovery-phase `listTree` — only the live event drain's. The sweep never reached enumeration, and 24 minutes is well past the 5-minute budget every `#bounded` call already carries, so the await it was held on had to be one outside that wrapper. There are exactly two. The sweep's own 90-minute deadline cannot substitute. It rejects the wait and leaves `runOnce()` running, so the next cycle coalesces onto the same wedged promise; a per-call rejection unwinds the pass and releases the discovery lease instead. Both feed methods now go through the existing `#bounded()` helper. The cursor-paged `getEvents` gets real cancellation — `GetEventsOptions` carries a `signal`. `listLastNChanges` cannot: its `ProactiveRequestContext` has no signal field, so it is bounded by `withRelayfileCallDeadline`'s race, the abandoned-wait backstop that module documents for exactly this case. Weaker (the socket stays live) but it is the half that matters: the rejection unwinds the sweep. Tests, must-fire and must-not-fire for each: - must-fire: with a 25 ms budget and a change feed that never answers, `getEventHighWatermark()`, `getEvents({ last })` and `getEvents({ cursor })` each reject as `RelayfileOperationTimeoutError` naming the operation, and the cursor path's signal is `aborted`. Verified fail-first: with the fix reverted all three fail by TIMING OUT at vitest's 5 s default — the production mechanism, not an assertion detail. - must-not-fire: a served read under a 60 s budget still returns its watermark and still makes exactly one call; with `operationTimeoutMs: 0` the hung read stays pending, so the rejections above are attributable to the budget and not to the wrapper. Not fixed here, and filed as separate defects on factory-cloud#55: `confirmWrite`'s `getOp` loop checks its deadline BETWEEN calls, so it can never interrupt one; and a `503 agent_host_unavailable` release against a host node offline for days is retried every second forever. Co-Authored-By: Claude Opus 5 Session-Id: 6534f313-3c75-412b-bfd4-6ac9b59b9405 --- .../relayfile-cloud-mount-client.test.ts | 125 +++++++++++++++++- src/mount/relayfile-cloud-mount-client.ts | 27 +++- 2 files changed, 148 insertions(+), 4 deletions(-) diff --git a/src/mount/relayfile-cloud-mount-client.test.ts b/src/mount/relayfile-cloud-mount-client.test.ts index 1b5ec8b2..0e6f34e3 100644 --- a/src/mount/relayfile-cloud-mount-client.test.ts +++ b/src/mount/relayfile-cloud-mount-client.test.ts @@ -1461,7 +1461,12 @@ describe('RelayfileCloudMountClient', () => { 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 } }) + // The change feed now carries the same deadline signal the tree read does. + expect(fake.getEventsCalls[0]).toEqual({ + workspaceId: 'rw_test', + opts: { cursor: 'evt-0', limit: 10, signal: expect.any(AbortSignal) }, + }) + expect(fake.getEventsCalls[0]?.opts?.signal?.aborted).toBe(false) }) it('paginates listTree to exhaustion', async () => { @@ -1617,6 +1622,124 @@ describe('RelayfileCloudMountClient', () => { expect(client.seenSignal?.aborted).toBe(true) }) + // The hole #354 left, and the one the deployed sweep hung in: the change + // feed. `#prepareDiscoverySession` calls `getEventHighWatermark()` as the + // FIRST network read after it claims the discovery lease, and both feed + // methods went straight to the SDK with no deadline and no signal. + class HangingChangeFeedClient extends FakeRelayFileClient { + seenSignal?: AbortSignal + + override async listLastNChanges( + limit: number, + context?: { workspaceId: string }, + ): Promise { + this.listLastNChangesCalls.push({ limit, context }) + return await new Promise(() => undefined) + } + + override async getEvents( + workspaceId: string, + opts?: { cursor?: string; limit?: number; provider?: string; last?: number; signal?: AbortSignal }, + ): Promise { + this.getEventsCalls.push({ workspaceId, opts }) + this.seenSignal = opts?.signal + return await new Promise((_resolve, reject) => { + opts?.signal?.addEventListener('abort', () => { + reject((opts.signal as AbortSignal & { reason?: unknown }).reason) + }) + }) + } + } + + it('bounds the discovery high-watermark read that wedged the sweep', async () => { + const client = new HangingChangeFeedClient() + const mount = new RelayfileCloudMountClient({ + workspaceId: 'rw_test', + client, + operationTimeoutMs: 25, + }) + + await expect(mount.getEventHighWatermark()).rejects.toMatchObject({ + name: 'RelayfileOperationTimeoutError', + operation: 'listLastNChanges', + timeoutMs: 25, + }) + }) + + it('bounds the change-log tail read behind a provider-filtered getEvents', async () => { + const client = new HangingChangeFeedClient() + const mount = new RelayfileCloudMountClient({ + workspaceId: 'rw_test', + client, + operationTimeoutMs: 25, + }) + + await expect(mount.getEvents({ last: 100, limit: 100 })).rejects.toMatchObject({ + name: 'RelayfileOperationTimeoutError', + operation: 'listLastNChanges', + }) + }) + + it('cancels a cursor-paged getEvents at the transport', async () => { + const client = new HangingChangeFeedClient() + const mount = new RelayfileCloudMountClient({ + workspaceId: 'rw_test', + client, + operationTimeoutMs: 25, + }) + + // `GetEventsOptions` carries a `signal`, unlike `listLastNChanges`'s + // context — so this half gets real cancellation, not just an abandoned + // wait. + await expect(mount.getEvents({ limit: 10 })).rejects.toMatchObject({ + name: 'RelayfileOperationTimeoutError', + operation: 'getEvents', + }) + expect(client.seenSignal?.aborted).toBe(true) + }) + + // MUST-NOT-FIRE. A deadline that rejected a healthy read would replace one + // wedge with an outage, and a deadline that fired with no budget + // configured would mean the rejections above prove nothing about the + // budget. + it('leaves a served change-feed read alone under a generous budget', async () => { + const client = new FakeRelayFileClient() + client.events = [{ + eventId: '11', + type: 'file.updated' as const, + path: '/github/repos/AgentWorkforce__factory/issues/by-id/1.json', + provider: 'github', + revision: '2', + timestamp: '2026-01-01T00:00:00.000Z', + contentType: 'application/json', + }] + const mount = new RelayfileCloudMountClient({ + workspaceId: 'rw_test', + client, + operationTimeoutMs: 60_000, + }) + + await expect(mount.getEventHighWatermark()).resolves.toBe('11') + expect(client.listLastNChangesCalls).toEqual([{ limit: 10, context: { workspaceId: 'rw_test' } }]) + }) + + it('leaves the change-feed read unbounded when no budget is configured', async () => { + const client = new HangingChangeFeedClient() + const mount = new RelayfileCloudMountClient({ + workspaceId: 'rw_test', + client, + operationTimeoutMs: 0, + }) + + const pending = mount.getEventHighWatermark() + 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') + void pending.catch(() => undefined) + }) + 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 05921b14..96adbe89 100644 --- a/src/mount/relayfile-cloud-mount-client.ts +++ b/src/mount/relayfile-cloud-mount-client.ts @@ -1013,9 +1013,29 @@ export class RelayfileCloudMountClient implements MountClient { }) } + /** + * The change-log tail read, under the same per-call deadline as every other + * relayfile read (#351 follow-up). + * + * `listLastNChanges` takes a `ProactiveRequestContext` with no `signal` + * field, so this one cannot be cancelled at the transport and is bounded by + * `withRelayfileCallDeadline`'s race instead — the abandoned-wait backstop + * that module documents for mounts which cannot honour a signal. That is + * weaker than `listTree`'s cancellation (the socket and the SDK's retry loop + * stay live) but it is the half that matters here: the rejection unwinds the + * sweep, which releases the discovery lease, so the next cycle starts clean + * instead of coalescing onto a wedged `runOnce()`. + */ + #boundedListLastNChanges(limit: number): Promise<{ events: ChangeEvent[] }> { + const listLastNChanges = this.#client.listLastNChanges + if (!listLastNChanges) throw new Error('relayfile client cannot list recent changes') + return this.#bounded('listLastNChanges', this.#operationTimeoutMs, () => + listLastNChanges.call(this.#client, limit, { workspaceId: this.workspaceId })) + } + async getEvents(opts: { cursor?: string; limit?: number; provider?: string; last?: number }): Promise { if (opts.last !== undefined && this.#client.listLastNChanges) { - const response = await this.#client.listLastNChanges(opts.last, { workspaceId: this.workspaceId }) + const response = await this.#boundedListLastNChanges(opts.last) const events = opts.provider ? response.events.filter((event) => eventProvider(event) === opts.provider) : response.events @@ -1024,7 +1044,8 @@ export class RelayfileCloudMountClient implements MountClient { nextCursor: null, } } - const response = await this.#client.getEvents(this.workspaceId, opts) + const response = await this.#bounded('getEvents', this.#operationTimeoutMs, (signal) => + this.#client.getEvents(this.workspaceId, { ...opts, ...(signal ? { signal } : {}) })) return { events: response.events as unknown as EventPage['events'], nextCursor: response.nextCursor, @@ -1033,7 +1054,7 @@ export class RelayfileCloudMountClient implements MountClient { async getEventHighWatermark(opts: { provider?: string } = {}): Promise { if (!this.#client.listLastNChanges) return undefined - const response = await this.#client.listLastNChanges(10, { workspaceId: this.workspaceId }) + const response = await this.#boundedListLastNChanges(10) const events = opts.provider ? response.events.filter((event) => event.resource.provider === opts.provider) : response.events