feat(chord): WAL reader, page checksums and segment container for SQLite replication

Portable modules under sqlite/replication: a WAL scanner that returns the newest page versions of committed transactions and stops at the last valid commit frame, incremental whole-database checksums from per-page hashes, the shared segment and base container with header validation, and object keys and start markers. A Node-only zstd page compressor sits beside them. None of it is exported from the package yet.

packages/chord/src/sqlite/replication/checksum.ts

@@ -0,0 +1,158 @@
+/**
+ * Page hashes and the whole-database checksum.
+ *
+ * A reader applying segments must know that each segment fits the database it has, and that the result is what the
+ * writer had. Hashing the whole file after every flush would cost time proportional to the database, so the checksum
+ * is built to update in time proportional to the change instead:
+ *
+ * - The hash of page `p` with contents `b` is the first 8 bytes of SHA-256 over `p` as a big-endian u32 followed by
+ *   `b`, read as a big-endian u64. Including the page number means two pages with equal contents hash differently.
+ * - The database checksum is the XOR of the hashes of pages `1..dbPageCount`. An empty database has checksum 0.
+ * - When page `p` changes, XOR out its old hash and XOR in the new one. When the database shrinks, XOR out the pages
+ *   that went away. Neither needs the old page contents, only the old hashes, which the writer keeps in a sidecar
+ *   file of 8 bytes per page.
+ *
+ * The checksum catches accidents such as a missing segment, a segment applied to the wrong database or a page written
+ * to the wrong offset. It is not a defence against a malicious writer: XOR of truncated hashes is easy to forge on
+ * purpose. Stores authenticate writers instead.
+ *
+ * SHA-256 comes from Web Crypto, which Node, browsers and Workers all have, so this module stays portable.
+ */
+
+/** Hash one page. */
+export async function pageHash(pageNo: number, page: Uint8Array): Promise<bigint> {
+	const input = new Uint8Array(4 + page.length);
+	new DataView(input.buffer).setUint32(0, pageNo);
+	input.set(page, 4);
+	const digest = await crypto.subtle.digest("SHA-256", input);
+	return new DataView(digest).getBigUint64(0);
+}
+
+/** Hash several pages, keyed like the input. */
+export async function hashPages(pages: ReadonlyMap<number, Uint8Array>): Promise<Map<number, bigint>> {
+	const entries = [...pages];
+	const hashes = await Promise.all(entries.map(([pageNo, page]) => pageHash(pageNo, page)));
+	return new Map(entries.map(([pageNo], index) => [pageNo, hashes[index] ?? 0n]));
+}
+
+/** A checksum as written in headers: 16 lowercase hex digits. */
+export function formatChecksum(checksum: bigint): string {
+	return checksum.toString(16).padStart(16, "0");
+}
+
+/** Parse a checksum written by `formatChecksum`. */
+export function parseChecksum(text: string): bigint {
+	if (!/^[0-9a-f]{16}$/.test(text)) throw new Error(`Invalid database checksum ${JSON.stringify(text)}`);
+	return BigInt(`0x${text}`);
+}
+
+/**
+ * The hash of every page of one database, and their XOR. This is the writer's sidecar and a restoring reader's
+ * running state; both update it from the pages of each segment they produce or apply.
+ */
+export class PageHashes {
+	private hashes: BigUint64Array;
+	private count: number;
+	private xor: bigint;
+
+	private constructor(hashes: BigUint64Array, count: number) {
+		this.hashes = hashes;
+		this.count = count;
+		let xor = 0n;
+		for (let index = 0; index < count; index++) xor ^= hashes[index] ?? 0n;
+		this.xor = xor;
+	}
+
+	/** The hashes of an empty database. */
+	static empty(): PageHashes {
+		return new PageHashes(new BigUint64Array(0), 0);
+	}
+
+	/** Hash every page of a complete database file. */
+	static async fromDatabase(bytes: Uint8Array, pageSize: number): Promise<PageHashes> {
+		if (bytes.length % pageSize !== 0) {
+			throw new Error(`Database of ${bytes.length} bytes is not a whole number of ${pageSize}-byte pages`);
+		}
+		const pages = new Map<number, Uint8Array>();
+		for (let offset = 0; offset < bytes.length; offset += pageSize) {
+			pages.set(offset / pageSize + 1, bytes.subarray(offset, offset + pageSize));
+		}
+		const hashes = PageHashes.empty();
+		hashes.apply(await hashPages(pages), pages.size);
+		return hashes;
+	}
+
+	/** Read a sidecar written by `encode`: one little-endian u64 per page, page `p` at offset `(p - 1) * 8`. */
+	static decode(bytes: Uint8Array): PageHashes {
+		if (bytes.length % 8 !== 0) throw new Error(`Page hash file of ${bytes.length} bytes is not a multiple of 8`);
+		const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength);
+		const count = bytes.length / 8;
+		const hashes = new BigUint64Array(count);
+		for (let index = 0; index < count; index++) hashes[index] = view.getBigUint64(index * 8, true);
+		return new PageHashes(hashes, count);
+	}
+
+	/** Database size in pages. */
+	get pageCount(): number {
+		return this.count;
+	}
+
+	/** The database checksum. */
+	get checksum(): bigint {
+		return this.xor;
+	}
+
+	/** The hash of page `pageNo`, or `undefined` beyond the database. */
+	hash(pageNo: number): bigint | undefined {
+		return pageNo >= 1 && pageNo <= this.count ? this.hashes[pageNo - 1] : undefined;
+	}
+
+	/**
+	 * Apply the pages of one segment: set the hash of every changed page, then resize to `dbPageCount`. Pages past
+	 * `dbPageCount` must not be present. When the database grows, every new page must be among `pages`; SQLite writes
+	 * every page it adds, and a page this database has never seen has no known contents to hash. A violation throws
+	 * and leaves the hashes unchanged.
+	 */
+	apply(pages: ReadonlyMap<number, bigint>, dbPageCount: number): void {
+		if (!Number.isSafeInteger(dbPageCount) || dbPageCount < 0) throw new Error(`Invalid page count ${dbPageCount}`);
+		for (const pageNo of pages.keys()) {
+			if (!Number.isSafeInteger(pageNo) || pageNo < 1 || pageNo > dbPageCount) {
+				throw new Error(`Page ${pageNo} is outside a database of ${dbPageCount} pages`);
+			}
+		}
+		for (let pageNo = this.count + 1; pageNo <= dbPageCount; pageNo++) {
+			if (!pages.has(pageNo)) {
+				throw new Error(`Database grew to ${dbPageCount} pages without page ${pageNo}`);
+			}
+		}
+
+		for (let pageNo = dbPageCount + 1; pageNo <= this.count; pageNo++) this.xor ^= this.hashes[pageNo - 1] ?? 0n;
+		if (dbPageCount > this.hashes.length) {
+			const grown = new BigUint64Array(Math.max(dbPageCount, this.hashes.length * 2));
+			grown.set(this.hashes.subarray(0, this.count));
+			this.hashes = grown;
+		} else if (dbPageCount < this.count) {
+			this.hashes.fill(0n, dbPageCount, this.count);
+		}
+		const previousCount = this.count;
+		this.count = dbPageCount;
+		for (const [pageNo, hash] of pages) {
+			if (pageNo <= previousCount) this.xor ^= this.hashes[pageNo - 1] ?? 0n;
+			this.hashes[pageNo - 1] = hash;
+			this.xor ^= hash;
+		}
+	}
+
+	/** Write the sidecar format read by `decode`. */
+	encode(): Uint8Array {
+		const bytes = new Uint8Array(this.count * 8);
+		const view = new DataView(bytes.buffer);
+		for (let index = 0; index < this.count; index++) view.setBigUint64(index * 8, this.hashes[index] ?? 0n, true);
+		return bytes;
+	}
+
+	/** An independent copy. */
+	clone(): PageHashes {
+		return new PageHashes(this.hashes.slice(0, this.count), this.count);
+	}
+}

packages/chord/src/sqlite/replication/container.ts

@@ -0,0 +1,359 @@
+/**
+ * The encoding shared by segments and bases.
+ *
+ * A segment holds the newest version of every page changed by a run of transactions; a base holds every page of the
+ * database at one seq. Both are immutable objects that a store keeps and readers apply to a database file, and both
+ * use one container:
+ *
+ * ```text
+ * offset  size  field
+ * 0       8     magic: ASCII "PISYNC01"
+ * 8       4     headerLength: u32, at most 65536
+ * 12      n     header: UTF-8 JSON
+ * 12+n    ...   page blobs, one per page, in ascending page number order
+ * ...     16*k  page index: k entries of { pageNo: u32, offset: u64, length: u32 }
+ * end-12  8     indexOffset: u64
+ * end-4   4     crc32: IEEE CRC-32 of every byte before this field
+ * ```
+ *
+ * All integers are little-endian, `k` is the header's `pageCount`, and index offsets are file offsets. The header comes
+ * first so that a store can validate an upload from its first bytes; the index comes last so that a reader can fetch
+ * one page with a suffix range request for the footer, one for the index and one for the page. Pages are compressed
+ * one at a time for the same reason.
+ *
+ * Encoding is deterministic: the same header and pages produce the same bytes with the same compressor. A writer that
+ * crashes after spooling a segment but before recording it re-encodes the same frames on restart, and the store
+ * recognizes the repeated upload by its content hash. Headers therefore carry no timestamps, and keys are written in
+ * a fixed order.
+ *
+ * This module does not compress. Callers pass a `PageCompression`; `NO_COMPRESSION` is built in, and the Node entry
+ * point provides zstd from `node:zlib`. Reading headers and indexes never needs the compressor, so a Worker can
+ * validate uploads without zstd.
+ */
+
+const MAGIC = new TextEncoder().encode("PISYNC01");
+const MAGIC_BYTES = 8;
+const PREFIX_BYTES = MAGIC_BYTES + 4;
+const MAX_HEADER_BYTES = 65536;
+const INDEX_ENTRY_BYTES = 16;
+const FOOTER_BYTES = 12;
+
+/** Largest encoded segment. A flush that would produce more becomes a base. */
+export const MAX_SEGMENT_BYTES = 32 * 1024 * 1024;
+
+/** How page blobs are stored. */
+export interface PageCompression {
+	readonly name: "zstd" | "none";
+	compress(page: Uint8Array): Uint8Array;
+	decompress(blob: Uint8Array, pageSize: number): Uint8Array;
+}
+
+/** Pages stored as they are. */
+export const NO_COMPRESSION: PageCompression = {
+	name: "none",
+	compress: (page) => page,
+	decompress: (blob) => blob,
+};
+
+type CommonHeader = {
+	readonly version: 1;
+	readonly replicaId: string;
+	readonly epoch: number;
+	readonly pageSize: number;
+	readonly dbPageCount: number;
+	readonly pageCount: number;
+	readonly compression: "zstd" | "none";
+};
+
+/** The newest version of every page changed by transactions `fromSeq..toSeq`. Checksums are 16 hex digits. */
+export type SegmentHeader = CommonHeader & {
+	readonly kind: "segment";
+	readonly fromSeq: number;
+	readonly toSeq: number;
+	readonly preChecksum: string;
+	readonly postChecksum: string;
+};
+
+/** Every page of the database at `seq`. */
+export type BaseHeader = CommonHeader & {
+	readonly kind: "base";
+	readonly seq: number;
+	readonly checksum: string;
+};
+
+export type ContainerHeader = SegmentHeader | BaseHeader;
+
+/** A header before encoding: `pageCount` and `compression` follow from the pages and the compressor. */
+export type ContainerHeaderInput =
+	| Omit<SegmentHeader, "pageCount" | "compression">
+	| Omit<BaseHeader, "pageCount" | "compression">;
+
+/** Where one page's blob lies in an encoded container. */
+export type PageIndexEntry = { readonly pageNo: number; readonly offset: number; readonly length: number };
+
+/** A container read from its complete bytes. */
+export type DecodedContainer = {
+	readonly header: ContainerHeader;
+	readonly index: readonly PageIndexEntry[];
+	/** Decompress every page, by page number. Throws if `compression` does not match the header. */
+	pages(compression: PageCompression): Map<number, Uint8Array>;
+};
+
+/** Thrown for bytes that are not a valid container. */
+export class ReplicationFormatError extends Error {
+	constructor(message: string) {
+		super(message);
+		this.name = "ReplicationFormatError";
+	}
+}
+
+/** Encode a segment or base. Pages may be given in any order; each must be exactly `pageSize` bytes. */
+export function encodeContainer(
+	input: ContainerHeaderInput,
+	pages: ReadonlyMap<number, Uint8Array>,
+	compression: PageCompression,
+): Uint8Array {
+	const header = validateHeader({ ...input, pageCount: pages.size, compression: compression.name });
+	const pageNumbers = [...pages.keys()].sort((a, b) => a - b);
+	if (header.kind === "base" && pageNumbers.some((pageNo, index) => pageNo !== index + 1)) {
+		throw new ReplicationFormatError(`A base must contain every page from 1 to ${header.dbPageCount}`);
+	}
+	const blobs: Uint8Array[] = [];
+	for (const pageNo of pageNumbers) {
+		const page = pages.get(pageNo);
+		if (page?.length !== header.pageSize) {
+			throw new ReplicationFormatError(`Page ${pageNo} is ${page?.length} bytes, not ${header.pageSize}`);
+		}
+		if (pageNo > header.dbPageCount) {
+			throw new ReplicationFormatError(`Page ${pageNo} is outside a database of ${header.dbPageCount} pages`);
+		}
+		blobs.push(compression.compress(page));
+	}
+
+	const headerBytes = new TextEncoder().encode(JSON.stringify(canonicalHeader(header)));
+	if (headerBytes.length > MAX_HEADER_BYTES) throw new ReplicationFormatError("Container header is too large");
+	const blobBytes = blobs.reduce((total, blob) => total + blob.length, 0);
+	const indexOffset = PREFIX_BYTES + headerBytes.length + blobBytes;
+	const total = indexOffset + pageNumbers.length * INDEX_ENTRY_BYTES + FOOTER_BYTES;
+	const bytes = new Uint8Array(total);
+	const view = new DataView(bytes.buffer);
+	bytes.set(MAGIC, 0);
+	view.setUint32(MAGIC_BYTES, headerBytes.length, true);
+	bytes.set(headerBytes, PREFIX_BYTES);
+
+	let offset = PREFIX_BYTES + headerBytes.length;
+	let entry = indexOffset;
+	pageNumbers.forEach((pageNo, index) => {
+		const blob = blobs[index] ?? new Uint8Array();
+		bytes.set(blob, offset);
+		view.setUint32(entry, pageNo, true);
+		view.setBigUint64(entry + 4, BigInt(offset), true);
+		view.setUint32(entry + 12, blob.length, true);
+		offset += blob.length;
+		entry += INDEX_ENTRY_BYTES;
+	});
+	view.setBigUint64(total - FOOTER_BYTES, BigInt(indexOffset), true);
+	view.setUint32(total - 4, crc32(bytes.subarray(0, total - 4)), true);
+	return bytes;
+}
+
+/**
+ * Read and validate the header from the first bytes of a container. `prefix` may be the whole object or just its
+ * beginning; it must contain at least the magic, the length and the header. This is what a store uses to validate an
+ * upload before accepting it.
+ */
+export function decodeContainerHeader(prefix: Uint8Array): ContainerHeader {
+	if (prefix.length < PREFIX_BYTES || !MAGIC.every((byte, index) => prefix[index] === byte)) {
+		throw new ReplicationFormatError("Not a replication container");
+	}
+	const headerLength = new DataView(prefix.buffer, prefix.byteOffset, prefix.byteLength).getUint32(MAGIC_BYTES, true);
+	if (headerLength > MAX_HEADER_BYTES) throw new ReplicationFormatError("Container header is too large");
+	if (prefix.length < PREFIX_BYTES + headerLength) throw new ReplicationFormatError("Container header is truncated");
+	let json: unknown;
+	try {
+		json = JSON.parse(
+			new TextDecoder("utf-8", { fatal: true }).decode(prefix.subarray(PREFIX_BYTES, PREFIX_BYTES + headerLength)),
+		);
+	} catch {
+		throw new ReplicationFormatError("Container header is not valid JSON");
+	}
+	return validateHeader(json);
+}
+
+/** Decode a complete container, checking its CRC, header and index. Pages are decompressed on request. */
+export function decodeContainer(bytes: Uint8Array): DecodedContainer {
+	if (bytes.length < PREFIX_BYTES + FOOTER_BYTES) throw new ReplicationFormatError("Container is truncated");
+	const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength);
+	if (crc32(bytes.subarray(0, bytes.length - 4)) !== view.getUint32(bytes.length - 4, true)) {
+		throw new ReplicationFormatError("Container CRC does not match");
+	}
+	const header = decodeContainerHeader(bytes);
+	const blobsStart = PREFIX_BYTES + view.getUint32(MAGIC_BYTES, true);
+	const indexOffset = Number(view.getBigUint64(bytes.length - FOOTER_BYTES, true));
+	const index = decodeIndex(bytes, indexOffset, bytes.length - FOOTER_BYTES, header.pageCount, blobsStart);
+	if (header.kind === "base" && index.some((entry, position) => entry.pageNo !== position + 1)) {
+		throw new ReplicationFormatError(`A base must contain every page from 1 to ${header.dbPageCount}`);
+	}
+	if (index.some((entry) => entry.pageNo > header.dbPageCount)) {
+		throw new ReplicationFormatError("Container has a page outside the database");
+	}
+	return {
+		header,
+		index,
+		pages(compression) {
+			if (compression.name !== header.compression) {
+				throw new ReplicationFormatError(`Container uses ${header.compression}, not ${compression.name}`);
+			}
+			const pages = new Map<number, Uint8Array>();
+			for (const entry of index) {
+				const page = compression.decompress(
+					bytes.subarray(entry.offset, entry.offset + entry.length),
+					header.pageSize,
+				);
+				if (page.length !== header.pageSize) {
+					throw new ReplicationFormatError(`Page ${entry.pageNo} decompressed to ${page.length} bytes`);
+				}
+				pages.set(entry.pageNo, page);
+			}
+			return pages;
+		},
+	};
+}
+
+function decodeIndex(
+	bytes: Uint8Array,
+	indexOffset: number,
+	indexEnd: number,
+	pageCount: number,
+	blobsStart: number,
+): PageIndexEntry[] {
+	if (indexOffset < blobsStart || indexEnd - indexOffset !== pageCount * INDEX_ENTRY_BYTES) {
+		throw new ReplicationFormatError("Container index does not match its page count");
+	}
+	const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength);
+	const index: PageIndexEntry[] = [];
+	let expectedOffset = blobsStart;
+	for (let entry = indexOffset; entry < indexEnd; entry += INDEX_ENTRY_BYTES) {
+		const pageNo = view.getUint32(entry, true);
+		const offset = Number(view.getBigUint64(entry + 4, true));
+		const length = view.getUint32(entry + 12, true);
+		const previous = index.at(-1);
+		if (pageNo === 0 || (previous !== undefined && pageNo <= previous.pageNo)) {
+			throw new ReplicationFormatError("Container index is not in ascending page order");
+		}
+		if (offset !== expectedOffset || length === 0) {
+			throw new ReplicationFormatError("Container index does not describe contiguous page blobs");
+		}
+		expectedOffset = offset + length;
+		index.push({ pageNo, offset, length });
+	}
+	if (expectedOffset !== indexOffset) throw new ReplicationFormatError("Container page blobs do not end at the index");
+	return index;
+}
+
+function canonicalHeader(header: ContainerHeader): Record<string, unknown> {
+	const common = { version: header.version, kind: header.kind, replicaId: header.replicaId, epoch: header.epoch };
+	const sizes = { pageSize: header.pageSize, dbPageCount: header.dbPageCount };
+	const tail = { pageCount: header.pageCount, compression: header.compression };
+	return header.kind === "segment"
+		? {
+				...common,
+				fromSeq: header.fromSeq,
+				toSeq: header.toSeq,
+				...sizes,
+				preChecksum: header.preChecksum,
+				postChecksum: header.postChecksum,
+				...tail,
+			}
+		: { ...common, seq: header.seq, ...sizes, checksum: header.checksum, ...tail };
+}
+
+const REPLICA_ID = /^[A-Za-z0-9._-]{1,128}$/;
+const CHECKSUM = /^[0-9a-f]{16}$/;
+
+/** Check a parsed header against the format rules. Unknown fields are ignored and dropped. */
+export function validateHeader(value: unknown): ContainerHeader {
+	if (typeof value !== "object" || value === null || Array.isArray(value)) {
+		throw new ReplicationFormatError("Container header is not an object");
+	}
+	const field = (name: string): unknown => (value as Record<string, unknown>)[name];
+	const integer = (name: string, minimum: number): number => {
+		const number = field(name);
+		if (typeof number !== "number" || !Number.isSafeInteger(number) || number < minimum) {
+			throw new ReplicationFormatError(`Container header field ${name} must be an integer of at least ${minimum}`);
+		}
+		return number;
+	};
+	const checksum = (name: string): string => {
+		const text = field(name);
+		if (typeof text !== "string" || !CHECKSUM.test(text)) {
+			throw new ReplicationFormatError(`Container header field ${name} must be 16 lowercase hex digits`);
+		}
+		return text;
+	};
+
+	if (field("version") !== 1) throw new ReplicationFormatError(`Unsupported container version ${field("version")}`);
+	const replicaId = field("replicaId");
+	if (typeof replicaId !== "string" || !REPLICA_ID.test(replicaId)) {
+		throw new ReplicationFormatError("Container header field replicaId is not a valid replica id");
+	}
+	const compression = field("compression");
+	if (compression !== "zstd" && compression !== "none") {
+		throw new ReplicationFormatError(`Unsupported page compression ${JSON.stringify(compression)}`);
+	}
+	const pageSize = integer("pageSize", 512);
+	if (pageSize > 65536 || (pageSize & (pageSize - 1)) !== 0) {
+		throw new ReplicationFormatError(`Invalid page size ${pageSize}`);
+	}
+	const common: CommonHeader = {
+		version: 1,
+		replicaId,
+		epoch: integer("epoch", 1),
+		pageSize,
+		dbPageCount: integer("dbPageCount", 0),
+		pageCount: integer("pageCount", 0),
+		compression,
+	};
+	if (common.pageCount > common.dbPageCount) {
+		throw new ReplicationFormatError("Container has more pages than the database");
+	}
+
+	const kind = field("kind");
+	if (kind === "segment") {
+		const fromSeq = integer("fromSeq", 1);
+		const toSeq = integer("toSeq", fromSeq);
+		if (common.pageCount < 1) throw new ReplicationFormatError("A segment must contain at least one page");
+		return {
+			...common,
+			kind,
+			fromSeq,
+			toSeq,
+			preChecksum: checksum("preChecksum"),
+			postChecksum: checksum("postChecksum"),
+		};
+	}
+	if (kind === "base") {
+		if (common.pageCount !== common.dbPageCount) {
+			throw new ReplicationFormatError("A base must contain every page of the database");
+		}
+		return { ...common, kind, seq: integer("seq", 0), checksum: checksum("checksum") };
+	}
+	throw new ReplicationFormatError(`Unknown container kind ${JSON.stringify(kind)}`);
+}
+
+let crcTable: Uint32Array | undefined;
+
+/** IEEE CRC-32, as in zlib. */
+export function crc32(bytes: Uint8Array): number {
+	if (crcTable === undefined) {
+		crcTable = new Uint32Array(256);
+		for (let n = 0; n < 256; n++) {
+			let c = n;
+			for (let bit = 0; bit < 8; bit++) c = c & 1 ? 0xedb88320 ^ (c >>> 1) : c >>> 1;
+			crcTable[n] = c >>> 0;
+		}
+	}
+	let crc = 0xffffffff;
+	for (const byte of bytes) crc = (crcTable[(crc ^ byte) & 0xff] ?? 0) ^ (crc >>> 8);
+	return (crc ^ 0xffffffff) >>> 0;
+}

packages/chord/src/sqlite/replication/index.ts

@@ -0,0 +1,38 @@
+export { formatChecksum, hashPages, PageHashes, pageHash, parseChecksum } from "./checksum.ts";
+export {
+	type BaseHeader,
+	type ContainerHeader,
+	type ContainerHeaderInput,
+	crc32,
+	type DecodedContainer,
+	decodeContainer,
+	decodeContainerHeader,
+	encodeContainer,
+	MAX_SEGMENT_BYTES,
+	NO_COMPRESSION,
+	type PageCompression,
+	type PageIndexEntry,
+	ReplicationFormatError,
+	type SegmentHeader,
+	validateHeader,
+} from "./container.ts";
+export {
+	baseKey,
+	decodeStartMarker,
+	encodeStartMarker,
+	type ObjectKey,
+	parseObjectKey,
+	type StartMarker,
+	segmentKey,
+	startMarkerKey,
+} from "./keys.ts";
+export {
+	parseWalHeader,
+	scanWal,
+	WAL_FRAME_HEADER_BYTES,
+	WAL_HEADER_BYTES,
+	type WalHeader,
+	type WalPosition,
+	type WalScan,
+	walStart,
+} from "./wal.ts";

packages/chord/src/sqlite/replication/keys.ts

@@ -0,0 +1,93 @@
+/**
+ * Object keys within one replica's prefix, and start markers.
+ *
+ * ```text
+ * epochs/{epoch}/start                    start marker, JSON
+ * epochs/{epoch}/segments/{fromSeq}       one segment, created only if absent
+ * bases/{seq}-{epoch}                     one base
+ * ```
+ *
+ * Numbers are decimal, zero-padded to 16 digits, so lexical order is numeric order; 16 digits hold every safe integer.
+ * Keys are relative: the store decides the prefix, for example `orgs/{org}/users/{user}/sessions/{replicaId}/`.
+ */
+
+const DIGITS = 16;
+const NUMBER = `(\\d{${DIGITS}})`;
+const SEGMENT = new RegExp(`^epochs/${NUMBER}/segments/${NUMBER}$`);
+const START = new RegExp(`^epochs/${NUMBER}/start$`);
+const BASE = new RegExp(`^bases/${NUMBER}-${NUMBER}$`);
+
+export type ObjectKey =
+	| { readonly type: "segment"; readonly epoch: number; readonly fromSeq: number }
+	| { readonly type: "base"; readonly seq: number; readonly epoch: number }
+	| { readonly type: "start"; readonly epoch: number };
+
+/** The first seq an epoch owns; its existence fences the epoch before it. Written only by the lease authority. */
+export type StartMarker = { readonly version: 1; readonly epoch: number; readonly startSeq: number };
+
+function pad(name: string, value: number, minimum: number): string {
+	if (!Number.isSafeInteger(value) || value < minimum) {
+		throw new Error(`${name} must be an integer of at least ${minimum}, got ${value}`);
+	}
+	return String(value).padStart(DIGITS, "0");
+}
+
+export function segmentKey(epoch: number, fromSeq: number): string {
+	return `epochs/${pad("epoch", epoch, 1)}/segments/${pad("fromSeq", fromSeq, 1)}`;
+}
+
+export function baseKey(seq: number, epoch: number): string {
+	return `bases/${pad("seq", seq, 0)}-${pad("epoch", epoch, 1)}`;
+}
+
+export function startMarkerKey(epoch: number): string {
+	return `epochs/${pad("epoch", epoch, 1)}/start`;
+}
+
+/** Parse a key written by the functions above. Anything else, including unpadded numbers, is `undefined`. */
+export function parseObjectKey(key: string): ObjectKey | undefined {
+	const number = (text: string | undefined): number => Number(text);
+	let match = SEGMENT.exec(key);
+	if (match !== null) {
+		const [epoch, fromSeq] = [number(match[1]), number(match[2])];
+		return epoch >= 1 && fromSeq >= 1 && Number.isSafeInteger(fromSeq) && Number.isSafeInteger(epoch)
+			? { type: "segment", epoch, fromSeq }
+			: undefined;
+	}
+	match = BASE.exec(key);
+	if (match !== null) {
+		const [seq, epoch] = [number(match[1]), number(match[2])];
+		return epoch >= 1 && Number.isSafeInteger(seq) && Number.isSafeInteger(epoch)
+			? { type: "base", seq, epoch }
+			: undefined;
+	}
+	match = START.exec(key);
+	if (match !== null) {
+		const epoch = number(match[1]);
+		return epoch >= 1 && Number.isSafeInteger(epoch) ? { type: "start", epoch } : undefined;
+	}
+	return undefined;
+}
+
+export function encodeStartMarker(marker: StartMarker): Uint8Array {
+	pad("epoch", marker.epoch, 1);
+	pad("startSeq", marker.startSeq, 1);
+	return new TextEncoder().encode(JSON.stringify({ version: 1, epoch: marker.epoch, startSeq: marker.startSeq }));
+}
+
+export function decodeStartMarker(bytes: Uint8Array): StartMarker {
+	let value: unknown;
+	try {
+		value = JSON.parse(new TextDecoder("utf-8", { fatal: true }).decode(bytes));
+	} catch {
+		throw new Error("Start marker is not valid JSON");
+	}
+	const marker = value as Partial<Record<keyof StartMarker, unknown>> | null;
+	const valid = (field: unknown): field is number =>
+		typeof field === "number" && Number.isSafeInteger(field) && field >= 1;
+	if (typeof marker !== "object" || marker === null || marker.version !== 1) {
+		throw new Error("Start marker has an unsupported version");
+	}
+	if (!valid(marker.epoch) || !valid(marker.startSeq)) throw new Error("Start marker has an invalid epoch or seq");
+	return { version: 1, epoch: marker.epoch, startSeq: marker.startSeq };
+}

packages/chord/src/sqlite/replication/wal.ts

@@ -0,0 +1,155 @@
+/**
+ * Reading SQLite's write-ahead log (WAL) from outside SQLite.
+ *
+ * A WAL file is a 32-byte header followed by frames, each a 24-byte header plus one page. The last frame of a
+ * transaction is its commit frame and records the database size in pages after the commit. Every frame carries the
+ * header's two salts and a checksum that covers the header and every frame before it, so a reader can tell where the
+ * valid log ends: at the first frame whose salts or checksum do not match. Frames past that point are either a
+ * transaction still being written, a torn write, or leftovers from before the WAL was last restarted; SQLite
+ * rewrites the file from the start after a checkpoint, with new salts, so old frames fail the salt check.
+ *
+ * `scanWal` reads forward from a known position and returns the committed transactions it found, as the newest version
+ * of each page they wrote. It stops at the last valid commit frame, so a caller that remembers the returned `end`
+ * position can call it again later and continue exactly where it stopped. It reads bytes it is given and never
+ * touches files; the Node layer reads the WAL and passes the bytes in.
+ *
+ * Format reference: https://www.sqlite.org/fileformat2.html#walformat
+ */
+
+export const WAL_HEADER_BYTES = 32;
+export const WAL_FRAME_HEADER_BYTES = 24;
+
+const WAL_MAGIC = 0x377f0682;
+const WAL_FORMAT_VERSION = 3007000;
+
+/** The parsed header of a WAL file. */
+export type WalHeader = {
+	/** Checksums read the log as big-endian 32-bit words (magic `0x377f0683`) rather than little-endian ones. */
+	readonly bigEndianChecksums: boolean;
+	readonly pageSize: number;
+	readonly checkpointSequence: number;
+	readonly salt1: number;
+	readonly salt2: number;
+	readonly checksum1: number;
+	readonly checksum2: number;
+};
+
+/**
+ * A point in a WAL between frames: the file offset of the next frame and the cumulative checksum of everything before
+ * it. A position is only meaningful together with the header (and so the salts) it was read under.
+ */
+export type WalPosition = {
+	readonly offset: number;
+	readonly checksum1: number;
+	readonly checksum2: number;
+};
+
+/** What `scanWal` found between a start position and the last valid commit frame after it. */
+export type WalScan = {
+	/** Committed transactions found. 0 when nothing new was committed. */
+	readonly transactions: number;
+	/**
+	 * The newest version of every page those transactions wrote, at most `dbPageCount`, by page number. Values are
+	 * views into the bytes passed to `scanWal`.
+	 */
+	readonly pages: ReadonlyMap<number, Uint8Array>;
+	/** Database size in pages after the last transaction, or 0 when there was none. */
+	readonly dbPageCount: number;
+	/** The position just after the last commit frame, or the start position when there was none. */
+	readonly end: WalPosition;
+};
+
+/**
+ * Parse a WAL header. Returns `undefined` when `bytes` is shorter than a header or the header is not valid (wrong
+ * magic, version, page size or checksum), which is also how an empty or freshly truncated WAL looks.
+ */
+export function parseWalHeader(bytes: Uint8Array): WalHeader | undefined {
+	if (bytes.length < WAL_HEADER_BYTES) return undefined;
+	const view = new DataView(bytes.buffer, bytes.byteOffset, WAL_HEADER_BYTES);
+	const magic = view.getUint32(0);
+	if ((magic & ~1) !== WAL_MAGIC || view.getUint32(4) !== WAL_FORMAT_VERSION) return undefined;
+	const pageSize = view.getUint32(8);
+	if (pageSize < 512 || pageSize > 65536 || (pageSize & (pageSize - 1)) !== 0) return undefined;
+	const bigEndianChecksums = (magic & 1) === 1;
+	const [checksum1, checksum2] = walChecksum(bytes.subarray(0, 24), bigEndianChecksums, 0, 0);
+	if (checksum1 !== view.getUint32(24) || checksum2 !== view.getUint32(28)) return undefined;
+	return {
+		bigEndianChecksums,
+		pageSize,
+		checkpointSequence: view.getUint32(12),
+		salt1: view.getUint32(16),
+		salt2: view.getUint32(20),
+		checksum1,
+		checksum2,
+	};
+}
+
+/** The position of the first frame under `header`. */
+export function walStart(header: WalHeader): WalPosition {
+	return { offset: WAL_HEADER_BYTES, checksum1: header.checksum1, checksum2: header.checksum2 };
+}
+
+/**
+ * Scan committed transactions from `from` to the last valid commit frame.
+ *
+ * `bytes` holds the WAL file from offset `from.offset` onwards, usually to its current end. Frames after the last
+ * valid commit frame are ignored; a later scan from `end` reads them once their transaction has committed.
+ *
+ * Pages are kept by their newest version, and pages beyond the final database size are dropped, which is exactly how
+ * SQLite itself reads the log: the newest frame of each page wins, and the size from the last commit applies.
+ */
+export function scanWal(header: WalHeader, bytes: Uint8Array, from: WalPosition): WalScan {
+	const frameBytes = WAL_FRAME_HEADER_BYTES + header.pageSize;
+	const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength);
+	const committed = new Map<number, Uint8Array>();
+	const pending = new Map<number, Uint8Array>();
+	let transactions = 0;
+	let dbPageCount = 0;
+	let end = from;
+	let checksum1 = from.checksum1;
+	let checksum2 = from.checksum2;
+
+	for (let local = 0; local + frameBytes <= bytes.length; local += frameBytes) {
+		const pageNo = view.getUint32(local);
+		const commitSize = view.getUint32(local + 4);
+		if (pageNo === 0 || view.getUint32(local + 8) !== header.salt1 || view.getUint32(local + 12) !== header.salt2) {
+			break;
+		}
+		const page = bytes.subarray(local + WAL_FRAME_HEADER_BYTES, local + frameBytes);
+		[checksum1, checksum2] = walChecksum(
+			bytes.subarray(local, local + 8),
+			header.bigEndianChecksums,
+			checksum1,
+			checksum2,
+		);
+		[checksum1, checksum2] = walChecksum(page, header.bigEndianChecksums, checksum1, checksum2);
+		if (checksum1 !== view.getUint32(local + 16) || checksum2 !== view.getUint32(local + 20)) break;
+
+		pending.set(pageNo, page);
+		if (commitSize === 0) continue;
+		for (const [number, data] of pending) committed.set(number, data);
+		pending.clear();
+		transactions++;
+		dbPageCount = commitSize;
+		end = { offset: from.offset + local + frameBytes, checksum1, checksum2 };
+	}
+
+	for (const pageNo of committed.keys()) {
+		if (pageNo > dbPageCount) committed.delete(pageNo);
+	}
+	return { transactions, pages: committed, dbPageCount, end };
+}
+
+/**
+ * SQLite's WAL checksum: Fibonacci-weighted sums over pairs of 32-bit words, continued from `s1` and `s2`. `bytes` must
+ * be a multiple of 8 long, which headers, frame headers and pages all are.
+ */
+function walChecksum(bytes: Uint8Array, bigEndian: boolean, s1: number, s2: number): [number, number] {
+	const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength);
+	const littleEndian = !bigEndian;
+	for (let offset = 0; offset < bytes.length; offset += 8) {
+		s1 = (s1 + view.getUint32(offset, littleEndian) + s2) >>> 0;
+		s2 = (s2 + view.getUint32(offset + 4, littleEndian) + s1) >>> 0;
+	}
+	return [s1, s2];
+}

packages/chord/src/sqlite/replication/zstd.ts

@@ -0,0 +1,12 @@
+import { constants, zstdCompressSync, zstdDecompressSync } from "node:zlib";
+import type { PageCompression } from "./container.ts";
+
+/** Higher levels gain under one point on p2 sessions and cost several times the CPU. */
+const LEVEL = 3;
+
+/** Per-page zstd from `node:zlib` (Node 22.15 and later). Node only. */
+export const ZSTD_COMPRESSION: PageCompression = {
+	name: "zstd",
+	compress: (page) => zstdCompressSync(page, { params: { [constants.ZSTD_c_compressionLevel]: LEVEL } }),
+	decompress: (blob, pageSize) => zstdDecompressSync(blob, { maxOutputLength: pageSize }),
+};

packages/chord/test/sqlite/replication-checksum.test.ts

@@ -0,0 +1,108 @@
+import { describe, expect, it } from "vitest";
+import {
+	formatChecksum,
+	hashPages,
+	PageHashes,
+	pageHash,
+	parseChecksum,
+} from "../../src/sqlite/replication/checksum.ts";
+
+const PAGE = 512;
+const page = (fill: number) => new Uint8Array(PAGE).fill(fill);
+
+function image(fills: readonly number[]): Uint8Array {
+	const bytes = new Uint8Array(fills.length * PAGE);
+	for (const [index, fill] of fills.entries()) bytes.fill(fill, index * PAGE, (index + 1) * PAGE);
+	return bytes;
+}
+
+describe("page hashes and database checksum", () => {
+	it("hashes the page number together with the contents", async () => {
+		const contents = page(7);
+		expect(await pageHash(1, contents)).not.toBe(await pageHash(2, contents));
+		expect(await pageHash(1, contents)).toBe(await pageHash(1, page(7)));
+		// The first 8 bytes of SHA-256(u32be pageNo || page), read big-endian.
+		const input = new Uint8Array([0, 0, 0, 1, ...contents]);
+		const digest = new DataView(await crypto.subtle.digest("SHA-256", input));
+		expect(await pageHash(1, contents)).toBe(digest.getBigUint64(0));
+	});
+
+	it("formats checksums as 16 hex digits", () => {
+		expect(formatChecksum(0n)).toBe("0000000000000000");
+		expect(formatChecksum(0xfedcba9876543210n)).toBe("fedcba9876543210");
+		expect(parseChecksum("00000000000000ff")).toBe(255n);
+		for (const invalid of ["ff", "FEDCBA9876543210", "0000000000000000 ", "zzzzzzzzzzzzzzzz"]) {
+			expect(() => parseChecksum(invalid)).toThrow("Invalid database checksum");
+		}
+	});
+
+	it("updates incrementally to the checksum of the whole database", async () => {
+		const hashes = PageHashes.empty();
+		expect(hashes.checksum).toBe(0n);
+		hashes.apply(await hashPages(new Map([1, 2, 3, 4].map((pageNo) => [pageNo, page(pageNo)]))), 4);
+		expect(hashes.checksum).toBe((await PageHashes.fromDatabase(image([1, 2, 3, 4]), PAGE)).checksum);
+
+		// Change one page and grow by one.
+		hashes.apply(
+			await hashPages(
+				new Map([
+					[2, page(9)],
+					[5, page(5)],
+				]),
+			),
+			5,
+		);
+		expect(hashes.checksum).toBe((await PageHashes.fromDatabase(image([1, 9, 3, 4, 5]), PAGE)).checksum);
+
+		// Shrink, changing a remaining page in the same step.
+		hashes.apply(await hashPages(new Map([[1, page(8)]])), 2);
+		expect(hashes.pageCount).toBe(2);
+		expect(hashes.hash(3)).toBeUndefined();
+		expect(hashes.checksum).toBe((await PageHashes.fromDatabase(image([8, 9]), PAGE)).checksum);
+
+		// Grow again into slots that held hashes before the shrink.
+		hashes.apply(
+			await hashPages(
+				new Map([
+					[3, page(3)],
+					[4, page(6)],
+				]),
+			),
+			4,
+		);
+		expect(hashes.checksum).toBe((await PageHashes.fromDatabase(image([8, 9, 3, 6]), PAGE)).checksum);
+	});
+
+	it("refuses pages it cannot account for and stays unchanged", async () => {
+		const hashes = await PageHashes.fromDatabase(image([1, 2]), PAGE);
+		const before = hashes.checksum;
+		await expect(async () => hashes.apply(await hashPages(new Map([[4, page(4)]])), 4)).rejects.toThrow(
+			"Database grew to 4 pages without page 3",
+		);
+		await expect(async () => hashes.apply(await hashPages(new Map([[3, page(3)]])), 2)).rejects.toThrow(
+			"Page 3 is outside a database of 2 pages",
+		);
+		expect(hashes.checksum).toBe(before);
+		expect(hashes.pageCount).toBe(2);
+	});
+
+	it("round-trips the sidecar file", async () => {
+		const hashes = await PageHashes.fromDatabase(image([1, 2, 3]), PAGE);
+		const encoded = hashes.encode();
+		expect(encoded.length).toBe(24);
+		const view = new DataView(encoded.buffer);
+		expect(view.getBigUint64(8, true)).toBe(hashes.hash(2));
+		const decoded = PageHashes.decode(encoded);
+		expect(decoded.checksum).toBe(hashes.checksum);
+		expect(decoded.pageCount).toBe(3);
+		expect(() => PageHashes.decode(new Uint8Array(7))).toThrow("not a multiple of 8");
+
+		const copy = hashes.clone();
+		copy.apply(new Map([[1, 1n]]), 1);
+		expect(hashes.pageCount).toBe(3);
+	});
+
+	it("refuses a database that is not whole pages", async () => {
+		await expect(PageHashes.fromDatabase(new Uint8Array(PAGE + 1), PAGE)).rejects.toThrow("whole number");
+	});
+});

packages/chord/test/sqlite/replication-container.test.ts

@@ -0,0 +1,173 @@
+import { crc32 as zlibCrc32 } from "node:zlib";
+import { describe, expect, it } from "vitest";
+import {
+	type ContainerHeaderInput,
+	crc32,
+	decodeContainer,
+	decodeContainerHeader,
+	encodeContainer,
+	NO_COMPRESSION,
+	ReplicationFormatError,
+	validateHeader,
+} from "../../src/sqlite/replication/container.ts";
+import { ZSTD_COMPRESSION } from "../../src/sqlite/replication/zstd.ts";
+
+const PAGE = 4096;
+
+function sqliteLikePage(seed: number): Uint8Array {
+	// Mostly structured text with some noise, roughly like a B-tree leaf of transcript rows.
+	const text = new TextEncoder().encode(`{"role":"assistant","content":"row ${seed} `.repeat(200));
+	const page = new Uint8Array(PAGE);
+	page.set(text.subarray(0, PAGE));
+	for (let index = 0; index < 64; index++) page[(index * 61 + seed) % PAGE] = (index * seed) & 0xff;
+	return page;
+}
+
+const segment: ContainerHeaderInput = {
+	version: 1,
+	kind: "segment",
+	replicaId: "82c89986-434b-4171-9c85-d1ee6a847a00",
+	epoch: 2,
+	fromSeq: 121,
+	toSeq: 133,
+	pageSize: PAGE,
+	dbPageCount: 622,
+	preChecksum: "9f3a0c71d2e45b80",
+	postChecksum: "41b7e2a90c3d6f15",
+};
+
+const pages = new Map([
+	[12, sqliteLikePage(12)],
+	[1, sqliteLikePage(1)],
+	[600, sqliteLikePage(600)],
+]);
+
+describe("segment and base container", () => {
+	it("round-trips segments with and without compression", () => {
+		for (const compression of [NO_COMPRESSION, ZSTD_COMPRESSION]) {
+			const bytes = encodeContainer(segment, pages, compression);
+			const decoded = decodeContainer(bytes);
+			expect(decoded.header).toEqual({ ...segment, pageCount: 3, compression: compression.name });
+			expect(decoded.index.map((entry) => entry.pageNo)).toEqual([1, 12, 600]);
+			const restored = decoded.pages(compression);
+			expect([...restored.keys()]).toEqual([1, 12, 600]);
+			for (const [pageNo, page] of pages) expect(Buffer.from(restored.get(pageNo) ?? [])).toEqual(Buffer.from(page));
+		}
+		const compressed = encodeContainer(segment, pages, ZSTD_COMPRESSION);
+		expect(compressed.length).toBeLessThan(encodeContainer(segment, pages, NO_COMPRESSION).length / 2);
+	});
+
+	it("encodes deterministically, with the header first and keys in a fixed order", () => {
+		const reordered = { ...segment, postChecksum: segment.postChecksum, replicaId: segment.replicaId };
+		const first = encodeContainer(segment, pages, ZSTD_COMPRESSION);
+		const second = encodeContainer(reordered, new Map([...pages].reverse()), ZSTD_COMPRESSION);
+		expect(Buffer.from(second)).toEqual(Buffer.from(first));
+		const text = new TextDecoder().decode(first.subarray(0, 400));
+		expect(text.startsWith("PISYNC01")).toBe(true);
+		expect(text).toContain(
+			'{"version":1,"kind":"segment","replicaId":"82c89986-434b-4171-9c85-d1ee6a847a00","epoch":2,"fromSeq":121,' +
+				'"toSeq":133,"pageSize":4096,"dbPageCount":622,"preChecksum":"9f3a0c71d2e45b80",' +
+				'"postChecksum":"41b7e2a90c3d6f15","pageCount":3,"compression":"zstd"}',
+		);
+	});
+
+	it("reads the header from a prefix, as a store validating an upload would", () => {
+		const bytes = encodeContainer(segment, pages, ZSTD_COMPRESSION);
+		expect(decodeContainerHeader(bytes.subarray(0, 300))).toMatchObject({ kind: "segment", fromSeq: 121 });
+		expect(() => decodeContainerHeader(bytes.subarray(0, 40))).toThrow("Container header is truncated");
+		expect(() => decodeContainerHeader(new TextEncoder().encode("PK\u0003\u0004 not ours"))).toThrow(
+			"Not a replication container",
+		);
+	});
+
+	it("allows reading one page from the footer and index alone", () => {
+		const bytes = encodeContainer(segment, pages, ZSTD_COMPRESSION);
+		const view = new DataView(bytes.buffer);
+		const indexOffset = Number(view.getBigUint64(bytes.length - 12, true));
+		const index = bytes.subarray(indexOffset, bytes.length - 12);
+		const entries = new DataView(index.buffer, index.byteOffset, index.byteLength);
+		// Second entry: page 12.
+		expect(entries.getUint32(16, true)).toBe(12);
+		const offset = Number(entries.getBigUint64(20, true));
+		const length = entries.getUint32(28, true);
+		const page = ZSTD_COMPRESSION.decompress(bytes.subarray(offset, offset + length), PAGE);
+		expect(Buffer.from(page)).toEqual(Buffer.from(sqliteLikePage(12)));
+	});
+
+	it("round-trips a base, which must hold every page", () => {
+		const base: ContainerHeaderInput = {
+			version: 1,
+			kind: "base",
+			replicaId: "r1",
+			epoch: 1,
+			seq: 0,
+			pageSize: PAGE,
+			dbPageCount: 3,
+			checksum: "0000000000000abc",
+		};
+		const all = new Map([1, 2, 3].map((pageNo) => [pageNo, sqliteLikePage(pageNo)]));
+		const decoded = decodeContainer(encodeContainer(base, all, ZSTD_COMPRESSION));
+		expect(decoded.header).toEqual({ ...base, pageCount: 3, compression: "zstd" });
+		expect(decoded.pages(ZSTD_COMPRESSION).size).toBe(3);
+
+		all.delete(2);
+		expect(() => encodeContainer({ ...base, dbPageCount: 3 }, all, NO_COMPRESSION)).toThrow(
+			"A base must contain every page of the database",
+		);
+		expect(() => encodeContainer({ ...base, dbPageCount: 2 }, all, NO_COMPRESSION)).toThrow(
+			"A base must contain every page from 1 to 2",
+		);
+	});
+
+	it("rejects pages that do not fit the header", () => {
+		expect(() => encodeContainer(segment, new Map([[1, new Uint8Array(100)]]), NO_COMPRESSION)).toThrow(
+			"Page 1 is 100 bytes, not 4096",
+		);
+		expect(() => encodeContainer(segment, new Map([[623, sqliteLikePage(1)]]), NO_COMPRESSION)).toThrow(
+			"Page 623 is outside a database of 622 pages",
+		);
+		expect(() => encodeContainer(segment, new Map(), NO_COMPRESSION)).toThrow("at least one page");
+	});
+
+	it("detects corruption and a mismatched compressor", () => {
+		const bytes = encodeContainer(segment, pages, ZSTD_COMPRESSION);
+		for (const position of [3, 30, 400, bytes.length - 20, bytes.length - 2]) {
+			const corrupted = bytes.slice();
+			corrupted[position] = (corrupted[position] ?? 0) ^ 1;
+			expect(() => decodeContainer(corrupted)).toThrow(ReplicationFormatError);
+		}
+		expect(() => decodeContainer(bytes.subarray(0, bytes.length - 1))).toThrow("CRC");
+		expect(() => decodeContainer(bytes).pages(NO_COMPRESSION)).toThrow("Container uses zstd, not none");
+	});
+
+	it("validates header fields", () => {
+		const valid = { ...segment, pageCount: 1, compression: "none" };
+		expect(validateHeader({ ...valid, unknownField: true })).toEqual(valid);
+		const invalid: [Record<string, unknown>, string][] = [
+			[{ ...valid, version: 2 }, "Unsupported container version 2"],
+			[{ ...valid, kind: "snapshot" }, "Unknown container kind"],
+			[{ ...valid, replicaId: "has/slash" }, "not a valid replica id"],
+			[{ ...valid, replicaId: "" }, "not a valid replica id"],
+			[{ ...valid, epoch: 0 }, "epoch must be an integer of at least 1"],
+			[{ ...valid, fromSeq: 1.5 }, "fromSeq must be an integer"],
+			[{ ...valid, toSeq: 120 }, "toSeq must be an integer of at least 121"],
+			[{ ...valid, pageSize: 3000 }, "Invalid page size 3000"],
+			[{ ...valid, pageSize: 131072 }, "Invalid page size 131072"],
+			[{ ...valid, preChecksum: "ABCDEF0123456789" }, "preChecksum must be 16 lowercase hex digits"],
+			[{ ...valid, compression: "gzip" }, "Unsupported page compression"],
+			[{ ...valid, pageCount: 700 }, "more pages than the database"],
+			[{ ...valid, toSeq: Number.MAX_SAFE_INTEGER + 2 }, "toSeq must be an integer"],
+		];
+		for (const [header, message] of invalid) {
+			expect(() => validateHeader(header), message).toThrow(message);
+		}
+		expect(() => validateHeader(null)).toThrow("not an object");
+		expect(() => validateHeader([])).toThrow("not an object");
+	});
+
+	it("computes the same CRC-32 as zlib", () => {
+		for (const bytes of [new Uint8Array(), new TextEncoder().encode("PISYNC01"), sqliteLikePage(3)]) {
+			expect(crc32(bytes)).toBe(zlibCrc32(bytes));
+		}
+	});
+});

packages/chord/test/sqlite/replication-keys.test.ts

@@ -0,0 +1,75 @@
+import { readFile } from "node:fs/promises";
+import { dirname, resolve } from "node:path";
+import { fileURLToPath } from "node:url";
+import { describe, expect, it } from "vitest";
+import {
+	baseKey,
+	decodeStartMarker,
+	encodeStartMarker,
+	parseObjectKey,
+	segmentKey,
+	startMarkerKey,
+} from "../../src/sqlite/replication/keys.ts";
+
+describe("replication object keys", () => {
+	it("zero-pads numbers so lexical order is numeric order", () => {
+		expect(segmentKey(2, 121)).toBe("epochs/0000000000000002/segments/0000000000000121");
+		expect(baseKey(0, 1)).toBe("bases/0000000000000000-0000000000000001");
+		expect(startMarkerKey(8)).toBe("epochs/0000000000000008/start");
+		const keys = [9, 10, 100, 2, Number.MAX_SAFE_INTEGER].map((seq) => segmentKey(1, seq));
+		expect([...keys].sort()).toEqual([2, 9, 10, 100, Number.MAX_SAFE_INTEGER].map((seq) => segmentKey(1, seq)));
+	});
+
+	it("parses the keys it writes and nothing else", () => {
+		expect(parseObjectKey(segmentKey(2, 121))).toEqual({ type: "segment", epoch: 2, fromSeq: 121 });
+		expect(parseObjectKey(baseKey(120, 2))).toEqual({ type: "base", seq: 120, epoch: 2 });
+		expect(parseObjectKey(startMarkerKey(3))).toEqual({ type: "start", epoch: 3 });
+		for (const key of [
+			"epochs/2/segments/121",
+			"epochs/0000000000000000/start",
+			"epochs/0000000000000001/segments/0000000000000000",
+			"bases/0000000000000001-0000000000000000",
+			"epochs/9999999999999999/start",
+			"epochs/0000000000000001/start/extra",
+			"other",
+		]) {
+			expect(parseObjectKey(key), key).toBeUndefined();
+		}
+	});
+
+	it("refuses numbers a key cannot hold", () => {
+		expect(() => segmentKey(0, 1)).toThrow("epoch must be an integer of at least 1");
+		expect(() => segmentKey(1, 0)).toThrow("fromSeq must be an integer of at least 1");
+		expect(() => baseKey(-1, 1)).toThrow("seq must be an integer of at least 0");
+		expect(() => startMarkerKey(1.5)).toThrow("epoch must be an integer");
+	});
+
+	it("round-trips start markers", () => {
+		const bytes = encodeStartMarker({ version: 1, epoch: 8, startSeq: 121 });
+		expect(new TextDecoder().decode(bytes)).toBe('{"version":1,"epoch":8,"startSeq":121}');
+		expect(decodeStartMarker(bytes)).toEqual({ version: 1, epoch: 8, startSeq: 121 });
+		const invalid = ['{"version":2,"epoch":8,"startSeq":121}', '{"version":1,"epoch":0,"startSeq":1}', "null", "{"];
+		for (const text of invalid) expect(() => decodeStartMarker(new TextEncoder().encode(text)), text).toThrow();
+	});
+});
+
+describe("portable replication modules", () => {
+	it("do not import Node modules", async () => {
+		// The portable entry point runs in Workers and browsers; only the Node entry point may use node:*.
+		const directory = resolve(dirname(fileURLToPath(import.meta.url)), "../../src/sqlite/replication");
+		const visited = new Set<string>();
+		const violations: string[] = [];
+		const visit = async (file: string): Promise<void> => {
+			if (visited.has(file)) return;
+			visited.add(file);
+			const source = await readFile(file, "utf8");
+			for (const [, specifier = ""] of source.matchAll(/(?:import|export)[^;]*?from\s*["']([^"']+)["']/gu)) {
+				if (specifier.startsWith(".")) await visit(resolve(dirname(file), specifier));
+				else violations.push(`${file}: ${specifier}`);
+			}
+		};
+		await visit(resolve(directory, "index.ts"));
+		expect(violations).toEqual([]);
+		expect(visited.size).toBeGreaterThan(4);
+	});
+});

packages/chord/test/sqlite/replication-wal.test.ts

@@ -0,0 +1,228 @@
+import { mkdtempSync, readFileSync, rmSync, statSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { DatabaseSync } from "node:sqlite";
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import {
+	parseWalHeader,
+	scanWal,
+	WAL_FRAME_HEADER_BYTES,
+	WAL_HEADER_BYTES,
+	type WalHeader,
+	type WalPosition,
+	walStart,
+} from "../../src/sqlite/replication/wal.ts";
+
+/** Apply scanned pages to a database image the way a checkpoint does: newest page wins, then truncate. */
+function applyPages(image: Uint8Array, pages: ReadonlyMap<number, Uint8Array>, pageSize: number, dbPageCount: number) {
+	const result = new Uint8Array(dbPageCount * pageSize);
+	result.set(image.subarray(0, Math.min(image.length, result.length)));
+	for (const [pageNo, page] of pages) result.set(page, (pageNo - 1) * pageSize);
+	return result;
+}
+
+function header(wal: Uint8Array): WalHeader {
+	const parsed = parseWalHeader(wal);
+	if (parsed === undefined) throw new Error("WAL has no valid header");
+	return parsed;
+}
+
+function scanFrom(wal: Uint8Array, from: WalPosition) {
+	return scanWal(header(wal), wal.subarray(from.offset), from);
+}
+
+/** A WAL built by hand, for byte orders and shapes SQLite on this machine does not produce. */
+function syntheticWal(
+	bigEndian: boolean,
+	pageSize: number,
+	frames: readonly { pageNo: number; commitSize: number; fill: number }[],
+) {
+	const checksum = (bytes: Uint8Array, s: [number, number]): [number, number] => {
+		const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength);
+		let [s1, s2] = s;
+		for (let offset = 0; offset < bytes.length; offset += 8) {
+			s1 = (s1 + view.getUint32(offset, !bigEndian) + s2) >>> 0;
+			s2 = (s2 + view.getUint32(offset + 4, !bigEndian) + s1) >>> 0;
+		}
+		return [s1, s2];
+	};
+	const bytes = new Uint8Array(WAL_HEADER_BYTES + frames.length * (WAL_FRAME_HEADER_BYTES + pageSize));
+	const view = new DataView(bytes.buffer);
+	view.setUint32(0, bigEndian ? 0x377f0683 : 0x377f0682);
+	view.setUint32(4, 3007000);
+	view.setUint32(8, pageSize);
+	view.setUint32(12, 7);
+	view.setUint32(16, 0x11111111);
+	view.setUint32(20, 0x22222222);
+	let s = checksum(bytes.subarray(0, 24), [0, 0]);
+	view.setUint32(24, s[0]);
+	view.setUint32(28, s[1]);
+	frames.forEach((frame, index) => {
+		const offset = WAL_HEADER_BYTES + index * (WAL_FRAME_HEADER_BYTES + pageSize);
+		view.setUint32(offset, frame.pageNo);
+		view.setUint32(offset + 4, frame.commitSize);
+		view.setUint32(offset + 8, 0x11111111);
+		view.setUint32(offset + 12, 0x22222222);
+		bytes.fill(frame.fill, offset + WAL_FRAME_HEADER_BYTES, offset + WAL_FRAME_HEADER_BYTES + pageSize);
+		s = checksum(bytes.subarray(offset, offset + 8), s);
+		s = checksum(bytes.subarray(offset + WAL_FRAME_HEADER_BYTES, offset + WAL_FRAME_HEADER_BYTES + pageSize), s);
+		view.setUint32(offset + 16, s[0]);
+		view.setUint32(offset + 20, s[1]);
+	});
+	return bytes;
+}
+
+describe("WAL reader", () => {
+	let directory: string;
+	let path: string;
+	let database: DatabaseSync;
+
+	beforeEach(() => {
+		directory = mkdtempSync(join(tmpdir(), "chord-wal-"));
+		path = join(directory, "replica.sqlite");
+		database = new DatabaseSync(path);
+		database.exec("PRAGMA journal_mode = WAL; PRAGMA wal_autocheckpoint = 0");
+		database.exec("CREATE TABLE rows (id INTEGER PRIMARY KEY, value BLOB)");
+		database.exec("PRAGMA wal_checkpoint(TRUNCATE)");
+	});
+
+	afterEach(() => {
+		if (database.isOpen) database.close();
+		rmSync(directory, { recursive: true, force: true });
+	});
+
+	const insert = (count: number, size = 300) => {
+		database.exec("BEGIN");
+		for (let row = 0; row < count; row++)
+			database.prepare("INSERT INTO rows (value) VALUES (randomblob(?))").run(size);
+		database.exec("COMMIT");
+	};
+	const readWal = () => new Uint8Array(readFileSync(`${path}-wal`));
+
+	it("treats an empty or missing header as no WAL", () => {
+		expect(parseWalHeader(new Uint8Array())).toBeUndefined();
+		expect(parseWalHeader(new Uint8Array(32))).toBeUndefined();
+		insert(1);
+		const wal = readWal();
+		expect(parseWalHeader(wal)).toMatchObject({ pageSize: 4096, bigEndianChecksums: false });
+		const corrupted = wal.slice(0, 32);
+		corrupted[17] = (corrupted[17] ?? 0) ^ 1;
+		expect(parseWalHeader(corrupted)).toBeUndefined();
+	});
+
+	it("reproduces the checkpointed database from the pages it scans", () => {
+		const before = new Uint8Array(readFileSync(path));
+		insert(40);
+		database.exec("UPDATE rows SET value = randomblob(900) WHERE id % 3 = 0");
+		insert(5, 5000);
+		database.exec("DELETE FROM rows WHERE id % 2 = 0");
+
+		const wal = readWal();
+		const parsed = header(wal);
+		const scan = scanWal(parsed, wal.subarray(WAL_HEADER_BYTES), walStart(parsed));
+		expect(scan.transactions).toBe(4);
+		expect(scan.end.offset).toBe(wal.length);
+
+		database.exec("PRAGMA wal_checkpoint(TRUNCATE)");
+		const after = new Uint8Array(readFileSync(path));
+		expect(scan.dbPageCount * parsed.pageSize).toBe(after.length);
+		expect(Buffer.from(applyPages(before, scan.pages, parsed.pageSize, scan.dbPageCount))).toEqual(
+			Buffer.from(after),
+		);
+	});
+
+	it("drops pages beyond the final size when the database shrinks", () => {
+		insert(30, 4000);
+		database.exec("PRAGMA wal_checkpoint(TRUNCATE)");
+		const before = new Uint8Array(readFileSync(path));
+		database.exec("DELETE FROM rows; VACUUM");
+		const wal = readWal();
+		const scan = scanFrom(wal, walStart(header(wal)));
+		expect(scan.dbPageCount * 4096).toBeLessThan(before.length);
+		expect([...scan.pages.keys()].every((pageNo) => pageNo <= scan.dbPageCount)).toBe(true);
+		database.exec("PRAGMA wal_checkpoint(TRUNCATE)");
+		expect(Buffer.from(applyPages(before, scan.pages, 4096, scan.dbPageCount))).toEqual(readFileSync(path));
+	});
+
+	it("continues from where the previous scan ended", () => {
+		insert(3);
+		insert(3);
+		const first = readWal();
+		const firstScan = scanFrom(first, walStart(header(first)));
+		expect(firstScan.transactions).toBe(2);
+
+		insert(3);
+		const second = readWal();
+		const secondScan = scanFrom(second, firstScan.end);
+		expect(secondScan.transactions).toBe(1);
+		expect(secondScan.end.offset).toBe(second.length);
+
+		const nothing = scanFrom(second, secondScan.end);
+		expect(nothing).toMatchObject({ transactions: 0, dbPageCount: 0, end: secondScan.end });
+		expect(nothing.pages.size).toBe(0);
+	});
+
+	it("stops at the last valid commit frame before a torn or uncommitted tail", () => {
+		insert(3);
+		const committed = readWal().length;
+		insert(20, 2000);
+		const wal = readWal();
+		const frameBytes = WAL_FRAME_HEADER_BYTES + 4096;
+		const start = walStart(header(wal));
+
+		// A transaction cut short: its frames without the commit frame, then half a frame.
+		for (const length of [wal.length - frameBytes, wal.length - frameBytes / 2, committed + frameBytes]) {
+			const scan = scanFrom(wal.subarray(0, length), start);
+			expect(scan.transactions).toBe(1);
+			expect(scan.end.offset).toBe(committed);
+		}
+
+		// A flipped byte in the second transaction's last page fails its checksum.
+		const corrupted = wal.slice();
+		const last = wal.length - 1;
+		corrupted[last] = (corrupted[last] ?? 0) ^ 0xff;
+		expect(scanFrom(corrupted, start).end.offset).toBe(committed);
+	});
+
+	it("ignores frames left over from before the WAL restarted", () => {
+		insert(20, 2000);
+		insert(20, 2000);
+		const longWal = statSync(`${path}-wal`).size;
+		database.exec("PRAGMA wal_checkpoint(RESTART)");
+		insert(1);
+		const wal = readWal();
+		// SQLite rewrote the WAL from the start with new salts; the old frames are still in the file.
+		expect(wal.length).toBe(longWal);
+		const scan = scanFrom(wal, walStart(header(wal)));
+		expect(scan.transactions).toBe(1);
+		expect(scan.end.offset).toBeLessThan(wal.length);
+	});
+
+	it("validates big-endian and little-endian checksums", () => {
+		const frames = [
+			{ pageNo: 1, commitSize: 0, fill: 1 },
+			{ pageNo: 2, commitSize: 2, fill: 2 },
+			{ pageNo: 2, commitSize: 0, fill: 3 },
+			{ pageNo: 3, commitSize: 3, fill: 4 },
+			{ pageNo: 1, commitSize: 0, fill: 5 },
+		];
+		for (const bigEndian of [true, false]) {
+			const wal = syntheticWal(bigEndian, 512, frames);
+			const parsed = header(wal);
+			expect(parsed).toMatchObject({ bigEndianChecksums: bigEndian, checkpointSequence: 7, pageSize: 512 });
+			const scan = scanWal(parsed, wal.subarray(WAL_HEADER_BYTES), walStart(parsed));
+			expect(scan.transactions).toBe(2);
+			expect(scan.dbPageCount).toBe(3);
+			expect([...scan.pages].map(([pageNo, page]) => [pageNo, page[0]])).toEqual([
+				[1, 1],
+				[2, 3],
+				[3, 4],
+			]);
+		}
+		// The same bytes read with the other byte order do not validate.
+		const wal = syntheticWal(true, 512, frames);
+		const view = new DataView(wal.buffer);
+		view.setUint32(0, 0x377f0682);
+		expect(parseWalHeader(wal)).toBeUndefined();
+	});
+});