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