fix(runtime-host): break reconnect failure cascades

apps/desktop/src/main/__tests__/runtime-host-desktop-manager.test.ts

@@ -141,6 +141,43 @@ test('quiesces reconnect and waits for the Host process before update install',
   await owner.close();
 });
 
+test('waits through a reconnect gap before quiescing Host retirement', async () => {
+  const first = candidateHarness();
+  const replacement = candidateHarness({ disconnectOnPrepare: true });
+  let starts = 0;
+  let reportReplacementStart!: () => void;
+  let releaseReplacement!: () => void;
+  const replacementStarted = new Promise<void>((resolve) => {
+    reportReplacementStart = resolve;
+  });
+  const replacementReleased = new Promise<void>((resolve) => {
+    releaseReplacement = resolve;
+  });
+  const owner = await startRuntimeHostDesktopManager({} as DesktopRuntimeHostCandidateStartInput, {
+    startCandidate: async () => {
+      starts += 1;
+      if (starts === 1) return ready(first.candidate);
+      reportReplacementStart();
+      await replacementReleased;
+      return ready(replacement.candidate);
+    },
+    waitForHostExit: async () => {},
+  });
+
+  first.disconnect();
+  await replacementStarted;
+  const retirement = owner.retireOwnedLocalHost('interrupt_active_work');
+  await new Promise<void>((resolve) => setImmediate(resolve));
+  assert.equal(replacement.prepareRetirementCalls, 0);
+
+  releaseReplacement();
+  assert.equal((await retirement).kind, 'retired');
+  assert.equal(replacement.prepareRetirementCalls, 1);
+  assert.deepEqual(replacement.retirementModes, ['interrupt_active_work']);
+  assert.equal(starts, 2);
+  await owner.close();
+});
+
 test('retires the owned ephemeral Host before Desktop quit', async () => {
   const events: string[] = [];
   const current = candidateHarness({

apps/desktop/src/main/__tests__/runtime-host-reconnecting-ipc-main.test.ts

@@ -128,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;
@@ -339,10 +339,10 @@ 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 () => {
+test("bounds an invocation while an active Runtime Host has no handler", async () => {
   const ipc = ipcHarness();
   const router = new RuntimeHostReconnectingIpcMain(ipc, {
-    handlerWaitTimeoutMs: 5,
+    replacementWaitTimeoutMs: 5,
   });
   const target = router.createTarget("target-a");
   target.handle("sessions:send", async () => "sent");
@@ -354,17 +354,14 @@ test("bounds invocation while an active Runtime Host has no handler", async () =
     RuntimeHostHandlerUnavailableError,
   );
   target.handle("sessions:send", async () => "retried");
-  assert.equal(
-    await ipc.invoke("sessions:send", scope("target-a")),
-    "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,
+    replacementWaitTimeoutMs: 5,
   });
   const target = router.createTarget("target-a");
   const failRead = deferred();
@@ -389,6 +386,79 @@ test("bounds reconnectable reads when no replacement handler becomes available",
   router.close();
 });
 
+test("settles concurrent invocations after one bounded replacement window", async () => {
+  const ipc = ipcHarness();
+  const router = new RuntimeHostReconnectingIpcMain(ipc, {
+    replacementWaitTimeoutMs: 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: 200 }, (_, index) =>
+    ipc.invoke("projects:getSnapshot", scope("target-a"), { index }).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 }), 100),
+      ),
+    ]);
+    assert.ok(Array.isArray(result), "Runtime Host invocations did not settle");
+    assert.equal(result.length, 200);
+    for (const read of result) {
+      assert.equal(read.ok, false);
+      if (!read.ok) assert.ok(read.error instanceof RuntimeHostHandlerUnavailableError);
+    }
+  } finally {
+    router.close();
+    await settled;
+  }
+});
+
+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);

apps/desktop/src/main/runtime-host-desktop-manager.ts

@@ -557,7 +557,7 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager {
     const lifecycle = this.#requireLifecycle(
       this.#requireTarget(LOCAL_RUNTIME_HOST_PROFILE.id),
     );
-    const quiescence = lifecycle.quiesce();
+    const quiescence = await lifecycle.quiesce();
     let hostPid = quiescence.current.hostPid;
     let launchBarrierPaused = false;
     const resume = () => {

apps/desktop/src/main/runtime-host-reconnecting-ipc-main.ts

@@ -33,18 +33,16 @@ type ReconcileIpcHandler = (
 
 type ReconciliationUnavailableIpcHandler = ReconcileIpcHandler;
 
-const DEFAULT_RECONCILIATION_WAIT_TIMEOUT_MS = 15_000;
-const DEFAULT_HANDLER_WAIT_TIMEOUT_MS = 5_000;
+const DEFAULT_REPLACEMENT_WAIT_TIMEOUT_MS = 15_000;
 
 export interface RuntimeHostReconnectingIpcMainOptions {
-  readonly reconciliationWaitTimeoutMs?: number;
-  readonly handlerWaitTimeoutMs?: 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";
   }
 }
 
@@ -99,27 +97,20 @@ export class RuntimeHostReconnectingIpcMain {
   readonly #ipcMain: Pick<IpcMain, "handle" | "removeHandler">;
   readonly #slots = new Map<string, HandlerSlot>();
   readonly #activeEpochs = new Set<string>();
-  readonly #reconciliationWaitTimeoutMs: number;
-  readonly #handlerWaitTimeoutMs: number;
+  readonly #replacementWaitTimeoutMs: number;
   #closed = false;
 
   constructor(
     ipcMain: Pick<IpcMain, "handle" | "removeHandler">,
     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;
-    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;
+    this.#replacementWaitTimeoutMs = replacementWaitTimeoutMs;
   }
 
   createTarget(epoch: string): RuntimeHostTargetIpcMain {
@@ -248,41 +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,
-        undefined,
-        this.#handlerWaitTimeoutMs,
-        () => new RuntimeHostHandlerUnavailableError(),
-      );
-    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,
-          this.#handlerWaitTimeoutMs,
-          () => new RuntimeHostHandlerUnavailableError(),
-        );
-      }
-      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,
@@ -309,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;
@@ -345,7 +322,6 @@ export class RuntimeHostReconnectingIpcMain {
     epoch: string,
     previous?: BoundHandler,
     timeoutMs?: number,
-    timeoutError: () => Error = () => new ReconciliationWaitExpiredError(),
   ): Promise<BoundHandler> {
     try {
       this.#assertActive(epoch);
@@ -357,7 +333,7 @@ export class RuntimeHostReconnectingIpcMain {
       return Promise.resolve(current);
     }
     if (timeoutMs !== undefined && timeoutMs <= 0) {
-      return Promise.reject(timeoutError());
+      return Promise.reject(new HandlerWaitExpiredError());
     }
     return new Promise((resolve, reject) => {
       let timeout: ReturnType<typeof setTimeout> | undefined;
@@ -376,7 +352,7 @@ export class RuntimeHostReconnectingIpcMain {
       if (timeoutMs !== undefined) {
         timeout = setTimeout(() => {
           if (!slot.waiters.delete(waiter)) return;
-          waiter.reject(timeoutError());
+          waiter.reject(new HandlerWaitExpiredError());
         }, timeoutMs);
       }
     });