From 36206b76f6820580e1eb16e51448ff7c43635e52 Mon Sep 17 00:00:00 2001 From: George Pickett Date: Sun, 1 Mar 2026 23:43:48 -0800 Subject: [PATCH] fix: disambiguate gateway event dedupe across reconnects --- src/lib/controlplane/contracts.ts | 1 + src/lib/controlplane/openclaw-adapter.ts | 65 ++++-- src/lib/controlplane/outbox.ts | 10 +- .../unit/controlPlaneProjectionStore.test.ts | 208 +++++++++++++++++ tests/unit/openclawAdapter.test.ts | 218 ++++++++++++++++++ 5 files changed, 478 insertions(+), 24 deletions(-) create mode 100644 tests/unit/openclawAdapter.test.ts diff --git a/src/lib/controlplane/contracts.ts b/src/lib/controlplane/contracts.ts index f32fc68..1136e46 100644 --- a/src/lib/controlplane/contracts.ts +++ b/src/lib/controlplane/contracts.ts @@ -16,6 +16,7 @@ export type ControlPlaneDomainEvent = type: "gateway.event"; event: string; seq: number | null; + connectionEpoch?: string | null; payload: unknown; asOf: string; }; diff --git a/src/lib/controlplane/openclaw-adapter.ts b/src/lib/controlplane/openclaw-adapter.ts index 6bbe33d..5c37d61 100644 --- a/src/lib/controlplane/openclaw-adapter.ts +++ b/src/lib/controlplane/openclaw-adapter.ts @@ -1,3 +1,5 @@ +import { randomUUID } from "node:crypto"; + import { WebSocket } from "ws"; import type { @@ -14,21 +16,22 @@ const REQUEST_TIMEOUT_MS = 15_000; const INITIAL_RECONNECT_DELAY_MS = 1_000; const MAX_RECONNECT_DELAY_MS = 15_000; const CONNECT_PROTOCOL = 3; +const CONNECT_CLIENT_ID = "gateway-client"; +const CONNECT_CLIENT_MODE = "backend"; const DEFAULT_METHOD_ALLOWLIST = new Set([ "status", "chat.send", "chat.abort", + "agents.create", "agents.update", "agents.delete", "agents.list", - "agents.create", "sessions.list", "sessions.preview", "sessions.patch", "sessions.reset", "config.get", - "config.patch", "config.set", "exec.approval.resolve", "exec.approvals.get", @@ -100,6 +103,7 @@ export class OpenClawGatewayAdapter { private reconnectAttempt = 0; private stopping = false; private nextRequestNumber = 1; + private connectionEpoch: string | null = null; private pending = new Map(); private loadSettings: () => ControlPlaneGatewaySettings; private createWebSocket: (url: string, opts: { origin: string }) => WebSocket; @@ -145,6 +149,7 @@ export class OpenClawGatewayAdapter { const ws = this.ws; this.ws = null; this.connectRequestId = null; + this.connectionEpoch = null; if (ws && ws.readyState === WebSocket.OPEN) { await new Promise((resolve) => { ws.once("close", () => resolve()); @@ -194,6 +199,7 @@ export class OpenClawGatewayAdapter { private async connect(): Promise { const settings = this.loadSettings(); + this.connectionEpoch = randomUUID(); const ws = this.createWebSocket(settings.url, { origin: resolveOriginForUpstream(settings.url) }); this.ws = ws; this.connectRequestId = null; @@ -230,6 +236,7 @@ export class OpenClawGatewayAdapter { type: "gateway.event", event: parsed.event, seq: typeof parsed.seq === "number" ? parsed.seq : null, + connectionEpoch: this.connectionEpoch, payload: parsed.payload, asOf: new Date().toISOString(), }); @@ -258,6 +265,8 @@ export class OpenClawGatewayAdapter { settle(() => reject(new Error("Control-plane gateway connection closed during connect."))); return; } + this.rejectPending("Control-plane gateway connection closed."); + this.connectionEpoch = null; this.updateStatus("reconnecting", "gateway_closed"); this.scheduleReconnect(); }); @@ -269,6 +278,7 @@ export class OpenClawGatewayAdapter { } }); }).catch((err) => { + this.connectionEpoch = null; this.updateStatus("error", err instanceof Error ? err.message : "connect_error"); this.scheduleReconnect(); throw err; @@ -294,27 +304,38 @@ export class OpenClawGatewayAdapter { if (!ws || ws.readyState !== WebSocket.OPEN || this.connectRequestId) return; const id = String(this.nextRequestNumber++); this.connectRequestId = id; - ws.send( - JSON.stringify({ - type: "req", - id, - method: "connect", - params: { - minProtocol: CONNECT_PROTOCOL, - maxProtocol: CONNECT_PROTOCOL, - client: { - id: "openclaw-studio-controlplane", - version: "dev", - platform: "node", - mode: "operator", + try { + ws.send( + JSON.stringify({ + type: "req", + id, + method: "connect", + params: { + minProtocol: CONNECT_PROTOCOL, + maxProtocol: CONNECT_PROTOCOL, + client: { + id: CONNECT_CLIENT_ID, + version: "dev", + platform: "node", + mode: CONNECT_CLIENT_MODE, + }, + role: "operator", + scopes: ["operator.admin", "operator.approvals", "operator.pairing"], + caps: [], + auth: { token }, }, - role: "operator", - scopes: ["operator.admin", "operator.approvals", "operator.pairing"], - caps: [], - auth: { token }, - }, - }) - ); + }) + ); + } catch (err) { + this.connectRequestId = null; + const reason = err instanceof Error ? err.message : "connect_send_failed"; + this.updateStatus("error", reason); + try { + ws.close(1011, "connect send failed"); + } catch (closeErr) { + console.error("Failed to close gateway socket after connect-send failure.", closeErr); + } + } } private parseFrame(raw: string): GatewayEventFrame | GatewayResponseFrame | null { diff --git a/src/lib/controlplane/outbox.ts b/src/lib/controlplane/outbox.ts index e2965be..5700470 100644 --- a/src/lib/controlplane/outbox.ts +++ b/src/lib/controlplane/outbox.ts @@ -6,8 +6,14 @@ export const deriveControlPlaneEventKey = (event: ControlPlaneDomainEvent): stri if (event.type === "runtime.status") { return ["runtime.status", event.status, safeString(event.reason), event.asOf].join(":"); } + const connectionEpoch = safeString(event.connectionEpoch).trim(); if (typeof event.seq === "number" && Number.isFinite(event.seq)) { - return ["gateway.event", event.event, "seq", String(event.seq)].join(":"); + if (connectionEpoch) { + return ["gateway.event", event.event, "epoch", connectionEpoch, "seq", String(event.seq)].join( + ":" + ); + } + return ["gateway.event", event.event, "seq", String(event.seq), safeString(event.asOf)].join(":"); } - return ["gateway.event", event.event, "", event.asOf].join(":"); + return ["gateway.event", event.event, connectionEpoch, event.asOf].join(":"); }; diff --git a/tests/unit/controlPlaneProjectionStore.test.ts b/tests/unit/controlPlaneProjectionStore.test.ts index 8f25a37..01197f9 100644 --- a/tests/unit/controlPlaneProjectionStore.test.ts +++ b/tests/unit/controlPlaneProjectionStore.test.ts @@ -4,6 +4,7 @@ import fs from "node:fs"; import os from "node:os"; import path from "node:path"; +import Database from "better-sqlite3"; import { afterEach, describe, expect, it } from "vitest"; import { SQLiteControlPlaneProjectionStore } from "@/lib/controlplane/projection-store"; @@ -48,6 +49,7 @@ describe("SQLiteControlPlaneProjectionStore", () => { type: "gateway.event" as const, event: "runtime.delta", seq: 42, + connectionEpoch: "conn-1", payload: { content: "a" }, asOf: "2026-02-28T02:01:00.000Z", }; @@ -55,6 +57,7 @@ describe("SQLiteControlPlaneProjectionStore", () => { type: "gateway.event" as const, event: "runtime.final", seq: 43, + connectionEpoch: "conn-1", payload: { content: "b" }, asOf: "2026-02-28T02:01:02.000Z", }; @@ -79,4 +82,209 @@ describe("SQLiteControlPlaneProjectionStore", () => { store.close(); }); + + it("stores same event+seq as distinct rows when connection epoch changes", () => { + const store = new SQLiteControlPlaneProjectionStore(makeDbPath()); + const first = store.applyDomainEvent({ + type: "gateway.event", + event: "agent", + seq: 1, + connectionEpoch: "conn-1", + payload: { state: "running" }, + asOf: "2026-02-28T02:01:00.000Z", + }); + const second = store.applyDomainEvent({ + type: "gateway.event", + event: "agent", + seq: 1, + connectionEpoch: "conn-2", + payload: { state: "idle" }, + asOf: "2026-02-28T02:01:10.000Z", + }); + + expect(first.id).toBe(1); + expect(second.id).toBe(2); + + const replay = store.readOutboxAfter(0, 10); + expect(replay.map((entry) => entry.id)).toEqual([1, 2]); + + store.close(); + }); + + it("reads outbox pages before a cursor in ascending order", () => { + const store = new SQLiteControlPlaneProjectionStore(makeDbPath()); + for (let index = 1; index <= 5; index += 1) { + store.applyDomainEvent({ + type: "gateway.event", + event: "runtime.delta", + seq: index, + payload: { index }, + asOf: `2026-02-28T02:01:0${index}.000Z`, + }); + } + + const newestTwo = store.readOutboxBefore(6, 2); + expect(newestTwo.map((entry) => entry.id)).toEqual([4, 5]); + + const olderTwo = store.readOutboxBefore(4, 2); + expect(olderTwo.map((entry) => entry.id)).toEqual([2, 3]); + + store.close(); + }); + + it("reads agent outbox pages before a cursor in ascending order and normalizes casing", () => { + const store = new SQLiteControlPlaneProjectionStore(makeDbPath()); + store.applyDomainEvent({ + type: "gateway.event", + event: "runtime.delta", + seq: 1, + payload: { sessionKey: "Agent:Alpha:Main", text: "a" }, + asOf: "2026-02-28T02:01:01.000Z", + }); + store.applyDomainEvent({ + type: "gateway.event", + event: "runtime.delta", + seq: 2, + payload: { sessionKey: "agent:beta:main", text: "b" }, + asOf: "2026-02-28T02:01:02.000Z", + }); + store.applyDomainEvent({ + type: "gateway.event", + event: "runtime.delta", + seq: 3, + payload: { agentId: "ALPHA", text: "c" }, + asOf: "2026-02-28T02:01:03.000Z", + }); + store.applyDomainEvent({ + type: "gateway.event", + event: "runtime.delta", + seq: 4, + payload: { sessionKey: "agent:alpha:main", text: "d" }, + asOf: "2026-02-28T02:01:04.000Z", + }); + + const newestTwo = store.readAgentOutboxBefore("ALPHA", 5, 2); + expect(newestTwo.map((entry) => entry.id)).toEqual([3, 4]); + + const olderOne = store.readAgentOutboxBefore("alpha", 3, 10); + expect(olderOne.map((entry) => entry.id)).toEqual([1]); + + store.close(); + }); + + it("backfills legacy outbox rows into agent index and marks non-agent rows", () => { + const dbPath = makeDbPath(); + const db = new Database(dbPath); + db.exec(` + CREATE TABLE runtime_projection ( + id INTEGER PRIMARY KEY CHECK (id = 1), + status TEXT NOT NULL, + reason TEXT, + as_of TEXT, + updated_at TEXT NOT NULL + ); + CREATE TABLE outbox ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + event_type TEXT NOT NULL, + event_json TEXT NOT NULL, + created_at TEXT NOT NULL + ); + CREATE TABLE processed_events ( + event_key TEXT PRIMARY KEY, + outbox_id INTEGER, + created_at TEXT NOT NULL, + FOREIGN KEY (outbox_id) REFERENCES outbox(id) ON DELETE SET NULL + ); + CREATE INDEX idx_outbox_id ON outbox(id); + `); + db.prepare("INSERT INTO outbox (event_type, event_json, created_at) VALUES (?, ?, ?)").run( + "gateway.event", + JSON.stringify({ + type: "gateway.event", + event: "runtime.delta", + seq: 10, + payload: { sessionKey: "Agent:Alpha:Main" }, + asOf: "2026-02-28T02:01:01.000Z", + }), + "2026-02-28T02:01:01.000Z" + ); + db.prepare("INSERT INTO outbox (event_type, event_json, created_at) VALUES (?, ?, ?)").run( + "runtime.status", + JSON.stringify({ + type: "runtime.status", + status: "connected", + reason: null, + asOf: "2026-02-28T02:01:02.000Z", + }), + "2026-02-28T02:01:02.000Z" + ); + db.prepare("INSERT INTO outbox (event_type, event_json, created_at) VALUES (?, ?, ?)").run( + "gateway.event", + JSON.stringify({ + type: "gateway.event", + event: "runtime.delta", + seq: 11, + payload: { sessionKey: "agent:beta:main" }, + asOf: "2026-02-28T02:01:03.000Z", + }), + "2026-02-28T02:01:03.000Z" + ); + db.pragma("user_version = 1"); + db.close(); + + const store = new SQLiteControlPlaneProjectionStore(dbPath); + const firstBackfill = store.backfillAgentOutboxBefore(4, 10); + expect(firstBackfill.scannedRows).toBe(3); + expect(firstBackfill.updatedRows).toBe(3); + + const alphaRows = store.readAgentOutboxBefore("alpha", 4, 10); + expect(alphaRows.map((entry) => entry.id)).toEqual([1]); + + const betaRows = store.readAgentOutboxBefore("beta", 4, 10); + expect(betaRows.map((entry) => entry.id)).toEqual([3]); + + const secondBackfill = store.backfillAgentOutboxBefore(4, 10); + expect(secondBackfill.scannedRows).toBe(0); + expect(secondBackfill.exhausted).toBe(true); + + store.close(); + + const verifyDb = new Database(dbPath, { readonly: true }); + const runtimeStatusRow = verifyDb + .prepare("SELECT agent_id FROM outbox WHERE id = 2") + .get() as { agent_id: string | null }; + expect(runtimeStatusRow.agent_id).toBe(""); + verifyDb.close(); + }); + + it("repairs missing agent index on dedupe replay", () => { + const dbPath = makeDbPath(); + const event = { + type: "gateway.event" as const, + event: "runtime.delta", + seq: 99, + payload: { sessionKey: "agent:alpha:main", text: "dedupe" }, + asOf: "2026-02-28T02:01:00.000Z", + }; + + const firstStore = new SQLiteControlPlaneProjectionStore(dbPath); + firstStore.applyDomainEvent(event); + firstStore.close(); + + const mutateDb = new Database(dbPath); + mutateDb.prepare("UPDATE outbox SET agent_id = NULL WHERE id = 1").run(); + mutateDb.close(); + + const secondStore = new SQLiteControlPlaneProjectionStore(dbPath); + const deduped = secondStore.applyDomainEvent(event); + expect(deduped.id).toBe(1); + secondStore.close(); + + const verifyDb = new Database(dbPath, { readonly: true }); + const repaired = verifyDb.prepare("SELECT agent_id FROM outbox WHERE id = 1").get() as { + agent_id: string | null; + }; + expect(repaired.agent_id).toBe("alpha"); + verifyDb.close(); + }); }); diff --git a/tests/unit/openclawAdapter.test.ts b/tests/unit/openclawAdapter.test.ts new file mode 100644 index 0000000..3973db0 --- /dev/null +++ b/tests/unit/openclawAdapter.test.ts @@ -0,0 +1,218 @@ +// @vitest-environment node + +import { afterEach, describe, expect, it } from "vitest"; +import { EventEmitter } from "node:events"; +import { WebSocket, WebSocketServer } from "ws"; + +import { OpenClawGatewayAdapter } from "@/lib/controlplane/openclaw-adapter"; +import type { ControlPlaneDomainEvent } from "@/lib/controlplane/contracts"; + +const closeWebSocketServer = (server: WebSocketServer) => + new Promise((resolve) => server.close(() => resolve())); + +const waitForCondition = async (predicate: () => boolean, timeoutMs: number = 3_000) => { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (predicate()) return; + await new Promise((resolve) => setTimeout(resolve, 10)); + } + throw new Error("Condition not met before timeout."); +}; + +describe("OpenClawGatewayAdapter", () => { + let upstream: WebSocketServer | null = null; + + afterEach(async () => { + if (upstream) { + await closeWebSocketServer(upstream); + upstream = null; + } + }); + + it("rejects in-flight requests immediately when the socket closes", async () => { + upstream = new WebSocketServer({ port: 0 }); + const address = upstream.address(); + if (!address || typeof address === "string") { + throw new Error("expected upstream server to provide a numeric port"); + } + const upstreamUrl = `ws://127.0.0.1:${address.port}`; + let observedConnectClientId: string | null = null; + let observedConnectClientMode: string | null = null; + + upstream.on("connection", (ws) => { + ws.send(JSON.stringify({ type: "event", event: "connect.challenge", payload: {} })); + ws.on("message", (raw) => { + const parsed = JSON.parse(String(raw ?? "")) as { + id?: string; + method?: string; + params?: { + client?: { id?: string; mode?: string }; + }; + }; + if (parsed?.method === "connect") { + observedConnectClientId = parsed.params?.client?.id ?? null; + observedConnectClientMode = parsed.params?.client?.mode ?? null; + ws.send( + JSON.stringify({ + type: "res", + id: parsed.id, + ok: true, + payload: { type: "hello-ok", protocol: 3 }, + }) + ); + return; + } + if (parsed?.method === "status") { + ws.close(1011, "upstream closed"); + } + }); + }); + + const adapter = new OpenClawGatewayAdapter({ + loadSettings: () => ({ url: upstreamUrl, token: "tkn" }), + }); + + await adapter.start(); + const startedAt = Date.now(); + await expect(adapter.request("status", {})).rejects.toThrow( + "Control-plane gateway connection closed." + ); + expect(Date.now() - startedAt).toBeLessThan(2_000); + expect(observedConnectClientId).toBe("gateway-client"); + expect(observedConnectClientMode).toBe("backend"); + + await adapter.stop(); + }); + + it("fails connect gracefully when sending the connect request throws", async () => { + class ThrowingConnectSocket extends EventEmitter { + readyState: number = WebSocket.OPEN; + + close() { + if (this.readyState === WebSocket.CLOSED) return; + this.readyState = WebSocket.CLOSED; + this.emit("close"); + } + + terminate() { + this.close(); + } + + send(raw: string) { + const parsed = JSON.parse(raw) as { method?: string }; + if (parsed.method === "connect") { + throw new Error("connect send failed"); + } + } + } + + const socket = new ThrowingConnectSocket(); + const adapter = new OpenClawGatewayAdapter({ + loadSettings: () => ({ url: "ws://127.0.0.1:9", token: "tkn" }), + createWebSocket: () => socket as unknown as WebSocket, + }); + + setTimeout(() => { + socket.emit("message", JSON.stringify({ type: "event", event: "connect.challenge", payload: {} })); + }, 0); + + await expect(adapter.start()).rejects.toThrow( + "Control-plane gateway connection closed during connect." + ); + + await adapter.stop(); + }); + + it("emits gateway events with unique connection epochs across reconnect cycles", async () => { + upstream = new WebSocketServer({ port: 0 }); + const address = upstream.address(); + if (!address || typeof address === "string") { + throw new Error("expected upstream server to provide a numeric port"); + } + const upstreamUrl = `ws://127.0.0.1:${address.port}`; + let acceptedConnections = 0; + + upstream.on("connection", (ws) => { + acceptedConnections += 1; + const connectionIndex = acceptedConnections; + ws.send(JSON.stringify({ type: "event", event: "connect.challenge", payload: {} })); + ws.on("message", (raw) => { + const parsed = JSON.parse(String(raw ?? "")) as { id?: string; method?: string }; + if (parsed?.method !== "connect" || !parsed.id) return; + ws.send( + JSON.stringify({ + type: "res", + id: parsed.id, + ok: true, + payload: { type: "hello-ok", protocol: 3 }, + }) + ); + ws.send( + JSON.stringify({ + type: "event", + event: "agent", + seq: 1, + payload: { connectionIndex }, + }) + ); + }); + }); + + const observedEvents: ControlPlaneDomainEvent[] = []; + const adapter = new OpenClawGatewayAdapter({ + loadSettings: () => ({ url: upstreamUrl, token: "tkn" }), + onDomainEvent: (event) => { + observedEvents.push(event); + }, + }); + + await adapter.start(); + await waitForCondition(() => + observedEvents.some( + (event) => + event.type === "gateway.event" && + event.event === "agent" && + typeof (event.payload as { connectionIndex?: unknown })?.connectionIndex === "number" && + (event.payload as { connectionIndex?: number }).connectionIndex === 1 + ) + ); + + await adapter.stop(); + + await adapter.start(); + await waitForCondition(() => + observedEvents.some( + (event) => + event.type === "gateway.event" && + event.event === "agent" && + typeof (event.payload as { connectionIndex?: unknown })?.connectionIndex === "number" && + (event.payload as { connectionIndex?: number }).connectionIndex === 2 + ) + ); + + const firstGatewayEvent = observedEvents.find( + (event) => + event.type === "gateway.event" && + (event.payload as { connectionIndex?: number })?.connectionIndex === 1 + ); + const secondGatewayEvent = observedEvents.find( + (event) => + event.type === "gateway.event" && + (event.payload as { connectionIndex?: number })?.connectionIndex === 2 + ); + + expect(firstGatewayEvent?.type).toBe("gateway.event"); + expect(secondGatewayEvent?.type).toBe("gateway.event"); + if (!firstGatewayEvent || firstGatewayEvent.type !== "gateway.event") { + throw new Error("Expected first gateway event."); + } + if (!secondGatewayEvent || secondGatewayEvent.type !== "gateway.event") { + throw new Error("Expected second gateway event."); + } + expect(firstGatewayEvent.connectionEpoch).toBeTruthy(); + expect(secondGatewayEvent.connectionEpoch).toBeTruthy(); + expect(firstGatewayEvent.connectionEpoch).not.toBe(secondGatewayEvent.connectionEpoch); + + await adapter.stop(); + }); +});