From 6e4f2e1ec46f05d50083ef3342e4499d334d59bb Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Mon, 27 Jul 2026 00:39:55 -0400 Subject: [PATCH] fix(gateway): settle account stops before failing (#114280) Co-authored-by: Qiong --- src/gateway/channel-health-monitor.test.ts | 15 ++ src/gateway/server-channels.test.ts | 195 ++++++++++++++++++ src/gateway/server-channels.ts | 180 +++++++++++----- .../server-methods/channels.start.test.ts | 43 ++++ 4 files changed, 381 insertions(+), 52 deletions(-) diff --git a/src/gateway/channel-health-monitor.test.ts b/src/gateway/channel-health-monitor.test.ts index adfecefc5706..a0a8d2ac2b0e 100644 --- a/src/gateway/channel-health-monitor.test.ts +++ b/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 } }); diff --git a/src/gateway/server-channels.test.ts b/src/gateway/server-channels.test.ts index e56adac865c0..a42c9c793a15 100644 --- a/src/gateway/server-channels.test.ts +++ b/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) => + await new Promise((resolve) => { + abortSignal.addEventListener( + "abort", + () => { + void taskReleases.get(accountId)?.promise.then(resolve); + }, + { once: true }, + ); + }), + ); + const stopAccount = vi.fn(async ({ accountId }: ChannelGatewayContext) => { + 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 })); diff --git a/src/gateway/server-channels.ts b/src/gateway/server-channels.ts index 307f27ad5b4c..f636f9959bd5 100644 --- a/src/gateway/server-channels.ts +++ b/src/gateway/server-channels.ts @@ -61,6 +61,7 @@ function waitForChannelStartupHandoff(): Promise { type ChannelRuntimeStore = { aborts: Map; starting: Map>; + stops: Map; tasks: Map>; runtimes: Map; }; @@ -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 } + | Extract; + async function waitForDeferredAccountStart( deferred: Promise, 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([ ...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 => { 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 => { + 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(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 () => { diff --git a/src/gateway/server-methods/channels.start.test.ts b/src/gateway/server-methods/channels.start.test.ts index 6121761a87dc..05466e1ba90a 100644 --- a/src/gateway/server-methods/channels.start.test.ts +++ b/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" }), + ); + }); });