diff --git a/src/gateway/node-connection-notifications.ts b/src/gateway/node-connection-notifications.ts index 8b464adc282c..7bd676fe73c7 100644 --- a/src/gateway/node-connection-notifications.ts +++ b/src/gateway/node-connection-notifications.ts @@ -13,7 +13,7 @@ type RouterOptions = { type PendingConnectionAlert = { nodeId: string; - generation: number; + timer?: ReturnType; }; const DEFAULT_PRIMARY_DELAY_MS = 750; @@ -44,9 +44,7 @@ function connectionLabel(node: NodeSession): string { class NodeConnectionNotificationRouter { private readonly primaryDelayMs: number; private readonly fallbackDelayMs: number; - private readonly timersByNodeId = new Map>(); private readonly pendingByNodeId = new Map(); - private nextGeneration = 0; constructor( private readonly registry: NotificationRegistry, @@ -62,22 +60,21 @@ class NodeConnectionNotificationRouter { if (!isFirstConnection && !this.pendingByNodeId.has(source.nodeId)) { return; } - const pending = { nodeId: source.nodeId, generation: ++this.nextGeneration }; + const previous = this.pendingByNodeId.get(source.nodeId); + if (previous?.timer) { + clearTimeout(previous.timer); + } + const pending: PendingConnectionAlert = { nodeId: source.nodeId }; this.pendingByNodeId.set(source.nodeId, pending); - this.replaceTimer( - source.nodeId, - setTimeout(() => { - this.timersByNodeId.delete(source.nodeId); - void this.deliverPrimary(pending); - }, this.primaryDelayMs), - ); + this.armTimer(pending, this.primaryDelayMs, () => this.deliverPrimary(pending)); } dispose(): void { - for (const timer of this.timersByNodeId.values()) { - clearTimeout(timer); + for (const pending of this.pendingByNodeId.values()) { + if (pending.timer) { + clearTimeout(pending.timer); + } } - this.timersByNodeId.clear(); this.pendingByNodeId.clear(); } @@ -99,12 +96,8 @@ class NodeConnectionNotificationRouter { this.finishAlert(pending); return; } - this.replaceTimer( - pending.nodeId, - setTimeout(() => { - this.timersByNodeId.delete(pending.nodeId); - void this.deliverFallback(pending, primary?.connId); - }, this.fallbackDelayMs), + this.armTimer(pending, this.fallbackDelayMs, () => + this.deliverFallback(pending, primary?.connId), ); } @@ -132,11 +125,16 @@ class NodeConnectionNotificationRouter { } private attemptIsCurrent(pending: PendingConnectionAlert): boolean { - return this.pendingByNodeId.get(pending.nodeId)?.generation === pending.generation; + // Object identity lets a replacement invalidate both staged timers and + // in-flight deliveries without a second generation bookkeeping path. + return this.pendingByNodeId.get(pending.nodeId) === pending; } private finishAlert(pending: PendingConnectionAlert): void { if (this.attemptIsCurrent(pending)) { + if (pending.timer) { + clearTimeout(pending.timer); + } this.pendingByNodeId.delete(pending.nodeId); } } @@ -166,12 +164,18 @@ class NodeConnectionNotificationRouter { } } - private replaceTimer(nodeId: string, timer: ReturnType): void { - const existing = this.timersByNodeId.get(nodeId); - if (existing) { - clearTimeout(existing); + private armTimer( + pending: PendingConnectionAlert, + delayMs: number, + deliver: () => Promise, + ): void { + if (pending.timer) { + clearTimeout(pending.timer); } - this.timersByNodeId.set(nodeId, timer); + pending.timer = setTimeout(() => { + pending.timer = undefined; + void deliver(); + }, delayMs); } } diff --git a/src/gateway/server-node-events.test.ts b/src/gateway/server-node-events.test.ts index 48793d913aa4..f63ac3551853 100644 --- a/src/gateway/server-node-events.test.ts +++ b/src/gateway/server-node-events.test.ts @@ -84,7 +84,6 @@ const sanitizeInboundSystemTagsMock = vi.hoisted(() => ), ); const updatePairedDeviceMetadataMock = vi.hoisted(() => vi.fn().mockResolvedValue(true)); -const updatePairedNodeMetadataMock = vi.hoisted(() => vi.fn().mockResolvedValue(true)); const runtimeMocks = vi.hoisted(() => ({ agentCommandFromIngress: ingressAgentCommandMock, @@ -154,10 +153,6 @@ vi.mock("./server-node-events.runtime.js", () => runtimeMocks); vi.mock("../infra/device-pairing.js", () => ({ updatePairedDeviceMetadata: updatePairedDeviceMetadataMock, })); -vi.mock("../infra/node-pairing.js", () => ({ - updatePairedNodeMetadata: updatePairedNodeMetadataMock, -})); - import type { CliDeps } from "../cli/deps.js"; import type { HealthSummary } from "../commands/health.js"; import type { NodeEventContext } from "./server-node-events-types.js"; @@ -347,8 +342,6 @@ describe("node exec events", () => { sanitizeInboundSystemTagsMock.mockClear(); updatePairedDeviceMetadataMock.mockClear(); updatePairedDeviceMetadataMock.mockResolvedValue(true); - updatePairedNodeMetadataMock.mockClear(); - updatePairedNodeMetadataMock.mockResolvedValue(true); }); it("enqueues exec.started events", async () => { @@ -1679,8 +1672,6 @@ describe("agent request events", () => { beforeEach(() => { updatePairedDeviceMetadataMock.mockClear(); updatePairedDeviceMetadataMock.mockResolvedValue(true); - updatePairedNodeMetadataMock.mockClear(); - updatePairedNodeMetadataMock.mockResolvedValue(true); }); it("persists authenticated node presence alive events", async () => { @@ -1701,7 +1692,6 @@ describe("agent request events", () => { handled: true, reason: "persisted", }); - expect(updatePairedNodeMetadataMock).not.toHaveBeenCalled(); expectPresencePersistCall( updatePairedDeviceMetadataMock, "ios-presence-persist", @@ -1722,12 +1712,10 @@ describe("agent request events", () => { handled: false, reason: "missing_device_identity", }); - expect(updatePairedNodeMetadataMock).not.toHaveBeenCalled(); expect(updatePairedDeviceMetadataMock).not.toHaveBeenCalled(); }); it("does not throttle unknown node presence alive identities", async () => { - updatePairedNodeMetadataMock.mockResolvedValue(false); updatePairedDeviceMetadataMock.mockResolvedValue(false); const ctx = buildCtx(); const result = await handleNodeEvent( @@ -1784,7 +1772,6 @@ describe("agent request events", () => { handled: true, reason: "throttled", }); - expect(updatePairedNodeMetadataMock).not.toHaveBeenCalled(); expect(updatePairedDeviceMetadataMock).toHaveBeenCalledTimes(1); }); @@ -1866,7 +1853,6 @@ describe("agent request events", () => { { deviceId: "ios-presence-normalize" }, ); - expect(updatePairedNodeMetadataMock).not.toHaveBeenCalled(); expectPresencePersistCall( updatePairedDeviceMetadataMock, "ios-presence-normalize", diff --git a/src/gateway/server/ws-connection/connect-hello.ts b/src/gateway/server/ws-connection/connect-hello.ts index 9e2b07e7c705..38771bd78f14 100644 --- a/src/gateway/server/ws-connection/connect-hello.ts +++ b/src/gateway/server/ws-connection/connect-hello.ts @@ -9,8 +9,8 @@ import { restoreDeviceBootstrapToken, } from "../../../infra/device-bootstrap.js"; import { - claimPairedNodeConnection, finalizeNodePairingCleanupClaim, + recordPairedNodeConnection, } from "../../../infra/node-pairing.js"; import { resolveRuntimeServiceVersion } from "../../../version.js"; import { listControlUiPluginTabs } from "../../control-ui-plugin-tabs.js"; @@ -192,15 +192,15 @@ export async function sendGatewayHello( // persistence waits; the router transfers the pending alert by node identity. if (nodeSession?.connId === connId) { try { - const claim = await claimPairedNodeConnection( + const connection = await recordPairedNodeConnection( nodeSession.nodeId, nodeSession.connectedAtMs, ); - if (!claim.recorded) { + if (!connection.recorded) { logGateway.warn(`failed to record last connect for ${nodeSession.nodeId}: not paired`); } else { scheduleNodeConnectionNotification(requestContext.nodeRegistry, nodeSession, { - isFirstConnection: claim.firstConnection, + isFirstConnection: connection.firstConnection, }); } } catch (err) { diff --git a/src/gateway/watch-node-http.ts b/src/gateway/watch-node-http.ts index 95d39da19c48..ed274924f8a1 100644 --- a/src/gateway/watch-node-http.ts +++ b/src/gateway/watch-node-http.ts @@ -37,7 +37,7 @@ import { finalizeNodePairingCleanupClaim, releaseNodePairingCleanupClaim, requestNodePairing, - updatePairedNodeMetadata, + recordPairedNodeConnection, type RequestNodePairingResult, } from "../infra/node-pairing.js"; import { isNodePairingSetupBootstrapProfile } from "../shared/device-bootstrap-profile.js"; @@ -902,9 +902,9 @@ export function createWatchNodeHttpRuntime(options: WatchNodeHttpRuntimeOptions) options.onError?.("watch node pending-pairing cleanup failed", error); } } - void updatePairedNodeMetadata( + void recordPairedNodeConnection( session.nodeId, - { lastConnectedAtMs: nodeSession.connectedAtMs }, + nodeSession.connectedAtMs, options.pairingBaseDir, ).catch((error: unknown) => options.onError?.("watch node last-connect metadata update failed", error), diff --git a/src/infra/node-pairing-migration.test.ts b/src/infra/node-pairing-migration.test.ts index fb9c26e77e37..3c5208c30442 100644 --- a/src/infra/node-pairing-migration.test.ts +++ b/src/infra/node-pairing-migration.test.ts @@ -7,8 +7,8 @@ import { approveDevicePairing, getPairedDevice, requestDevicePairing } from "./d import { migrateLegacyNodePairingStore } from "./node-pairing-migration.js"; import { approveNodePairing, - claimPairedNodeConnection, listNodePairing, + recordPairedNodeConnection, requestNodePairing, } from "./node-pairing.js"; import { resolvePairingPaths } from "./pairing-files.js"; @@ -136,10 +136,9 @@ describe("migrateLegacyNodePairingStore", () => { orphaned: 1, }); await expect( - claimPairedNodeConnection("legacy-client-instance-id", 4_000, baseDir), + recordPairedNodeConnection("legacy-client-instance-id", 4_000, baseDir), ).resolves.toEqual({ recorded: false, - firstConnection: false, }); const pending = await requestNodePairing( @@ -153,17 +152,17 @@ describe("migrateLegacyNodePairingStore", () => { baseDir, ), ).resolves.toMatchObject({ node: { nodeId: "canonical-device-id" } }); - await expect(claimPairedNodeConnection("canonical-device-id", 4_000, baseDir)).resolves.toEqual( - { - recorded: true, - firstConnection: true, - }, - ); - await expect(claimPairedNodeConnection("canonical-device-id", 5_000, baseDir)).resolves.toEqual( - { - recorded: true, - firstConnection: false, - }, - ); + await expect( + recordPairedNodeConnection("canonical-device-id", 4_000, baseDir), + ).resolves.toEqual({ + recorded: true, + firstConnection: true, + }); + await expect( + recordPairedNodeConnection("canonical-device-id", 5_000, baseDir), + ).resolves.toEqual({ + recorded: true, + firstConnection: false, + }); }); }); diff --git a/src/infra/node-pairing.test.ts b/src/infra/node-pairing.test.ts index 4ad9345cc090..a13596fd6b78 100644 --- a/src/infra/node-pairing.test.ts +++ b/src/infra/node-pairing.test.ts @@ -5,14 +5,14 @@ import { approveDevicePairing, requestDevicePairing } from "./device-pairing.js" import { approveNodePairing, beginNodePairingConnect, - claimPairedNodeConnection, finalizeNodePairingCleanupClaim, listNodePairing, + recordPairedNodeConnection, releaseNodePairingCleanupClaim, renamePairedNode, requestNodePairing, reusePendingNodePairingForReconnect, - updatePairedNodeMetadata, + updatePairedNodeBins, } from "./node-pairing.js"; const tempDirs = createSuiteTempRootTracker({ prefix: "openclaw-node-pairing-" }); @@ -569,19 +569,19 @@ describe("node surface approvals", () => { }); }); - test("updates node runtime metadata and reports missing nodes", async () => { + test("updates remote skill bins and reports missing nodes", async () => { await withNodePairingDir(async (baseDir) => { await setupPairedNode(baseDir); - await expect( - updatePairedNodeMetadata("node-1", { lastConnectedAtMs: 1234, bins: ["ffmpeg"] }, baseDir), - ).resolves.toBe(true); - await expect( - updatePairedNodeMetadata("missing", { lastConnectedAtMs: 1 }, baseDir), - ).resolves.toBe(false); + await expect(recordPairedNodeConnection("node-1", 1_234, baseDir)).resolves.toEqual({ + recorded: true, + firstConnection: true, + }); + await expect(updatePairedNodeBins("node-1", ["ffmpeg"], baseDir)).resolves.toBe(true); + await expect(updatePairedNodeBins("missing", ["ffmpeg"], baseDir)).resolves.toBe(false); const pairedNode = await findPairedNode("node-1", baseDir); - expect(pairedNode?.lastConnectedAtMs).toBe(1234); + expect(pairedNode?.lastConnectedAtMs).toBe(1_234); expect(pairedNode?.bins).toEqual(["ffmpeg"]); }); }); @@ -591,25 +591,24 @@ describe("node surface approvals", () => { await setupPairedNode(baseDir); const claims = await Promise.all([ - claimPairedNodeConnection("node-1", 1_000, baseDir), - claimPairedNodeConnection("node-1", 2_000, baseDir), + recordPairedNodeConnection("node-1", 1_000, baseDir), + recordPairedNodeConnection("node-1", 2_000, baseDir), ]); - expect(claims.filter((claim) => claim.firstConnection)).toHaveLength(1); + expect(claims.filter((claim) => claim.recorded && claim.firstConnection)).toHaveLength(1); expect(claims.every((claim) => claim.recorded)).toBe(true); expect((await findPairedNode("node-1", baseDir))?.lastConnectedAtMs).toBe(2_000); - await expect(claimPairedNodeConnection("node-1", 1_500, baseDir)).resolves.toEqual({ + await expect(recordPairedNodeConnection("node-1", 1_500, baseDir)).resolves.toEqual({ recorded: true, firstConnection: false, }); expect((await findPairedNode("node-1", baseDir))?.lastConnectedAtMs).toBe(2_000); - await expect(claimPairedNodeConnection("node-1", 3_000, baseDir)).resolves.toEqual({ + await expect(recordPairedNodeConnection("node-1", 3_000, baseDir)).resolves.toEqual({ recorded: true, firstConnection: false, }); - await expect(claimPairedNodeConnection("missing", 4_000, baseDir)).resolves.toEqual({ + await expect(recordPairedNodeConnection("missing", 4_000, baseDir)).resolves.toEqual({ recorded: false, - firstConnection: false, }); expect((await findPairedNode("node-1", baseDir))?.lastConnectedAtMs).toBe(3_000); }); diff --git a/src/infra/node-pairing.ts b/src/infra/node-pairing.ts index 93ce4d92cf5a..88463d9d6ac5 100644 --- a/src/infra/node-pairing.ts +++ b/src/infra/node-pairing.ts @@ -578,10 +578,10 @@ export async function getPendingNodePairing( }); } -/** Update runtime node-surface metadata (connect stamps, remote skill bins). */ -export async function updatePairedNodeMetadata( +/** Update the remote skill bins advertised by a paired node. */ +export async function updatePairedNodeBins( nodeId: string, - patch: { lastConnectedAtMs?: number; bins?: string[] }, + bins: string[], baseDir?: string, ): Promise { return await withPairedDeviceRecords(baseDir, (pairedByDeviceId) => { @@ -591,27 +591,28 @@ export async function updatePairedNodeMetadata( } device.nodeSurface = { ...device.nodeSurface, - ...(patch.lastConnectedAtMs !== undefined - ? { lastConnectedAtMs: patch.lastConnectedAtMs } - : {}), - ...(patch.bins !== undefined ? { bins: patch.bins } : {}), + bins, }; return { value: true, persist: true }; }); } +type RecordPairedNodeConnectionResult = + | { recorded: false } + | { recorded: true; firstConnection: boolean }; + /** Atomically classify and persist one successful node connection. */ -export async function claimPairedNodeConnection( +export async function recordPairedNodeConnection( nodeId: string, connectedAtMs: number, baseDir?: string, -): Promise<{ recorded: boolean; firstConnection: boolean }> { - return await withPairedDeviceRecords<{ recorded: boolean; firstConnection: boolean }>( +): Promise { + return await withPairedDeviceRecords( baseDir, (pairedByDeviceId) => { const device = nodeSurfaceDevice(pairedByDeviceId, nodeId); if (!device?.nodeSurface) { - return { value: { recorded: false, firstConnection: false }, persist: false }; + return { value: { recorded: false }, persist: false }; } // Read and write under the pairing lock. Concurrent rehandshakes must not // both claim the same node's first connection and schedule duplicate alerts. diff --git a/src/skills/runtime/remote.ts b/src/skills/runtime/remote.ts index fe85e39af40e..597083b8bf62 100644 --- a/src/skills/runtime/remote.ts +++ b/src/skills/runtime/remote.ts @@ -7,7 +7,7 @@ import { normalizeStringEntries } from "@openclaw/normalization-core/string-norm import { listAgentWorkspaceDirs } from "../../agents/workspace-dirs.js"; import type { OpenClawConfig } from "../../config/types.openclaw.js"; import type { NodeRegistry } from "../../gateway/node-registry.js"; -import { listNodePairing, updatePairedNodeMetadata } from "../../infra/node-pairing.js"; +import { listNodePairing, updatePairedNodeBins } from "../../infra/node-pairing.js"; import { createSubsystemLogger } from "../../logging/subsystem.js"; import { loadWorkspaceSkillEntries } from "../loading/workspace.js"; import type { SkillEligibilityContext, SkillEntry } from "../types.js"; @@ -668,7 +668,7 @@ async function refreshRemoteNodeBinsUncoalesced(params: { if (!hasChanged) { return; } - await updatePairedNodeMetadata(params.nodeId, { bins }); + await updatePairedNodeBins(params.nodeId, bins); bumpSkillsSnapshotVersion({ reason: "remote-node" }); } catch (err) { const recorded = markRemoteNodeProbeFailure({