fix(channels): retire buses before shutdown (#126637)

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<void>((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<void>((resolve) => {
+      resolveClose = resolve;
+    });
+    gatewayMocks.close.mockImplementationOnce(() => closePending);
+    const first = startTestGateway();
+    let replacement: ReturnType<typeof startTestGateway> | 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());

extensions/buzz/src/gateway.ts

@@ -174,10 +174,11 @@ export async function startBuzzGatewayAccount(ctx: ChannelGatewayContext<Resolve
       }
       cycleError = error instanceof Error ? error : new Error(String(error));
     } finally {
-      await bus?.close();
+      // Retire before fallible async shutdown so new work cannot reacquire this bus.
       if (activeBuses.get(account.accountId) === bus) {
         activeBuses.delete(account.accountId);
       }
+      await bus?.close();
       ctx.setStatus({
         accountId: account.accountId,
         running: false,

extensions/nostr/src/channel.lifecycle.test.ts

@@ -8,7 +8,7 @@ import {
   waitForStartedMocks,
 } from "openclaw/plugin-sdk/channel-test-helpers";
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
-import { getActiveNostrBuses, startNostrGatewayAccount } from "./gateway.js";
+import { getActiveNostrBuses, nostrOutboundAdapter, startNostrGatewayAccount } from "./gateway.js";
 import { setNostrRuntime } from "./runtime.js";
 import { buildResolvedNostrAccount } from "./test-fixtures.js";
 
@@ -92,6 +92,134 @@ describe("nostr gateway lifecycle", () => {
     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<void>((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<void>((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);

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<ChannelOutboundAdapter["sanitizeText"]>;
 };
 const activeBuses = new Map<string, NostrBusHandle>();
-const metricsSnapshots = new Map<string, MetricsSnapshot>();
 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<string>();
 
   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`);
         },
       };