diff --git a/src/commands/doctor-session-canonical-keys.test.ts b/src/commands/doctor-session-canonical-keys.test.ts index 338e60448338..106bded77475 100644 --- a/src/commands/doctor-session-canonical-keys.test.ts +++ b/src/commands/doctor-session-canonical-keys.test.ts @@ -79,6 +79,7 @@ describe("doctor canonical session-key repair", () => { expect(await repairCanonicalSessionKeys({ apply: true, cfg, env })).toMatchObject({ foundGroups: 2, + repairBatches: 1, removedRows: 1, repairedGroups: 2, }); @@ -117,6 +118,48 @@ describe("doctor canonical session-key repair", () => { }); }); + it("bounds same-database repair batches while collapsing whole-store projections", async () => { + await withStateDirEnv("openclaw-doctor-canonical-batches-", async ({ stateDir }) => { + const env = { ...process.env, OPENCLAW_STATE_DIR: stateDir }; + const storeTemplate = path.join(stateDir, "agents", "{agentId}", "sessions.json"); + const storePath = resolveStorePath(storeTemplate, { agentId: "main", env }); + const cfg = { + agents: { list: [{ id: "main", default: true }] }, + session: { store: storeTemplate }, + } as OpenClawConfig; + for (let index = 0; index < 65; index += 1) { + const target = `!BatchRoom${index}:example.org`; + const canonicalKey = `agent:main:matrix:channel:${target}`; + insertLegacySession({ + agentId: "main", + entry: { + chatType: "channel", + delivery: normalizeSessionDeliveryState({ + context: { channel: "matrix", to: target }, + }), + sessionId: `batch-session-${index}`, + updatedAt: index, + }, + env, + sessionKey: canonicalKey.toLowerCase(), + storePath, + }); + } + + expect(await repairCanonicalSessionKeys({ apply: true, cfg, env })).toMatchObject({ + foundGroups: 65, + repairBatches: 2, + removedRows: 65, + repairedGroups: 65, + }); + expect(await repairCanonicalSessionKeys({ apply: true, cfg, env })).toMatchObject({ + foundGroups: 0, + repairBatches: 0, + repairedGroups: 0, + }); + }); + }); + it("is a no-op for fresh stores and remains idempotent after repair", async () => { await withStateDirEnv("openclaw-doctor-canonical-fresh-", async ({ stateDir }) => { const env = { ...process.env, OPENCLAW_STATE_DIR: stateDir }; diff --git a/src/commands/doctor-session-canonical-keys.ts b/src/commands/doctor-session-canonical-keys.ts index c35d3bf50d66..cb6ea0e614c8 100644 --- a/src/commands/doctor-session-canonical-keys.ts +++ b/src/commands/doctor-session-canonical-keys.ts @@ -8,6 +8,7 @@ import { listSessionGenerationIdsForCanonicalRepair, loadTranscriptEvents, rehomeSessionDeliveryReferencesForCanonicalRepair, + rehomeSessionDeliveryReferencesForCanonicalRepairBatch, } from "../config/sessions/session-accessor.js"; import type { SessionEntryLifecycleRemoval } from "../config/sessions/session-accessor.lifecycle-types.js"; import { writeSqliteTranscriptArchive } from "../config/sessions/session-accessor.sqlite-archive.js"; @@ -28,7 +29,10 @@ import { resolveStoredSessionKeyForAgentStore, } from "../gateway/session-store-key.js"; import { normalizeAgentId, parseAgentSessionKey } from "../routing/session-key.js"; -import { openOpenClawAgentDatabase } from "../state/openclaw-agent-db.js"; +import { + openOpenClawAgentDatabase, + type OpenClawAgentDatabase, +} from "../state/openclaw-agent-db.js"; import { resolveTargetSqlitePath } from "./doctor-session-sqlite-readers.js"; type CanonicalSessionCandidate = { @@ -67,11 +71,19 @@ function createCanonicalRepairRemoval( export type CanonicalSessionKeyRepairReport = { archivedTranscriptDirectories: string[]; foundGroups: number; + repairBatches: number; removedRows: number; repairedGroups: number; scannedStores: number; }; +type CanonicalSessionRepairGroup = { + candidates: CanonicalSessionCandidate[]; + removedRows: number; +}; + +const CANONICAL_SESSION_REPAIR_BATCH_GROUP_LIMIT = 64; + type CanonicalSessionStore = { agentId: string; sqlitePath: string; @@ -305,7 +317,7 @@ function selectCanonicalSessionCandidate( function groupRepairCandidates( candidates: readonly CanonicalSessionCandidate[], params: { cfg: OpenClawConfig; env: NodeJS.ProcessEnv }, -) { +): CanonicalSessionRepairGroup[] { const byCanonicalKey = new Map(); for (const candidate of candidates) { const sentinelOwner = @@ -349,6 +361,139 @@ function groupRepairCandidates( }); } +type SingleDatabaseCanonicalRepairGroup = { + candidates: readonly CanonicalSessionCandidate[]; + selected: NonNullable>; +}; + +function resolveSingleDatabaseCanonicalRepairGroup( + candidates: readonly CanonicalSessionCandidate[], + params: { cfg: OpenClawConfig; env: NodeJS.ProcessEnv }, +): SingleDatabaseCanonicalRepairGroup | undefined { + const selected = selectCanonicalSessionCandidate(candidates, params); + if ( + !selected || + selected.winner.sqlitePath !== selected.destination.sqlitePath || + candidates.some((candidate) => candidate.sqlitePath !== selected.destination.sqlitePath) + ) { + return undefined; + } + return { candidates, selected }; +} + +function createCanonicalDestinationRemovals( + candidates: readonly CanonicalSessionCandidate[], + selected: NonNullable>, +): SessionEntryLifecycleRemoval[] { + const relatedSessionIds = new Set( + [selected.entry.sessionId, selected.entry.previousSessionId].filter( + (value): value is string => typeof value === "string" && value.length > 0, + ), + ); + return candidates + .filter( + (candidate) => + candidate.sessionKey !== selected.winner.canonicalKey || + candidate.rawEntryJson !== undefined, + ) + .map((candidate) => + createCanonicalRepairRemoval(candidate, { + archiveRemovedTranscript: !relatedSessionIds.has(candidate.entry.sessionId), + deleteOwnedWindows: false, + }), + ); +} + +function listCanonicalDestinationAliasKeys( + destinationStore: readonly CanonicalSessionCandidate[], + winner: CanonicalSessionCandidate, +): string[] { + return destinationStore + .map((candidate) => candidate.sessionKey) + .filter((sessionKey) => sessionKey !== winner.canonicalKey); +} + +function applyCanonicalDestinationArtifacts(params: { + copyWinnerAlias: boolean; + database: OpenClawAgentDatabase; + destinationStore: readonly CanonicalSessionCandidate[]; + rehomeDeliveries: boolean; + winner: CanonicalSessionCandidate; +}): void { + const destinationAliasKeys = listCanonicalDestinationAliasKeys( + params.destinationStore, + params.winner, + ); + if (destinationAliasKeys.length > 0) { + if (params.rehomeDeliveries) { + rehomeSessionDeliveryReferencesForCanonicalRepair( + params.database, + params.winner.canonicalKey, + destinationAliasKeys, + ); + } + copySessionNodeArtifactsForRepair( + params.database, + params.database, + destinationAliasKeys, + params.winner.canonicalKey, + { includeMembers: false }, + ); + } + if (!params.copyWinnerAlias || params.winner.sessionKey === params.winner.canonicalKey) { + return; + } + deleteSessionMembersForRepair(params.database, params.winner.canonicalKey); + copySessionNodeArtifactsForRepair( + params.database, + params.database, + [params.winner.sessionKey], + params.winner.canonicalKey, + ); +} + +async function repairCanonicalSessionGroupsInSingleDatabase( + groups: readonly SingleDatabaseCanonicalRepairGroup[], +): Promise { + const first = groups[0]; + if (!first) { + return []; + } + const destination = first.selected.destination; + const result = await applySessionEntryLifecycleMutation({ + agentId: destination.agentId, + allowCanonicalRepair: true, + afterUpsertsInTransaction: (database) => { + rehomeSessionDeliveryReferencesForCanonicalRepairBatch( + database, + groups.map((group) => ({ + canonicalKey: group.selected.winner.canonicalKey, + previousKeys: listCanonicalDestinationAliasKeys(group.candidates, group.selected.winner), + })), + ); + for (const group of groups) { + applyCanonicalDestinationArtifacts({ + copyWinnerAlias: true, + database, + destinationStore: group.candidates, + rehomeDeliveries: false, + winner: group.selected.winner, + }); + } + }, + removals: groups.flatMap((group) => + createCanonicalDestinationRemovals(group.candidates, group.selected), + ), + skipMaintenance: true, + storePath: destination.storePath, + upserts: groups.map((group) => ({ + entry: group.selected.entry, + sessionKey: group.selected.winner.canonicalKey, + })), + }); + return result.archivedTranscriptDirectories; +} + async function repairCanonicalSessionGroup( candidates: readonly CanonicalSessionCandidate[], params: { cfg: OpenClawConfig; env: NodeJS.ProcessEnv }, @@ -422,11 +567,6 @@ async function repairCanonicalSessionGroup( } } } - const relatedSessionIds = new Set( - [selected.entry.sessionId, selected.entry.previousSessionId].filter( - (value): value is string => typeof value === "string" && value.length > 0, - ), - ); setCanonicalSqliteSessionMainKey( openOpenClawAgentDatabase({ agentId: destination.agentId, path: destination.sqlitePath }), params.cfg.session?.mainKey, @@ -435,35 +575,13 @@ async function repairCanonicalSessionGroup( agentId: destination.agentId, allowCanonicalRepair: true, afterUpsertsInTransaction: (destinationDatabase) => { - const destinationAliasKeys = destinationStore - .map((candidate) => candidate.sessionKey) - .filter((sessionKey) => sessionKey !== winner.canonicalKey); - if (destinationAliasKeys.length > 0) { - rehomeSessionDeliveryReferencesForCanonicalRepair( - destinationDatabase, - winner.canonicalKey, - destinationAliasKeys, - ); - copySessionNodeArtifactsForRepair( - destinationDatabase, - destinationDatabase, - destinationAliasKeys, - winner.canonicalKey, - { includeMembers: false }, - ); - } - if ( - winner.sqlitePath === destination.sqlitePath && - winner.sessionKey !== winner.canonicalKey - ) { - deleteSessionMembersForRepair(destinationDatabase, winner.canonicalKey); - copySessionNodeArtifactsForRepair( - destinationDatabase, - destinationDatabase, - [winner.sessionKey], - winner.canonicalKey, - ); - } + applyCanonicalDestinationArtifacts({ + copyWinnerAlias: winner.sqlitePath === destination.sqlitePath, + database: destinationDatabase, + destinationStore, + rehomeDeliveries: true, + winner, + }); if (winner.sqlitePath !== destination.sqlitePath) { copySessionOwnedStateForCanonicalRepair({ canonicalKey: winner.canonicalKey, @@ -476,17 +594,7 @@ async function repairCanonicalSessionGroup( }); } }, - removals: destinationStore - .filter( - (candidate) => - candidate.sessionKey !== winner.canonicalKey || candidate.rawEntryJson !== undefined, - ) - .map((candidate) => - createCanonicalRepairRemoval(candidate, { - archiveRemovedTranscript: !relatedSessionIds.has(candidate.entry.sessionId), - deleteOwnedWindows: false, - }), - ), + removals: createCanonicalDestinationRemovals(destinationStore, selected), skipMaintenance: true, storePath: destination.storePath, upserts: [{ entry: selected.entry, sessionKey: winner.canonicalKey }], @@ -538,6 +646,7 @@ export async function repairCanonicalSessionKeys(params: { env, }); const archivedTranscriptDirectories = new Set(); + let repairBatches = 0; let repairedGroups = 0; if (params.apply) { for (const store of stores) { @@ -550,19 +659,65 @@ export async function repairCanonicalSessionKeys(params: { const candidates = collectCanonicalSessionCandidates({ cfg: params.cfg, env }, stores); const repairGroups = groupRepairCandidates(candidates, { cfg: params.cfg, env }); if (params.apply) { - for (const group of repairGroups) { - for (const directory of await repairCanonicalSessionGroup(group.candidates, { + let index = 0; + while (index < repairGroups.length) { + const group = repairGroups[index]; + if (!group) { + break; + } + const singleDatabaseGroup = resolveSingleDatabaseCanonicalRepairGroup(group.candidates, { cfg: params.cfg, env, - })) { + }); + if (!singleDatabaseGroup) { + for (const directory of await repairCanonicalSessionGroup(group.candidates, { + cfg: params.cfg, + env, + })) { + archivedTranscriptDirectories.add(directory); + } + index += 1; + repairBatches += 1; + repairedGroups += 1; + continue; + } + const batch = [singleDatabaseGroup]; + index += 1; + // Keep commits bounded and preserve the original order around cross-store moves, while + // collapsing the repeated whole-store projections for the common same-database path. + while ( + index < repairGroups.length && + batch.length < CANONICAL_SESSION_REPAIR_BATCH_GROUP_LIMIT + ) { + const nextGroup = repairGroups[index]; + if (!nextGroup) { + break; + } + const nextSingleDatabaseGroup = resolveSingleDatabaseCanonicalRepairGroup( + nextGroup.candidates, + { cfg: params.cfg, env }, + ); + if ( + !nextSingleDatabaseGroup || + nextSingleDatabaseGroup.selected.destination.sqlitePath !== + singleDatabaseGroup.selected.destination.sqlitePath + ) { + break; + } + batch.push(nextSingleDatabaseGroup); + index += 1; + } + for (const directory of await repairCanonicalSessionGroupsInSingleDatabase(batch)) { archivedTranscriptDirectories.add(directory); } - repairedGroups += 1; + repairBatches += 1; + repairedGroups += batch.length; } } return { archivedTranscriptDirectories: [...archivedTranscriptDirectories].toSorted(), foundGroups: repairGroups.length, + repairBatches, removedRows: repairGroups.reduce((total, group) => total + group.removedRows, 0), repairedGroups, scannedStores: stores.length, diff --git a/src/commands/doctor-session-transcripts.test.ts b/src/commands/doctor-session-transcripts.test.ts index 96c4c098375f..3307f51c922c 100644 --- a/src/commands/doctor-session-transcripts.test.ts +++ b/src/commands/doctor-session-transcripts.test.ts @@ -116,6 +116,7 @@ describe("doctor session transcript repair", () => { repairCanonicalSessionKeys.mockReset().mockResolvedValue({ archivedTranscriptDirectories: [], foundGroups: 0, + repairBatches: 0, removedRows: 0, repairedGroups: 0, scannedStores: 0, diff --git a/src/commands/doctor-session-transcripts.ts b/src/commands/doctor-session-transcripts.ts index 0e5eb6718343..1d15920d7875 100644 --- a/src/commands/doctor-session-transcripts.ts +++ b/src/commands/doctor-session-transcripts.ts @@ -526,6 +526,7 @@ async function noteSessionSqliteMigrationHealth(params: { let canonicalKeyReport: CanonicalSessionKeyRepairReport = { archivedTranscriptDirectories: [], foundGroups: 0, + repairBatches: 0, removedRows: 0, repairedGroups: 0, scannedStores: 0, @@ -585,7 +586,7 @@ async function noteSessionSqliteMigrationHealth(params: { if (canonicalKeyReport.foundGroups > 0) { note( params.shouldRepair - ? `- Canonicalized ${canonicalKeyReport.repairedGroups} session-key group(s), removed ${canonicalKeyReport.removedRows} duplicate or alias row(s), and preserved cross-store history in ${canonicalKeyReport.archivedTranscriptDirectories.length} archive director${canonicalKeyReport.archivedTranscriptDirectories.length === 1 ? "y" : "ies"}.` + ? `- Canonicalized ${canonicalKeyReport.repairedGroups} session-key group(s) in ${canonicalKeyReport.repairBatches} transaction batch(es), removed ${canonicalKeyReport.removedRows} duplicate or alias row(s), and preserved cross-store history in ${canonicalKeyReport.archivedTranscriptDirectories.length} archive director${canonicalKeyReport.archivedTranscriptDirectories.length === 1 ? "y" : "ies"}.` : `- Found ${canonicalKeyReport.foundGroups} non-canonical or duplicate session-key group(s). Run "openclaw doctor --fix" to preserve their history and canonicalize the rows.`, "Session SQLite", ); diff --git a/src/config/sessions/session-accessor.entry.ts b/src/config/sessions/session-accessor.entry.ts index 057acaaf6933..0eb2d32dd95f 100644 --- a/src/config/sessions/session-accessor.entry.ts +++ b/src/config/sessions/session-accessor.entry.ts @@ -16,6 +16,7 @@ import { listSqliteSessionEntries, listSqliteSessionEntriesForCanonicalRepair as listSessionEntriesForCanonicalRepair, rehomeSqliteSessionDeliveryReferencesForCanonicalRepair as rehomeSessionDeliveryReferencesForCanonicalRepair, + rehomeSqliteSessionDeliveryReferencesForCanonicalRepairBatch as rehomeSessionDeliveryReferencesForCanonicalRepairBatch, listSqliteSessionEntriesReadOnly as listSessionEntriesReadOnly, listSqliteSessionEntryKeysReadOnly as listSessionEntryKeysReadOnly, loadExactSqliteSessionEntry as loadExactSessionEntry, @@ -63,6 +64,7 @@ export { listSessionEntriesReadOnly, listSessionEntriesForCanonicalRepair, rehomeSessionDeliveryReferencesForCanonicalRepair, + rehomeSessionDeliveryReferencesForCanonicalRepairBatch, listSessionEntryKeysReadOnly, loadExactSessionEntry, loadExactSessionEntryReadOnly, diff --git a/src/config/sessions/session-accessor.sqlite-canonical-repair.ts b/src/config/sessions/session-accessor.sqlite-canonical-repair.ts index 0dcb0dbcc432..98f6d4d95658 100644 --- a/src/config/sessions/session-accessor.sqlite-canonical-repair.ts +++ b/src/config/sessions/session-accessor.sqlite-canonical-repair.ts @@ -140,31 +140,64 @@ export function rehomeSqliteSessionDeliveryReferencesForCanonicalRepair( canonicalKey: string, previousKeys: readonly string[], ): void { - const ownedKeys = new Set([canonicalKey, ...previousKeys]); - const db = getSessionKysely(database.db); - const competingIdentities = new Set( - executeSqliteQuerySync( - database.db, - db.selectFrom("session_nodes").select("session_key"), - ).rows.flatMap((row) => - ownedKeys.has(row.session_key) ? [] : [normalizeStoreSessionKey(row.session_key.trim())], - ), - ); - const aliases = resolveSqliteCanonicalRepairLookupKeys(canonicalKey, previousKeys).filter( - (key) => - key !== canonicalKey && - (ownedKeys.has(key) || !competingIdentities.has(normalizeStoreSessionKey(key.trim()))), - ); - if (aliases.length === 0) { + rehomeSqliteSessionDeliveryReferencesForCanonicalRepairBatch(database, [ + { canonicalKey, previousKeys }, + ]); +} + +/** Doctor-only batched delivery rewrite with one session identity inventory per database. */ +export function rehomeSqliteSessionDeliveryReferencesForCanonicalRepairBatch( + database: OpenClawAgentDatabase, + repairs: readonly { canonicalKey: string; previousKeys: readonly string[] }[], +): void { + if (repairs.length === 0) { return; } - executeSqliteQuerySync( + const db = getSessionKysely(database.db); + const storedSessionKeys = executeSqliteQuerySync( database.db, - db - .updateTable("conversation_deliveries") - .set({ source_session_key: canonicalKey }) - .where("source_session_key", "in", aliases), - ); + db.selectFrom("session_nodes").select("session_key"), + ).rows.map((row) => row.session_key); + const storedSessionKeySet = new Set(storedSessionKeys); + const identityCounts = new Map(); + for (const sessionKey of storedSessionKeys) { + const identity = normalizeStoreSessionKey(sessionKey.trim()); + identityCounts.set(identity, (identityCounts.get(identity) ?? 0) + 1); + } + for (const repair of repairs) { + const ownedKeys = new Set([repair.canonicalKey, ...repair.previousKeys]); + const ownedIdentityCounts = new Map(); + for (const sessionKey of ownedKeys) { + if (!storedSessionKeySet.has(sessionKey)) { + continue; + } + const identity = normalizeStoreSessionKey(sessionKey.trim()); + ownedIdentityCounts.set(identity, (ownedIdentityCounts.get(identity) ?? 0) + 1); + } + const aliases = resolveSqliteCanonicalRepairLookupKeys( + repair.canonicalKey, + repair.previousKeys, + ).filter((key) => { + if (key === repair.canonicalKey) { + return false; + } + if (ownedKeys.has(key)) { + return true; + } + const identity = normalizeStoreSessionKey(key.trim()); + return (identityCounts.get(identity) ?? 0) <= (ownedIdentityCounts.get(identity) ?? 0); + }); + if (aliases.length === 0) { + continue; + } + executeSqliteQuerySync( + database.db, + db + .updateTable("conversation_deliveries") + .set({ source_session_key: repair.canonicalKey }) + .where("source_session_key", "in", aliases), + ); + } } type CanonicalRepairRow = Selectable & { diff --git a/src/config/sessions/session-accessor.sqlite.ts b/src/config/sessions/session-accessor.sqlite.ts index 0ea690b39ae6..6a1b80373e36 100644 --- a/src/config/sessions/session-accessor.sqlite.ts +++ b/src/config/sessions/session-accessor.sqlite.ts @@ -26,6 +26,7 @@ export { listSqliteSessionEntriesForCanonicalRepair, listSqliteSessionGenerationIdsForCanonicalRepair, rehomeSqliteSessionDeliveryReferencesForCanonicalRepair, + rehomeSqliteSessionDeliveryReferencesForCanonicalRepairBatch, } from "./session-accessor.sqlite-canonical-repair.js"; export { cleanupSqliteSessionLifecycleArtifacts, diff --git a/src/config/sessions/session-accessor.ts b/src/config/sessions/session-accessor.ts index d8383bd69085..dcfe5cdeef3b 100644 --- a/src/config/sessions/session-accessor.ts +++ b/src/config/sessions/session-accessor.ts @@ -127,6 +127,7 @@ export { listSessionEntriesReadOnly, listSessionEntriesForCanonicalRepair, rehomeSessionDeliveryReferencesForCanonicalRepair, + rehomeSessionDeliveryReferencesForCanonicalRepairBatch, listSessionEntryKeysReadOnly, loadExactSessionEntry, loadExactSessionEntryReadOnly,