diff --git a/packages/chord/src/sqlite/replication/checksum.ts b/packages/chord/src/sqlite/replication/checksum.ts new file mode 100644 index 000000000..24aadc694 --- /dev/null +++ b/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 { + 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): Promise> { + 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 { + 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(); + 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, 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); + } +} diff --git a/packages/chord/src/sqlite/replication/container.ts b/packages/chord/src/sqlite/replication/container.ts new file mode 100644 index 000000000..e8c043193 --- /dev/null +++ b/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 + | Omit; + +/** 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; +}; + +/** 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, + 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(); + 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 { + 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)[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; +} diff --git a/packages/chord/src/sqlite/replication/index.ts b/packages/chord/src/sqlite/replication/index.ts new file mode 100644 index 000000000..5b0375753 --- /dev/null +++ b/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"; diff --git a/packages/chord/src/sqlite/replication/keys.ts b/packages/chord/src/sqlite/replication/keys.ts new file mode 100644 index 000000000..9cc3dd75a --- /dev/null +++ b/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> | 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 }; +} diff --git a/packages/chord/src/sqlite/replication/wal.ts b/packages/chord/src/sqlite/replication/wal.ts new file mode 100644 index 000000000..aaa0dab1d --- /dev/null +++ b/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; + /** 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(); + const pending = new Map(); + 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]; +} diff --git a/packages/chord/src/sqlite/replication/zstd.ts b/packages/chord/src/sqlite/replication/zstd.ts new file mode 100644 index 000000000..188104be7 --- /dev/null +++ b/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 }), +}; diff --git a/packages/chord/test/sqlite/replication-checksum.test.ts b/packages/chord/test/sqlite/replication-checksum.test.ts new file mode 100644 index 000000000..2953a30f9 --- /dev/null +++ b/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"); + }); +}); diff --git a/packages/chord/test/sqlite/replication-container.test.ts b/packages/chord/test/sqlite/replication-container.test.ts new file mode 100644 index 000000000..cddf02526 --- /dev/null +++ b/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][] = [ + [{ ...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)); + } + }); +}); diff --git a/packages/chord/test/sqlite/replication-keys.test.ts b/packages/chord/test/sqlite/replication-keys.test.ts new file mode 100644 index 000000000..a098c6a43 --- /dev/null +++ b/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(); + const violations: string[] = []; + const visit = async (file: string): Promise => { + 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); + }); +}); diff --git a/packages/chord/test/sqlite/replication-wal.test.ts b/packages/chord/test/sqlite/replication-wal.test.ts new file mode 100644 index 000000000..a53daf1fe --- /dev/null +++ b/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, 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(); + }); +});