diff --git a/extensions/buzz/src/directory-relay.test.ts b/extensions/buzz/src/directory-relay.test.ts index 3f709bafe1ed..0fa3fff0b511 100644 --- a/extensions/buzz/src/directory-relay.test.ts +++ b/extensions/buzz/src/directory-relay.test.ts @@ -107,6 +107,48 @@ describe("Buzz directory relay", () => { expect(subscriptions).toHaveLength(1); }); + it("closes sibling profile subscriptions when one chunk fails", () => { + const subscriptions: SubscriptionRecord[] = []; + const relay = { + idleSince: undefined, + ongoingOperations: 0, + prepareSubscription: vi.fn( + ( + filters: Filter[], + handlers: SubscriptionRecord["handlers"], + ): ReturnType => { + const close = vi.fn(); + subscriptions.push({ filters, handlers, close }); + return { + id: `sub:${subscriptions.length}`, + close, + } as ReturnType; + }, + ), + send: vi.fn(async () => {}), + } as unknown as Relay; + const directory = startBuzzDirectoryRelay({ + relay, + relayPublicKey: RELAY_PUBLIC_KEY, + state: new BuzzDirectoryState({ + publicKey: BOT_PUBLIC_KEY, + fallbackProfileName: "OpenClaw", + channelIds: [], + }), + }); + + directory.replaceProfilePublicKeys( + Array.from({ length: 201 }, (_, index) => index.toString(16).padStart(64, "0")), + ); + expect(subscriptions).toHaveLength(2); + + subscriptions[0]?.handlers.onclose("relay rejected subscription"); + + expect(subscriptions[1]?.close).toHaveBeenCalledWith( + "directory profile subscription generation failed", + ); + }); + it("defers query cleanup until the relay confirms EOSE", async () => { const abort = new AbortController(); let handlers: SubscriptionRecord["handlers"] | undefined; @@ -144,4 +186,37 @@ describe("Buzz directory relay", () => { handlers?.oneose(); expect(close).toHaveBeenCalledWith("directory query complete"); }); + + it("recycles the relay instead of closing a query before EOSE", async () => { + vi.useFakeTimers(); + const subscriptionClose = vi.fn(); + const relayClose = vi.fn(); + const relay = { + close: relayClose, + idleSince: undefined, + ongoingOperations: 0, + prepareSubscription: vi.fn( + (): ReturnType => + ({ id: "sub:1", close: subscriptionClose }) as ReturnType, + ), + send: vi.fn(async () => {}), + } as unknown as Relay; + const query = queryBuzzDirectoryRooms({ + relay, + relayPublicKey: RELAY_PUBLIC_KEY, + state: new BuzzDirectoryState({ + publicKey: BOT_PUBLIC_KEY, + fallbackProfileName: "OpenClaw", + channelIds: [], + }), + channelIds: ["7c4a6d2a-2ed9-4b4e-a5e2-4d705ee9b34c"], + }); + + const rejection = expect(query).rejects.toThrow("Timed out loading Buzz directory snapshot"); + await vi.advanceTimersByTimeAsync(10_000); + await rejection; + expect(subscriptionClose).not.toHaveBeenCalled(); + expect(relayClose).toHaveBeenCalledOnce(); + vi.useRealTimers(); + }); }); diff --git a/extensions/buzz/src/directory-relay.ts b/extensions/buzz/src/directory-relay.ts index 28f7ff0a7a30..6f3da762178e 100644 --- a/extensions/buzz/src/directory-relay.ts +++ b/extensions/buzz/src/directory-relay.ts @@ -9,8 +9,10 @@ import { openBuzzRelaySubscription } from "./relay-subscription.js"; const BUZZ_ROOM_QUERY_CHUNK_SIZE = 1_000; const PROFILE_SUBSCRIPTION_REPLACED_REASON = "directory profile subscription replaced"; +const PROFILE_SUBSCRIPTION_FAILED_REASON = "directory profile subscription generation failed"; const DIRECTORY_SHUTDOWN_REASON = "directory shutdown"; const DIRECTORY_QUERY_COMPLETE_REASON = "directory query complete"; +const DIRECTORY_QUERY_TIMEOUT_MS = 10_000; type BuzzSubscription = ReturnType; type ProfileSubscriptionGeneration = { @@ -37,12 +39,18 @@ async function queryBuzzDirectoryBatch(params: { await new Promise((resolve, reject) => { let settled = false; let receivedEose = false; + const timeout = setTimeout(() => { + const error = new Error("Timed out loading Buzz directory snapshot"); + finish(error); + params.relay.close(); + }, DIRECTORY_QUERY_TIMEOUT_MS); const subscriptionRef: { current?: BuzzSubscription } = {}; const finish = (error?: unknown) => { if (settled) { return; } settled = true; + clearTimeout(timeout); params.signal?.removeEventListener("abort", onAbort); if (receivedEose) { subscriptionRef.current?.close(DIRECTORY_QUERY_COMPLETE_REASON); @@ -159,14 +167,16 @@ export function startBuzzDirectoryRelay(params: { ); }; - const closeProfileGeneration = (reason: string) => { + const closeProfileGeneration = (reason: string, skip?: BuzzSubscription) => { const current = profileGeneration; profileGeneration = undefined; if (!current) { return; } for (const subscription of current.subscriptions) { - subscription.close(reason); + if (subscription !== skip) { + subscription.close(reason); + } } }; @@ -206,7 +216,8 @@ export function startBuzzDirectoryRelay(params: { applyQueuedProfilePublicKeys(); } }; - const subscription = openBuzzRelaySubscription( + let subscription: BuzzSubscription; + subscription = openBuzzRelaySubscription( params.relay, [ { @@ -222,11 +233,12 @@ export function startBuzzDirectoryRelay(params: { oneose: markReady, onclose: (reason) => { if (profileGeneration === generation) { - profileGeneration = undefined; queuedProfilePublicKeys = undefined; + closeProfileGeneration(PROFILE_SUBSCRIPTION_FAILED_REASON, subscription); } if ( reason !== PROFILE_SUBSCRIPTION_REPLACED_REASON && + reason !== PROFILE_SUBSCRIPTION_FAILED_REASON && reason !== DIRECTORY_SHUTDOWN_REASON && reason !== "relay connection closed by us" ) { diff --git a/extensions/buzz/src/directory.test.ts b/extensions/buzz/src/directory.test.ts index b6c227460fb0..1c6c1bef1527 100644 --- a/extensions/buzz/src/directory.test.ts +++ b/extensions/buzz/src/directory.test.ts @@ -225,4 +225,35 @@ describe("Buzz live directory", () => { expect(refreshDirectory).toHaveBeenCalledOnce(); expect(relayMocks.connect).not.toHaveBeenCalled(); }); + + it("returns the active directory snapshot when room metadata refresh fails", async () => { + const refreshDirectory = vi.fn(async () => { + throw new Error("relay stalled"); + }); + gatewayMocks.activeBus = { + directory: { + self: () => null, + listPeers: () => [], + listGroupMembers: () => [], + listGroups: () => [{ kind: "group", id: `buzz:${ROOM_ID}`, name: "Cached Engineering" }], + }, + refreshDirectory, + }; + const { listBuzzDirectoryGroupsLive } = await import("./directory.js"); + + await expect( + listBuzzDirectoryGroupsLive({ + cfg: { + channels: { + buzz: { + relayUrl: "wss://buzz.example.com", + privateKey: PRIVATE_KEY, + groups: { [ROOM_ID]: {} }, + }, + }, + } as unknown as OpenClawConfig, + accountId: "default", + }), + ).resolves.toEqual([{ kind: "group", id: `buzz:${ROOM_ID}`, name: "Cached Engineering" }]); + }); }); diff --git a/extensions/buzz/src/directory.ts b/extensions/buzz/src/directory.ts index e07dbd3cce4e..7644cbae60b0 100644 --- a/extensions/buzz/src/directory.ts +++ b/extensions/buzz/src/directory.ts @@ -50,7 +50,12 @@ async function loadBuzzDirectoryState( const activeBus = getActiveBuzzBus(configured.account.accountId); if (activeBus) { if (options.refreshRooms) { - await activeBus.refreshDirectory(); + try { + await activeBus.refreshDirectory(); + } catch { + // A stalled metadata refresh recycles the relay session. Directory + // reads can still return the last complete in-memory snapshot. + } } return activeBus.directory; }