From 08ac5b987fb02e65882fb767ed8a8dbae7bd12c2 Mon Sep 17 00:00:00 2001 From: Shakker Date: Fri, 31 Jul 2026 11:40:34 +0100 Subject: [PATCH] fix: cancel Buzz turns during reconnect --- .../buzz/src/buzz-bus.lifecycle.test.ts | 45 ++++++++++++++++++- extensions/buzz/src/buzz-bus.ts | 13 +++--- extensions/buzz/src/gateway.lifecycle.test.ts | 7 ++- extensions/buzz/src/gateway.ts | 4 +- extensions/buzz/src/inbound.test.ts | 17 +++++++ extensions/buzz/src/inbound.ts | 6 ++- 6 files changed, 81 insertions(+), 11 deletions(-) diff --git a/extensions/buzz/src/buzz-bus.lifecycle.test.ts b/extensions/buzz/src/buzz-bus.lifecycle.test.ts index 81716e5b9600..b07d018a06c6 100644 --- a/extensions/buzz/src/buzz-bus.lifecycle.test.ts +++ b/extensions/buzz/src/buzz-bus.lifecycle.test.ts @@ -108,7 +108,7 @@ vi.mock("nostr-tools", async (importOriginal) => { }; }); -import { sendBuzzTextOneShot, startBuzzBus } from "./buzz-bus.js"; +import { sendBuzzTextOneShot, startBuzzBus, type BuzzBus } from "./buzz-bus.js"; import { BUZZ_DIFF_MESSAGE_KIND, BUZZ_INBOUND_MESSAGE_KINDS, @@ -493,6 +493,49 @@ describe("Buzz bus lifecycle", () => { expect(relayMocks.close).toHaveBeenCalledOnce(); }); + it("aborts active inbound dispatch before waiting for bus shutdown", async () => { + relayMocks.auth.mockResolvedValue("ok"); + relayMocks.roomHistoryEvents = [ + { + id: "d".repeat(64), + kind: 9, + pubkey: SENDER_PUBLIC_KEY, + created_at: 1_700_000_000, + content: "historical message", + sig: "e".repeat(128), + tags: [["h", CHANNEL_ID]], + }, + ]; + let dispatchSignal: AbortSignal | undefined; + const onMessage = vi.fn( + async (_message: BuzzInboundMessage, _bus: BuzzBus, signal: AbortSignal) => { + dispatchSignal = signal; + await new Promise((resolve) => { + if (signal.aborted) { + resolve(); + return; + } + signal.addEventListener("abort", () => resolve(), { once: true }); + }); + }, + ); + const bus = await startBuzzBus({ + accountId: ACCOUNT_ID, + relayUrl: "wss://buzz.example.com", + privateKey: PRIVATE_KEY, + channelIds: [CHANNEL_ID], + onMessage, + }); + + await vi.waitFor(() => expect(onMessage).toHaveBeenCalledOnce()); + expect(dispatchSignal?.aborted).toBe(false); + + await bus.close(); + + expect(dispatchSignal?.aborted).toBe(true); + expect(relayMocks.close).toHaveBeenCalledOnce(); + }); + it("leaves room subscription shutdown to the relay", async () => { relayMocks.auth.mockResolvedValue("ok"); const bus = await startBuzzBus({ diff --git a/extensions/buzz/src/buzz-bus.ts b/extensions/buzz/src/buzz-bus.ts index 6f5fe45b6ddf..b922b4802932 100644 --- a/extensions/buzz/src/buzz-bus.ts +++ b/extensions/buzz/src/buzz-bus.ts @@ -510,7 +510,7 @@ export async function startBuzzBus(options: { authTag?: string; channelIds: string[]; since?: number; - onMessage: (message: BuzzInboundMessage, bus: BuzzBus) => Promise; + onMessage: (message: BuzzInboundMessage, bus: BuzzBus, signal: AbortSignal) => Promise; onMessageError?: (error: Error) => void; onFatalError?: (error: Error) => void; onDedupeError?: (error: Error) => void; @@ -578,6 +578,7 @@ export async function startBuzzBus(options: { await directoryRelay?.refreshRooms(options.channelIds); }, sendText: async ({ channelId, text, threadId, replyToId }) => { + signal.throwIfAborted(); const event = buildBuzzTextEvent({ secretKey, channelId, text, threadId, replyToId }); await relay.publish(event); return event.id; @@ -597,10 +598,10 @@ export async function startBuzzBus(options: { await relay.send(JSON.stringify(["EVENT", event])); }, close: async () => { - // A replacement bus must not overlap handlers admitted by this relay - // generation. Let those handlers finish while their reply socket is live. - await dispatchQueue.close(); lifecycleAbort.abort(new Error("Buzz bus closed")); + // Agent turns receive this generation signal. Abort them before draining + // so reconnect and Gateway shutdown do not wait on stale work. + await dispatchQueue.close(); stopPresenceHeartbeat(); directoryRelay?.close(); replayGuard.clearMemory(); @@ -630,7 +631,7 @@ export async function startBuzzBus(options: { // only by active workers, so replay cannot create unbounded in-flight state. const admission = dispatchQueue.enqueue(async () => { await replayGuard.processGuarded(event, async () => { - await options.onMessage(message, bus); + await options.onMessage(message, bus, signal); }); }); if (admission === "overflow") { @@ -722,8 +723,8 @@ export async function startBuzzBus(options: { } catch (error) { // Every failed startup must release the socket before ownership returns to // the gateway-level reconnect loop. - await dispatchQueue.close(); lifecycleAbort.abort(error); + await dispatchQueue.close(); directoryRelay?.close(); relay.close(); throw error; diff --git a/extensions/buzz/src/gateway.lifecycle.test.ts b/extensions/buzz/src/gateway.lifecycle.test.ts index ec3fe4bc4c13..78048f7996ac 100644 --- a/extensions/buzz/src/gateway.lifecycle.test.ts +++ b/extensions/buzz/src/gateway.lifecycle.test.ts @@ -10,7 +10,11 @@ const gatewayMocks = vi.hoisted(() => ({ busSendTyping: vi.fn(async () => undefined), sendBuzzTextOneShot: vi.fn(async () => "standalone-event-id"), onMessage: undefined as - | ((message: import("./message-event.js").BuzzInboundMessage, bus: BuzzBus) => Promise) + | (( + message: import("./message-event.js").BuzzInboundMessage, + bus: BuzzBus, + signal: AbortSignal, + ) => Promise) | undefined, onMessageError: undefined as ((error: Error) => void) | undefined, onFatalError: undefined as ((error: Error) => void) | undefined, @@ -83,6 +87,7 @@ describe("Buzz gateway lifecycle", () => { onMessage: ( message: import("./message-event.js").BuzzInboundMessage, bus: BuzzBus, + signal: AbortSignal, ) => Promise; onMessageError?: (error: Error) => void; onFatalError?: (error: Error) => void; diff --git a/extensions/buzz/src/gateway.ts b/extensions/buzz/src/gateway.ts index f52c144e5475..8574d448210a 100644 --- a/extensions/buzz/src/gateway.ts +++ b/extensions/buzz/src/gateway.ts @@ -97,12 +97,12 @@ export async function startBuzzGatewayAccount(ctx: ChannelGatewayContext { + onMessage: async (message, sessionBus, signal) => { // Subscription filters reduce traffic, but relay events remain untrusted. if (!isConfiguredBuzzChannel(configuredChannelIds, message.channelId)) { return; } - await handleBuzzInbound({ account, cfg: ctx.cfg, bus: sessionBus, message }); + await handleBuzzInbound({ account, cfg: ctx.cfg, bus: sessionBus, message, signal }); }, onMessageError: (error) => { ctx.log?.error?.(`[${account.accountId}] Buzz message failed: ${error.message}`); diff --git a/extensions/buzz/src/inbound.test.ts b/extensions/buzz/src/inbound.test.ts index b954f3ffb6f4..4d302c998252 100644 --- a/extensions/buzz/src/inbound.test.ts +++ b/extensions/buzz/src/inbound.test.ts @@ -55,6 +55,10 @@ function createMessage(overrides: Partial = {}): BuzzInbound }; } +function createSignal(): AbortSignal { + return new AbortController().signal; +} + function createBus(): BuzzBus { return { publicKey: BOT_PUBLIC_KEY, @@ -88,15 +92,18 @@ describe("handleBuzzInbound", () => { it("accepts a native Nostr public-key mention", async () => { const runtime = createPluginRuntimeMock(); setBuzzRuntime(runtime); + const signal = createSignal(); await handleBuzzInbound({ account: createAccount(), cfg: {} satisfies OpenClawConfig, bus: createBus(), message: createMessage({ mentionedPubkeys: [BOT_PUBLIC_KEY] }), + signal, }); expect(runtime.channel.inbound.dispatch).toHaveBeenCalledTimes(1); + expect(firstDispatch(runtime).replyOptions?.abortSignal).toBe(signal); expect(firstDispatch(runtime).ctxPayload).toMatchObject({ WasMentioned: true, SenderId: SENDER_PUBLIC_KEY, @@ -158,6 +165,7 @@ describe("handleBuzzInbound", () => { cfg: {} satisfies OpenClawConfig, bus, message: createMessage(), + signal: createSignal(), }); expect(firstDispatch(runtime).ctxPayload).toMatchObject({ @@ -178,6 +186,7 @@ describe("handleBuzzInbound", () => { cfg: {} satisfies OpenClawConfig, bus: createBus(), message: createMessage({ text: "@openclaw status" }), + signal: createSignal(), }); expect(runtime.channel.inbound.dispatch).toHaveBeenCalledTimes(1); @@ -193,6 +202,7 @@ describe("handleBuzzInbound", () => { cfg: {} satisfies OpenClawConfig, bus: createBus(), message: createMessage(), + signal: createSignal(), }); expect(runtime.channel.inbound.dispatch).not.toHaveBeenCalled(); @@ -210,6 +220,7 @@ describe("handleBuzzInbound", () => { cfg: {} satisfies OpenClawConfig, bus: createBus(), message: createMessage({ mentionedPubkeys: [BOT_PUBLIC_KEY] }), + signal: createSignal(), }); expect(runtime.channel.inbound.dispatch).not.toHaveBeenCalled(); @@ -231,6 +242,7 @@ describe("handleBuzzInbound", () => { message: createMessage({ text: "/status", }), + signal: createSignal(), }); expect(runtime.channel.inbound.dispatch).toHaveBeenCalledTimes(1); @@ -251,6 +263,7 @@ describe("handleBuzzInbound", () => { cfg: {} satisfies OpenClawConfig, bus: createBus(), message: createMessage({ text: "/status" }), + signal: createSignal(), }); expect(runtime.channel.inbound.dispatch).not.toHaveBeenCalled(); @@ -270,6 +283,7 @@ describe("handleBuzzInbound", () => { threadId: "event-root", mentionedPubkeys: [BOT_PUBLIC_KEY], }), + signal: createSignal(), }); const dispatch = firstDispatch(runtime); @@ -327,6 +341,7 @@ describe("handleBuzzInbound", () => { truncated: true, }, }), + signal: createSignal(), }); expect(runtime.channel.commands.shouldComputeCommandAuthorized).not.toHaveBeenCalled(); @@ -365,6 +380,7 @@ describe("handleBuzzInbound", () => { truncated: false, }, }), + signal: createSignal(), }); expect(runtime.channel.mentions.matchesMentionPatterns).not.toHaveBeenCalled(); @@ -380,6 +396,7 @@ describe("handleBuzzInbound", () => { cfg: {} satisfies OpenClawConfig, bus: createBus(), message: createMessage({ mentionedPubkeys: [BOT_PUBLIC_KEY] }), + signal: createSignal(), }); const dispatch = firstDispatch(runtime); diff --git a/extensions/buzz/src/inbound.ts b/extensions/buzz/src/inbound.ts index 44fd4e471152..f558c7925c28 100644 --- a/extensions/buzz/src/inbound.ts +++ b/extensions/buzz/src/inbound.ts @@ -22,9 +22,10 @@ export async function handleBuzzInbound(params: { cfg: OpenClawConfig; bus: BuzzBus; message: BuzzInboundMessage; + signal: AbortSignal; }) { const runtime = getBuzzRuntime(); - const { account, cfg, bus, message } = params; + const { account, cfg, bus, message, signal } = params; const channelId = parseBuzzTarget(message.channelId); const target = buildBuzzTarget(channelId); const textForAgent = formatBuzzMessageForAgent(message); @@ -161,6 +162,9 @@ export async function handleBuzzInbound(params: { throw error instanceof Error ? error : new Error(String(error)); }, }, + replyOptions: { + abortSignal: signal, + }, replyPipeline: { typing: { start: async () => {