diff --git a/src/sync/index.ts b/src/sync/index.ts index d490cc5..6b8b509 100644 --- a/src/sync/index.ts +++ b/src/sync/index.ts @@ -86,55 +86,68 @@ function syncPiSession( } } - // Upsert session - upsertSession(db, { - id: sessionId, - file_path: disc.filePath, - project: disc.project, - source: disc.source, - cwd, - parent_session: parentSession, - started_at: startedAt, - last_line: cursor?.last_line ?? 0, - last_modified: disc.mtime, - analyzed_at: null, - message_count: 0, - branch_count: branchCount, - }); - - // Parse messages from resume point - const resumeLine = cursor?.last_line ?? 0; - let msgCount = 0; - - for (let i = resumeLine; i < lines.length; i++) { - const line = lines[i]?.trim(); - if (!line) continue; - - const parsed = parseLine(line); - if (!parsed || parsed.kind === "session") continue; - - const entry = parsed.entry; - insertMessage(db, { - id: entry.id, - session_id: sessionId, + // Upsert, insert, and update the cursor/count as ONE atomic unit per session + // (issue #59). Committing once per session instead of once per row amortises + // commit cycles. The cursor update lives inside the same transaction so it can + // never advance past messages that were rolled back — an interrupted sync + // leaves no partial session and is re-runnable. + const sessionTx = db.transaction((): number => { + // Upsert session + upsertSession(db, { + id: sessionId, + file_path: disc.filePath, + project: disc.project, source: disc.source, - parent_id: entry.parentId, - timestamp: entry.timestamp, - role: entry.role, - content_text: entry.text, - content_thinking: entry.thinking, - tool_calls: entry.tool_calls ? JSON.stringify(entry.tool_calls) : null, - tool_results: entry.tool_results ? JSON.stringify(entry.tool_results) : null, - usage: entry.usage ? JSON.stringify(entry.usage) : null, + cwd, + parent_session: parentSession, + started_at: startedAt, + last_line: cursor?.last_line ?? 0, + last_modified: disc.mtime, + analyzed_at: null, + message_count: 0, + branch_count: branchCount, }); - msgCount++; - } - // Update cursor and message count - updateCursor(db, sessionId, lines.length, disc.mtime); - const total = countMessages(db, sessionId); - updateMessageCount(db, sessionId, total); + // Parse messages from resume point + const resumeLine = cursor?.last_line ?? 0; + let msgCount = 0; + + for (let i = resumeLine; i < lines.length; i++) { + const line = lines[i]?.trim(); + if (!line) continue; + + const parsed = parseLine(line); + if (!parsed || parsed.kind === "session") continue; + + const entry = parsed.entry; + insertMessage(db, { + id: entry.id, + session_id: sessionId, + source: disc.source, + parent_id: entry.parentId, + timestamp: entry.timestamp, + role: entry.role, + content_text: entry.text, + content_thinking: entry.thinking, + tool_calls: entry.tool_calls ? JSON.stringify(entry.tool_calls) : null, + tool_results: entry.tool_results ? JSON.stringify(entry.tool_results) : null, + usage: entry.usage ? JSON.stringify(entry.usage) : null, + }); + msgCount++; + } + + // Update cursor and message count (same transaction — see above). If any + // insert throws, the whole session rolls back including the cursor. + updateCursor(db, sessionId, lines.length, disc.mtime); + const total = countMessages(db, sessionId); + updateMessageCount(db, sessionId, total); + + return msgCount; + }); + // Commit; a thrown error rolls back the session and propagates to runSync's + // per-session catch. Only count a session once its transaction commits. + const msgCount = sessionTx(); result.sessionsProcessed++; result.messagesInserted += msgCount; } @@ -153,61 +166,72 @@ function syncClaudeSession( const startedAt = meta?.timestamp ?? null; const cwd = (meta?.cwd ?? disc.project) || ""; - // Upsert session - upsertSession(db, { - id: sessionId, - file_path: disc.filePath, - project: disc.project, - source: disc.source, - cwd, - parent_session: null, - started_at: startedAt ?? "", - last_line: cursor?.last_line ?? 0, - last_modified: disc.mtime, - analyzed_at: null, - message_count: 0, - branch_count: 0, - }); - - // Claude tool_result blocks carry only a tool_use_id; resolve the tool name - // from the matching tool_use in the preceding assistant message (issue #30). - // Built from ALL lines (not just the resume point) so a tool_use/tool_result - // pair that straddles the cursor still resolves on an incremental sync. - const toolNamesById = buildClaudeToolNameMap(lines); - - // Parse messages from resume point - const resumeLine = cursor?.last_line ?? 0; - let msgCount = 0; - - for (let i = resumeLine; i < lines.length; i++) { - const line = lines[i]?.trim(); - if (!line) continue; - - const parsed = parseLine(line, "claude", toolNamesById); - if (!parsed || parsed.kind !== "message") continue; - - const entry = parsed.entry; - insertMessage(db, { - id: entry.id, - session_id: sessionId, + // Upsert, insert, and update the cursor/count as ONE atomic unit per session + // (issue #59) — see syncPiSession for the reasoning. The cursor update must be + // inside the same transaction as the inserts so a rollback can never leave the + // cursor advanced past messages that were discarded. + const sessionTx = db.transaction((): number => { + // Upsert session + upsertSession(db, { + id: sessionId, + file_path: disc.filePath, + project: disc.project, source: disc.source, - parent_id: entry.parentId, - timestamp: entry.timestamp, - role: entry.role, - content_text: entry.text, - content_thinking: entry.thinking, - tool_calls: entry.tool_calls ? JSON.stringify(entry.tool_calls) : null, - tool_results: entry.tool_results ? JSON.stringify(entry.tool_results) : null, - usage: entry.usage ? JSON.stringify(entry.usage) : null, + cwd, + parent_session: null, + started_at: startedAt ?? "", + last_line: cursor?.last_line ?? 0, + last_modified: disc.mtime, + analyzed_at: null, + message_count: 0, + branch_count: 0, }); - msgCount++; - } - // Update cursor and message count - updateCursor(db, sessionId, lines.length, disc.mtime); - const total = countMessages(db, sessionId); - updateMessageCount(db, sessionId, total); + // Claude tool_result blocks carry only a tool_use_id; resolve the tool name + // from the matching tool_use in the preceding assistant message (issue #30). + // Built from ALL lines (not just the resume point) so a tool_use/tool_result + // pair that straddles the cursor still resolves on an incremental sync. + const toolNamesById = buildClaudeToolNameMap(lines); + + // Parse messages from resume point + const resumeLine = cursor?.last_line ?? 0; + let msgCount = 0; + + for (let i = resumeLine; i < lines.length; i++) { + const line = lines[i]?.trim(); + if (!line) continue; + + const parsed = parseLine(line, "claude", toolNamesById); + if (!parsed || parsed.kind !== "message") continue; + + const entry = parsed.entry; + insertMessage(db, { + id: entry.id, + session_id: sessionId, + source: disc.source, + parent_id: entry.parentId, + timestamp: entry.timestamp, + role: entry.role, + content_text: entry.text, + content_thinking: entry.thinking, + tool_calls: entry.tool_calls ? JSON.stringify(entry.tool_calls) : null, + tool_results: entry.tool_results ? JSON.stringify(entry.tool_results) : null, + usage: entry.usage ? JSON.stringify(entry.usage) : null, + }); + msgCount++; + } + + // Update cursor and message count (same transaction — see above). + updateCursor(db, sessionId, lines.length, disc.mtime); + const total = countMessages(db, sessionId); + updateMessageCount(db, sessionId, total); + + return msgCount; + }); + // Commit; a thrown error rolls back the session and propagates to runSync's + // per-session catch. Only count a session once its transaction commits. + const msgCount = sessionTx(); result.sessionsProcessed++; result.messagesInserted += msgCount; } diff --git a/tests/component/sync.test.ts b/tests/component/sync.test.ts index ad914a4..f420435 100644 --- a/tests/component/sync.test.ts +++ b/tests/component/sync.test.ts @@ -1,5 +1,7 @@ import { describe, it } from "node:test"; import assert from "node:assert/strict"; +import * as fs from "node:fs"; +import * as os from "node:os"; import * as path from "node:path"; import { runSync } from "../../src/sync/index.js"; import { getStats } from "../../src/db/queries.js"; @@ -7,6 +9,34 @@ import { tempDb, NO_CLAUDE_DIR } from "./helpers.js"; const FIXTURES = path.resolve(import.meta.dirname, "..", "fixtures"); +/** + * A real, reachable mid-transaction failure: any message whose content is + * exactly POISON aborts the INSERT. RAISE(FAIL) is never suppressed by the + * INSERT OR IGNORE in insertMessage, so it always throws. + */ +function armPoisonTrigger(db: import("better-sqlite3").Database): void { + db.exec(` + CREATE TRIGGER poison_message + BEFORE INSERT ON messages + WHEN NEW.content_text = 'POISON' + BEGIN + SELECT RAISE(FAIL, 'poisoned'); + END; + `); +} + +function makeSessionDir(): { root: string; file: string } { + const root = fs.mkdtempSync(path.join(os.tmpdir(), "prospect-sync-tx-")); + const proj = path.join(root, "--proj"); + fs.mkdirSync(proj, { recursive: true }); + return { root, file: path.join(proj, "session-tx.jsonl") }; +} + +const SESSION_HEADER = JSON.stringify({ type: "session", version: 3, id: "tx-sess", timestamp: "2026-01-01T00:00:00Z", cwd: "/x" }); +const GOOD_M1 = JSON.stringify({ type: "message", id: "m1", timestamp: "2026-01-01T00:00:01Z", message: { role: "user", content: "first" } }); +const GOOD_M2 = JSON.stringify({ type: "message", id: "m2", timestamp: "2026-01-01T00:00:02Z", message: { role: "user", content: "second" } }); +const POISON = JSON.stringify({ type: "message", id: "m3", timestamp: "2026-01-01T00:00:03Z", message: { role: "user", content: "POISON" } }); + describe("end-to-end sync", () => { it("syncs simple.jsonl into database", () => { const { db, close } = tempDb(); @@ -50,3 +80,60 @@ describe("end-to-end sync", () => { } }); }); + +describe("a session commits atomically (issue #59)", () => { + it("a session that fails mid-sync leaves no partial session", () => { + const { db, close } = tempDb(); + const { root, file } = makeSessionDir(); + try { + armPoisonTrigger(db); + // Good message first, then a POISON message that aborts the transaction. + fs.writeFileSync(file, `${SESSION_HEADER}\n${GOOD_M1}\n${POISON}`); + + const result = runSync(db, root, NO_CLAUDE_DIR); + + // The whole session — including the upsert and the good message — rolled + // back; nothing partial survives. + assert.ok(result.errors.length >= 1, "the poisoned file is reported"); + assert.equal(result.sessionsProcessed, 0, "no session commits"); + assert.equal(result.messagesInserted, 0, "no messages commit"); + assert.equal(db.prepare("SELECT count(*) c FROM sessions WHERE id = 'tx-sess'").get()!.c, 0, "no session row"); + assert.equal(db.prepare("SELECT count(*) c FROM messages").get()!.c, 0, "no message rows"); + } finally { + db.close(); + fs.rmSync(root, { recursive: true, force: true }); + } + }); + + it("a partial failure does not advance the resume cursor past rolled-back rows", () => { + const { db, close } = tempDb(); + const { root, file } = makeSessionDir(); + try { + armPoisonTrigger(db); + fs.writeFileSync(file, `${SESSION_HEADER}\n${GOOD_M1}\n${GOOD_M2}`); + + // First sync commits cleanly. "header\nm1\nm2" splits to 3 elements, so + // last_line lands on 3 with both messages present. + const first = runSync(db, root, NO_CLAUDE_DIR); + assert.equal(first.sessionsProcessed, 1); + + // Append a poisoned line and touch the file so it is re-read. The new line + // lands at index 3 == the stored resume cursor (previous lines.length). + fs.appendFileSync(file, `\n${POISON}`); + const later = new Date().getTime() / 1000 + 10; + fs.utimesSync(file, later, later); + + const second = runSync(db, root, NO_CLAUDE_DIR); + assert.ok(second.errors.length >= 1, "the poisoned append is reported"); + + const row = db.prepare("SELECT last_line, message_count FROM sessions WHERE id = 'tx-sess'").get()! as { last_line: number; message_count: number }; + assert.equal(row.last_line, 3, "cursor stays at the previously committed line, not the poisoned one"); + assert.equal(row.message_count, 2, "count stays at the committed messages"); + const msgCount = db.prepare("SELECT count(*) c FROM messages WHERE session_id = 'tx-sess'").get()! as { c: number }; + assert.equal(msgCount.c, 2, "no partial message from the failed append"); + } finally { + db.close(); + fs.rmSync(root, { recursive: true, force: true }); + } + }); +});