diff --git a/scripts/check-kysely-guardrails.mjs b/scripts/check-kysely-guardrails.mjs index bd636cc5c48e..d53ba8e1d0f4 100644 --- a/scripts/check-kysely-guardrails.mjs +++ b/scripts/check-kysely-guardrails.mjs @@ -108,6 +108,7 @@ const rawSqliteAllowPathGroups = { "session entry cache connection-local validity counters": [ "src/config/sessions/session-accessor.sqlite-entry-cache.ts", ], + "device pairing cache connection-local validity counters": ["src/infra/device-pairing-store.ts"], "Kysely-backed stores that own a DatabaseSync boundary": [ "src/acp/event-ledger.ts", "src/state/user-profiles.ts", diff --git a/src/gateway/server-methods/nodes.read.ts b/src/gateway/server-methods/nodes.read.ts index 67094483d5a7..43f9ae7b988f 100644 --- a/src/gateway/server-methods/nodes.read.ts +++ b/src/gateway/server-methods/nodes.read.ts @@ -10,7 +10,7 @@ import { import type { OpenClawConfig } from "../../config/types.openclaw.js"; import { listDevicePairing, resolveNodePairingState } from "../../infra/device-pairing.js"; import { formatErrorMessage } from "../../infra/errors.js"; -import { listNodePairing } from "../../infra/node-pairing.js"; +import { projectNodePairing } from "../../infra/node-pairing.js"; import type { NodeListNode } from "../../shared/node-list-types.js"; import { replaceRemoteNodeSkills } from "../../skills/runtime/remote-skills.js"; import { recordRemoteNodeInfo, refreshRemoteNodeBins } from "../../skills/runtime/remote.js"; @@ -58,8 +58,8 @@ function isVisibleNode(node: NodeListNode | null): node is NodeListNode { function listNodesForClient(params: { client: GatewayClient | null; pairedDevices: Awaited>["paired"]; - pairedNodes: Awaited>["paired"]; - pendingNodes: Awaited>["pending"]; + pairedNodes: ReturnType["paired"]; + pendingNodes: ReturnType["pending"]; connectedNodes: readonly NodeSession[]; }): NodeListNode[] { const catalog = createKnownNodeCatalog({ @@ -220,10 +220,8 @@ export const nodeReadHandlers: GatewayRequestHandlers = { return; } await respondUnavailableOnThrow(respond, async () => { - const [devicePairing, nodePairing] = await Promise.all([ - listDevicePairing(), - listNodePairing(), - ]); + const devicePairing = await listDevicePairing(); + const nodePairing = projectNodePairing(devicePairing.paired); const connectedNodes = listCurrentConnectedNodes(context, devicePairing.paired); const nodes = listNodesForClient({ client, @@ -255,10 +253,8 @@ export const nodeReadHandlers: GatewayRequestHandlers = { return; } await respondUnavailableOnThrow(respond, async () => { - const [devicePairing, nodePairing] = await Promise.all([ - listDevicePairing(), - listNodePairing(), - ]); + const devicePairing = await listDevicePairing(); + const nodePairing = projectNodePairing(devicePairing.paired); const connectedNodes = listCurrentConnectedNodes(context, devicePairing.paired); const catalog = createKnownNodeCatalog({ pairedDevices: devicePairing.paired, diff --git a/src/gateway/server.node-pairing-authz.test.ts b/src/gateway/server.node-pairing-authz.test.ts index b2d1125b28c9..467f9c12fa0c 100644 --- a/src/gateway/server.node-pairing-authz.test.ts +++ b/src/gateway/server.node-pairing-authz.test.ts @@ -2,15 +2,9 @@ // command scopes, and gateway enforcement around node client identity. import { afterAll, beforeAll, describe, expect, test, vi } from "vitest"; import { WebSocket } from "ws"; -import { - approveDevicePairing, - getPairedDevice, - listDevicePairing, - requestDevicePairing, -} from "../infra/device-pairing.js"; +import { getPairedDevice, listDevicePairing } from "../infra/device-pairing.js"; import { NODE_MCP_TOOLS_CALL_COMMAND } from "../infra/node-commands.js"; import { approveNodePairing, listNodePairing, requestNodePairing } from "../infra/node-pairing.js"; -import { createSuiteTempRootTracker } from "../test-helpers/temp-dir.js"; import { GATEWAY_CLIENT_MODES, GATEWAY_CLIENT_NAMES, @@ -23,6 +17,10 @@ import { openTrackedWs, pairDeviceIdentity, } from "./device-authz.test-helpers.js"; +import { + createNodePairingTestState, + describeWithGatewayServer, +} from "./server.node-pairing.test-support.js"; import { connectGatewayClient } from "./test-helpers.e2e.js"; import { connectOk, @@ -33,20 +31,12 @@ import { installGatewayTestHooks({ scope: "suite" }); -const tempDirs = createSuiteTempRootTracker({ prefix: "openclaw-node-pair-authz-" }); - -async function makeNodePairingStateDir(): Promise { - return await tempDirs.make("case"); -} - -// Node surfaces attach to paired devices, so tests seed device pairing first. -async function seedNodeDevice(nodeId: string, baseDir?: string): Promise { - const request = await requestDevicePairing( - { deviceId: nodeId, publicKey: `pk-${nodeId}`, role: "node", roles: ["node"], scopes: [] }, - baseDir, - ); - await approveDevicePairing(request.request.requestId, { callerScopes: [] }, baseDir); -} +const { + cleanup: cleanupNodePairingTestState, + makeStateDir: makeNodePairingStateDir, + seedNodeDevice, + setup: setupNodePairingTestState, +} = createNodePairingTestState("openclaw-node-pair-authz-"); async function findPairedNode(nodeId: string, baseDir?: string) { const pairing = await listNodePairing(baseDir); @@ -241,39 +231,13 @@ async function expectRpcNodePairingApprovalRejected(params: { } } -function describeWithGatewayServer( - name: string, - defineTests: (getStarted: () => Awaited>) => void, -): void { - describe(name, () => { - let started: Awaited> | undefined; - - beforeAll(async () => { - started = await startServerWithClient("secret"); - }); - - afterAll(async () => { - started?.ws.close(); - await started?.server.close(); - started?.envSnapshot.restore(); - }); - - defineTests(() => { - if (!started) { - throw new Error("gateway test server was not started"); - } - return started; - }); - }); -} - describe("gateway node pairing authorization", () => { beforeAll(async () => { - await tempDirs.setup(); + await setupNodePairingTestState(); }); afterAll(async () => { - await tempDirs.cleanup(); + await cleanupNodePairingTestState(); }); describe("approval scopes", () => { diff --git a/src/gateway/server.node-pairing-memo.test.ts b/src/gateway/server.node-pairing-memo.test.ts new file mode 100644 index 000000000000..342813adc089 --- /dev/null +++ b/src/gateway/server.node-pairing-memo.test.ts @@ -0,0 +1,141 @@ +import { DatabaseSync } from "node:sqlite"; +import { afterAll, beforeAll, describe, expect, test, vi } from "vitest"; +import { listDevicePairing } from "../infra/device-pairing.js"; +import { requestNodePairing } from "../infra/node-pairing.js"; +import { configureSqliteConnectionPragmas } from "../infra/sqlite-wal.js"; +import { + closeOpenClawStateDatabaseForTest, + openOpenClawStateDatabase, +} from "../state/openclaw-state-db.js"; +import { openTrackedWs } from "./device-authz.test-helpers.js"; +import { + createNodePairingTestState, + describeWithGatewayServer, +} from "./server.node-pairing.test-support.js"; +import { connectOk, installGatewayTestHooks, rpcReq } from "./test-helpers.js"; + +installGatewayTestHooks({ scope: "suite" }); + +const { + cleanup: cleanupNodePairingTestState, + makeStateDir: makeNodePairingStateDir, + seedNodeDevice, + setup: setupNodePairingTestState, +} = createNodePairingTestState("openclaw-node-pair-memo-"); + +describe("gateway node pairing memoization", () => { + beforeAll(async () => { + await setupNodePairingTestState(); + }); + + afterAll(async () => { + closeOpenClawStateDatabaseForTest(); + await cleanupNodePairingTestState(); + }); + + describeWithGatewayServer("node.list pairing snapshots", (getStarted) => { + test("scans pairing tables once across two node.list dispatches", async () => { + const ws = await openTrackedWs(getStarted().port); + try { + await connectOk(ws, { + token: "secret", + scopes: ["operator.read", "operator.pairing"], + deviceIdentityPath: `${await makeNodePairingStateDir()}/memo-scan-count.sqlite`, + }); + await seedNodeDevice("node-list-memo-scan-count"); + const database = openOpenClawStateDatabase(); + const originalPrepare = database.db.prepare.bind(database.db); + const tableSelects = { paired: 0, pending: 0 }; + const prepareSpy = vi.spyOn(database.db, "prepare").mockImplementation((sql) => { + if (sql.includes('from "device_pairing_pending"')) { + tableSelects.pending += 1; + } + if (sql.includes('from "device_pairing_paired"')) { + tableSelects.paired += 1; + } + return originalPrepare(sql); + }); + try { + expect((await rpcReq(ws, "node.list", {})).ok).toBe(true); + expect((await rpcReq(ws, "node.list", {})).ok).toBe(true); + expect(tableSelects).toEqual({ paired: 1, pending: 1 }); + } finally { + prepareSpy.mockRestore(); + } + } finally { + ws.close(); + } + }); + + test("reflects a pairing mutation on the next node.list dispatch", async () => { + const nodeId = "node-list-memo-mutation"; + await seedNodeDevice(nodeId); + const ws = await openTrackedWs(getStarted().port); + try { + await connectOk(ws, { + token: "secret", + scopes: ["operator.read", "operator.pairing"], + deviceIdentityPath: `${await makeNodePairingStateDir()}/memo-mutation.sqlite`, + }); + const before = await rpcReq<{ + nodes?: Array<{ nodeId: string; pendingRequestId?: string }>; + }>(ws, "node.list", {}); + expect(before.payload?.nodes?.find((node) => node.nodeId === nodeId)).not.toHaveProperty( + "pendingRequestId", + ); + + const pending = await requestNodePairing({ + nodeId, + platform: "macos", + commands: ["system.run"], + }); + const after = await rpcReq<{ + nodes?: Array<{ nodeId: string; pendingRequestId?: string }>; + }>(ws, "node.list", {}); + expect(after.payload?.nodes).toContainEqual( + expect.objectContaining({ + nodeId, + pendingRequestId: pending.request.requestId, + }), + ); + } finally { + ws.close(); + } + }); + }); + + test("reloads cached pairing tables after another connection commits", async () => { + const nodeId = "node-pairing-memo-external-writer"; + const baseDir = await makeNodePairingStateDir(); + await seedNodeDevice(nodeId, baseDir); + expect( + (await listDevicePairing(baseDir)).paired.find((device) => device.deviceId === nodeId) + ?.displayName, + ).toBeUndefined(); + + const database = openOpenClawStateDatabase({ + env: { ...process.env, OPENCLAW_STATE_DIR: baseDir }, + }); + const external = new DatabaseSync(database.path); + const maintenance = configureSqliteConnectionPragmas(external, { + checkpointIntervalMs: 0, + databaseLabel: "device-pairing-memo-external-writer", + databasePath: database.path, + foreignKeys: true, + synchronous: "NORMAL", + }); + try { + external + .prepare("UPDATE device_pairing_paired SET display_name = ? WHERE device_id = ?") + .run("external name", nodeId); + + expect( + (await listDevicePairing(baseDir)).paired.find((device) => device.deviceId === nodeId) + ?.displayName, + ).toBe("external name"); + } finally { + maintenance.close(); + external.close(); + } + }); +}); diff --git a/src/gateway/server.node-pairing.test-support.ts b/src/gateway/server.node-pairing.test-support.ts new file mode 100644 index 000000000000..fea15c99fa2b --- /dev/null +++ b/src/gateway/server.node-pairing.test-support.ts @@ -0,0 +1,47 @@ +import { afterAll, beforeAll, describe } from "vitest"; +import { approveDevicePairing, requestDevicePairing } from "../infra/device-pairing.js"; +import { createSuiteTempRootTracker } from "../test-helpers/temp-dir.js"; +import { startServerWithClient } from "./test-helpers.js"; + +export function createNodePairingTestState(prefix: string) { + const tempDirs = createSuiteTempRootTracker({ prefix }); + + return { + setup: async () => await tempDirs.setup(), + cleanup: async () => await tempDirs.cleanup(), + makeStateDir: async () => await tempDirs.make("case"), + seedNodeDevice: async (nodeId: string, baseDir?: string) => { + const request = await requestDevicePairing( + { deviceId: nodeId, publicKey: `pk-${nodeId}`, role: "node", roles: ["node"], scopes: [] }, + baseDir, + ); + await approveDevicePairing(request.request.requestId, { callerScopes: [] }, baseDir); + }, + }; +} + +export function describeWithGatewayServer( + name: string, + defineTests: (getStarted: () => Awaited>) => void, +): void { + describe(name, () => { + let started: Awaited> | undefined; + + beforeAll(async () => { + started = await startServerWithClient("secret"); + }); + + afterAll(async () => { + started?.ws.close(); + await started?.server.close(); + started?.envSnapshot.restore(); + }); + + defineTests(() => { + if (!started) { + throw new Error("gateway test server was not started"); + } + return started; + }); + }); +} diff --git a/src/infra/device-pairing-store.ts b/src/infra/device-pairing-store.ts index 588368ba2482..89fc6ea34ab8 100644 --- a/src/infra/device-pairing-store.ts +++ b/src/infra/device-pairing-store.ts @@ -5,6 +5,7 @@ // snapshot semantics the retired devices/*.json files had (including // cross-process last-writer-wins per store) while WAL + busy_timeout make // concurrent gateway/CLI access safe at the statement level. +import type { DatabaseSync } from "node:sqlite"; import type { DB as OpenClawStateKyselyDatabase, DevicePairingPaired, @@ -40,6 +41,23 @@ type DevicePairingStoreState = { type DevicePairingStoreTarget = "pending" | "paired" | "both"; +type DevicePairingStoreValidityToken = { + dataVersion: number; + totalChanges: number; +}; + +type DevicePairingStoreCache = { + connection: DatabaseSync; + path: string; + state: DevicePairingStoreState; + validityToken: DevicePairingStoreValidityToken; +}; + +type DevicePairingStoreMutation = { + mutated: boolean; + value: T; +}; + type PairedDeviceNodeSurfaceUpdate = | { value: T; persist: false } | { value: T; persist: true; nodeSurface: PairedDeviceNodeSurface }; @@ -53,11 +71,70 @@ type PairedDevicePresenceUpdate = lastSeenReason: string; }; +// One materialized pairing snapshot avoids rescanning both tables for every node catalog read. +// The connection token detects other-process writes, and store-owned writes clear it post-commit; +// without both paths, Gateway and CLI pairing mutations could leave node.list serving stale rows. +let devicePairingStoreCache: DevicePairingStoreCache | undefined; + /** Route an explicit pairing base dir (tests, alternate state roots) to that dir's DB. */ function resolveDevicePairingStateDbOptions(baseDir?: string): OpenClawStateDatabaseOptions { return baseDir ? { env: { ...process.env, OPENCLAW_STATE_DIR: baseDir } } : {}; } +function readDataVersion(database: DatabaseSync): number { + const row = database.prepare("PRAGMA data_version").get() as { data_version?: unknown }; + if (typeof row.data_version !== "number") { + throw new Error("SQLite did not return a numeric PRAGMA data_version"); + } + return row.data_version; +} + +function readTotalChanges(database: DatabaseSync): number { + const row = database.prepare("SELECT total_changes() AS value").get() as { value?: unknown }; + if (typeof row.value !== "number") { + throw new Error("SQLite did not return a numeric total_changes() value"); + } + return row.value; +} + +function readDevicePairingStoreValidityToken( + database: DatabaseSync, +): DevicePairingStoreValidityToken { + return { + dataVersion: readDataVersion(database), + totalChanges: readTotalChanges(database), + }; +} + +function devicePairingStoreValidityTokensEqual( + left: DevicePairingStoreValidityToken, + right: DevicePairingStoreValidityToken, +): boolean { + return left.dataVersion === right.dataVersion && left.totalChanges === right.totalChanges; +} + +function invalidateDevicePairingStoreCache(database: OpenClawStateDatabase): void { + if ( + devicePairingStoreCache?.connection === database.db && + devicePairingStoreCache.path === database.path + ) { + devicePairingStoreCache = undefined; + } +} + +function runDevicePairingStoreMutation( + baseDir: string | undefined, + mutate: (database: OpenClawStateDatabase) => DevicePairingStoreMutation, +): T { + const databaseOptions = resolveDevicePairingStateDbOptions(baseDir); + const database = openOpenClawStateDatabase(databaseOptions); + const result = runOpenClawStateWriteTransaction(mutate, { ...databaseOptions, database }); + if (result.mutated) { + invalidateDevicePairingStoreCache(database); + } + return result.value; +} + // Read-back allowlist for the approved_via column. The Record type forces // every PairedDeviceApprovalKind to appear here at compile time: omit one and // this object is a type error, instead of the stored provenance silently @@ -245,7 +322,16 @@ function fromBootstrapRow(row: DeviceBootstrapTokens): DeviceBootstrapTokenRecor /** Load the full pending + paired device snapshot from the shared state DB. */ export function loadDevicePairingStoreState(baseDir?: string): DevicePairingStoreState { - const { db } = openOpenClawStateDatabase(resolveDevicePairingStateDbOptions(baseDir)); + const database = openOpenClawStateDatabase(resolveDevicePairingStateDbOptions(baseDir)); + const { db } = database; + const validityToken = readDevicePairingStoreValidityToken(db); + if ( + devicePairingStoreCache?.connection === db && + devicePairingStoreCache.path === database.path && + devicePairingStoreValidityTokensEqual(devicePairingStoreCache.validityToken, validityToken) + ) { + return structuredClone(devicePairingStoreCache.state); + } const kysely = getNodeSqliteKysely(db); const pendingById: Record = {}; for (const row of executeSqliteQuerySync( @@ -261,7 +347,14 @@ export function loadDevicePairingStoreState(baseDir?: string): DevicePairingStor ).rows) { pairedByDeviceId[row.device_id] = fromPairedRow(row); } - return { pendingById, pairedByDeviceId }; + const state = { pendingById, pairedByDeviceId }; + devicePairingStoreCache = { + connection: db, + path: database.path, + state: structuredClone(state), + validityToken, + }; + return state; } /** Load one paired-device row without materializing either pairing table. */ @@ -299,14 +392,14 @@ export function updatePairedDeviceNodeSurfaceInTransaction( baseDir: string | undefined, update: (device: PairedDevice | null) => PairedDeviceNodeSurfaceUpdate, ): T { - return runOpenClawStateWriteTransaction(({ db }) => { + return runDevicePairingStoreMutation(baseDir, ({ db }) => { const normalizedDeviceId = deviceId.trim(); const device = normalizedDeviceId ? loadPairedDevicePairingStoreRecordFromDatabase(db, normalizedDeviceId) : null; const result = update(device); if (!result.persist) { - return result.value; + return { mutated: false, value: result.value }; } if (!device) { throw new Error("cannot update a missing paired-device node surface"); @@ -319,8 +412,8 @@ export function updatePairedDeviceNodeSurfaceInTransaction( .set({ node_surface_json: toJsonColumn(result.nodeSurface) }) .where("device_id", "=", normalizedDeviceId), ); - return result.value; - }, resolveDevicePairingStateDbOptions(baseDir)); + return { mutated: true, value: result.value }; + }); } /** Read, validate, and update one paired-device presence row in one transaction. */ @@ -329,14 +422,14 @@ export function updatePairedDevicePresenceInTransaction( baseDir: string | undefined, update: (device: PairedDevice | null) => PairedDevicePresenceUpdate, ): T { - return runOpenClawStateWriteTransaction(({ db }) => { + return runDevicePairingStoreMutation(baseDir, ({ db }) => { const normalizedDeviceId = deviceId.trim(); const device = normalizedDeviceId ? loadPairedDevicePairingStoreRecordFromDatabase(db, normalizedDeviceId) : null; const result = update(device); if (!result.persist) { - return result.value; + return { mutated: false, value: result.value }; } if (!device) { throw new Error("cannot update presence for a missing paired device"); @@ -352,8 +445,8 @@ export function updatePairedDevicePresenceInTransaction( }) .where("device_id", "=", normalizedDeviceId), ); - return result.value; - }, resolveDevicePairingStateDbOptions(baseDir)); + return { mutated: true, value: result.value }; + }); } /** Replace the pending and/or paired table contents with the given snapshot. */ @@ -363,7 +456,7 @@ export function persistDevicePairingStoreState( target: DevicePairingStoreTarget, options?: { clearApnsNodeIds?: readonly string[] }, ): void { - runOpenClawStateWriteTransaction(({ db }) => { + runDevicePairingStoreMutation(baseDir, ({ db }) => { const kysely = getNodeSqliteKysely(db); if (target !== "paired") { executeSqliteQuerySync(db, kysely.deleteFrom("device_pairing_pending")); @@ -382,7 +475,8 @@ export function persistDevicePairingStoreState( for (const nodeId of new Set(options?.clearApnsNodeIds ?? [])) { clearApnsRegistrationFromDatabase(db, nodeId); } - }, resolveDevicePairingStateDbOptions(baseDir)); + return { mutated: true, value: undefined }; + }); } /** Load all bootstrap token records keyed by token key. */ diff --git a/src/infra/node-pairing.ts b/src/infra/node-pairing.ts index 901b1919901a..0e6e9fe28acc 100644 --- a/src/infra/node-pairing.ts +++ b/src/infra/node-pairing.ts @@ -348,23 +348,34 @@ export async function listNodePairing( options?: { includePairingGeneration?: boolean }, ): Promise { return await withPairedDeviceRecords(baseDir, (pairedByDeviceId) => { - const pending: NodePairingPendingEntry[] = []; - const paired: NodePairingPairedNode[] = []; - for (const device of Object.values(pairedByDeviceId)) { - if (device.pendingNodeSurface) { - pending.push(toPendingEntry(device, device.pendingNodeSurface)); - } - const node = toPairedNode(device, options); - if (node) { - paired.push(node); - } - } - pending.sort((a, b) => b.ts - a.ts); - paired.sort((a, b) => b.approvedAtMs - a.approvedAtMs); - return { value: { pending, paired }, persist: false }; + return { + value: projectNodePairing(Object.values(pairedByDeviceId), options), + persist: false, + }; }); } +/** Project node pairing state from an already-loaded device pairing snapshot. */ +export function projectNodePairing( + pairedDevices: readonly PairedDevice[], + options?: { includePairingGeneration?: boolean }, +): NodePairingListWithGeneration { + const pending: NodePairingPendingEntry[] = []; + const paired: NodePairingPairedNode[] = []; + for (const device of pairedDevices) { + if (device.pendingNodeSurface) { + pending.push(toPendingEntry(device, device.pendingNodeSurface)); + } + const node = toPairedNode(device, options); + if (node) { + paired.push(node); + } + } + pending.sort((a, b) => b.ts - a.ts); + paired.sort((a, b) => b.approvedAtMs - a.approvedAtMs); + return { pending, paired }; +} + /** Snapshot pairing state and claim current pending revisions for one paired reconnect. */ export async function beginNodePairingConnect( nodeId: string,