diff --git a/src/gateway/server-channels.test.ts b/src/gateway/server-channels.test.ts index fb1703f91b22..91cab0ba3484 100644 --- a/src/gateway/server-channels.test.ts +++ b/src/gateway/server-channels.test.ts @@ -546,6 +546,47 @@ describe("server-channels auto restart", () => { expect(account?.lastError).toBeNull(); }); + it("keeps the second recovery task running when the stale task rejects", async () => { + const releaseFirstTask = createDeferred(); + let startCount = 0; + const startAccount = vi.fn(async ({ abortSignal }: { abortSignal: AbortSignal }) => { + startCount += 1; + abortSignal.addEventListener("abort", () => {}, { once: true }); + if (startCount === 1) { + await releaseFirstTask.promise; + throw new Error("late stale worker exit"); + } + await new Promise(() => {}); + }); + installTestRegistry( + createTestPlugin({ + startAccount, + }), + ); + const manager = createManager(); + + await manager.startChannels(); + const recoveryStopTask = manager.stopChannel("discord", DEFAULT_ACCOUNT_ID, { + manual: false, + }); + await vi.advanceTimersByTimeAsync(5_000); + await recoveryStopTask; + + await manager.startChannel("discord", DEFAULT_ACCOUNT_ID); + await manager.startChannel("discord", DEFAULT_ACCOUNT_ID); + expect(startAccount).toHaveBeenCalledTimes(2); + + releaseFirstTask.resolve(); + await flushMicrotasks(); + + const account = manager.getRuntimeSnapshot().channelAccounts.discord?.[DEFAULT_ACCOUNT_ID]; + expect(startAccount).toHaveBeenCalledTimes(2); + expect(account?.running).toBe(true); + expect(account?.restartPending).toBe(false); + expect(account?.lastError).toBeNull(); + expect(hoisted.sleepWithAbort).not.toHaveBeenCalled(); + }); + it("restarts immediately when recovery stop timeout settles with an error", async () => { const rejectFirstTask = createDeferred(); let startCount = 0; diff --git a/src/gateway/server-channels.ts b/src/gateway/server-channels.ts index b129534f5eb0..bf654573a79d 100644 --- a/src/gateway/server-channels.ts +++ b/src/gateway/server-channels.ts @@ -570,6 +570,8 @@ export function createChannelManager(opts: ChannelManagerOptions): ChannelManage return; } + let trackedPromise: Promise; + const isCurrentTask = () => store.tasks.get(id) === trackedPromise; scopedChannelRuntime = await measureStartup(`channels.${channelId}.runtime`, async () => createTaskScopedChannelRuntime({ channelRuntime: await getChannelRuntime(), @@ -628,7 +630,10 @@ export function createChannelManager(opts: ChannelManagerOptions): ChannelManage abortSignal: abort.signal, log, getStatus: () => getRuntime(channelId, id), - setStatus: (next) => setRuntimeFromTaskStatus(channelId, id, next, abort.signal), + setStatus: (next) => + isCurrentTask() + ? setRuntimeFromTaskStatus(channelId, id, next, abort.signal) + : getRuntime(channelId, id), ...(channelRuntimeForTask ? { channelRuntime: channelRuntimeForTask } : {}), }); const routeRegistry = getPluginHttpRouteRegistry?.(); @@ -641,9 +646,11 @@ export function createChannelManager(opts: ChannelManagerOptions): ChannelManage } await startAccountTask; }); - const trackedPromise = task + // Recovery can replace a timed-out task before the old promise settles. + // Only the task that still owns the store slot may write lifecycle state. + trackedPromise = task .then(() => { - if (abort.signal.aborted || manuallyStopped.has(rKey)) { + if (abort.signal.aborted || manuallyStopped.has(rKey) || !isCurrentTask()) { return; } const message = "channel exited without an error"; @@ -651,17 +658,26 @@ export function createChannelManager(opts: ChannelManagerOptions): ChannelManage log.error?.(`[${id}] ${message}`); }) .catch((err: unknown) => { + if (!isCurrentTask()) { + return; + } const message = formatErrorMessage(err); setRuntime(channelId, id, { accountId: id, lastError: message }); log.error?.(`[${id}] channel exited: ${message}`); }) .then(async () => { await cleanupTaskScopedApprovalRuntime("channel cleanup failed"); + if (!isCurrentTask()) { + return; + } setStoppedRuntime(channelId, id, { lastStopAt: Date.now(), }); }) .then(async () => { + if (!isCurrentTask()) { + return; + } if (manuallyStopped.has(rKey)) { recoveryStopTimedOut.delete(rKey); recoveryStartRequested.delete(rKey);