fix(gateway): settle account stops before failing (#114280)

Co-authored-by: Qiong <yang.ji2@xydigit.com>

src/gateway/channel-health-monitor.test.ts

@@ -238,6 +238,21 @@ describe("channel-health-monitor", () => {
     monitor.stop();
   });
 
+  it("does not start a replacement when channel teardown fails", async () => {
+    const manager = createSlackSnapshotManager(disconnectedAccount(Date.now() - 300_000), {
+      stopChannel: vi.fn(async () => {
+        throw new Error("stop failed");
+      }),
+    });
+
+    const monitor = await startAndRunCheck(manager, { cooldownCycles: 0 });
+
+    expect(manager.stopChannel).toHaveBeenCalledWith("slack", "default", { manual: false });
+    expect(manager.resetRestartAttempts).not.toHaveBeenCalled();
+    expect(manager.startChannel).not.toHaveBeenCalled();
+    monitor.stop();
+  });
+
   it("accepts timing.monitorStartupGraceMs", async () => {
     const manager = createMockChannelManager();
     const monitor = startDefaultMonitor(manager, { timing: { monitorStartupGraceMs: 60_000 } });

src/gateway/server-channels.test.ts

@@ -471,6 +471,201 @@ describe("server-channels auto restart", () => {
     expect(account?.lastError).toBeNull();
   });
 
+  it("settles every account before surfacing a stop hook failure", async () => {
+    const accountIds = ["broken", "healthy"];
+    const taskReleases = new Map(accountIds.map((accountId) => [accountId, createDeferred()]));
+    const startAccount = vi.fn(
+      async ({ abortSignal, accountId }: ChannelGatewayContext<TestAccount>) =>
+        await new Promise<void>((resolve) => {
+          abortSignal.addEventListener(
+            "abort",
+            () => {
+              void taskReleases.get(accountId)?.promise.then(resolve);
+            },
+            { once: true },
+          );
+        }),
+    );
+    const stopAccount = vi.fn(async ({ accountId }: ChannelGatewayContext<TestAccount>) => {
+      if (accountId === "broken") {
+        throw new Error("stop hook failed");
+      }
+    });
+    installTestRegistry(
+      createTestPlugin({
+        listAccountIds: () => accountIds,
+        resolveAccount: () => ({ enabled: true, configured: true }),
+        startAccount,
+        stopAccount,
+      }),
+    );
+    const manager = createManager();
+
+    await manager.startChannels();
+    await flushMicrotasks();
+    const stopTask = manager.stopChannel("discord");
+    let stopSettled = false;
+    void stopTask.then(
+      () => {
+        stopSettled = true;
+      },
+      () => {
+        stopSettled = true;
+      },
+    );
+    try {
+      await flushMicrotasks();
+      expect(stopSettled).toBe(false);
+
+      taskReleases.get("healthy")?.resolve();
+      await flushMicrotasks();
+      expect(stopSettled).toBe(false);
+
+      taskReleases.get("broken")?.resolve();
+      await expect(stopTask).rejects.toThrow("stop hook failed");
+      const accounts = manager.getRuntimeSnapshot().channelAccounts.discord;
+      expect(stopAccount.mock.calls.map(([context]) => context.accountId)).toEqual(accountIds);
+      expect(accounts?.broken).toMatchObject({
+        running: true,
+        restartPending: false,
+        lastError: "stop hook failed",
+      });
+      expect(accounts?.healthy).toMatchObject({ running: false, lastError: null });
+
+      await manager.startChannel("discord", "broken");
+      expect(startAccount).toHaveBeenCalledTimes(2);
+    } finally {
+      for (const release of taskReleases.values()) {
+        release.resolve();
+      }
+    }
+  });
+
+  it("blocks replacement while a stop hook outlives the old account task", async () => {
+    const releaseTask = createDeferred();
+    const releaseStopHook = createDeferred();
+    const startAccount = vi.fn(async () => await releaseTask.promise);
+    const stopAccount = vi.fn(async () => {
+      await releaseStopHook.promise;
+      throw new Error("stop hook failed");
+    });
+    installTestRegistry(createTestPlugin({ startAccount, stopAccount }));
+    const manager = createManager();
+
+    await manager.startChannels();
+    const stopTask = manager.stopChannel("discord", DEFAULT_ACCOUNT_ID, { manual: false });
+    const stopFailure = expect(stopTask).rejects.toThrow("stop hook failed");
+    await flushMicrotasks();
+    expect(stopAccount).toHaveBeenCalledOnce();
+
+    releaseTask.resolve();
+    await flushMicrotasks();
+    await manager.startChannel("discord", DEFAULT_ACCOUNT_ID);
+    expect(startAccount).toHaveBeenCalledTimes(1);
+
+    releaseStopHook.resolve();
+    await stopFailure;
+    await manager.startChannel("discord", DEFAULT_ACCOUNT_ID);
+    expect(startAccount).toHaveBeenCalledTimes(1);
+    expect(
+      manager.getRuntimeSnapshot().channelAccounts.discord?.[DEFAULT_ACCOUNT_ID],
+    ).toMatchObject({
+      running: true,
+      restartPending: false,
+      lastError: "stop hook failed",
+    });
+  });
+
+  it("serializes overlapping stops until the last teardown settles", async () => {
+    const releaseTask = createDeferred();
+    const stopHooks = [createDeferred(), createDeferred()];
+    const startAccount = vi.fn(async () => await releaseTask.promise);
+    const stopAccount = vi.fn(async () => {
+      const callIndex = stopAccount.mock.calls.length - 1;
+      await stopHooks[callIndex]?.promise;
+      if (callIndex === 1) {
+        throw new Error("second stop failed");
+      }
+    });
+    installTestRegistry(createTestPlugin({ startAccount, stopAccount }));
+    const manager = createManager();
+
+    await manager.startChannels();
+    const firstStop = manager.stopChannel("discord", DEFAULT_ACCOUNT_ID, { manual: false });
+    const secondStop = manager.stopChannel("discord", DEFAULT_ACCOUNT_ID, { manual: false });
+    const secondFailure = expect(secondStop).rejects.toThrow("second stop failed");
+
+    releaseTask.resolve();
+    stopHooks[0]?.resolve();
+    await expect(firstStop).resolves.toBeUndefined();
+    await flushMicrotasks();
+    expect(stopAccount).toHaveBeenCalledTimes(2);
+
+    await manager.startChannel("discord", DEFAULT_ACCOUNT_ID);
+    expect(startAccount).toHaveBeenCalledTimes(1);
+
+    stopHooks[1]?.resolve();
+    await secondFailure;
+    await manager.startChannel("discord", DEFAULT_ACCOUNT_ID);
+    expect(startAccount).toHaveBeenCalledTimes(1);
+    expect(
+      manager.getRuntimeSnapshot().channelAccounts.discord?.[DEFAULT_ACCOUNT_ID],
+    ).toMatchObject({
+      running: true,
+      restartPending: false,
+      lastError: "second stop failed",
+    });
+  });
+
+  it("keeps a timed-out stop hook failure authoritative after late task settlement", async () => {
+    const releaseTask = createDeferred();
+    const startAccount = vi.fn(async () => {
+      await releaseTask.promise;
+      throw new Error("late task failure");
+    });
+    let stopShouldFail = true;
+    const stopAccount = vi.fn(async () => {
+      if (stopShouldFail) {
+        throw new Error("stop hook failed");
+      }
+    });
+    let accountIds = [DEFAULT_ACCOUNT_ID];
+    installTestRegistry(
+      createTestPlugin({ startAccount, stopAccount, listAccountIds: () => accountIds }),
+    );
+    const manager = createManager();
+
+    await manager.startChannels();
+    const stopTask = manager.stopChannel("discord", DEFAULT_ACCOUNT_ID, { manual: false });
+    const stopFailure = expect(stopTask).rejects.toThrow("stop hook failed");
+    await vi.advanceTimersByTimeAsync(5_000);
+    await stopFailure;
+
+    releaseTask.resolve();
+    await flushMicrotasks();
+    expect(
+      manager.getRuntimeSnapshot().channelAccounts.discord?.[DEFAULT_ACCOUNT_ID],
+    ).toMatchObject({
+      running: true,
+      restartPending: false,
+      lastError: "stop hook failed",
+    });
+
+    await manager.startChannel("discord", DEFAULT_ACCOUNT_ID);
+    expect(startAccount).toHaveBeenCalledTimes(1);
+
+    accountIds = [];
+    await manager.startChannels();
+    accountIds = [DEFAULT_ACCOUNT_ID];
+    await manager.startChannels();
+    expect(startAccount).toHaveBeenCalledTimes(1);
+
+    stopShouldFail = false;
+    await manager.stopChannel("discord", DEFAULT_ACCOUNT_ID);
+    await manager.startChannel("discord", DEFAULT_ACCOUNT_ID);
+    expect(startAccount).toHaveBeenCalledTimes(2);
+  });
+
   it("does not enumerate configured accounts when stopping a never-started channel", async () => {
     const listAccountIds = vi.fn(() => [DEFAULT_ACCOUNT_ID]);
     const resolveAccount = vi.fn(() => ({ enabled: true, configured: true }));

src/gateway/server-channels.ts

@@ -61,6 +61,7 @@ function waitForChannelStartupHandoff(): Promise<void> {
 type ChannelRuntimeStore = {
   aborts: Map<string, AbortController>;
   starting: Map<string, Promise<void>>;
+  stops: Map<string, ChannelAccountStopState>;
   tasks: Map<string, Promise<unknown>>;
   runtimes: Map<string, ChannelAccountSnapshot>;
 };
@@ -119,6 +120,7 @@ function createRuntimeStore(): ChannelRuntimeStore {
   return {
     aborts: new Map(),
     starting: new Map(),
+    stops: new Map(),
     tasks: new Map(),
     runtimes: new Map(),
   };
@@ -231,6 +233,12 @@ type StopChannelOptions = {
   manual?: boolean;
 };
 
+type ChannelAccountStopOutcome = { status: "fulfilled" } | { status: "rejected"; error: unknown };
+
+type ChannelAccountStopState =
+  | { status: "stopping"; attempt: Promise<ChannelAccountStopOutcome> }
+  | Extract<ChannelAccountStopOutcome, { status: "rejected" }>;
+
 async function waitForDeferredAccountStart(
   deferred: Promise<void>,
   abortSignal: AbortSignal,
@@ -431,6 +439,7 @@ export function createChannelManager(opts: ChannelManagerOptions): ChannelManage
         activeAccountIds.has(id) ||
         store.aborts.has(id) ||
         store.starting.has(id) ||
+        store.stops.has(id) ||
         store.tasks.has(id)
       ) {
         continue;
@@ -497,6 +506,11 @@ export function createChannelManager(opts: ChannelManagerOptions): ChannelManage
       limit: CHANNEL_STARTUP_CONCURRENCY,
       tasks: accountIds.map((id) => async () => {
         const rKey = restartKey(channelId, id);
+        // An in-flight or failed plugin teardown may still own resources. Only
+        // the last queued attempt or a later successful stop clears this gate.
+        if (store.stops.has(id)) {
+          return;
+        }
         if (store.tasks.has(id)) {
           let clearedTimedOutRecoveryTask = false;
           if (recoveryStopTimedOut.has(rKey)) {
@@ -737,7 +751,7 @@ export function createChannelManager(opts: ChannelManagerOptions): ChannelManage
               log.error?.(`[${id}] ${message}`);
             })
             .catch((err: unknown) => {
-              if (!isCurrentTask()) {
+              if (!isCurrentTask() || store.stops.has(id)) {
                 return;
               }
               const message = formatErrorMessage(err);
@@ -746,7 +760,9 @@ export function createChannelManager(opts: ChannelManagerOptions): ChannelManage
             })
             .then(async () => {
               await cleanupTaskScopedApprovalRuntime("channel cleanup failed");
-              if (!isCurrentTask()) {
+              // stopChannel owns the failed-teardown snapshot until a later
+              // successful stop proves replacement is safe.
+              if (!isCurrentTask() || store.stops.has(id)) {
                 return;
               }
               setStoppedRuntime(channelId, id, {
@@ -754,7 +770,7 @@ export function createChannelManager(opts: ChannelManagerOptions): ChannelManage
               });
             })
             .then(async () => {
-              if (!isCurrentTask()) {
+              if (!isCurrentTask() || store.stops.has(id)) {
                 return;
               }
               if (manuallyStopped.has(rKey)) {
@@ -932,6 +948,7 @@ export function createChannelManager(opts: ChannelManagerOptions): ChannelManage
     const lifecycleIds = new Set<string>([
       ...store.aborts.keys(),
       ...store.starting.keys(),
+      ...store.stops.keys(),
       ...store.tasks.keys(),
     ]);
     if (!accountId && lifecycleIds.size === 0) {
@@ -951,69 +968,128 @@ export function createChannelManager(opts: ChannelManagerOptions): ChannelManage
       knownIds.add(accountId);
     }
 
-    await Promise.all(
-      Array.from(knownIds.values()).map(async (id) => {
-        const abort = store.aborts.get(id);
-        const task = store.tasks.get(id);
-        if (!abort && !task && !plugin?.gateway?.stopAccount) {
-          return;
-        }
+    // Gate replacement starts before teardown begins. Failures still reject only
+    // after every sibling account has finished its independent lifecycle cleanup.
+    const stopOutcomes = await Promise.all(
+      Array.from(knownIds.values()).map(async (id): Promise<ChannelAccountStopOutcome> => {
         const rKey = restartKey(channelId, id);
         if (manual) {
           manuallyStopped.add(rKey);
         }
-        abort?.abort();
-        const log = ensureChannelLog(channelId);
-        const runtime = ensureChannelRuntime(channelId);
-        if (plugin?.gateway?.stopAccount) {
-          const account = plugin.config.resolveAccount(cfg, id);
-          await plugin.gateway.stopAccount({
-            cfg,
-            accountId: id,
-            account,
-            runtime,
-            abortSignal: abort?.signal ?? new AbortController().signal,
-            log,
-            getStatus: () => getRuntime(channelId, id),
-            setStatus: (next) => setRuntime(channelId, id, next),
-          });
-        }
-        const stoppedCleanly = await waitForChannelStopGracefully(
-          task,
-          CHANNEL_STOP_ABORT_TIMEOUT_MS,
-        );
-        if (!stoppedCleanly) {
-          log.warn?.(
-            `[${id}] channel stop exceeded ${CHANNEL_STOP_ABORT_TIMEOUT_MS}ms after abort; continuing shutdown`,
+
+        const runStopAttempt = async (
+          previousOutcome: ChannelAccountStopOutcome,
+        ): Promise<ChannelAccountStopOutcome> => {
+          const abort = store.aborts.get(id);
+          const task = store.tasks.get(id);
+          if (!abort && !task && !plugin?.gateway?.stopAccount) {
+            return previousOutcome;
+          }
+          abort?.abort();
+          const log = ensureChannelLog(channelId);
+          const runtime = ensureChannelRuntime(channelId);
+          let outcome: ChannelAccountStopOutcome = { status: "fulfilled" };
+          if (plugin?.gateway?.stopAccount) {
+            try {
+              const account = plugin.config.resolveAccount(cfg, id);
+              await plugin.gateway.stopAccount({
+                cfg,
+                accountId: id,
+                account,
+                runtime,
+                abortSignal: abort?.signal ?? new AbortController().signal,
+                log,
+                getStatus: () => getRuntime(channelId, id),
+                setStatus: (next) => setRuntime(channelId, id, next),
+              });
+            } catch (error) {
+              outcome = { status: "rejected", error };
+              log.warn?.(`[${id}] stopAccount failed: ${formatErrorMessage(error)}`);
+            }
+          }
+          const stoppedCleanly = await waitForChannelStopGracefully(
+            task,
+            CHANNEL_STOP_ABORT_TIMEOUT_MS,
           );
-          const stoppedPatch = {
-            restartPending: !manual,
-            lastError: `channel stop timed out after ${CHANNEL_STOP_ABORT_TIMEOUT_MS}ms`,
-          };
-          if (manual) {
+          if (!stoppedCleanly) {
+            log.warn?.(
+              `[${id}] channel stop exceeded ${CHANNEL_STOP_ABORT_TIMEOUT_MS}ms after abort; continuing shutdown`,
+            );
+          }
+          if (outcome.status === "rejected") {
+            recoveryStopTimedOut.delete(rKey);
+            recoveryStartRequested.delete(rKey);
+            if (stoppedCleanly) {
+              if (store.aborts.get(id) === abort) {
+                store.aborts.delete(id);
+              }
+              if (store.tasks.get(id) === task) {
+                store.tasks.delete(id);
+              }
+            }
             setRuntime(channelId, id, {
               accountId: id,
               running: true,
-              ...stoppedPatch,
+              restartPending: false,
+              lastError: formatErrorMessage(outcome.error),
             });
+            return outcome;
+          }
+          if (!stoppedCleanly) {
+            const stoppedPatch = {
+              restartPending: !manual,
+              lastError: `channel stop timed out after ${CHANNEL_STOP_ABORT_TIMEOUT_MS}ms`,
+            };
+            if (manual) {
+              setRuntime(channelId, id, {
+                accountId: id,
+                running: true,
+                ...stoppedPatch,
+              });
+            } else {
+              setStoppedRuntime(channelId, id, stoppedPatch);
+              recoveryStopTimedOut.add(rKey);
+            }
+            return outcome;
+          }
+          recoveryStopTimedOut.delete(rKey);
+          recoveryStartRequested.delete(rKey);
+          if (store.aborts.get(id) === abort) {
+            store.aborts.delete(id);
+          }
+          if (store.tasks.get(id) === task) {
+            store.tasks.delete(id);
+          }
+          setStoppedRuntime(channelId, id, {
+            restartPending: false,
+            lastStopAt: Date.now(),
+          });
+          return outcome;
+        };
+
+        const currentStop = store.stops.get(id);
+        const previousStop =
+          currentStop?.status === "stopping"
+            ? currentStop.attempt
+            : Promise.resolve<ChannelAccountStopOutcome>(currentStop ?? { status: "fulfilled" });
+        const stopAttempt = previousStop.then(runStopAttempt);
+        store.stops.set(id, { status: "stopping", attempt: stopAttempt });
+        const outcome = await stopAttempt;
+        const latestStop = store.stops.get(id);
+        if (latestStop?.status === "stopping" && latestStop.attempt === stopAttempt) {
+          if (outcome.status === "rejected") {
+            store.stops.set(id, outcome);
           } else {
-            setStoppedRuntime(channelId, id, stoppedPatch);
+            store.stops.delete(id);
           }
-          if (!manual) {
-            recoveryStopTimedOut.add(rKey);
-          }
-          return;
         }
-        recoveryStopTimedOut.delete(rKey);
-        recoveryStartRequested.delete(rKey);
-        store.aborts.delete(id);
-        store.tasks.delete(id);
-        setStoppedRuntime(channelId, id, {
-          restartPending: false,
-          lastStopAt: Date.now(),
-        });
+        return outcome;
       }),
     );
+    const failedStop = stopOutcomes.find((outcome) => outcome.status === "rejected");
+    if (failedStop?.status === "rejected") {
+      throw failedStop.error;
+    }
   };
 
   const startChannels = async () => {

src/gateway/server-methods/channels.start.test.ts

@@ -278,4 +278,47 @@ describe("channelsHandlers channels.logout", () => {
       undefined,
     );
   });
+
+  it("does not clear channel auth when runtime teardown fails", async () => {
+    const stopChannel = vi.fn(async () => {
+      throw new Error("stop failed");
+    });
+    const logoutAccount = vi.fn(async () => ({ cleared: true, loggedOut: true }));
+    const markChannelLoggedOut = vi.fn();
+    const respond = vi.fn();
+    mocks.getChannelPlugin.mockReturnValue({
+      id: "whatsapp",
+      gateway: { logoutAccount },
+      config: {
+        defaultAccountId: () => "default-account",
+        listAccountIds: () => ["default-account"],
+        resolveAccount: () => ({}),
+      },
+    });
+
+    await expectDefined(
+      channelsHandlers["channels.logout"],
+      'channelsHandlers["channels.logout"] test invariant',
+    )(
+      createOptions(
+        { channel: "whatsapp" },
+        {
+          respond,
+          context: {
+            getRuntimeConfig: mocks.getRuntimeConfig,
+            stopChannel,
+            markChannelLoggedOut,
+          } as unknown as GatewayRequestHandlerOptions["context"],
+        },
+      ),
+    );
+
+    expect(logoutAccount).not.toHaveBeenCalled();
+    expect(markChannelLoggedOut).not.toHaveBeenCalled();
+    expect(respond).toHaveBeenCalledWith(
+      false,
+      undefined,
+      expect.objectContaining({ code: "UNAVAILABLE", message: "Error: stop failed" }),
+    );
+  });
 });