fix: bound Buzz directory refresh lifecycle

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<Relay["prepareSubscription"]> => {
+          const close = vi.fn();
+          subscriptions.push({ filters, handlers, close });
+          return {
+            id: `sub:${subscriptions.length}`,
+            close,
+          } as ReturnType<Relay["prepareSubscription"]>;
+        },
+      ),
+      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<Relay["prepareSubscription"]> =>
+          ({ id: "sub:1", close: subscriptionClose }) as ReturnType<Relay["prepareSubscription"]>,
+      ),
+      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();
+  });
 });

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<Relay["prepareSubscription"]>;
 type ProfileSubscriptionGeneration = {
@@ -37,12 +39,18 @@ async function queryBuzzDirectoryBatch(params: {
   await new Promise<void>((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"
               ) {

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" }]);
+  });
 });

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;
   }