diff --git a/src/config/sessions/session-accessor.sqlite-active-events.test.ts b/src/config/sessions/session-accessor.sqlite-active-events.test.ts index 3d35af06e782..48c050d6b7f1 100644 --- a/src/config/sessions/session-accessor.sqlite-active-events.test.ts +++ b/src/config/sessions/session-accessor.sqlite-active-events.test.ts @@ -571,11 +571,12 @@ describe("SQLite active transcript event projection", () => { } }); - it("awaits queued completion work after the preparation worker exits", async () => { + it("skips the preparation worker when the projection is already current", async () => { await persistSessionTranscriptTurn(scope, { messages: [{ eventId: "seed", message: { role: "user", content: "seed" } }], touchSessionEntry: false, }); + queuedSessionWrite.mockClear(); let resolveCompletionQueued!: () => void; const completionQueued = new Promise((resolve) => { resolveCompletionQueued = resolve; @@ -601,21 +602,26 @@ describe("SQLite active transcript event projection", () => { }, ); await entered; + const createWorker = vi.fn(() => { + throw new Error("clean projection must not spawn a worker"); + }); const outcome = reconcileSessionTranscriptIndexes({ agentId: scope.agentId, + createWorker, env: scope.env, }).then( (value) => ({ value }), (error: unknown) => ({ error }), ); - // The second queued write is the orphan sweep issued after the worker's done message. + // The second queued write is the preflight transaction waiting behind the held writer. await completionQueued; expect(queuedSessionWrite).toHaveBeenCalledTimes(2); releaseWriter(); await heldWriter; expect(await outcome).toEqual({ value: { reconciledSessions: 0 } }); + expect(createWorker).not.toHaveBeenCalled(); }, 10_000); it("keeps dirty batch appends off the synchronous writer stack", async () => { diff --git a/src/config/sessions/session-transcript-reconcile.ts b/src/config/sessions/session-transcript-reconcile.ts index 4cb6efb1445e..0dc9d3a2d74f 100644 --- a/src/config/sessions/session-transcript-reconcile.ts +++ b/src/config/sessions/session-transcript-reconcile.ts @@ -3,7 +3,7 @@ import { randomInt } from "node:crypto"; import path from "node:path"; import { fileURLToPath, pathToFileURL } from "node:url"; -import { Worker } from "node:worker_threads"; +import { Worker, type WorkerOptions } from "node:worker_threads"; import { createSubsystemLogger } from "../../logging/subsystem.js"; import { resolveOpenClawAgentSqlitePath, @@ -17,7 +17,10 @@ import { runExclusiveSqliteSessionWrite, toDatabaseOptions, } from "./session-accessor.sqlite-scope.js"; -import { deleteOrphanedTranscriptIndexRowsInTransaction } from "./session-transcript-index.js"; +import { + deleteOrphanedTranscriptIndexRowsInTransaction, + listSessionsNeedingTranscriptIndexReconcile, +} from "./session-transcript-index.js"; import { appendPreparedSessionTranscriptProjectionChunkInTransaction, claimPreparedSessionTranscriptProjectionInTransaction, @@ -47,6 +50,7 @@ export type SessionTranscriptReconcileResult = { }; type SessionTranscriptReconcileParams = OpenClawAgentDatabaseOptions & { + createWorker?: (filename: string | URL, options: WorkerOptions) => Worker; preferredSessionId?: string; }; @@ -198,7 +202,7 @@ async function finalizePreparedProjection( } /** Prepares full trees off-thread, then commits bounded chunks through the runtime writer owner. */ -export function reconcileSessionTranscriptIndexes( +export async function reconcileSessionTranscriptIndexes( params: SessionTranscriptReconcileParams, ): Promise { const databasePath = resolveOpenClawAgentSqlitePath(params); @@ -207,6 +211,19 @@ export function reconcileSessionTranscriptIndexes( ...(params.env ? { env: params.env } : {}), path: databasePath, }; + // The SQLite owner can cheaply prove a clean projection before paying for a + // Worker. Keep the post-worker sweep too, because request-time writers may race. + const needsWorker = await runProjectionWrite( + databaseOptions, + "sessions.transcript-index.preflight", + (database) => { + deleteOrphanedTranscriptIndexRowsInTransaction(database.db); + return listSessionsNeedingTranscriptIndexReconcile(database.db).length > 0; + }, + ); + if (!needsWorker) { + return { reconciledSessions: 0 }; + } const workerUrl = resolveSessionTranscriptReconcileWorkerUrl(); const sourceWorkerExecArgv = workerUrl.pathname.endsWith(".ts") ? ["--import", "tsx"] : undefined; const input: SessionTranscriptReconcileWorkerInput = { @@ -216,9 +233,12 @@ export function reconcileSessionTranscriptIndexes( }; let worker: Worker; try { - worker = new Worker(workerUrl, { workerData: input, execArgv: sourceWorkerExecArgv }); + worker = (params.createWorker ?? ((filename, options) => new Worker(filename, options)))( + workerUrl, + { workerData: input, execArgv: sourceWorkerExecArgv }, + ); } catch (error) { - return Promise.reject(normalizeReconcileError(error)); + throw normalizeReconcileError(error); } return new Promise((resolve, reject) => {