test(ledger): prove compaction crash safety with fault injection and against the real bucket

src/lib/server/ledger/ledger.ts

@@ -71,7 +71,7 @@ export class Ledger {
 			try {
 				pending.set(entry.path, decodeEvent(bytes));
 			} catch (e) {
-				throw new Error(`Corrupt event file ${entry.path}: ${(e as Error).message}`);
+				throw new Error(`Corrupt event file ${entry.path}`, { cause: e });
 			}
 		}
 		// A compactor deleted an event after publishing a ledger that contains it: re-read the ledger.
@@ -110,7 +110,7 @@ export class Ledger {
 	/** `id` may be supplied so callers can store a photo under the expense id before the event is written. */
 	createExpense(actor: string, body: ExpenseBody, id = `e_${this.newId()}`): Promise<Expense> {
 		return this.run(async () => {
-			if (this.state.latest.has(id)) throw new ConflictError(id, this.state.expenses.get(id)?.version ?? 0);
+			if (this.state.latest.has(id)) throw new ConflictError(id, this.state.latest.get(id)!.version);
 			const now = this.clock().toISOString();
 			const expense: Expense = { ...body, id, version: 1, createdBy: actor, createdAt: now, updatedBy: actor, updatedAt: now };
 			const written = await this.writeEvent({ kind: 'expense', op: 'upsert', entityId: id, version: 1, data: expense }, actor, now);

tests/integration/ledger.test.ts

@@ -0,0 +1,39 @@
+import { afterAll, describe, expect, it } from 'vitest';
+import type { ExpenseBody } from '../../src/lib/domain/types';
+import { ulid } from '../../src/lib/domain/ulid';
+import { Ledger } from '../../src/lib/server/ledger/ledger';
+import { HfBucketStore } from '../../src/lib/server/storage/hf-bucket-store';
+import { PrefixedStore } from '../../src/lib/server/storage/prefixed-store';
+import { realBucket } from './env';
+
+const creds = realBucket();
+const body = (description: string): ExpenseBody => ({
+	description, amountCents: 300, currency: 'USD', date: '2026-08-01', paidBy: 'u_a',
+	split: { mode: 'equal', participants: ['u_a', 'u_b', 'u_c'] }, shares: { u_a: 100, u_b: 100, u_c: 100 }
+});
+
+describe.skipIf(!creds)(
+	'Ledger against the real bucket',
+	() => {
+		const store = creds
+			? new PrefixedStore(new HfBucketStore(creds.bucket, creds.token), `test/${ulid()}`)
+			: (undefined as unknown as PrefixedStore);
+
+		afterAll(() => store.wipe(), 60_000);
+
+		it('writes events, compacts twice (append path), and reloads identically', async () => {
+			const ledger = await Ledger.load(store);
+			await ledger.createExpense('u_a', body('one'));
+			await ledger.createExpense('u_a', body('two'));
+			expect(await ledger.compact()).toMatchObject({ appended: 2, deleted: 2 });
+			await ledger.createExpense('u_a', body('three'));
+			expect(await ledger.compact()).toMatchObject({ appended: 1, deleted: 1, backedUp: true });
+			expect(await store.list('events/')).toEqual([]);
+
+			const reloaded = await Ledger.load(store);
+			expect(reloaded.listExpenses().map((e) => e.description).sort()).toEqual(['one', 'three', 'two']);
+			expect(reloaded.stats()).toEqual({ pendingEvents: 0, ledgerEvents: 3, ledgerExists: true });
+		});
+	},
+	120_000
+);

tests/unit/helpers/failing-store.ts

@@ -0,0 +1,55 @@
+import type { BucketStore, StoreEntry } from '../../../src/lib/server/storage/store';
+
+type Method = keyof BucketStore;
+export interface Fault {
+	method: Method;
+	/** 'before' = operation never happens; 'after' = operation happens but the caller sees an error (lost ack). */
+	when: 'before' | 'after';
+	/** Fail only the nth call (1-based) of that method; default every call. */
+	nth?: number;
+}
+
+/** Wraps a store and injects one fault, then behaves normally. */
+export class FailingStore implements BucketStore {
+	private calls = new Map<Method, number>();
+	constructor(
+		private readonly inner: BucketStore,
+		public fault: Fault | null
+	) {}
+
+	private async wrap<T>(method: Method, op: () => Promise<T>): Promise<T> {
+		const n = (this.calls.get(method) ?? 0) + 1;
+		this.calls.set(method, n);
+		const f = this.fault;
+		const hit = f && f.method === method && (f.nth === undefined || f.nth === n);
+		if (hit && f.when === 'before') {
+			this.fault = null;
+			throw new Error(`injected failure before ${method}`);
+		}
+		const result = await op();
+		if (hit && f.when === 'after') {
+			this.fault = null;
+			throw new Error(`injected failure after ${method}`);
+		}
+		return result;
+	}
+
+	get(path: string): Promise<Uint8Array | null> {
+		return this.wrap('get', () => this.inner.get(path));
+	}
+	put(path: string, bytes: Uint8Array, contentType?: string): Promise<void> {
+		return this.wrap('put', () => this.inner.put(path, bytes, contentType));
+	}
+	append(path: string, bytes: Uint8Array): Promise<void> {
+		return this.wrap('append', () => this.inner.append(path, bytes));
+	}
+	list(prefix: string): Promise<StoreEntry[]> {
+		return this.wrap('list', () => this.inner.list(prefix));
+	}
+	delete(paths: string[]): Promise<void> {
+		return this.wrap('delete', () => this.inner.delete(paths));
+	}
+	copy(src: string, dst: string): Promise<void> {
+		return this.wrap('copy', () => this.inner.copy(src, dst));
+	}
+}

tests/unit/ledger/compaction.test.ts

@@ -0,0 +1,117 @@
+import { mkdtemp, rm } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { afterEach, beforeEach, describe, expect, it } from 'vitest';
+import type { ExpenseBody } from '../../../src/lib/domain/types';
+import { Ledger } from '../../../src/lib/server/ledger/ledger';
+import { LocalFsStore } from '../../../src/lib/server/storage/local-fs-store';
+import { decodeText } from '../../../src/lib/server/storage/store';
+import { FailingStore, type Fault } from '../helpers/failing-store';
+
+const body = (description: string): ExpenseBody => ({
+	description, amountCents: 300, currency: 'USD', date: '2026-08-01', paidBy: 'u_a',
+	split: { mode: 'equal', participants: ['u_a', 'u_b', 'u_c'] }, shares: { u_a: 100, u_b: 100, u_c: 100 }
+});
+
+describe('Ledger.compact', () => {
+	let dir: string;
+	let base: LocalFsStore;
+	let day = Date.UTC(2026, 7, 29, 12);
+	const clock = () => new Date(day);
+
+	beforeEach(async () => {
+		dir = await mkdtemp(join(tmpdir(), 'splitwise-compact-'));
+		base = new LocalFsStore(dir);
+	});
+	afterEach(() => rm(dir, { recursive: true, force: true }));
+
+	async function seed(n: number): Promise<Ledger> {
+		const ledger = await Ledger.load(base, { clock });
+		for (let i = 0; i < n; i++) await ledger.createExpense('u_a', body(`e${i}`));
+		return ledger;
+	}
+
+	it('does nothing when there are no events', async () => {
+		const ledger = await Ledger.load(base, { clock });
+		expect(await ledger.compact()).toEqual({ appended: 0, deleted: 0, backedUp: false });
+		expect(await base.get('ledger.jsonl')).toBeNull();
+	});
+
+	it('moves events into the ledger in ULID order and deletes the files', async () => {
+		const ledger = await seed(3);
+		const result = await ledger.compact();
+		expect(result).toEqual({ appended: 3, deleted: 3, backedUp: false });
+		expect(await base.list('events/')).toEqual([]);
+		const lines = decodeText((await base.get('ledger.jsonl'))!).trim().split('\n');
+		expect(lines).toHaveLength(3);
+		expect(lines.map((l) => JSON.parse(l).data.description)).toEqual(['e0', 'e1', 'e2']);
+		expect(ledger.stats()).toEqual({ pendingEvents: 0, ledgerEvents: 3, ledgerExists: true });
+
+		const reloaded = await Ledger.load(base, { clock });
+		expect(reloaded.listExpenses().map((e) => e.description).sort()).toEqual(['e0', 'e1', 'e2']);
+	});
+
+	it('appends on the second compaction and backs up once per day', async () => {
+		const ledger = await seed(2);
+		await ledger.compact();
+		await ledger.createExpense('u_a', body('later'));
+		const second = await ledger.compact();
+		expect(second).toEqual({ appended: 1, deleted: 1, backedUp: true });
+		const backup = decodeText((await base.get('backups/ledger-2026-08-29.jsonl'))!).trim().split('\n');
+		expect(backup).toHaveLength(2);
+		expect(decodeText((await base.get('ledger.jsonl'))!).trim().split('\n')).toHaveLength(3);
+
+		await ledger.createExpense('u_a', body('same day'));
+		expect((await ledger.compact()).backedUp).toBe(false);
+		day += 24 * 3600 * 1000;
+		await ledger.createExpense('u_a', body('next day'));
+		expect((await ledger.compact()).backedUp).toBe(true);
+		expect(await base.get('backups/ledger-2026-08-30.jsonl')).not.toBeNull();
+	});
+
+	const faults: Fault[] = [
+		{ method: 'copy', when: 'before' },
+		{ method: 'append', when: 'before' },
+		{ method: 'append', when: 'after' },
+		{ method: 'delete', when: 'before' },
+		{ method: 'delete', when: 'after' }
+	];
+
+	for (const fault of faults) {
+		it(`loses nothing when the store fails ${fault.when} ${fault.method}`, async () => {
+			// First compaction creates the ledger so later ones exercise copy+append.
+			const seeded = await seed(2);
+			await seeded.compact();
+			await seeded.createExpense('u_a', body('x'));
+			await seeded.createExpense('u_a', body('y'));
+
+			const failing = new FailingStore(base, fault);
+			const ledger = await Ledger.load(failing, { clock });
+			await expect(ledger.compact()).rejects.toThrow('injected failure');
+
+			// Simulate a restart from whatever the store now contains.
+			const after = await Ledger.load(base, { clock });
+			expect(after.listExpenses().map((e) => e.description).sort()).toEqual(['e0', 'e1', 'x', 'y']);
+
+			// A retry finishes the job.
+			const retried = await Ledger.load(base, { clock });
+			await retried.compact();
+			expect(await base.list('events/')).toEqual([]);
+			const lines = decodeText((await base.get('ledger.jsonl'))!).trim().split('\n').map((l) => JSON.parse(l));
+			// Duplicated lines are allowed on disk (lost ack) but never lost, and fold dedups them.
+			expect(new Set(lines.map((l) => l.id)).size).toBe(4);
+			const final = await Ledger.load(base, { clock });
+			expect(final.listExpenses().map((e) => e.description).sort()).toEqual(['e0', 'e1', 'x', 'y']);
+			expect(final.stats().pendingEvents).toBe(0);
+		});
+	}
+
+	it('writes made while a compaction is in flight are not lost', async () => {
+		const ledger = await seed(1);
+		const compaction = ledger.compact();
+		const write = ledger.createExpense('u_a', body('during'));
+		await Promise.all([compaction, write]);
+		const reloaded = await Ledger.load(base, { clock });
+		expect(reloaded.listExpenses().map((e) => e.description).sort()).toEqual(['during', 'e0']);
+	});
+});