You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by wy...@apache.org on 2026/08/26 09:33:36 UTC
(maka) 03/03: fix(desktop): bound one IPC replacement lifetime
This is an automated email from the ASF dual-hosted git repository.
M4n5ter pushed a commit to branch fix/runtime-host-flapping-root
in repository https://gitbox.apache.org/repos/asf/maka.git
commit 70f366df78e7cee8160856833a3bcf7191fa9fd2
Author: M4n5ter <m4...@gmail.com>
AuthorDate: Wed Aug 26 17:31:52 2026 +0800
fix(desktop): bound one IPC replacement lifetime
Co-authored-by: me2seeks <me...@gmail.com>
Co-authored-by: hqhq1025 <15...@qq.com>
Generated-by: Codex
---
.../runtime-host-reconnecting-ipc-main.test.ts | 59 ++++++++++++++++++++-
.../src/main/runtime-host-reconnecting-ipc-main.ts | 60 ++++++++++++----------
2 files changed, 92 insertions(+), 27 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 75181fca2..6cd4575bd 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";
@@ -127,7 +128,7 @@ test("reconciles a dispatched control on a replacement without replaying it", as
test("bounds reconciliation when no replacement candidate becomes available", async () => {
const ipc = ipcHarness();
const router = new RuntimeHostReconnectingIpcMain(ipc, {
- reconciliationWaitTimeoutMs: 5,
+ replacementWaitTimeoutMs: 5,
});
const firstTarget = router.createTarget("target-a") as ReconciledControlTarget;
let dispatches = 0;
@@ -338,6 +339,62 @@ test("holds an invocation across a Runtime Host candidate replacement", async ()
assert.equal(ipc.size, 0);
});
+test("bounds an invocation while an active Runtime Host has no handler", async () => {
+ const ipc = ipcHarness();
+ const router = new RuntimeHostReconnectingIpcMain(ipc, {
+ replacementWaitTimeoutMs: 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("does not reset one reconnectable read deadline across failed replacements", async (t) => {
+ const ipc = ipcHarness();
+ const router = new RuntimeHostReconnectingIpcMain(ipc, {
+ replacementWaitTimeoutMs: 15,
+ });
+ let monotonicNow = 0;
+ t.mock.method(performance, "now", () => monotonicNow);
+ 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;
+ monotonicNow += 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")),
+ RuntimeHostHandlerUnavailableError,
+ );
+ assert.equal(attempts, 4);
+ } finally {
+ 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 3875fb5f4..22a5390b3 100644
--- a/apps/desktop/src/main/runtime-host-reconnecting-ipc-main.ts
+++ b/apps/desktop/src/main/runtime-host-reconnecting-ipc-main.ts
@@ -33,16 +33,16 @@ type ReconcileIpcHandler = (
type ReconciliationUnavailableIpcHandler = ReconcileIpcHandler;
-const DEFAULT_RECONCILIATION_WAIT_TIMEOUT_MS = 15_000;
+const DEFAULT_REPLACEMENT_WAIT_TIMEOUT_MS = 15_000;
export interface RuntimeHostReconnectingIpcMainOptions {
- readonly reconciliationWaitTimeoutMs?: number;
+ readonly replacementWaitTimeoutMs?: 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";
}
}
@@ -81,6 +81,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,
@@ -90,7 +97,7 @@ export class RuntimeHostReconnectingIpcMain {
readonly #ipcMain: Pick<IpcMain, "handle" | "removeHandler">;
readonly #slots = new Map<string, HandlerSlot>();
readonly #activeEpochs = new Set<string>();
- readonly #reconciliationWaitTimeoutMs: number;
+ readonly #replacementWaitTimeoutMs: number;
#closed = false;
constructor(
@@ -98,12 +105,12 @@ export class RuntimeHostReconnectingIpcMain {
options: RuntimeHostReconnectingIpcMainOptions = {},
) {
this.#ipcMain = ipcMain;
- const reconciliationWaitTimeoutMs =
- options.reconciliationWaitTimeoutMs ?? DEFAULT_RECONCILIATION_WAIT_TIMEOUT_MS;
- if (!Number.isSafeInteger(reconciliationWaitTimeoutMs) || reconciliationWaitTimeoutMs <= 0) {
- throw new TypeError("Runtime Host reconciliation wait timeout must be positive");
+ const replacementWaitTimeoutMs =
+ options.replacementWaitTimeoutMs ?? DEFAULT_REPLACEMENT_WAIT_TIMEOUT_MS;
+ if (!Number.isSafeInteger(replacementWaitTimeoutMs) || replacementWaitTimeoutMs <= 0) {
+ throw new TypeError("Runtime Host replacement wait timeout must be positive");
}
- this.#reconciliationWaitTimeoutMs = reconciliationWaitTimeoutMs;
+ this.#replacementWaitTimeoutMs = replacementWaitTimeoutMs;
}
createTarget(epoch: string): RuntimeHostTargetIpcMain {
@@ -232,26 +239,28 @@ export class RuntimeHostReconnectingIpcMain {
args: readonly unknown[],
): Promise<unknown> {
const epoch = this.#requireTargetEpoch(args[0]);
- let handler: BoundHandler =
- slot.handlers.get(epoch) ?? await this.#waitForHandler(slot, epoch);
- let reconciliationContext: unknown;
let reconciling = false;
- let reconciliationDeadline: number | undefined;
+ let replacementDeadline: number | undefined;
const waitForReplacement = async (
- previous: BoundHandler,
+ previous?: BoundHandler,
): Promise<BoundHandler | undefined> => {
- if (!reconciling) return this.#waitForHandler(slot, epoch, previous);
- const remainingMs = Math.max(
- 0,
- (reconciliationDeadline ?? Date.now()) - Date.now(),
- );
+ // One invocation gets one monotonic replacement window across every
+ // candidate it visits; flapping must not restart its lifetime.
+ replacementDeadline ??= performance.now() + this.#replacementWaitTimeoutMs;
+ const remainingMs = Math.max(0, replacementDeadline - performance.now());
try {
+ if (remainingMs <= 0) throw new HandlerWaitExpiredError();
return await this.#waitForHandler(slot, epoch, previous, remainingMs);
} catch (error) {
- if (error instanceof ReconciliationWaitExpiredError) return undefined;
- throw error;
+ if (!(error instanceof HandlerWaitExpiredError)) throw error;
+ if (reconciling) return undefined;
+ throw new RuntimeHostHandlerUnavailableError();
}
};
+ const initialHandler = slot.handlers.get(epoch) ?? await waitForReplacement();
+ if (!initialHandler) throw new RuntimeHostHandlerUnavailableError();
+ let handler: BoundHandler = initialHandler;
+ let reconciliationContext: unknown;
const unavailable = (): Promise<unknown> =>
requireReconciliationUnavailableHandler(handler)(
reconciliationContext,
@@ -278,7 +287,6 @@ export class RuntimeHostReconnectingIpcMain {
if (step.kind === "completed") return step.value;
reconciliationContext = step.context;
reconciling = true;
- reconciliationDeadline = Date.now() + this.#reconciliationWaitTimeoutMs;
const replacement = await waitForReplacement(handler);
if (!replacement) return unavailable();
handler = replacement;
@@ -325,7 +333,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<typeof setTimeout> | undefined;
@@ -344,7 +352,7 @@ export class RuntimeHostReconnectingIpcMain {
if (timeoutMs !== undefined) {
timeout = setTimeout(() => {
if (!slot.waiters.delete(waiter)) return;
- waiter.reject(new ReconciliationWaitExpiredError());
+ waiter.reject(new HandlerWaitExpiredError());
}, timeoutMs);
}
});
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org