From 29deb75ec5837b77dd3d54bf9f19e5a1e4a41b3d Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sun, 16 Aug 2026 12:26:32 -0700 Subject: [PATCH] refactor(gateway): drop dead session-subscriber registry surface (#124754) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * refactor(gateway): drop dead session-subscriber registry surface getForConnection and both registry clear() methods had no production callers — only server-chat-state.test.ts consumed them (verified by repo-wide grep incl. plugin-sdk). The recency-sorted connToSessionKeys index existed solely so getForConnection returned keys in subscription order, an ordering nothing read, rebuilt via toSorted() on every subscribe/unsubscribe/settle — including the provisional-replay settle path. unsubscribeAll now reads the reverse index directly from connToSessionRecency's keys (same contents, no ordering), removing the per-call O(k log k) rebuild and the duplicate map. Tests assert through get()/getApprovals(), the production read paths, per test-audit norms. Behavior-neutral: subscription state transitions and event-visible delivery unchanged; chat-state, broadcast, node-subscription, instance-runtime, agent-events, session-events suites pass. * test(gateway): fix ETXTBSY flake in workspace-sync candidate bounding The test rewrote its mock git script between two createGitTransferList calls. On Linux CI the second spawn can race a lingering fork of the first child still holding the script's write fd, failing exec with ETXTBSY. Write the script once; vary the entry count through a data file the script reads instead. --- src/gateway/server-chat-state.test.ts | 19 ++----- src/gateway/server-chat-state.ts | 53 ++----------------- .../workspace-sync-local.test.ts | 21 +++++--- 3 files changed, 21 insertions(+), 72 deletions(-) diff --git a/src/gateway/server-chat-state.test.ts b/src/gateway/server-chat-state.test.ts index 29286e738524..0f7052569321 100644 --- a/src/gateway/server-chat-state.test.ts +++ b/src/gateway/server-chat-state.test.ts @@ -168,15 +168,10 @@ describe("createSessionMessageSubscriberRegistry", () => { subscribers.subscribe("conn-other", "agent:main:child", { includeApprovals: true }); subscribers.unsubscribeAll("conn-reviewer"); - expect([...subscribers.getForConnection("conn-reviewer")]).toEqual([]); expect([...subscribers.get("agent:main:main")]).toEqual([]); expect([...subscribers.getApprovals("agent:main:main")]).toEqual([]); expect([...subscribers.get("agent:main:child")]).toEqual(["conn-other"]); expect([...subscribers.getApprovals("agent:main:child")]).toEqual(["conn-other"]); - - subscribers.clear(); - expect([...subscribers.get("agent:main:child")]).toEqual([]); - expect([...subscribers.getApprovals("agent:main:child")]).toEqual([]); }); it.each(["first", "second"])( @@ -195,7 +190,6 @@ describe("createSessionMessageSubscriberRegistry", () => { } expect([...subscribers.get("agent:main:main")]).toEqual([]); - expect([...subscribers.getForConnection("conn")]).toEqual([]); }, ); @@ -215,10 +209,8 @@ describe("createSessionMessageSubscriberRegistry", () => { first(); } - expect([...subscribers.getForConnection("conn")]).toEqual([ - "agent:main:other", - "agent:main:main", - ]); + expect([...subscribers.get("agent:main:other")]).toEqual(["conn"]); + expect([...subscribers.get("agent:main:main")]).toEqual(["conn"]); }, ); @@ -230,10 +222,8 @@ describe("createSessionMessageSubscriberRegistry", () => { rollback(); - expect([...subscribers.getForConnection("conn")]).toEqual([ - "agent:main:main", - "agent:main:child", - ]); + expect([...subscribers.get("agent:main:main")]).toEqual(["conn"]); + expect([...subscribers.get("agent:main:child")]).toEqual(["conn"]); }); it("does not restore a replay invalidated by unsubscribe", () => { @@ -245,7 +235,6 @@ describe("createSessionMessageSubscriberRegistry", () => { subscribers.unsubscribe("conn", "agent:main:main"); subscription.commit(); - expect([...subscribers.getForConnection("conn")]).toEqual([]); expect([...subscribers.get("agent:main:main")]).toEqual([]); }); }); diff --git a/src/gateway/server-chat-state.ts b/src/gateway/server-chat-state.ts index abba9721a596..02893bdf8d2e 100644 --- a/src/gateway/server-chat-state.ts +++ b/src/gateway/server-chat-state.ts @@ -338,7 +338,6 @@ export type SessionEventSubscriberRegistry = { subscribe: (connId: string) => void; unsubscribe: (connId: string) => void; getAll: () => ReadonlySet; - clear: () => void; }; export type SessionMessageSubscriberRegistry = { @@ -350,10 +349,8 @@ export type SessionMessageSubscriberRegistry = { unsubscribe: (connId: string, sessionKey: string) => void; unsubscribeAll: (connId: string) => void; get: (sessionKey: string) => ReadonlySet; - getForConnection: (connId: string) => ReadonlySet; getApprovals: (sessionKey: string) => ReadonlySet; onChange: (listener: (sessionKey: string) => void) => () => void; - clear: () => void; }; type SessionMessageSubscription = (() => void) & { commit: () => void }; @@ -391,18 +388,15 @@ export function createSessionEventSubscriberRegistry(): SessionEventSubscriberRe connIds.delete(normalized); }, getAll: () => (connIds.size > 0 ? connIds : empty), - clear: () => { - connIds.clear(); - }, }; } /** Create the per-session message subscriber registry. */ export function createSessionMessageSubscriberRegistry(): SessionMessageSubscriberRegistry { const sessionToConnIds = new Map>(); - const connToSessionKeys = new Map>(); // The final state after overlapping replays settles to their latest success // or the original committed base; failed provisionals cannot leave ghosts. + // Its keys double as the per-connection reverse index for unsubscribeAll. const connToSessionRecency = new Map>(); const provisionalSubscriptions = new Map>(); const approvalSessionToConnIds = new Map>(); @@ -412,17 +406,6 @@ export function createSessionMessageSubscriberRegistry(): SessionMessageSubscrib let subscriptionSequence = 0; const normalize = (value: string): string => value.trim(); - const rebuildConnectionSessionKeys = (connId: string) => { - const recency = connToSessionRecency.get(connId); - if (!recency || recency.size === 0) { - connToSessionKeys.delete(connId); - return; - } - connToSessionKeys.set( - connId, - new Set([...recency.entries()].toSorted(([, a], [, b]) => a - b).map(([key]) => key)), - ); - }; const setMessageSubscription = (connId: string, sessionKey: string, subscribed: boolean) => { const connIds = sessionToConnIds.get(sessionKey); const wasSubscribed = connIds?.has(connId) === true; @@ -497,7 +480,6 @@ export function createSessionMessageSubscriberRegistry(): SessionMessageSubscrib setMessageSubscription(normalizedConnId, normalizedSessionKey, true); recency.set(normalizedSessionKey, provisionalRecency); connToSessionRecency.set(normalizedConnId, recency); - rebuildConnectionSessionKeys(normalizedConnId); setApprovalSubscription( normalizedConnId, @@ -537,7 +519,6 @@ export function createSessionMessageSubscriberRegistry(): SessionMessageSubscrib if (recency.size === 0) { connToSessionRecency.delete(normalizedConnId); } - rebuildConnectionSessionKeys(normalizedConnId); states.delete(normalizedSessionKey); if (states.size === 0) { provisionalSubscriptions.delete(normalizedConnId); @@ -573,7 +554,6 @@ export function createSessionMessageSubscriberRegistry(): SessionMessageSubscrib if (recency.size === 0) { connToSessionRecency.delete(normalizedConnId); } - rebuildConnectionSessionKeys(normalizedConnId); } const approvalConnIds = approvalSessionToConnIds.get(normalizedSessionKey); if (approvalConnIds) { @@ -600,14 +580,13 @@ export function createSessionMessageSubscriberRegistry(): SessionMessageSubscrib state.active = false; } provisionalSubscriptions.delete(normalizedConnId); - const sessionKeys = connToSessionKeys.get(normalizedConnId); + const sessionKeys = connToSessionRecency.get(normalizedConnId); if (!sessionKeys) { return; } - for (const sessionKey of sessionKeys) { + for (const sessionKey of sessionKeys.keys()) { setMessageSubscription(normalizedConnId, sessionKey, false); } - connToSessionKeys.delete(normalizedConnId); connToSessionRecency.delete(normalizedConnId); const approvalSessionKeys = connToApprovalSessionKeys.get(normalizedConnId); @@ -627,13 +606,6 @@ export function createSessionMessageSubscriberRegistry(): SessionMessageSubscrib } return sessionToConnIds.get(normalizedSessionKey) ?? empty; }, - getForConnection: (connId: string) => { - const normalizedConnId = normalize(connId); - if (!normalizedConnId) { - return empty; - } - return connToSessionKeys.get(normalizedConnId) ?? empty; - }, getApprovals: (sessionKey: string) => { const normalizedSessionKey = normalize(sessionKey); if (!normalizedSessionKey) { @@ -645,25 +617,6 @@ export function createSessionMessageSubscriberRegistry(): SessionMessageSubscrib changeListeners.add(listener); return () => changeListeners.delete(listener); }, - clear: () => { - const changedSessionKeys = [...sessionToConnIds.keys()].toSorted(); - sessionToConnIds.clear(); - connToSessionKeys.clear(); - connToSessionRecency.clear(); - for (const states of provisionalSubscriptions.values()) { - for (const state of states.values()) { - state.active = false; - } - } - provisionalSubscriptions.clear(); - approvalSessionToConnIds.clear(); - connToApprovalSessionKeys.clear(); - for (const sessionKey of changedSessionKeys) { - for (const listener of changeListeners) { - listener(sessionKey); - } - } - }, }; return registry; } diff --git a/src/gateway/worker-environments/workspace-sync-local.test.ts b/src/gateway/worker-environments/workspace-sync-local.test.ts index 9f66959e2410..41e21dae848d 100644 --- a/src/gateway/worker-environments/workspace-sync-local.test.ts +++ b/src/gateway/worker-environments/workspace-sync-local.test.ts @@ -268,19 +268,26 @@ describe("runLocalCommandToFile", () => { const root = tempDirs.make("openclaw-workspace-candidates-"); const bin = path.join(root, "bin"); const mockGit = path.join(bin, "git"); + const countFile = path.join(bin, "git-entry-count"); const firstTransfer = `${root}-transfer-accepted`; const secondTransfer = `${root}-transfer-rejected`; + // Rewriting an executed script between spawns races exec against a forked + // child still holding the write fd (ETXTBSY). Write the script once and + // vary only a data file it reads. const writeMockGit = async (count: number) => { - await fs.writeFile( - mockGit, - `#!/usr/bin/env node -process.stdout.write("eligible.txt\\0".repeat(${count})); -`, - { mode: 0o755 }, - ); + await fs.writeFile(countFile, String(count)); }; try { await fs.mkdir(bin); + await fs.writeFile( + mockGit, + `#!/usr/bin/env node +const fs = require("node:fs"); +const count = Number(fs.readFileSync(${JSON.stringify(countFile)}, "utf8")); +process.stdout.write("eligible.txt\\0".repeat(count)); +`, + { mode: 0o755 }, + ); await fs.writeFile(path.join(root, "eligible.txt"), "eligible\n"); vi.stubEnv("PATH", `${bin}${path.delimiter}${process.env.PATH ?? ""}`); await writeMockGit(MAX_WORKSPACE_INVENTORY_ENTRIES + 1);