diff --git a/src/storage/mysql.ts b/src/storage/mysql.ts index b3005bb5f4..948cf6df75 100644 --- a/src/storage/mysql.ts +++ b/src/storage/mysql.ts @@ -1794,7 +1794,10 @@ export class MysqlBackend implements StorageBackend { if (c !== null) content[hash] = c; } mkdirSync(opts.archiveDir, { recursive: true }); - appendFileSync(exportPath, JSON.stringify({ version: 1, sessionId, archivedAt: Date.now(), session: row, content, messages, mediaMaps, blocks, refs: refsRows, deadRefs, ccrEntries: ccr, sessionExt: ext, checkpoints }) + "\n"); + // sqlite twin: appended as the last step inside the sweep + // transaction, so a failed sweep leaves no line for a session that + // is still active. + const archiveLine = JSON.stringify({ version: 1, sessionId, archivedAt: Date.now(), session: row, content, messages, mediaMaps, blocks, refs: refsRows, deadRefs, ccrEntries: ccr, sessionExt: ext, checkpoints }) + "\n"; let contentRowsDeleted = 0; await this.tx(async (c) => { const rel: string[] = []; @@ -1823,6 +1826,7 @@ export class MysqlBackend implements StorageBackend { // #4796 follow-up: scoped collection (see collectZeroRefIn) — // the reported count is what THIS archive collected. contentRowsDeleted = await this.collectZeroRefIn(c, rel); + appendFileSync(exportPath, archiveLine); }); loggerLog("info", `[#3931] archived ${sessionId} → ${exportPath} (${contentRowsDeleted} content rows collected)`); return { ok: true, exportPath, contentRowsDeleted }; @@ -1842,11 +1846,12 @@ export class MysqlBackend implements StorageBackend { sessionExt: Record[]; checkpoints: Record[]; }; + // sqlite twin: newest matching line wins. let lineNo = -1; let rawLines: string[] = []; try { rawLines = readFileSync(exportPath, "utf8").split("\n"); - for (let i = 0; i < rawLines.length; i++) { + for (let i = rawLines.length - 1; i >= 0; i--) { if (!rawLines[i].trim()) continue; let candidate: { sessionId?: string }; try { @@ -1929,11 +1934,15 @@ export class MysqlBackend implements StorageBackend { ); }); const kept: string[] = []; + let ownLines = 0; for (const l of rawLines) { if (!l.trim()) continue; try { const head = JSON.parse(l) as { sessionId?: string }; - if (head.sessionId === sessionId) continue; + if (head.sessionId === sessionId) { + if (++ownLines === 2) loggerLog("warn", `[storage] archive log holds more than one line for session ${sessionId}: the newest was restored, older ones are dropped`); + continue; + } } catch { continue; } diff --git a/src/storage/store.ts b/src/storage/store.ts index 97c674f438..be95a5a3b9 100644 --- a/src/storage/store.ts +++ b/src/storage/store.ts @@ -2079,7 +2079,15 @@ export class UnifiedStore { // One JSONL line per archived session: append-only (cheap, crash = // at worst one truncated line which the reader skips), restore // rewrites the file without the restored session's lines. - appendFileSync(exportPath, JSON.stringify({ version: 1, sessionId, archivedAt: Date.now(), session: row, content, messages, mediaMaps, blocks, refs: refsRows, deadRefs, ccrEntries: ccr, sessionExt: ext, checkpoints }) + "\n"); + // Serialized here, appended as the LAST step inside the sweep + // transaction below: a sweep that cannot start or fails midway + // (write lock held by another process, constraint error) must not + // leave a line behind for a session that is still active — that + // stale line would shadow the next successful archive on restore, + // and every failed attempt would add another full copy. The line + // still reaches the log before COMMIT, so a committed sweep always + // has its export. + const archiveLine = JSON.stringify({ version: 1, sessionId, archivedAt: Date.now(), session: row, content, messages, mediaMaps, blocks, refs: refsRows, deadRefs, ccrEntries: ccr, sessionExt: ext, checkpoints }) + "\n"; // --- sweep the row bodies (refcount discipline; pinned checkpoints survive) --- let contentRowsDeleted = 0; this.db.transaction(() => { @@ -2101,6 +2109,7 @@ export class UnifiedStore { // #4796 follow-up: scoped collection (see collectZeroRefIn) — // the reported count is what THIS archive collected. contentRowsDeleted = this.collectZeroRefIn(rel); + appendFileSync(exportPath, archiveLine); }); loggerLog("info", `[#3901] archived ${sessionId} → ${exportPath} (${contentRowsDeleted} content rows collected)`); return { ok: true, exportPath, contentRowsDeleted }; @@ -2127,12 +2136,15 @@ export class UnifiedStore { }; // Scan the JSONL for the session's line (skip malformed/truncated // trailing lines — an archive append interrupted by a crash must not - // block restoring OTHER sessions). + // block restoring OTHER sessions). Newest first: when the log holds + // more than one line for this session (an archive whose sweep never + // committed, then a later successful one), the last line appended is + // the state the session was actually archived in. let lineNo = -1; let rawLines: string[] = []; try { rawLines = readFileSync(exportPath, "utf8").split("\n"); - for (let i = 0; i < rawLines.length; i++) { + for (let i = rawLines.length - 1; i >= 0; i--) { if (!rawLines[i].trim()) continue; let candidate: { sessionId?: string }; try { @@ -2195,11 +2207,15 @@ export class UnifiedStore { // archive can theoretically appear more than once across reinstalls; // rewrite removes them all). Other sessions' lines are untouched. const kept: string[] = []; + let ownLines = 0; for (const l of rawLines) { if (!l.trim()) continue; try { const head = JSON.parse(l) as { sessionId?: string }; - if (head.sessionId === sessionId) continue; + if (head.sessionId === sessionId) { + if (++ownLines === 2) loggerLog("warn", `[storage] archive log holds more than one line for session ${sessionId}: the newest was restored, older ones are dropped`); + continue; + } } catch { continue; } kept.push(l); } diff --git a/tests/storage-archive-failed-sweep.test.ts b/tests/storage-archive-failed-sweep.test.ts new file mode 100644 index 0000000000..04a603f352 --- /dev/null +++ b/tests/storage-archive-failed-sweep.test.ts @@ -0,0 +1,215 @@ +import { test } 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 { planIngest, type LegacySessionLike } from "../src/storage/ingest.ts"; +import { UnifiedStore } from "../src/storage/store.ts"; +import { openSqliteDatabase } from "../src/storage/driver.ts"; +import { MysqlBackend, type Mysql2Module } from "../src/storage/mysql.ts"; + +/** + * archiveSession used to append the session's export line to archive.jsonl + * BEFORE the sweep transaction ran. A sweep that then failed (write lock held + * by another process, constraint error) left the line behind while the + * session stayed active and kept growing. The next successful archive + * appended a second line; restoreArchived took the FIRST matching line (the + * stale one), then dropped every line of that session — the newer export, + * with everything said since the failed attempt, was gone. + * + * Pinned here, for both backends: + * - a failed sweep leaves no line in the archive log; + * - when the log does hold several lines for one session (logs written by + * earlier versions), restore takes the newest. + */ + +process.env.BILI_PERSIST_ZSTD = "0"; +process.env.BILI_STORAGE_UNIFIED = "full"; +process.env.BILI_CONFIG_FILE = path.join( + fs.mkdtempSync(path.join(os.tmpdir(), "bili-archive-failed-sweep-")), + "no-config.json", +); + +interface CoreMsg { id: string; role: string; contentType: string; text: string } +const mkMsg = (id: string, text: string): CoreMsg => ({ id, role: "user", contentType: "text", text }); + +const mkSession = (id: string, msgs: CoreMsg[], lastSeen = 2_000): LegacySessionLike => ({ + id, + createdAt: 1_000, + lastSeen, + meta: { protocol: "openai" }, + stats: {}, + metadata: {}, + state: { blocks: [], nextBlockId: 1, nextRunId: 1, messageRefs: { byRaw: {}, byRef: {} } }, + blockContents: new Map(), + lastMessages: msgs as unknown as LegacySessionLike["lastMessages"], + lastMessagesFolded: false, + pluginSnapshot: undefined, + contentStore: undefined, +} as unknown as LegacySessionLike); + +const two = [mkMsg("m1", "one"), mkMsg("m2", "two")]; +const five = [...two, mkMsg("m3", "three"), mkMsg("m4", "four"), mkMsg("m5", "five")]; + +const logLines = (file: string): string[] => (fs.existsSync(file) ? fs.readFileSync(file, "utf8").split("\n").filter((l) => l.trim() !== "") : []); +const linesOf = (file: string, sessionId: string): number => logLines(file).filter((l) => (JSON.parse(l) as { sessionId?: string }).sessionId === sessionId).length; + +test("sqlite: a failed sweep leaves no archive line; the later archive restores in full", () => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "archive-failed-sweep-")); + const dbPath = path.join(dir, "index.db"); + const archiveDir = path.join(dir, "archive"); + const file = path.join(archiveDir, UnifiedStore.ARCHIVE_LOG_NAME); + const store = UnifiedStore.open(dbPath); + try { + store.ingestSession(planIngest(mkSession("sess-A", two))); + store.ingestSession(planIngest(mkSession("sess-B", [mkMsg("b1", "other session")]))); + assert.ok(store.archiveSession("sess-B", { archiveDir }).ok); + const otherLine = logLines(file)[0]; + + // Make the sweep fail at its last statement, after the row deletes. + const side = openSqliteDatabase(dbPath).db; + side.exec("CREATE TRIGGER fail_archive BEFORE UPDATE OF status ON sessions WHEN NEW.status = 'archived' BEGIN SELECT RAISE(ABORT, 'injected sweep failure'); END"); + assert.throws(() => store.archiveSession("sess-A", { archiveDir }), /injected sweep failure/); + side.exec("DROP TRIGGER fail_archive"); + side.close(); + + assert.equal((store.getSessionRow("sess-A") as unknown as { status?: string }).status, "active", "the session is still active"); + assert.equal(store.getMessages("sess-A").length, 2, "the sweep rolled back"); + assert.equal(linesOf(file, "sess-A"), 0, "a failed sweep leaves no line behind"); + assert.deepEqual(logLines(file), [otherLine], "the other session's line is untouched"); + + // The session lives on, then archives for real. + store.ingestSession(planIngest(mkSession("sess-A", five, 3_000))); + assert.equal(store.getMessages("sess-A").length, 5); + assert.ok(store.archiveSession("sess-A", { archiveDir }).ok); + assert.equal(linesOf(file, "sess-A"), 1, "exactly one line for the one committed archive"); + assert.equal(store.getMessages("sess-A").length, 0); + + assert.deepEqual(store.restoreArchived("sess-A", { archiveDir }), { ok: true }); + assert.deepEqual(store.getMessages("sess-A").map((m) => m.msgId), ["m1", "m2", "m3", "m4", "m5"], "the messages added after the failed attempt are restored"); + assert.deepEqual(logLines(file), [otherLine], "only the restored session's line is removed"); + } finally { + store.close(); + fs.rmSync(dir, { recursive: true, force: true }); + } +}); + +test("sqlite: with several lines for one session in the log, restore takes the newest", () => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "archive-newest-line-")); + const archiveDir = path.join(dir, "archive"); + const file = path.join(archiveDir, UnifiedStore.ARCHIVE_LOG_NAME); + const store = UnifiedStore.open(path.join(dir, "index.db")); + try { + // A log as earlier versions could leave it: the stale line of an + // archive whose sweep never committed, then the real one. + store.ingestSession(planIngest(mkSession("sess-A", two))); + assert.ok(store.archiveSession("sess-A", { archiveDir }).ok); + const staleLine = logLines(file)[0]; + assert.deepEqual(store.restoreArchived("sess-A", { archiveDir }), { ok: true }); + store.ingestSession(planIngest(mkSession("sess-A", five, 3_000))); + store.ingestSession(planIngest(mkSession("sess-B", [mkMsg("b1", "other session")]))); + assert.ok(store.archiveSession("sess-B", { archiveDir }).ok); + assert.ok(store.archiveSession("sess-A", { archiveDir }).ok); + const [otherLine, newestLine] = logLines(file); + fs.writeFileSync(file, [staleLine, otherLine, newestLine].join("\n") + "\n"); + + assert.deepEqual(store.restoreArchived("sess-A", { archiveDir }), { ok: true }); + assert.deepEqual(store.getMessages("sess-A").map((m) => m.msgId), ["m1", "m2", "m3", "m4", "m5"], "restored from the newest line, not the first"); + assert.deepEqual(logLines(file), [otherLine], "every line of the restored session is removed, nothing else"); + } finally { + store.close(); + fs.rmSync(dir, { recursive: true, force: true }); + } +}); + +/** A mysql2 stand-in: the pool answers the archive reads from fixed rows, + * the transaction connection records what it is asked and can be told to + * fail one statement. */ +function fakeMysql(opts: { messages: Record[]; failOn?: () => RegExp | null }) { + const calls: Array<{ sql: string; params: unknown[] }> = []; + const sessionRow = { + session_id: "sess-A", status: "active", namespace: "proxy", created_at: 1_000, updated_at: 2_000, next_block_id: 1, next_run_id: 1, + meta_json: "{}", stats_json: "{}", metadata_json: "{}", kstate_json: "{}", tail_json: "{}", + }; + const pool = { + async query(sql: string, params: unknown[] = []) { + if (/FROM sessions WHERE session_id = \?/.test(sql)) return [[sessionRow]]; + if (/FROM messages WHERE session_id = \?/.test(sql)) return [opts.messages]; + if (/SELECT bytes FROM content WHERE hash = \?/.test(sql)) return [[{ bytes: `body of ${String(params[0])}` }]]; + return [[]]; + }, + async getConnection() { + return { + async beginTransaction() { calls.push({ sql: "begin", params: [] }); }, + async commit() { calls.push({ sql: "commit", params: [] }); }, + async rollback() { calls.push({ sql: "rollback", params: [] }); }, + release() { /* nothing to free */ }, + async query(sql: string, params: unknown[] = []) { + calls.push({ sql, params }); + const fail = opts.failOn?.() ?? null; + if (fail !== null && fail.test(sql)) throw new Error("injected sweep failure"); + return [{ affectedRows: 1 }]; + }, + }; + }, + }; + const driver = { createPool: () => pool } as unknown as Mysql2Module; + return { backend: new MysqlBackend("unused", driver), calls }; +} + +test("mysql: a failed sweep leaves no archive line; a committed one writes it before COMMIT", async () => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "archive-failed-sweep-mysql-")); + const archiveDir = path.join(dir, "archive"); + const file = path.join(archiveDir, MysqlBackend.ARCHIVE_LOG_NAME); + try { + let fail: RegExp | null = /^UPDATE sessions SET status = 'archived'/; + const { backend, calls } = fakeMysql({ + messages: [{ seq: 1, msg_id: "m1", ref: null, ref_num: null, role: "user", kind: "message", content_hash: "h1", identity: "i1", retired_at: null, ts: null }], + failOn: () => fail, + }); + await assert.rejects(backend.archiveSession("sess-A", { archiveDir }), /injected sweep failure/); + assert.equal(calls[calls.length - 1].sql, "rollback"); + assert.equal(linesOf(file, "sess-A"), 0, "a failed sweep leaves no line behind"); + + fail = null; + calls.length = 0; + let linesAtCommit = -1; + const realPush = calls.push.bind(calls); + calls.push = (...items) => { + for (const it of items) if (it.sql === "commit") linesAtCommit = linesOf(file, "sess-A"); + return realPush(...items); + }; + assert.equal((await backend.archiveSession("sess-A", { archiveDir })).ok, true); + assert.equal(linesAtCommit, 1, "the export line is on disk before the sweep commits"); + assert.equal(linesOf(file, "sess-A"), 1); + } finally { + fs.rmSync(dir, { recursive: true, force: true }); + } +}); + +test("mysql: with several lines for one session in the log, restore takes the newest", async () => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "archive-newest-line-mysql-")); + const archiveDir = path.join(dir, "archive"); + const file = path.join(archiveDir, MysqlBackend.ARCHIVE_LOG_NAME); + const source = UnifiedStore.open(path.join(dir, "source.db")); + try { + // Build the two exports with the sqlite twin (same line format). + source.ingestSession(planIngest(mkSession("sess-A", two))); + assert.ok(source.archiveSession("sess-A", { archiveDir }).ok); + const staleLine = logLines(file)[0]; + assert.deepEqual(source.restoreArchived("sess-A", { archiveDir }), { ok: true }); + source.ingestSession(planIngest(mkSession("sess-A", five, 3_000))); + assert.ok(source.archiveSession("sess-A", { archiveDir }).ok); + const newestLine = logLines(file)[0]; + fs.writeFileSync(file, [staleLine, newestLine].join("\n") + "\n"); + + const { backend, calls } = fakeMysql({ messages: [] }); + assert.deepEqual(await backend.restoreArchived("sess-A", { archiveDir }), { ok: true }); + const restored = calls.filter((c) => c.sql.startsWith("INSERT INTO messages")).map((c) => c.params[2]); + assert.deepEqual(restored, ["m1", "m2", "m3", "m4", "m5"], "restored from the newest line, not the first"); + assert.deepEqual(logLines(file), [], "every line of the restored session is removed"); + } finally { + source.close(); + fs.rmSync(dir, { recursive: true, force: true }); + } +});