refactor(gateway): simplify node connection state (#110452)

2026-07-28 08:01:17 +00:00 · 2026-07-18 07:13:11 +01:00
parent 021d2526f0
commit 5bfd03b845
8 changed files with 81 additions and 92 deletions
--- 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<typeof setTimeout>;
 };

 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<string, ReturnType<typeof setTimeout>>();
  private readonly pendingByNodeId = new Map<string, PendingConnectionAlert>();
-  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<typeof setTimeout>): void {
-    const existing = this.timersByNodeId.get(nodeId);
-    if (existing) {
-      clearTimeout(existing);
+  private armTimer(
+    pending: PendingConnectionAlert,
+    delayMs: number,
+    deliver: () => Promise<void>,
+  ): void {
+    if (pending.timer) {
+      clearTimeout(pending.timer);
    }
-    this.timersByNodeId.set(nodeId, timer);
+    pending.timer = setTimeout(() => {
+      pending.timer = undefined;
+      void deliver();
+    }, delayMs);
  }
 }

--- 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",
--- 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) {
--- 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),
--- 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,
+    });
  });
 });
--- 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);
    });
--- 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<boolean> {
  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<RecordPairedNodeConnectionResult> {
+  return await withPairedDeviceRecords<RecordPairedNodeConnectionResult>(
    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.
--- 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({