From fbc73ca1b7cddd40849917f5033322100d614b5d Mon Sep 17 00:00:00 2001 From: Masato Hoshino Date: Fri, 10 Jul 2026 11:13:27 +0900 Subject: [PATCH] fix(outbound): publish conversation bindings after SQLite commit (#101023) * fix(outbound): restore current-conversation binding map when persist fails The generic current-conversation binding mutators (bind/touch/unbind/ prune-on-read) changed the process-wide in-memory map before calling writePersistedBindings(), with no rollback. A throwing SQLite write left the map ahead of disk; because bindingsLoaded is a one-time flag, the diverged bindings were served until restart, routing inbound messages to wrong or deleted session targets while the caller already saw the throw. Add persistBindingsOrRestore(): snapshot the map before mutation and, on write failure, restore it and rethrow. Covers all six write sites. Mirrors the cron rollback precedent (#99960, src/cron/service/ops.ts persistOrRestore). Fault-injection tests assert the map reverts for each path. * fix(outbound): publish conversation bindings after SQLite commit * chore(outbound): align binding files with current main * fix(outbound): publish conversation bindings after SQLite commit --------- Co-authored-by: Peter Steinberger Co-authored-by: Peter Steinberger --- .../current-conversation-bindings.test.ts | 192 ++++++++++++++++++ .../outbound/current-conversation-bindings.ts | 119 ++++++----- 2 files changed, 263 insertions(+), 48 deletions(-) diff --git a/src/infra/outbound/current-conversation-bindings.test.ts b/src/infra/outbound/current-conversation-bindings.test.ts index 057dcfb6897..341cd732dd8 100644 --- a/src/infra/outbound/current-conversation-bindings.test.ts +++ b/src/infra/outbound/current-conversation-bindings.test.ts @@ -8,6 +8,7 @@ import { setActivePluginRegistry } from "../../plugins/runtime.js"; import type { DB as OpenClawStateKyselyDatabase } from "../../state/openclaw-state-db.generated.js"; import { closeOpenClawStateDatabaseForTest, + openOpenClawStateDatabase, runOpenClawStateWriteTransaction, } from "../../state/openclaw-state-db.js"; import { createTestRegistry } from "../../test-utils/channel-plugins.js"; @@ -130,6 +131,45 @@ function setMinimalCurrentConversationRegistry(): void { ); } +async function withReadOnlyStateDatabase(run: () => T | Promise): Promise { + const { db } = openOpenClawStateDatabase(); + db.exec("PRAGMA query_only = ON"); + try { + return await run(); + } finally { + db.exec("PRAGMA query_only = OFF"); + } +} + +function workspaceConversation(conversationId: string) { + return { + channel: "workspace", + accountId: "default", + conversationId, + }; +} + +async function bindWorkspaceConversation( + conversationId: string, + options: { + targetSessionKey?: string; + ttlMs?: number; + metadata?: Record; + } = {}, +): Promise { + return bindGenericCurrentConversation({ + targetSessionKey: options.targetSessionKey ?? "agent:codex:acp:workspace-dm", + targetKind: "session", + conversation: workspaceConversation(conversationId), + ...(options.ttlMs === undefined ? {} : { ttlMs: options.ttlMs }), + ...(options.metadata === undefined ? {} : { metadata: options.metadata }), + }); +} + +function resolveWorkspaceConversation(conversationId: string): SessionBindingRecord | null { + return resolveGenericCurrentConversationBinding(workspaceConversation(conversationId)); +} + describe("generic current-conversation bindings", () => { let previousStateDir: string | undefined; let testStateDir = ""; @@ -458,4 +498,156 @@ describe("generic current-conversation bindings", () => { }, ); }); + + describe("SQLite write failures", () => { + it("keeps a replacement bind out of memory and disk", async () => { + await bindWorkspaceConversation("user:U1", { + targetSessionKey: "agent:codex:acp:session-a", + }); + + await expect( + withReadOnlyStateDatabase(() => + bindWorkspaceConversation("user:U1", { + targetSessionKey: "agent:codex:acp:session-b", + }), + ), + ).rejects.toThrow(); + + expect(resolveWorkspaceConversation("user:U1")?.targetSessionKey).toBe( + "agent:codex:acp:session-a", + ); + testing.resetCurrentConversationBindingsForTests(); + closeOpenClawStateDatabaseForTest(); + expect(resolveWorkspaceConversation("user:U1")?.targetSessionKey).toBe( + "agent:codex:acp:session-a", + ); + }); + + it("keeps a failed touch out of memory and disk", async () => { + const bound = expectSessionBinding( + await bindWorkspaceConversation("user:U1", { metadata: { label: "workspace-dm" } }), + ); + const originalActivity = bound.metadata?.lastActivityAt; + + await expect( + withReadOnlyStateDatabase(() => + touchGenericCurrentConversationBinding(bound.bindingId, 9_999_999), + ), + ).rejects.toThrow(); + + expect(resolveWorkspaceConversation("user:U1")?.metadata?.lastActivityAt).toBe( + originalActivity, + ); + testing.resetCurrentConversationBindingsForTests(); + expect(resolveWorkspaceConversation("user:U1")?.metadata?.lastActivityAt).toBe( + originalActivity, + ); + }); + + it("keeps a binding when unbind by id fails", async () => { + const bound = expectSessionBinding(await bindWorkspaceConversation("user:U1")); + + await expect( + withReadOnlyStateDatabase(() => + unbindGenericCurrentConversationBindings({ + bindingId: bound.bindingId, + reason: "test cleanup", + }), + ), + ).rejects.toThrow(); + + expect(resolveWorkspaceConversation("user:U1")).not.toBeNull(); + testing.resetCurrentConversationBindingsForTests(); + expect(resolveWorkspaceConversation("user:U1")).not.toBeNull(); + }); + + it("keeps every matching binding when unbind by session fails", async () => { + const targetSessionKey = "agent:codex:acp:shared"; + await bindWorkspaceConversation("user:U1", { targetSessionKey }); + await bindWorkspaceConversation("user:U2", { targetSessionKey }); + + await expect( + withReadOnlyStateDatabase(() => + unbindGenericCurrentConversationBindings({ + targetSessionKey, + reason: "test cleanup", + }), + ), + ).rejects.toThrow(); + + expect(listGenericCurrentConversationBindingsBySession(targetSessionKey)).toHaveLength(2); + testing.resetCurrentConversationBindingsForTests(); + expect(listGenericCurrentConversationBindingsBySession(targetSessionKey)).toHaveLength(2); + }); + + it("keeps an expired binding when prune-on-resolve fails", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date(1_000_000)); + await bindWorkspaceConversation("user:U1", { ttlMs: 1_000 }); + + vi.setSystemTime(new Date(1_002_000)); + await expect( + withReadOnlyStateDatabase(() => resolveWorkspaceConversation("user:U1")), + ).rejects.toThrow(); + + vi.setSystemTime(new Date(1_000_500)); + expect(resolveWorkspaceConversation("user:U1")).not.toBeNull(); + testing.resetCurrentConversationBindingsForTests(); + expect(resolveWorkspaceConversation("user:U1")).not.toBeNull(); + }); + + it("keeps expired list entries when their cleanup write fails", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date(1_000_000)); + const targetSessionKey = "agent:codex:acp:shared"; + await bindWorkspaceConversation("user:U1", { targetSessionKey }); + await bindWorkspaceConversation("user:U2", { targetSessionKey, ttlMs: 1_000 }); + + vi.setSystemTime(new Date(1_002_000)); + await expect( + withReadOnlyStateDatabase(() => + listGenericCurrentConversationBindingsBySession(targetSessionKey), + ), + ).rejects.toThrow(); + + vi.setSystemTime(new Date(1_000_500)); + expect(listGenericCurrentConversationBindingsBySession(targetSessionKey)).toHaveLength(2); + testing.resetCurrentConversationBindingsForTests(); + expect(listGenericCurrentConversationBindingsBySession(targetSessionKey)).toHaveLength(2); + }); + + it("does not partially prune an unbind-by-session batch", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date(1_000_000)); + const targetSessionKey = "agent:codex:acp:shared"; + await bindWorkspaceConversation("user:U1", { targetSessionKey }); + await bindWorkspaceConversation("user:U2", { targetSessionKey, ttlMs: 1_000 }); + + vi.setSystemTime(new Date(1_002_000)); + await expect( + withReadOnlyStateDatabase(() => + unbindGenericCurrentConversationBindings({ + targetSessionKey, + reason: "test cleanup", + }), + ), + ).rejects.toThrow(); + + vi.setSystemTime(new Date(1_000_500)); + expect(listGenericCurrentConversationBindingsBySession(targetSessionKey)).toHaveLength(2); + testing.resetCurrentConversationBindingsForTests(); + expect(listGenericCurrentConversationBindingsBySession(targetSessionKey)).toHaveLength(2); + }); + + it("retries the initial cache load after its SQLite cleanup fails", async () => { + await bindWorkspaceConversation("user:U1"); + testing.resetCurrentConversationBindingsForTests(); + + await expect( + withReadOnlyStateDatabase(() => resolveWorkspaceConversation("user:U1")), + ).rejects.toThrow(); + + expect(resolveWorkspaceConversation("user:U1")).not.toBeNull(); + }); + }); }); diff --git a/src/infra/outbound/current-conversation-bindings.ts b/src/infra/outbound/current-conversation-bindings.ts index c8289ab8440..6639def7c39 100644 --- a/src/infra/outbound/current-conversation-bindings.ts +++ b/src/infra/outbound/current-conversation-bindings.ts @@ -34,7 +34,7 @@ type CurrentConversationBindingDatabase = Pick< >; let bindingsLoaded = false; -const bindingsByConversationKey = new Map(); +let bindingsByConversationKey = new Map(); function buildConversationKey(ref: ConversationRef): string { const normalized = normalizeConversationRef(ref); @@ -122,8 +122,8 @@ function targetAgentIdForSessionKey(targetSessionKey: string): string { return resolveAgentIdFromSessionKey(targetSessionKey); } -function writePersistedBindings(): void { - const records = [...bindingsByConversationKey.values()] +function writePersistedBindings(nextBindings: ReadonlyMap): void { + const records = [...nextBindings.values()] .filter((record) => !isBindingExpired(record)) .toSorted((a, b) => a.bindingId.localeCompare(b.bindingId)); const updatedAt = Date.now(); @@ -163,29 +163,23 @@ function writePersistedBindings(): void { }); } +function commitBindings(nextBindings: Map): void { + // SQLite is canonical: publish the prepared map only after its transaction + // commits, so a storage error cannot leave runtime routing ahead of disk. + writePersistedBindings(nextBindings); + bindingsByConversationKey = nextBindings; +} + function loadBindingsIntoMemory(): void { if (bindingsLoaded) { return; } - bindingsLoaded = true; - bindingsByConversationKey.clear(); + const nextBindings = new Map(); for (const record of readPersistedBindings()) { - bindingsByConversationKey.set(buildConversationKey(record.conversation), record); + nextBindings.set(buildConversationKey(record.conversation), record); } -} - -function pruneExpiredBinding(key: string): SessionBindingRecord | null { - loadBindingsIntoMemory(); - const record = bindingsByConversationKey.get(key) ?? null; - if (!record) { - return null; - } - if (!isBindingExpired(record)) { - return record; - } - bindingsByConversationKey.delete(key); - writePersistedBindings(); - return null; + bindingsByConversationKey = nextBindings; + bindingsLoaded = true; } function resolveChannelSupportsCurrentConversationBinding(params: { @@ -295,7 +289,8 @@ export async function bindGenericCurrentConversation( return null; } const key = buildConversationKey(conversation); - const existing = pruneExpiredBinding(key); + const existing = bindingsByConversationKey.get(key); + const activeExisting = existing && !isBindingExpired(existing) ? existing : undefined; const record: SessionBindingRecord = { bindingId: buildBindingId(conversation), targetSessionKey, @@ -305,13 +300,14 @@ export async function bindGenericCurrentConversation( boundAt: now, ...(expiresAt !== undefined ? { expiresAt } : {}), metadata: { - ...existing?.metadata, + ...activeExisting?.metadata, ...input.metadata, lastActivityAt: now, }, }; - bindingsByConversationKey.set(key, record); - writePersistedBindings(); + const nextBindings = new Map(bindingsByConversationKey); + nextBindings.set(key, record); + commitBindings(nextBindings); return record; } @@ -322,7 +318,16 @@ export function resolveGenericCurrentConversationBinding( if (!supportsGenericCurrentConversationBinding(ref)) { return null; } - return pruneExpiredBinding(buildConversationKey(ref)); + loadBindingsIntoMemory(); + const key = buildConversationKey(ref); + const record = bindingsByConversationKey.get(key) ?? null; + if (!record || !isBindingExpired(record)) { + return record; + } + const nextBindings = new Map(bindingsByConversationKey); + nextBindings.delete(key); + commitBindings(nextBindings); + return null; } /** Lists non-expired current-conversation bindings owned by one target session. */ @@ -331,10 +336,14 @@ export function listGenericCurrentConversationBindingsBySession( ): SessionBindingRecord[] { loadBindingsIntoMemory(); const results: SessionBindingRecord[] = []; - for (const key of bindingsByConversationKey.keys()) { - const record = pruneExpiredBinding(key); + let nextBindings: Map | undefined; + for (const [key, record] of bindingsByConversationKey) { + if (isBindingExpired(record)) { + nextBindings ??= new Map(bindingsByConversationKey); + nextBindings.delete(key); + continue; + } if ( - !record || record.targetSessionKey !== targetSessionKey || !supportsGenericCurrentConversationBinding(record.conversation) ) { @@ -342,6 +351,9 @@ export function listGenericCurrentConversationBindingsBySession( } results.push(record); } + if (nextBindings) { + commitBindings(nextBindings); + } return results; } @@ -353,18 +365,23 @@ export function touchGenericCurrentConversationBinding(bindingId: string, at = D } loadBindingsIntoMemory(); const key = bindingId.slice(CURRENT_BINDINGS_ID_PREFIX.length); - const record = pruneExpiredBinding(key); + const record = bindingsByConversationKey.get(key); if (!record) { return; } - bindingsByConversationKey.set(key, { - ...record, - metadata: { - ...record.metadata, - lastActivityAt: at, - }, - }); - writePersistedBindings(); + const nextBindings = new Map(bindingsByConversationKey); + if (isBindingExpired(record)) { + nextBindings.delete(key); + } else { + nextBindings.set(key, { + ...record, + metadata: { + ...record.metadata, + lastActivityAt: at, + }, + }); + } + commitBindings(nextBindings); } /** Removes generic current-conversation bindings by binding id or target session key. */ @@ -381,11 +398,14 @@ export async function unbindGenericCurrentConversationBindings( } loadBindingsIntoMemory(); const key = normalizedBindingId.slice(CURRENT_BINDINGS_ID_PREFIX.length); - const record = pruneExpiredBinding(key); + const record = bindingsByConversationKey.get(key); if (record) { - bindingsByConversationKey.delete(key); - removed.push(record); - writePersistedBindings(); + const nextBindings = new Map(bindingsByConversationKey); + nextBindings.delete(key); + if (!isBindingExpired(record)) { + removed.push(record); + } + commitBindings(nextBindings); } return removed; } @@ -393,20 +413,23 @@ export async function unbindGenericCurrentConversationBindings( return removed; } loadBindingsIntoMemory(); - for (const key of bindingsByConversationKey.keys()) { - const record = pruneExpiredBinding(key); + const nextBindings = new Map(bindingsByConversationKey); + for (const [key, record] of bindingsByConversationKey) { + if (isBindingExpired(record)) { + nextBindings.delete(key); + continue; + } if ( - !record || record.targetSessionKey !== normalizedTargetSessionKey || !supportsGenericCurrentConversationBinding(record.conversation) ) { continue; } - bindingsByConversationKey.delete(key); + nextBindings.delete(key); removed.push(record); } - if (removed.length > 0) { - writePersistedBindings(); + if (nextBindings.size !== bindingsByConversationKey.size) { + commitBindings(nextBindings); } return removed; } @@ -417,7 +440,7 @@ export const testing = { env?: NodeJS.ProcessEnv; }) { bindingsLoaded = false; - bindingsByConversationKey.clear(); + bindingsByConversationKey = new Map(); if (params?.deletePersistedFile) { runOpenClawStateWriteTransaction( ({ db }) => {