From a4eba6c606580cd64dab72c2415b5acaa66575db Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Thu, 20 Aug 2026 03:38:39 -0700 Subject: [PATCH] fix(channels): retire buses before shutdown (#126637) --- extensions/buzz/src/gateway.lifecycle.test.ts | 119 +++++++++++++++- extensions/buzz/src/gateway.ts | 3 +- .../nostr/src/channel.lifecycle.test.ts | 130 +++++++++++++++++- extensions/nostr/src/gateway.ts | 15 +- 4 files changed, 252 insertions(+), 15 deletions(-) diff --git a/extensions/buzz/src/gateway.lifecycle.test.ts b/extensions/buzz/src/gateway.lifecycle.test.ts index a9082375e758..08ddcba5ca41 100644 --- a/extensions/buzz/src/gateway.lifecycle.test.ts +++ b/extensions/buzz/src/gateway.lifecycle.test.ts @@ -34,7 +34,12 @@ vi.mock("./inbound.js", () => ({ })); import { BuzzDirectoryState } from "./directory-state.js"; -import { buzzOutboundAdapter, sendBuzzTyping, startBuzzGatewayAccount } from "./gateway.js"; +import { + buzzOutboundAdapter, + getActiveBuzzBus, + sendBuzzTyping, + startBuzzGatewayAccount, +} from "./gateway.js"; import { BUZZ_NORMAL_MESSAGE_KIND } from "./message-event.js"; import { setBuzzRuntime } from "./runtime.js"; import { resolveBuzzAccount } from "./types.js"; @@ -243,6 +248,118 @@ describe("Buzz gateway lifecycle", () => { expect(gatewayMocks.sendBuzzTextOneShot).not.toHaveBeenCalled(); }); + it.each(["resolves", "rejects"] as const)( + "retires the active bus before asynchronous shutdown %s", + async (closeOutcome) => { + let resolveClose: (() => void) | undefined; + let rejectClose: ((error: Error) => void) | undefined; + const closePending = new Promise((resolve, reject) => { + resolveClose = resolve; + rejectClose = reject; + }); + gatewayMocks.close.mockImplementationOnce(() => closePending); + const { abortController, cfg, account, lifecycle, setStatus } = startTestGateway(); + + try { + await vi.waitFor(() => expect(getActiveBuzzBus(account.accountId)).toBeDefined()); + abortController.abort(); + await vi.waitFor(() => expect(gatewayMocks.close).toHaveBeenCalledOnce()); + + expect(getActiveBuzzBus(account.accountId)).toBeUndefined(); + expect(setStatus).not.toHaveBeenCalledWith({ + accountId: account.accountId, + running: false, + }); + + const pendingResult = await buzzOutboundAdapter.sendText({ + cfg, + to: `buzz:${CHANNEL_ID}`, + text: "while closing", + accountId: account.accountId, + }); + await sendBuzzTyping({ + cfg, + to: `buzz:${CHANNEL_ID}`, + accountId: account.accountId, + }); + + expect(pendingResult.messageId).toBe("standalone-event-id"); + expect(gatewayMocks.sendBuzzTextOneShot).toHaveBeenCalledOnce(); + expect(gatewayMocks.busSendText).not.toHaveBeenCalled(); + expect(gatewayMocks.busSendTyping).not.toHaveBeenCalled(); + + if (closeOutcome === "rejects") { + const closeError = new Error("Buzz close failed"); + rejectClose?.(closeError); + await expect(lifecycle).rejects.toBe(closeError); + expect(setStatus).not.toHaveBeenCalledWith({ + accountId: account.accountId, + running: false, + }); + } else { + resolveClose?.(); + await expect(lifecycle).resolves.toBeUndefined(); + expect(setStatus).toHaveBeenLastCalledWith({ + accountId: account.accountId, + running: false, + }); + } + + expect(getActiveBuzzBus(account.accountId)).toBeUndefined(); + await buzzOutboundAdapter.sendText({ + cfg, + to: `buzz:${CHANNEL_ID}`, + text: "after closing", + accountId: account.accountId, + }); + await sendBuzzTyping({ + cfg, + to: `buzz:${CHANNEL_ID}`, + accountId: account.accountId, + }); + expect(gatewayMocks.sendBuzzTextOneShot).toHaveBeenCalledTimes(2); + expect(gatewayMocks.busSendText).not.toHaveBeenCalled(); + expect(gatewayMocks.busSendTyping).not.toHaveBeenCalled(); + } finally { + abortController.abort(); + resolveClose?.(); + await lifecycle.catch(() => undefined); + } + }, + ); + + it("does not retire a replacement bus when an earlier generation finishes closing", async () => { + let resolveClose: (() => void) | undefined; + const closePending = new Promise((resolve) => { + resolveClose = resolve; + }); + gatewayMocks.close.mockImplementationOnce(() => closePending); + const first = startTestGateway(); + let replacement: ReturnType | undefined; + + try { + await vi.waitFor(() => expect(getActiveBuzzBus(first.account.accountId)).toBeDefined()); + replacement = startTestGateway(); + await vi.waitFor(() => expect(gatewayMocks.startBuzzBus).toHaveBeenCalledTimes(2)); + const replacementBus = getActiveBuzzBus(first.account.accountId); + expect(replacementBus).toBeDefined(); + + first.abortController.abort(); + await vi.waitFor(() => expect(gatewayMocks.close).toHaveBeenCalledOnce()); + expect(getActiveBuzzBus(first.account.accountId)).toBe(replacementBus); + + resolveClose?.(); + await expect(first.lifecycle).resolves.toBeUndefined(); + expect(getActiveBuzzBus(first.account.accountId)).toBe(replacementBus); + } finally { + first.abortController.abort(); + resolveClose?.(); + await first.lifecycle.catch(() => undefined); + replacement?.abortController.abort(); + await replacement?.lifecycle.catch(() => undefined); + } + }); + it("reuses the gateway bus for sends in the running process", async () => { const { abortController, cfg, lifecycle } = startTestGateway({ profileName: "BuzzClaw" }); await vi.waitFor(() => expect(gatewayMocks.startBuzzBus).toHaveBeenCalledOnce()); diff --git a/extensions/buzz/src/gateway.ts b/extensions/buzz/src/gateway.ts index c66f6af029fb..e4c9f895a7f7 100644 --- a/extensions/buzz/src/gateway.ts +++ b/extensions/buzz/src/gateway.ts @@ -174,10 +174,11 @@ export async function startBuzzGatewayAccount(ctx: ChannelGatewayContext { expect(getActiveNostrBuses().has("default")).toBe(false); }); + it.each([ + { outcome: "resolves", closeFails: false }, + { outcome: "rejects", closeFails: true }, + ])("retires the active bus before shutdown $outcome", async ({ closeFails }) => { + const bus = createMockBus(); + let finishClose!: () => void; + let rejectClose!: (reason: Error) => void; + bus.close.mockReturnValueOnce( + new Promise((resolve, reject) => { + finishClose = resolve; + rejectClose = reject; + }), + ); + mocks.startNostrBus.mockResolvedValueOnce(bus as never); + const abort = new AbortController(); + const context = bindChannelRuntime( + createStartAccountContext({ + account: buildResolvedNostrAccount(), + abortSignal: abort.signal, + }), + ); + const lifecycle = startNostrGatewayAccount(context); + + await vi.waitFor(() => expect(getActiveNostrBuses().get("default")).toBe(bus)); + abort.abort(); + await vi.waitFor(() => expect(bus.close).toHaveBeenCalledOnce()); + + const activeBusWhileClosing = getActiveNostrBuses().get("default"); + const sendWhileClosing = await nostrOutboundAdapter + .sendText({ + cfg: context.cfg, + to: context.account.publicKey, + text: "hello", + accountId: context.account.accountId, + }) + .then( + () => undefined, + (error: unknown) => error, + ); + const sendsWhileClosing = bus.sendDm.mock.calls.length; + expect(context.log?.info).not.toHaveBeenCalledWith("[default] Nostr provider stopped"); + + if (closeFails) { + const closeError = new Error("Nostr relay shutdown failed"); + rejectClose(closeError); + await expect(lifecycle).rejects.toBe(closeError); + expect(context.log?.info).not.toHaveBeenCalledWith("[default] Nostr provider stopped"); + } else { + finishClose(); + await expect(lifecycle).resolves.toBeUndefined(); + expect(context.log?.info).toHaveBeenCalledWith("[default] Nostr provider stopped"); + } + + expect(activeBusWhileClosing).toBeUndefined(); + expect(sendWhileClosing).toEqual(new Error("Nostr bus not running for account default")); + expect(sendsWhileClosing).toBe(0); + expect(getActiveNostrBuses().has("default")).toBe(false); + + if (closeFails) { + await expect( + nostrOutboundAdapter.sendText({ + cfg: context.cfg, + to: context.account.publicKey, + text: "hello again", + accountId: context.account.accountId, + }), + ).rejects.toThrow("Nostr bus not running for account default"); + expect(bus.sendDm).not.toHaveBeenCalled(); + } + }); + + it("does not retire a replacement bus while the previous generation closes", async () => { + const firstBus = createMockBus(); + const replacementBus = createMockBus(); + let finishFirstClose!: () => void; + firstBus.close.mockReturnValueOnce( + new Promise((resolve) => { + finishFirstClose = resolve; + }), + ); + mocks.startNostrBus + .mockResolvedValueOnce(firstBus as never) + .mockResolvedValueOnce(replacementBus as never); + const firstAbort = new AbortController(); + const firstLifecycle = startNostrGatewayAccount( + bindChannelRuntime( + createStartAccountContext({ + account: buildResolvedNostrAccount(), + abortSignal: firstAbort.signal, + }), + ), + ); + let replacementAbort: AbortController | undefined; + let replacementLifecycle: typeof firstLifecycle | undefined; + + try { + await vi.waitFor(() => expect(getActiveNostrBuses().get("default")).toBe(firstBus)); + replacementAbort = new AbortController(); + replacementLifecycle = startNostrGatewayAccount( + bindChannelRuntime( + createStartAccountContext({ + account: buildResolvedNostrAccount(), + abortSignal: replacementAbort.signal, + }), + ), + ); + await vi.waitFor(() => expect(getActiveNostrBuses().get("default")).toBe(replacementBus)); + + firstAbort.abort(); + await vi.waitFor(() => expect(firstBus.close).toHaveBeenCalledOnce()); + expect(getActiveNostrBuses().get("default")).toBe(replacementBus); + + finishFirstClose(); + await expect(firstLifecycle).resolves.toBeUndefined(); + expect(getActiveNostrBuses().get("default")).toBe(replacementBus); + + replacementAbort.abort(); + await expect(replacementLifecycle).resolves.toBeUndefined(); + expect(getActiveNostrBuses().has("default")).toBe(false); + } finally { + firstAbort.abort(); + finishFirstClose(); + await firstLifecycle.catch(() => undefined); + replacementAbort?.abort(); + await replacementLifecycle?.catch(() => undefined); + } + }); + it("stops immediately when startAccount receives an already-aborted signal", async () => { const bus = createMockBus(); mocks.startNostrBus.mockResolvedValueOnce(bus as never); diff --git a/extensions/nostr/src/gateway.ts b/extensions/nostr/src/gateway.ts index f9032a2a3b9a..ac85ec43fc96 100644 --- a/extensions/nostr/src/gateway.ts +++ b/extensions/nostr/src/gateway.ts @@ -18,7 +18,7 @@ import { } from "openclaw/plugin-sdk/text-chunking"; import type { PluginRuntime } from "../runtime-api.js"; import type { ChannelOutboundAdapter, ChannelPlugin } from "./channel-api.js"; -import type { MetricEvent, MetricsSnapshot } from "./metrics.js"; +import type { MetricEvent } from "./metrics.js"; import { startNostrBus, type NostrBusHandle } from "./nostr-bus.js"; import { normalizePubkey } from "./nostr-key-utils.js"; import { getNostrRuntime } from "./runtime.js"; @@ -35,7 +35,6 @@ type NostrOutboundAdapter = Pick< sanitizeText: NonNullable; }; const activeBuses = new Map(); -const metricsSnapshots = new Map(); const ACCESS_GROUP_PREFIX = "accessGroup:"; function normalizeRelayLifecycleKey(relay: string): string { @@ -135,7 +134,6 @@ export const startNostrGatewayAccount: NostrGatewayStart = async (ctx) => { : undefined, }); - let busHandle: NostrBusHandle | null = null; const connectedRelays = new Set(); const authorizeSender = async (input: { @@ -295,12 +293,8 @@ export const startNostrGatewayAccount: NostrGatewayStart = async (ctx) => { } else if (event.name === "relay.error") { ctx.log?.debug?.(`[${account.accountId}] Relay error: ${event.labels?.relay}`); } - if (busHandle) { - metricsSnapshots.set(account.accountId, busHandle.getMetrics()); - } }, }); - busHandle = bus; activeBuses.set(account.accountId, bus); ctx.log?.info?.( @@ -309,14 +303,11 @@ export const startNostrGatewayAccount: NostrGatewayStart = async (ctx) => { return { stop: async () => { - await bus.close(); - if (busHandle === bus) { - busHandle = null; - } + // Retire before fallible async shutdown so new work cannot reacquire this bus. if (activeBuses.get(account.accountId) === bus) { activeBuses.delete(account.accountId); } - metricsSnapshots.delete(account.accountId); + await bus.close(); ctx.log?.info?.(`[${account.accountId}] Nostr provider stopped`); }, };