From 835d3cec3f7de60588fd53e89917dfdd876579c3 Mon Sep 17 00:00:00 2001 From: me2seeks Date: Wed, 26 Aug 2026 10:59:03 +0800 Subject: [PATCH 1/2] fix(desktop): bound reconnectable read retries Generated-by: Codex --- ...runtime-host-reconnecting-ipc-main.test.ts | 76 +++++++++++++++++++ .../runtime-host-reconnecting-ipc-main.ts | 55 ++++++++++++-- 2 files changed, 123 insertions(+), 8 deletions(-) diff --git a/apps/desktop/src/main/__tests__/runtime-host-reconnecting-ipc-main.test.ts b/apps/desktop/src/main/__tests__/runtime-host-reconnecting-ipc-main.test.ts index 75181fca24..46077bc0e4 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-reconnecting-ipc-main.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-reconnecting-ipc-main.test.ts @@ -169,6 +169,82 @@ test("bounds reconciliation when no replacement candidate becomes available", as } }); +test("bounds reconnectable reads while no replacement candidate is available", async () => { + const ipc = ipcHarness(); + const router = new RuntimeHostReconnectingIpcMain(ipc, { + reconnectableReadWaitTimeoutMs: 5, + }); + const target = router.createTarget("target-a"); + target.handleReconnectableRead?.("projects:getSnapshot", async () => ({ projects: [] })); + router.activate("target-a"); + target.removeHandler("projects:getSnapshot"); + + const reads = Array.from({ length: 64 }, () => + ipc.invoke("projects:getSnapshot", scope("target-a")).then( + (value) => ({ ok: true as const, value }), + (error: unknown) => ({ ok: false as const, error }), + ), + ); + const settled = Promise.all(reads); + try { + const result = await Promise.race([ + settled, + new Promise<{ readonly timedOut: true }>((resolve) => + setTimeout(() => resolve({ timedOut: true }), 50), + ), + ]); + assert.ok(Array.isArray(result), "Reconnectable reads did not settle at their retry deadline"); + assert.equal(result.length, 64); + for (const read of result) { + assert.equal(read.ok, false); + if (!read.ok) { + assert.ok(read.error instanceof Error); + assert.match(read.error.message, /did not reconnect before the read retry deadline/); + } + } + } finally { + router.close(); + await settled; + } +}); + +test("does not reset a reconnectable read deadline across failed replacements", async (t) => { + const ipc = ipcHarness(); + const router = new RuntimeHostReconnectingIpcMain(ipc, { + reconnectableReadWaitTimeoutMs: 15, + }); + let now = 0; + t.mock.method(Date, "now", () => now); + router.activate("target-a"); + let attempts = 0; + const maximumAttempts = 20; + const installFailingTarget = (): void => { + const target = router.createTarget("target-a"); + target.handleReconnectableRead?.("projects:getSnapshot", async () => { + attempts += 1; + now += 5; + target.removeHandler("projects:getSnapshot"); + if (attempts < maximumAttempts) installFailingTarget(); + throw new RuntimeHostOperationError( + "project.catalog.query", + "host_draining", + "Runtime Host is draining", + ); + }); + }; + installFailingTarget(); + + try { + await assert.rejects( + () => ipc.invoke("projects:getSnapshot", scope("target-a")), + /did not reconnect before the read retry deadline/, + ); + assert.equal(attempts, 4); + } finally { + router.close(); + } +}); + test("retries only reconciliation when its replacement connection is lost", async () => { const ipc = ipcHarness(); const router = new RuntimeHostReconnectingIpcMain(ipc); diff --git a/apps/desktop/src/main/runtime-host-reconnecting-ipc-main.ts b/apps/desktop/src/main/runtime-host-reconnecting-ipc-main.ts index 3875fb5f4d..9bc1d2bec4 100644 --- a/apps/desktop/src/main/runtime-host-reconnecting-ipc-main.ts +++ b/apps/desktop/src/main/runtime-host-reconnecting-ipc-main.ts @@ -34,15 +34,24 @@ type ReconcileIpcHandler = ( type ReconciliationUnavailableIpcHandler = ReconcileIpcHandler; const DEFAULT_RECONCILIATION_WAIT_TIMEOUT_MS = 15_000; +const DEFAULT_RECONNECTABLE_READ_WAIT_TIMEOUT_MS = 15_000; export interface RuntimeHostReconnectingIpcMainOptions { readonly reconciliationWaitTimeoutMs?: number; + readonly reconnectableReadWaitTimeoutMs?: number; } -class ReconciliationWaitExpiredError extends Error { +class HandlerWaitExpiredError extends Error { constructor() { - super("Runtime Host reconciliation replacement wait expired"); - this.name = "ReconciliationWaitExpiredError"; + super("Runtime Host replacement wait expired"); + this.name = "HandlerWaitExpiredError"; + } +} + +class ReconnectableReadWaitExpiredError extends Error { + constructor() { + super("Runtime Host did not reconnect before the read retry deadline"); + this.name = "ReconnectableReadWaitExpiredError"; } } @@ -91,6 +100,7 @@ export class RuntimeHostReconnectingIpcMain { readonly #slots = new Map(); readonly #activeEpochs = new Set(); readonly #reconciliationWaitTimeoutMs: number; + readonly #reconnectableReadWaitTimeoutMs: number; #closed = false; constructor( @@ -104,6 +114,15 @@ export class RuntimeHostReconnectingIpcMain { throw new TypeError("Runtime Host reconciliation wait timeout must be positive"); } this.#reconciliationWaitTimeoutMs = reconciliationWaitTimeoutMs; + const reconnectableReadWaitTimeoutMs = + options.reconnectableReadWaitTimeoutMs ?? DEFAULT_RECONNECTABLE_READ_WAIT_TIMEOUT_MS; + if ( + !Number.isSafeInteger(reconnectableReadWaitTimeoutMs) || + reconnectableReadWaitTimeoutMs <= 0 + ) { + throw new TypeError("Runtime Host reconnectable read wait timeout must be positive"); + } + this.#reconnectableReadWaitTimeoutMs = reconnectableReadWaitTimeoutMs; } createTarget(epoch: string): RuntimeHostTargetIpcMain { @@ -232,15 +251,35 @@ export class RuntimeHostReconnectingIpcMain { args: readonly unknown[], ): Promise { const epoch = this.#requireTargetEpoch(args[0]); + let reconnectableReadDeadline: number | undefined; + const waitForReconnectableReadHandler = async ( + previous?: BoundHandler, + ): Promise => { + if (!slot.reconnectableRead) return this.#waitForHandler(slot, epoch, previous); + // One invocation gets one replacement window across every candidate it + // visits. Resetting it per generation would let Host flapping retain the + // renderer request indefinitely. + reconnectableReadDeadline ??= Date.now() + this.#reconnectableReadWaitTimeoutMs; + const remainingMs = Math.max(0, reconnectableReadDeadline - Date.now()); + if (remainingMs <= 0) throw new ReconnectableReadWaitExpiredError(); + try { + return await this.#waitForHandler(slot, epoch, previous, remainingMs); + } catch (error) { + if (error instanceof HandlerWaitExpiredError) { + throw new ReconnectableReadWaitExpiredError(); + } + throw error; + } + }; let handler: BoundHandler = - slot.handlers.get(epoch) ?? await this.#waitForHandler(slot, epoch); + slot.handlers.get(epoch) ?? await waitForReconnectableReadHandler(); let reconciliationContext: unknown; let reconciling = false; let reconciliationDeadline: number | undefined; const waitForReplacement = async ( previous: BoundHandler, ): Promise => { - if (!reconciling) return this.#waitForHandler(slot, epoch, previous); + if (!reconciling) return waitForReconnectableReadHandler(previous); const remainingMs = Math.max( 0, (reconciliationDeadline ?? Date.now()) - Date.now(), @@ -248,7 +287,7 @@ export class RuntimeHostReconnectingIpcMain { try { return await this.#waitForHandler(slot, epoch, previous, remainingMs); } catch (error) { - if (error instanceof ReconciliationWaitExpiredError) return undefined; + if (error instanceof HandlerWaitExpiredError) return undefined; throw error; } }; @@ -325,7 +364,7 @@ export class RuntimeHostReconnectingIpcMain { return Promise.resolve(current); } if (timeoutMs !== undefined && timeoutMs <= 0) { - return Promise.reject(new ReconciliationWaitExpiredError()); + return Promise.reject(new HandlerWaitExpiredError()); } return new Promise((resolve, reject) => { let timeout: ReturnType | undefined; @@ -344,7 +383,7 @@ export class RuntimeHostReconnectingIpcMain { if (timeoutMs !== undefined) { timeout = setTimeout(() => { if (!slot.waiters.delete(waiter)) return; - waiter.reject(new ReconciliationWaitExpiredError()); + waiter.reject(new HandlerWaitExpiredError()); }, timeoutMs); } }); From 3961c6d7d5e2e3b53d56fde139de4470b317687c Mon Sep 17 00:00:00 2001 From: me2seeks Date: Wed, 26 Aug 2026 12:04:34 +0800 Subject: [PATCH 2/2] fix(desktop): use monotonic read retry deadlines Keep the shared reconnectable-read window independent of wall-clock adjustments, and cover rollback while replacements keep failing. Generated-by: Codex --- .../__tests__/runtime-host-reconnecting-ipc-main.test.ts | 9 ++++++--- .../src/main/runtime-host-reconnecting-ipc-main.ts | 4 ++-- 2 files changed, 8 insertions(+), 5 deletions(-) diff --git a/apps/desktop/src/main/__tests__/runtime-host-reconnecting-ipc-main.test.ts b/apps/desktop/src/main/__tests__/runtime-host-reconnecting-ipc-main.test.ts index 46077bc0e4..ed973944a3 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-reconnecting-ipc-main.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-reconnecting-ipc-main.test.ts @@ -213,8 +213,10 @@ test("does not reset a reconnectable read deadline across failed replacements", const router = new RuntimeHostReconnectingIpcMain(ipc, { reconnectableReadWaitTimeoutMs: 15, }); - let now = 0; - t.mock.method(Date, "now", () => now); + let monotonicNow = 0; + let wallClockNow = 1_000; + t.mock.method(performance, "now", () => monotonicNow); + t.mock.method(Date, "now", () => wallClockNow); router.activate("target-a"); let attempts = 0; const maximumAttempts = 20; @@ -222,7 +224,8 @@ test("does not reset a reconnectable read deadline across failed replacements", const target = router.createTarget("target-a"); target.handleReconnectableRead?.("projects:getSnapshot", async () => { attempts += 1; - now += 5; + monotonicNow += 5; + wallClockNow -= 100; target.removeHandler("projects:getSnapshot"); if (attempts < maximumAttempts) installFailingTarget(); throw new RuntimeHostOperationError( diff --git a/apps/desktop/src/main/runtime-host-reconnecting-ipc-main.ts b/apps/desktop/src/main/runtime-host-reconnecting-ipc-main.ts index 9bc1d2bec4..466d68d059 100644 --- a/apps/desktop/src/main/runtime-host-reconnecting-ipc-main.ts +++ b/apps/desktop/src/main/runtime-host-reconnecting-ipc-main.ts @@ -259,8 +259,8 @@ export class RuntimeHostReconnectingIpcMain { // One invocation gets one replacement window across every candidate it // visits. Resetting it per generation would let Host flapping retain the // renderer request indefinitely. - reconnectableReadDeadline ??= Date.now() + this.#reconnectableReadWaitTimeoutMs; - const remainingMs = Math.max(0, reconnectableReadDeadline - Date.now()); + reconnectableReadDeadline ??= performance.now() + this.#reconnectableReadWaitTimeoutMs; + const remainingMs = Math.max(0, reconnectableReadDeadline - performance.now()); if (remainingMs <= 0) throw new ReconnectableReadWaitExpiredError(); try { return await this.#waitForHandler(slot, epoch, previous, remainingMs);