From c40c08e071dd0be2cc3ef06eb6bc43f8a447a893 Mon Sep 17 00:00:00 2001 From: Galin Iliev Date: Wed, 19 Aug 2026 11:04:47 -0700 Subject: [PATCH] fix(memory): fence brokered runtime during gateway updates --- docs/concepts/memory-multiplayer.md | 7 +++ src/gateway/server-methods/update.test.ts | 42 +++++++++++++ src/gateway/server-methods/update.ts | 72 +++++++++++++---------- src/plugins/memory-broker-runtime.test.ts | 30 ++++++++++ src/plugins/memory-broker-runtime.ts | 35 ++++++++++- 5 files changed, 154 insertions(+), 32 deletions(-) diff --git a/docs/concepts/memory-multiplayer.md b/docs/concepts/memory-multiplayer.md index 8c0c9214b918..73b14b041046 100644 --- a/docs/concepts/memory-multiplayer.md +++ b/docs/concepts/memory-multiplayer.md @@ -1815,6 +1815,13 @@ allow. restart, upgrade, backup, and recovery to this broker lifecycle. A restart rotates its epoch and rejects every pre-restart envelope; Gateway must re-authorize rather than reuse a continuation. +- For an unsupervised direct Gateway source update, Gateway takes the broker + lifecycle writer lease, fences and retires every selected broker before it + mutates the package tree, then keeps memory unavailable in that old process. + Core update and post-core plugin convergence share that fenced window; only + the replacement Gateway can start a child with a new epoch and secret. A + managed-service handoff instead waits for Gateway shutdown before its helper + mutates the install root. - Run agent processes in per-session sandboxes with only authorized virtual mounts. The agent never receives the broker database handle or real artifact root. diff --git a/src/gateway/server-methods/update.test.ts b/src/gateway/server-methods/update.test.ts index 96b314b041bc..9d3361770b24 100644 --- a/src/gateway/server-methods/update.test.ts +++ b/src/gateway/server-methods/update.test.ts @@ -77,6 +77,9 @@ type PostCoreFinalizeOutcome = Awaited< const runPostCoreFinalizeAfterGatewayUpdateMock = vi.fn<() => Promise>( async () => ({ status: "skipped", reason: "not-git-update" }), ); +const withBrokeredMemoryUpgradeMock = vi.fn( + async (run: () => Promise): Promise => await run(), +); type UpdateRunPayload = { ok: boolean; @@ -187,6 +190,10 @@ vi.mock("../../infra/update-runner.js", () => ({ runGatewayUpdate: runGatewayUpdateMock, })); +vi.mock("../../plugins/memory-broker-runtime.js", () => ({ + withBrokeredMemoryUpgrade: withBrokeredMemoryUpgradeMock, +})); + // Keep the real `foldPostCoreFinalizeIntoResult` so the restart-gate behavior on // finalize failure is exercised; only stub the subprocess-spawning finalizer. vi.mock("../../infra/update-post-core-finalize.js", async () => { @@ -312,6 +319,10 @@ beforeEach(() => { status: "skipped", reason: "not-git-update", }); + withBrokeredMemoryUpgradeMock.mockClear(); + withBrokeredMemoryUpgradeMock.mockImplementation( + async (run: () => Promise): Promise => await run(), + ); }); async function invokeUpdateRun( @@ -982,6 +993,37 @@ describe("update.run post-core plugin finalize", () => { expect(payload?.result?.status).toBe("ok"); }); + it("fences brokered memory around the full unsupervised git update lifecycle", async () => { + const order: string[] = []; + withBrokeredMemoryUpgradeMock.mockImplementationOnce(async (run: () => Promise) => { + order.push("upgrade:start"); + const result = await run(); + order.push("upgrade:finish"); + return result; + }); + runGatewayUpdateMock.mockImplementationOnce(async () => { + order.push("core:update"); + return { + status: "ok", + mode: "git", + root: "/tmp/openclaw-git", + after: { version: "2026.6.1" }, + steps: [], + durationMs: 100, + }; + }); + runPostCoreFinalizeAfterGatewayUpdateMock.mockImplementationOnce(async () => { + order.push("plugin:finalize"); + return { status: "ok", entrypoint: "/tmp/openclaw-git/dist/index.mjs" }; + }); + mockGitInstallSurface("/tmp/openclaw-git"); + + await captureUpdateRunPayload(); + + expect(withBrokeredMemoryUpgradeMock).toHaveBeenCalledOnce(); + expect(order).toEqual(["upgrade:start", "core:update", "plugin:finalize", "upgrade:finish"]); + }); + it("carries the pre-doctor source config into the git finalizer", async () => { const preUpdateConfig = { channels: { diff --git a/src/gateway/server-methods/update.ts b/src/gateway/server-methods/update.ts index 4eb0aa648225..3ae6a459b256 100644 --- a/src/gateway/server-methods/update.ts +++ b/src/gateway/server-methods/update.ts @@ -506,41 +506,51 @@ export const updateHandlers: GatewayRequestHandlers = { return undefined; }) : undefined; + const runDirectGatewayUpdate = async () => { + // The package tree can change beneath this process. Retire selected-memory children + // first, then keep this Gateway fenced until its replacement owns a fresh broker epoch. + result = await runGatewayUpdate({ + timeoutMs, + cwd: root, + argv1: process.argv[1], + channel: + installSurface.kind === "git" + ? (configChannel ?? undefined) + : effectiveChannel === "extended-stable" + ? effectiveChannel + : (configChannel ?? undefined), + ...(adoptedPackageTargetVersion ? { tag: adoptedPackageTargetVersion } : {}), + ...(adoptedDevTarget ? { devTarget: adoptedDevTarget } : {}), + allowGatewayServiceRepair: false, + allowGatewayActivation: false, + }); + // The CLI `openclaw update` resumes post-core plugin convergence after a + // git/source core update; the RPC path did not, leaving official managed + // plugins stale on the new core. Run the finalizer here to match. + const finalizeOutcome = await runPostCoreFinalizeAfterGatewayUpdate({ + result, + channel: configChannel ?? undefined, + serviceRepairPolicy: "external", + ...(timeoutMs === undefined ? {} : { timeoutMs }), + ...(preUpdateConfig ? { preUpdateConfig } : {}), + }); + if (finalizeOutcome.status === "error") { + context?.logGateway?.warn( + `update.run post-core plugin finalize failed ${formatControlPlaneActor(actor)} reason=${finalizeOutcome.reason}`, + ); + } + return foldPostCoreFinalizeIntoResult(result, finalizeOutcome); + }; // Supervised Windows gateways, including Startup-folder fallbacks, take // the detached handoff above. This direct path is unsupervised, so keep // doctor service mutation disabled: it could rewrite or terminate the // RPC server before the response and restart sentinel become durable. - result = await runGatewayUpdate({ - timeoutMs, - cwd: root, - argv1: process.argv[1], - channel: - installSurface.kind === "git" - ? (configChannel ?? undefined) - : effectiveChannel === "extended-stable" - ? effectiveChannel - : (configChannel ?? undefined), - ...(adoptedPackageTargetVersion ? { tag: adoptedPackageTargetVersion } : {}), - ...(adoptedDevTarget ? { devTarget: adoptedDevTarget } : {}), - allowGatewayServiceRepair: false, - allowGatewayActivation: false, - }); - // The CLI `openclaw update` resumes post-core plugin convergence after a - // git/source core update; the RPC path did not, leaving official managed - // plugins stale on the new core. Run the finalizer here to match. - const finalizeOutcome = await runPostCoreFinalizeAfterGatewayUpdate({ - result, - channel: configChannel ?? undefined, - serviceRepairPolicy: "external", - ...(timeoutMs === undefined ? {} : { timeoutMs }), - ...(preUpdateConfig ? { preUpdateConfig } : {}), - }); - if (finalizeOutcome.status === "error") { - context?.logGateway?.warn( - `update.run post-core plugin finalize failed ${formatControlPlaneActor(actor)} reason=${finalizeOutcome.reason}`, - ); - } - result = foldPostCoreFinalizeIntoResult(result, finalizeOutcome); + result = + installSurface.kind === "git" + ? await ( + await import("../../plugins/memory-broker-runtime.js") + ).withBrokeredMemoryUpgrade(runDirectGatewayUpdate) + : await runDirectGatewayUpdate(); } } catch { result = { diff --git a/src/plugins/memory-broker-runtime.test.ts b/src/plugins/memory-broker-runtime.test.ts index 8a1f40c45540..cad2d38c4439 100644 --- a/src/plugins/memory-broker-runtime.test.ts +++ b/src/plugins/memory-broker-runtime.test.ts @@ -5,6 +5,7 @@ import { startBrokeredMemoryRuntimeSupervisor, testing, withBrokeredMemoryMaintenance, + withBrokeredMemoryUpgrade, } from "./memory-broker-runtime.js"; import type { MemoryPluginCapability } from "./registry-contribution-types.js"; @@ -45,6 +46,7 @@ const brokerCapability = { afterEach(async () => { await closeBrokeredMemoryRuntimes(); + testing.clearBrokeredMemoryUpgradeFenceForTest(); startMemoryBrokerProcess.mockReset(); }); @@ -242,4 +244,32 @@ describe("brokered memory maintenance", () => { await Promise.all([supervisor, maintenance]); expect(run).toHaveBeenCalledOnce(); }); + + it("retires the broker before direct Gateway update work and keeps the old process fenced", async () => { + const order: string[] = []; + const broker = { + ...brokerProcess(), + close: vi.fn(async () => { + order.push("broker:closed"); + }), + }; + startMemoryBrokerProcess.mockResolvedValueOnce(broker); + await startBrokeredMemoryRuntimeSupervisor(brokerCapability); + + const update = vi.fn(async () => { + order.push("update"); + }); + await withBrokeredMemoryUpgrade(update); + + expect(order).toEqual(["broker:closed", "update"]); + expect(broker.close).toHaveBeenCalledOnce(); + + // Shutdown cannot reopen a broker after the old Gateway has started replacing its own code. + // The following readiness attempt models a late lifecycle callback in that same process. + await closeBrokeredMemoryRuntimes(); + await expect(startBrokeredMemoryRuntimeSupervisor(brokerCapability)).rejects.toThrow( + "selected memory broker did not become ready", + ); + expect(startMemoryBrokerProcess).toHaveBeenCalledOnce(); + }); }); diff --git a/src/plugins/memory-broker-runtime.ts b/src/plugins/memory-broker-runtime.ts index 98388d5fcd2c..f5fc0f129539 100644 --- a/src/plugins/memory-broker-runtime.ts +++ b/src/plugins/memory-broker-runtime.ts @@ -20,6 +20,7 @@ import type { type BrokerRuntimeState = { processes: Map>; agentIdsByModule: Map; + unavailableAfterUpgrade: Set; supervisors: Set; leases: Map>; maintenance: BrokerMaintenanceGate; @@ -47,6 +48,7 @@ const state = resolveGlobalSingleton( (): BrokerRuntimeState => ({ processes: new Map(), agentIdsByModule: new Map(), + unavailableAfterUpgrade: new Set(), supervisors: new Set(), leases: new Map(), maintenance: createBrokerMaintenanceGate(), @@ -252,6 +254,11 @@ async function resolveProcess( } async function resolveProcessOnLease(moduleUrl: string): Promise { + // The Gateway may update its own package tree while it is still answering the update RPC. Do + // not fork a child from a half-replaced tree: only the replacement Gateway may clear this fence. + if (state.unavailableAfterUpgrade.has(moduleUrl)) { + return undefined; + } const existing = state.processes.get(moduleUrl); if (existing) { try { @@ -328,7 +335,10 @@ export async function startBrokeredMemoryRuntimeSupervisor( if (!moduleUrl) { return undefined; } - state.agentIdsByModule.set(moduleUrl, Object.freeze([...new Set(params.agentIds ?? [])].toSorted())); + state.agentIdsByModule.set( + moduleUrl, + Object.freeze([...new Set(params.agentIds ?? [])].toSorted()), + ); const supervisor = await startMemoryBrokerSupervisor({ ensureProcess: () => resolveProcess(capability), retireProcess: () => retireProcess(capability), @@ -586,6 +596,28 @@ export async function closeBrokeredMemoryRuntimes(): Promise { } } +/** + * An in-process Gateway update must retire every selected broker before it rewrites code or + * plugins. The current process stays memory-unavailable afterward; only its replacement may + * create a child, run startup recovery, and mint a new epoch/secret. + */ +export async function withBrokeredMemoryUpgrade(run: () => Promise): Promise { + return await withGatewayBrokeredMemoryMaintenanceLease(state.maintenance, async () => { + const moduleUrls = [ + ...new Set([...state.processes.keys(), ...state.agentIdsByModule.keys()]), + ].toSorted(); + for (const moduleUrl of moduleUrls) { + await withBrokerLease(state.leases, moduleUrl, async () => { + // Fence resolution before closing the old child. A failed close must abort the update; + // proceeding could leave an old broker serving from code the update is about to replace. + state.unavailableAfterUpgrade.add(moduleUrl); + await retireProcessOnLease(state, moduleUrl); + }); + } + return await run(); + }); +} + type BrokerMaintenanceProcess = Pick; async function runBrokeredMemoryMaintenance(params: { @@ -679,6 +711,7 @@ export async function withBrokeredMemoryMaintenance(run: () => Promise): P } export const testing = { + clearBrokeredMemoryUpgradeFenceForTest: () => state.unavailableAfterUpgrade.clear(), createBrokerMaintenanceGate, withBrokerLease, withBrokerLifecycleOperation,