Skip to content

perf: batch message inserts in a per-session transaction during sync (#59) #64

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
elecnix wants to merge 1 commit into from
Closed

perf: batch message inserts in a per-session transaction during sync (#59) #64

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
216 changes: 120 additions & 96 deletions src/sync/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
Expand All @@ -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;
}
Expand Down
87 changes: 87 additions & 0 deletions tests/component/sync.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,12 +1,42 @@
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";
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();
Expand Down Expand Up @@ -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 });
}
});
});