From f668156cd0d222729cfb8467db1864be937d6cbc Mon Sep 17 00:00:00 2001 From: hqhq1025 <1506751656@qq.com> Date: Tue, 25 Aug 2026 20:14:55 +0800 Subject: [PATCH] fix(desktop): bound Runtime Host handler waits --- ...runtime-host-reconnecting-ipc-main.test.ts | 51 +++++++++++++++++++ .../runtime-host-reconnecting-ipc-main.ts | 40 +++++++++++++-- 2 files changed, 87 insertions(+), 4 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..2223d93af6 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 @@ -30,6 +30,7 @@ import { } from "../ipc-reconnect-policy.js"; import * as ipcReconnectPolicy from "../ipc-reconnect-policy.js"; import { + RuntimeHostHandlerUnavailableError, RuntimeHostReconnectingIpcMain, RuntimeHostTargetChangedError, } from "../runtime-host-reconnecting-ipc-main.js"; @@ -338,6 +339,56 @@ test("holds an invocation across a Runtime Host candidate replacement", async () assert.equal(ipc.size, 0); }); +test("bounds invocation while an active Runtime Host has no handler", async () => { + const ipc = ipcHarness(); + const router = new RuntimeHostReconnectingIpcMain(ipc, { + handlerWaitTimeoutMs: 5, + }); + const target = router.createTarget("target-a"); + target.handle("sessions:send", async () => "sent"); + router.activate("target-a"); + target.removeHandler("sessions:send"); + + await assert.rejects( + () => ipc.invoke("sessions:send", scope("target-a")), + RuntimeHostHandlerUnavailableError, + ); + target.handle("sessions:send", async () => "retried"); + assert.equal( + await ipc.invoke("sessions:send", scope("target-a")), + "retried", + ); + router.close(); +}); + +test("bounds reconnectable reads when no replacement handler becomes available", async () => { + const ipc = ipcHarness(); + const router = new RuntimeHostReconnectingIpcMain(ipc, { + handlerWaitTimeoutMs: 5, + }); + const target = router.createTarget("target-a"); + const failRead = deferred(); + target.handleReconnectableRead?.("taskReadiness:getSnapshot", async () => { + await failRead.promise; + throw new RuntimeHostOperationError( + "session.catalog.query", + "host_draining", + "Runtime Host is draining", + ); + }); + router.activate("target-a"); + + const reading = ipc.invoke("taskReadiness:getSnapshot", scope("target-a")); + target.removeHandler("taskReadiness:getSnapshot"); + failRead.resolve(); + + await assert.rejects( + () => reading, + RuntimeHostHandlerUnavailableError, + ); + router.close(); +}); + test("does not return a late read from a replaced Runtime Host candidate", 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..0e62489137 100644 --- a/apps/desktop/src/main/runtime-host-reconnecting-ipc-main.ts +++ b/apps/desktop/src/main/runtime-host-reconnecting-ipc-main.ts @@ -34,9 +34,11 @@ type ReconcileIpcHandler = ( type ReconciliationUnavailableIpcHandler = ReconcileIpcHandler; const DEFAULT_RECONCILIATION_WAIT_TIMEOUT_MS = 15_000; +const DEFAULT_HANDLER_WAIT_TIMEOUT_MS = 5_000; export interface RuntimeHostReconnectingIpcMainOptions { readonly reconciliationWaitTimeoutMs?: number; + readonly handlerWaitTimeoutMs?: number; } class ReconciliationWaitExpiredError extends Error { @@ -81,6 +83,13 @@ export class RuntimeHostTargetChangedError extends Error { } } +export class RuntimeHostHandlerUnavailableError extends Error { + constructor() { + super("Runtime Host handler remained unavailable after the reconnection window"); + this.name = "RuntimeHostHandlerUnavailableError"; + } +} + /** * Keeps Electron IPC registration stable across reconnects while fencing each * target generation. Reconnectable reads may move to a replacement candidate, @@ -91,6 +100,7 @@ export class RuntimeHostReconnectingIpcMain { readonly #slots = new Map(); readonly #activeEpochs = new Set(); readonly #reconciliationWaitTimeoutMs: number; + readonly #handlerWaitTimeoutMs: number; #closed = false; constructor( @@ -104,6 +114,12 @@ export class RuntimeHostReconnectingIpcMain { throw new TypeError("Runtime Host reconciliation wait timeout must be positive"); } this.#reconciliationWaitTimeoutMs = reconciliationWaitTimeoutMs; + const handlerWaitTimeoutMs = + options.handlerWaitTimeoutMs ?? DEFAULT_HANDLER_WAIT_TIMEOUT_MS; + if (!Number.isSafeInteger(handlerWaitTimeoutMs) || handlerWaitTimeoutMs <= 0) { + throw new TypeError("Runtime Host handler wait timeout must be positive"); + } + this.#handlerWaitTimeoutMs = handlerWaitTimeoutMs; } createTarget(epoch: string): RuntimeHostTargetIpcMain { @@ -233,14 +249,29 @@ export class RuntimeHostReconnectingIpcMain { ): Promise { const epoch = this.#requireTargetEpoch(args[0]); let handler: BoundHandler = - slot.handlers.get(epoch) ?? await this.#waitForHandler(slot, epoch); + slot.handlers.get(epoch) ?? + await this.#waitForHandler( + slot, + epoch, + undefined, + this.#handlerWaitTimeoutMs, + () => new RuntimeHostHandlerUnavailableError(), + ); 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 this.#waitForHandler( + slot, + epoch, + previous, + this.#handlerWaitTimeoutMs, + () => new RuntimeHostHandlerUnavailableError(), + ); + } const remainingMs = Math.max( 0, (reconciliationDeadline ?? Date.now()) - Date.now(), @@ -314,6 +345,7 @@ export class RuntimeHostReconnectingIpcMain { epoch: string, previous?: BoundHandler, timeoutMs?: number, + timeoutError: () => Error = () => new ReconciliationWaitExpiredError(), ): Promise { try { this.#assertActive(epoch); @@ -325,7 +357,7 @@ export class RuntimeHostReconnectingIpcMain { return Promise.resolve(current); } if (timeoutMs !== undefined && timeoutMs <= 0) { - return Promise.reject(new ReconciliationWaitExpiredError()); + return Promise.reject(timeoutError()); } return new Promise((resolve, reject) => { let timeout: ReturnType | undefined; @@ -344,7 +376,7 @@ export class RuntimeHostReconnectingIpcMain { if (timeoutMs !== undefined) { timeout = setTimeout(() => { if (!slot.waiters.delete(waiter)) return; - waiter.reject(new ReconciliationWaitExpiredError()); + waiter.reject(timeoutError()); }, timeoutMs); } });