diff --git a/src/gateway/worker-environments/workspace-mutation-lock-remote-script.ts b/src/gateway/worker-environments/workspace-mutation-lock-remote-script.ts index 2fd3735fb4d2..fbdbfbb5dd30 100644 --- a/src/gateway/worker-environments/workspace-mutation-lock-remote-script.ts +++ b/src/gateway/worker-environments/workspace-mutation-lock-remote-script.ts @@ -3,33 +3,49 @@ export const REMOTE_WORKSPACE_MUTATION_LOCK_JS = String.raw`const lockRoot = pat ".openclaw-accepted-lock-" + workspaceKey, ); const lockToken = crypto.randomBytes(16).toString("hex"); -const lockOwner = { action, nonce, pid: lockOwnerPid, token: lockToken }; +const lockOwner = { + action, + nonce, + pid: lockOwnerPid, + controllerPid: process.pid, + token: lockToken, +}; const lockWait = new Int32Array(new SharedArrayBuffer(4)); const lockDeadlineMs = Date.now() + 9 * 60 * 1000; let acquiredLock; function encodeLockIdentity(identity) { - return [identity.action, identity.nonce, identity.pid, identity.token].join("."); + return [ + identity.action, + identity.nonce, + identity.pid, + identity.controllerPid, + identity.token, + ].join("."); } function parseLockIdentity(parts) { - if (parts.length !== 4) return null; - const [entryAction, entryNonce, rawPid, token] = parts; + if (parts.length !== 5) return null; + const [entryAction, entryNonce, rawPid, rawControllerPid, token] = parts; const pid = Number(rawPid); + const controllerPid = Number(rawControllerPid); if ( !mutationActions.includes(entryAction) || !/^[a-f0-9]{32}$/.test(entryNonce || "") || !/^[1-9][0-9]*$/.test(rawPid || "") || !Number.isSafeInteger(pid) || + !/^[1-9][0-9]*$/.test(rawControllerPid || "") || + !Number.isSafeInteger(controllerPid) || !/^[a-f0-9]{32}$/.test(token || "") ) { return null; } - return { action: entryAction, nonce: entryNonce, pid, token }; + return { action: entryAction, nonce: entryNonce, pid, controllerPid, token }; } function sameLockIdentity(left, right) { return ( left.action === right.action && left.nonce === right.nonce && left.pid === right.pid && + left.controllerPid === right.controllerPid && left.token === right.token ); } @@ -54,7 +70,7 @@ function processGroupIsAlive(pid) { } } function lockIdentityIsAlive(identity) { - return processIsAlive(identity.pid) || + return processIsAlive(identity.controllerPid) || (identity.action === "receiver" && processGroupIsAlive(identity.pid)); } function ownerEntryName(owner) { @@ -65,13 +81,13 @@ function reclaimEntryName(owner, reclaimer) { } function parseLockEntry(name) { const parts = name.split("."); - if (parts[0] === "owner" && parts.length === 5) { + if (parts[0] === "owner" && parts.length === 6) { const owner = parseLockIdentity(parts.slice(1)); return owner ? { kind: "owner", owner } : null; } - if (parts[0] === "reclaim" && parts.length === 9) { - const owner = parseLockIdentity(parts.slice(1, 5)); - const reclaimer = parseLockIdentity(parts.slice(5)); + if (parts[0] === "reclaim" && parts.length === 11) { + const owner = parseLockIdentity(parts.slice(1, 6)); + const reclaimer = parseLockIdentity(parts.slice(6)); return owner && reclaimer ? { kind: "reclaim", owner, reclaimer } : null; } return null; diff --git a/src/gateway/worker-environments/workspace-mutation-remote-script.test.ts b/src/gateway/worker-environments/workspace-mutation-remote-script.test.ts index ae930325e9e4..8831d3ba9d98 100644 --- a/src/gateway/worker-environments/workspace-mutation-remote-script.test.ts +++ b/src/gateway/worker-environments/workspace-mutation-remote-script.test.ts @@ -25,7 +25,7 @@ function spawnTransaction(argv: string[], env: NodeJS.ProcessEnv) { signal, stderr, })); - return { exited }; + return { pid: child.pid, exited }; } describe("remote workspace mutation receiver script", () => { @@ -60,7 +60,7 @@ describe("remote workspace mutation receiver script", () => { String.raw`const fs = require("node:fs"); const kill = process.kill.bind(process); process.kill = function(pid, signal) { - if (signal === 0 && pid < 0 && process.argv[4] === process.env.OPENCLAW_TEST_RESET_NONCE) { + if (signal === 0 && pid > 0 && process.argv[4] === process.env.OPENCLAW_TEST_RESET_NONCE) { fs.writeFileSync(process.env.OPENCLAW_TEST_CONTENDER_MARKER, ""); } return kill(pid, signal); @@ -100,10 +100,16 @@ process.kill = function(pid, signal) { const workspaceKey = createHash("sha256").update(workspace).digest("hex"); const lock = path.join(path.dirname(workspace), `.openclaw-accepted-lock-${workspaceKey}`); const [ownerName] = await fs.readdir(lock); - const receiverPid = Number( - /^owner\.receiver\.[a-f0-9]{32}\.([1-9][0-9]*)\./u.exec(ownerName!)?.[1], - ); + const owner = + /^owner\.receiver\.[a-f0-9]{32}\.([1-9][0-9]*)\.([1-9][0-9]*)\.[a-f0-9]{32}$/u.exec( + ownerName!, + ); + const receiverPid = Number(owner?.[1]); + const controllerPid = Number(owner?.[2]); expect(Number.isSafeInteger(receiverPid)).toBe(true); + expect(Number.isSafeInteger(controllerPid)).toBe(true); + expect(controllerPid).toBe(receiver.pid); + expect(controllerPid).not.toBe(receiverPid); await waitForDead(receiverPid, 10_000); const reset = runCommandWithTimeout( diff --git a/src/gateway/worker-environments/workspace-sync-scripts.test.ts b/src/gateway/worker-environments/workspace-sync-scripts.test.ts index 3d3513b66765..d598fc3d3a83 100644 --- a/src/gateway/worker-environments/workspace-sync-scripts.test.ts +++ b/src/gateway/worker-environments/workspace-sync-scripts.test.ts @@ -605,9 +605,9 @@ process.kill = function(pid, signal) { const deadPid = 2_147_483_647; const token = "9".repeat(32); await fs.mkdir(lock); - const ownerIdentity = ["apply", nonce, deadPid, token].join("."); + const ownerIdentity = ["apply", nonce, deadPid, deadPid, token].join("."); const reclaimToken = "a".repeat(32); - const reclaimerIdentity = ["settle", nonce, deadPid, reclaimToken].join("."); + const reclaimerIdentity = ["settle", nonce, deadPid, deadPid, reclaimToken].join("."); await fs.writeFile(path.join(lock, `reclaim.${ownerIdentity}.${reclaimerIdentity}`), ""); const settled = await runTransaction("settle"); diff --git a/src/infra/outbound/channel-bootstrap.runtime.test.ts b/src/infra/outbound/channel-bootstrap.runtime.test.ts index a9f3644d99cf..aed8be09f096 100644 --- a/src/infra/outbound/channel-bootstrap.runtime.test.ts +++ b/src/infra/outbound/channel-bootstrap.runtime.test.ts @@ -109,8 +109,10 @@ describe("bootstrapOutboundChannelPlugin", () => { ] as never; loaderMocks.loadPluginRegistryHandle.mockReturnValue(handle); + expect(bootstrapOutboundChannelPlugin({ channel: "discord", cfg: discordConfig })).toBe(handle); expect(bootstrapOutboundChannelPlugin({ channel: "discord", cfg: discordConfig })).toBe(handle); expect(getActivePluginRegistry()).toBe(root); + expect(loaderMocks.loadPluginRegistryHandle).toHaveBeenCalledTimes(1); expect(loaderMocks.loadPluginRegistryHandle).toHaveBeenCalledWith( expect.objectContaining({ onlyPluginIds: ["discord"] }), ); @@ -170,6 +172,24 @@ describe("bootstrapOutboundChannelPlugin", () => { expect(loaderMocks.loadPluginRegistryHandle).toHaveBeenCalledTimes(1); }); + it("bounds failed channel outcomes and refreshes misses by LRU recency", () => { + installDiscordSetupShell(); + loaderMocks.loadPluginRegistryHandle.mockReturnValue(createEmptyPluginRegistry()); + + for (let index = 0; index < 64; index += 1) { + bootstrapOutboundChannelPlugin({ channel: `channel-${index}`, cfg: discordConfig }); + } + bootstrapOutboundChannelPlugin({ channel: "channel-0", cfg: discordConfig }); + bootstrapOutboundChannelPlugin({ channel: "channel-64", cfg: discordConfig }); + bootstrapOutboundChannelPlugin({ channel: "channel-0", cfg: discordConfig }); + + expect(loaderMocks.loadPluginRegistryHandle).toHaveBeenCalledTimes(65); + + bootstrapOutboundChannelPlugin({ channel: "channel-1", cfg: discordConfig }); + + expect(loaderMocks.loadPluginRegistryHandle).toHaveBeenCalledTimes(66); + }); + it("retries after the runtime config changes", () => { installDiscordSetupShell(); bootstrapOutboundChannelPlugin({ channel: "discord", cfg: discordConfig }); diff --git a/src/infra/outbound/channel-bootstrap.runtime.ts b/src/infra/outbound/channel-bootstrap.runtime.ts index 07f3471be5d9..90d1fd250eee 100644 --- a/src/infra/outbound/channel-bootstrap.runtime.ts +++ b/src/infra/outbound/channel-bootstrap.runtime.ts @@ -13,9 +13,21 @@ import { getActivePluginRegistry, getActivePluginRegistryVersion } from "../../p import { pruneMapToMaxSize } from "../map-size.js"; const MAX_BOOTSTRAP_CONFIG_GENERATIONS = 64; +const MAX_BOOTSTRAP_CHANNEL_OUTCOMES_PER_CONFIG = 64; let bootstrapRegistryGeneration: string | undefined; const bootstrapRegistriesByConfig = new Map>(); +function cacheBootstrapOutcome( + registries: Map, + channel: string, + outcome: PluginRegistry | null, +): void { + // Reinsert every outcome, including null, so reads and writes share LRU ordering. + registries.delete(channel); + registries.set(channel, outcome); + pruneMapToMaxSize(registries, MAX_BOOTSTRAP_CHANNEL_OUTCOMES_PER_CONFIG); +} + function resolveBootstrapRegistryGeneration(): string { return String(getActivePluginRegistryVersion()); } @@ -84,8 +96,10 @@ export function bootstrapOutboundChannelPlugin(params: { } const registries = resolveBootstrapRegistries(cfg); - if (registries.has(params.channel)) { - return resolveSendCapableRegistry(registries.get(params.channel), params.channel); + const cachedRegistry = registries.get(params.channel); + if (cachedRegistry !== undefined) { + cacheBootstrapOutcome(registries, params.channel, cachedRegistry); + return resolveSendCapableRegistry(cachedRegistry, params.channel); } const autoEnabled = applyPluginAutoEnable({ config: cfg }); @@ -101,6 +115,7 @@ export function bootstrapOutboundChannelPlugin(params: { const activatedConfig = withActivatedPluginIds({ config: autoEnabled.config, pluginIds }) ?? autoEnabled.config; const activatedSourceConfig = withActivatedPluginIds({ config: cfg, pluginIds }) ?? cfg; + let sendRegistry: PluginRegistry | undefined; try { const registry = loadPluginRegistryHandle({ config: activatedConfig, @@ -112,12 +127,10 @@ export function bootstrapOutboundChannelPlugin(params: { allowGatewaySubagentBinding: true, }, }); - const sendRegistry = resolveSendCapableRegistry(registry, params.channel); - registries.set(params.channel, sendRegistry ?? null); - return sendRegistry; + sendRegistry = resolveSendCapableRegistry(registry, params.channel); } catch { // Best-effort bootstrap; the caller reports the unavailable channel. - registries.set(params.channel, null); - return undefined; } + cacheBootstrapOutcome(registries, params.channel, sendRegistry ?? null); + return sendRegistry; }