From 89e309581b9e084de1d2f72f7afc2768d1fafb91 Mon Sep 17 00:00:00 2001 From: Armin Ronacher Date: Wed, 21 Oct 2026 14:26:02 +0200 Subject: [PATCH] fix(chord): remove WAL page replication Reverts 48c8f3228 and 4759f09fb. Session sync replicates pi-durable commits instead of SQLite pages, so NodeSqliteDatabase.exclusive(), the replicated connection wrapper, the WAL reader, page checksums, the segment container, start-marker keys and the zstd page compressor have no user. --- packages/chord/CHANGELOG.md | 1 - packages/chord/README.md | 2 +- packages/chord/src/sqlite/node.ts | 60 +-- .../chord/src/sqlite/replication/checksum.ts | 158 -------- .../chord/src/sqlite/replication/container.ts | 359 ------------------ .../chord/src/sqlite/replication/index.ts | 38 -- packages/chord/src/sqlite/replication/keys.ts | 93 ----- packages/chord/src/sqlite/replication/node.ts | 139 ------- packages/chord/src/sqlite/replication/wal.ts | 155 -------- packages/chord/src/sqlite/replication/zstd.ts | 12 - packages/chord/test/sqlite/node.test.ts | 63 --- .../test/sqlite/replication-checksum.test.ts | 108 ------ .../test/sqlite/replication-container.test.ts | 173 --------- .../test/sqlite/replication-keys.test.ts | 75 ---- .../test/sqlite/replication-node.test.ts | 142 ------- .../chord/test/sqlite/replication-wal.test.ts | 228 ----------- 16 files changed, 7 insertions(+), 1799 deletions(-) delete mode 100644 packages/chord/src/sqlite/replication/checksum.ts delete mode 100644 packages/chord/src/sqlite/replication/container.ts delete mode 100644 packages/chord/src/sqlite/replication/index.ts delete mode 100644 packages/chord/src/sqlite/replication/keys.ts delete mode 100644 packages/chord/src/sqlite/replication/node.ts delete mode 100644 packages/chord/src/sqlite/replication/wal.ts delete mode 100644 packages/chord/src/sqlite/replication/zstd.ts delete mode 100644 packages/chord/test/sqlite/replication-checksum.test.ts delete mode 100644 packages/chord/test/sqlite/replication-container.test.ts delete mode 100644 packages/chord/test/sqlite/replication-keys.test.ts delete mode 100644 packages/chord/test/sqlite/replication-node.test.ts delete mode 100644 packages/chord/test/sqlite/replication-wal.test.ts diff --git a/packages/chord/CHANGELOG.md b/packages/chord/CHANGELOG.md index fcf7e6881..59ea796d3 100644 --- a/packages/chord/CHANGELOG.md +++ b/packages/chord/CHANGELOG.md @@ -11,7 +11,6 @@ - `@earendil-works/chord/telemetry-sqlite`: portable SQLite telemetry recording with bounded settled-span queues, ten-second metric aggregates, periodic flushes, and retention cleanup. - `Context.telemetry` is an explicit, non-optional telemetry parent (no-op by default). `withTelemetry(telemetry, context)` derives it without changing values or cancellation; `traceIn(context, options, fn)` starts a child span and passes its derived context to the callback. - `@earendil-works/chord/sqlite`: portable SQLite interfaces, atomic migration runner, and schema-prefix validation, with Node (`/sqlite/node`) and Cloudflare Durable Object (`/sqlite/cloudflare`) drivers. Moved from Durable, which keeps its storage implementations and re-exports the moved public names. -- `@earendil-works/chord/sqlite/node`: `NodeSqliteDatabase.exclusive(operation)` runs an operation alone on the connection, outside a transaction, after every earlier call and before every later one. - `@earendil-works/chord/output`: `Progress` takes an optional `ProgressClock` (`now`, `setTimer`) to pace commits on, by default `Date.now` and `setTimeout`. ## [1.3.0] - 2026-10-15 diff --git a/packages/chord/README.md b/packages/chord/README.md index 2b732c448..20f595ccf 100644 --- a/packages/chord/README.md +++ b/packages/chord/README.md @@ -21,7 +21,7 @@ Contracts are runtime values described with [TypeBox](https://github.com/sinclai | `@earendil-works/chord/output` | Bounded output streams and paced progress: `OutputBuffer` (head or tail window of a stream, `OutputLimits`, `OutputSkip`), `boundOutput`, `sanitizeOutput`, `utf8ByteLength`, and `Progress` (adaptive commits of a changing value). | Node and browsers | | `@earendil-works/chord/telemetry-sqlite` | `createSqliteTelemetry(db, { retentionMs?, maxQueuedSpans? })`: buffered recording and backend-independent `TelemetryStore` queries. | Any SQLite adapter | | `@earendil-works/chord/sqlite` | Portable `SqliteDatabase`, `SqliteExecutor`, and `SqliteValue`, atomic `applySqliteMigrations(database, migrations, { table })`, and `validateSqliteSchemaPrefix`. The metadata table must be a valid, non-reserved SQLite identifier; migration versions start at 1 and are contiguous. | Any SQLite adapter | -| `@earendil-works/chord/sqlite/node` | `NodeSqliteDatabase` (with `exclusive(operation)` for work that must run alone on the connection outside a transaction, such as checkpoints), `openNodeSqliteDatabase`, and `NodeSqliteDatabaseOptions`. | Node | +| `@earendil-works/chord/sqlite/node` | `NodeSqliteDatabase`, `openNodeSqliteDatabase`, and `NodeSqliteDatabaseOptions`. | Node | | `@earendil-works/chord/sqlite/cloudflare` | `DurableObjectSqliteDatabase` and the structural `DurableObjectSqliteStorage` binding type. | Cloudflare Durable Objects | ## Quick start diff --git a/packages/chord/src/sqlite/node.ts b/packages/chord/src/sqlite/node.ts index 818f5ac87..b156477e8 100644 --- a/packages/chord/src/sqlite/node.ts +++ b/packages/chord/src/sqlite/node.ts @@ -15,7 +15,7 @@ export type NodeSqliteDatabaseOptions = { const DEFAULT_WAL_AUTO_CHECKPOINT_PAGES = 1_000; const DEFAULT_BUSY_TIMEOUT_MS = 5_000; -type HandleScope = { active: boolean }; +type TransactionScope = { active: boolean }; const ignore = (): void => {}; @@ -113,19 +113,16 @@ abstract class NodeSqliteExecutor implements SqliteExecutor { } } -/** The handle passed to a transaction or exclusive operation; it runs SQL directly and is invalid once its scope ends. */ -class NodeSqliteScopedExecutor extends NodeSqliteExecutor { - private readonly scope: HandleScope; - private readonly kind: string; +class NodeSqliteTransaction extends NodeSqliteExecutor { + private readonly scope: TransactionScope; - constructor(database: DatabaseSync, statements: Map, scope: HandleScope, kind: string) { + constructor(database: DatabaseSync, statements: Map, scope: TransactionScope) { super(database, statements); this.scope = scope; - this.kind = kind; } protected async runOperation(operation: () => T): Promise { - if (!this.scope.active) throw new Error(`SQLite ${this.kind} handle is no longer active`); + if (!this.scope.active) throw new Error("SQLite transaction handle is no longer active"); return operation(); } } @@ -144,9 +141,7 @@ export class NodeSqliteDatabase extends NodeSqliteExecutor implements SqliteData this.database.exec("BEGIN IMMEDIATE"); const scope = { active: true }; try { - const result = await callback( - new NodeSqliteScopedExecutor(this.database, this.statements, scope, "transaction"), - ); + const result = await callback(new NodeSqliteTransaction(this.database, this.statements, scope)); scope.active = false; this.database.exec("COMMIT"); return result; @@ -162,49 +157,6 @@ export class NodeSqliteDatabase extends NodeSqliteExecutor implements SqliteData }); } - /** - * Runs `operation` alone on the connection, outside a transaction: it starts after every earlier operation and - * transaction has settled, and later calls wait until it settles. The handle runs SQL immediately and is invalid - * afterwards. Use it for work that must not interleave with commits but cannot run inside a transaction, such as - * `PRAGMA wal_checkpoint`, or to hold back all later operations while awaiting something else. As with - * `transaction`, calling the database itself from inside the operation waits for the operation and never settles. - * - * An operation that begins a transaction must end it. Otherwise later statements would silently join it and - * never commit, so a transaction still open when the operation settles is rolled back and the call rejects, with - * the operation's own error as the `cause` if it failed. - */ - exclusive(operation: (executor: SqliteExecutor) => Promise): Promise { - return this.access.runAsync(async () => { - const scope = { active: true }; - let outcome: { readonly ok: true; readonly value: T } | { readonly ok: false; readonly error: unknown }; - try { - const handle = new NodeSqliteScopedExecutor(this.database, this.statements, scope, "exclusive"); - outcome = { ok: true, value: await operation(handle) }; - } catch (error) { - outcome = { ok: false, error }; - } finally { - scope.active = false; - } - if (this.database.isTransaction) { - const leftOpen = new Error( - "SQLite exclusive operation left a transaction open; it was rolled back", - outcome.ok ? undefined : { cause: outcome.error }, - ); - try { - this.database.exec("ROLLBACK"); - } catch (rollbackError) { - throw new AggregateError( - [leftOpen, rollbackError], - "SQLite exclusive operation left a transaction open and rollback failed", - ); - } - throw leftOpen; - } - if (!outcome.ok) throw outcome.error; - return outcome.value; - }); - } - close(): Promise { return this.access.run(() => { if (this.closed) return; diff --git a/packages/chord/src/sqlite/replication/checksum.ts b/packages/chord/src/sqlite/replication/checksum.ts deleted file mode 100644 index 24aadc694..000000000 --- a/packages/chord/src/sqlite/replication/checksum.ts +++ /dev/null @@ -1,158 +0,0 @@ -/** - * 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 deleted file mode 100644 index e8c043193..000000000 --- a/packages/chord/src/sqlite/replication/container.ts +++ /dev/null @@ -1,359 +0,0 @@ -/** - * 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 deleted file mode 100644 index 5b0375753..000000000 --- a/packages/chord/src/sqlite/replication/index.ts +++ /dev/null @@ -1,38 +0,0 @@ -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 deleted file mode 100644 index 9cc3dd75a..000000000 --- a/packages/chord/src/sqlite/replication/keys.ts +++ /dev/null @@ -1,93 +0,0 @@ -/** - * 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/node.ts b/packages/chord/src/sqlite/replication/node.ts deleted file mode 100644 index 51d0431c0..000000000 --- a/packages/chord/src/sqlite/replication/node.ts +++ /dev/null @@ -1,139 +0,0 @@ -import { resolve } from "node:path"; -import type { SqliteDatabase, SqliteExecutor, SqliteValue } from "../database.ts"; -import { type NodeSqliteDatabase, openNodeSqliteDatabase } from "../node.ts"; - -/** What the replication engine observes on a replicated connection. See `ReplicatedNodeSqliteDatabase`. */ -export interface ReplicationHooks { - /** - * Called synchronously after any operation through the connection settles, whether it succeeded or not, because - * it may have committed. Must not throw and must not call the database. - */ - afterOperation(): void; - /** - * Runs during `close`, alone on the connection, after every earlier operation and before SQLite checkpoints and - * deletes the WAL. Everything the engine has not taken out of the WAL by the time this settles is lost to it. - */ - beforeClose(executor: SqliteExecutor): Promise; -} - -/** Connection settings for a replicated database file. Automatic WAL checkpoints are always off. */ -export type ReplicatedNodeSqliteDatabaseOptions = { - /** Time SQLite waits for a competing file lock; see `NodeSqliteDatabaseOptions.busyTimeoutMs`. */ - readonly busyTimeoutMs?: number; -}; - -/** - * A Node SQLite connection whose write-ahead log (WAL) the replication engine reads from outside SQLite. - * - * The engine ships the pages SQLite appends to `{path}-wal`. That only works if no page leaves the WAL before the - * engine has read it, and a page leaves the WAL whenever anything checkpoints: SQLite's automatic checkpoint after a - * commit, a checkpoint by another connection, and the checkpoint SQLite runs when the last connection closes. This - * wrapper exists so that the engine, and only the engine, decides when that happens on this connection: - * - * - It opens the file itself, with automatic checkpoints disabled, so that the engine is attached before anyone else - * can write. pi-durable runs schema migrations on open, and those commits must be replicated like any other. - * - It owns the underlying connection and never hands it out, so every operation goes through the wrapper. After - * each one it calls `afterOperation`, which is how the engine learns that the WAL may have grown. That includes - * statements outside a transaction, which commit on their own. - * - It exposes `exclusive` so the engine can checkpoint, or hold back every later operation, without interleaving - * with a commit. - * - `close` first runs `beforeClose` in the connection's queue, after everything queued earlier, and only then lets - * the connection close. Operations called after `close` reject, so nothing can commit between the two steps. - * - * The wrapper does not read the WAL, schedule flushes or checkpoint; that is the engine's work, through the hooks. - * It also cannot stop another process from opening the file, which may checkpoint behind the engine's back. The - * engine detects that from the WAL salts and recovers by taking a new base. - * - * The class is exported as a type only, so the only way to get an instance is `openReplicatedNodeSqliteDatabase`. - * A constructor that accepted any `NodeSqliteDatabase` would let a caller pass one with automatic checkpoints on, or - * keep writing through the connection directly, and either would drop WAL pages before the engine read them. - */ -class ReplicatedNodeSqliteDatabase implements SqliteDatabase { - /** Absolute path of the database file. The WAL is `${path}-wal`. */ - readonly path: string; - private readonly database: NodeSqliteDatabase; - private readonly hooks: ReplicationHooks; - private closing: Promise | undefined; - - constructor(path: string, database: NodeSqliteDatabase, hooks: ReplicationHooks) { - this.path = path; - this.database = database; - this.hooks = hooks; - } - - exec(sql: string): Promise { - return this.observe(() => this.database.exec(sql)); - } - - run(sql: string, ...params: SqliteValue[]): Promise { - return this.observe(() => this.database.run(sql, ...params)); - } - - get(sql: string, ...params: SqliteValue[]): Promise { - return this.observe(() => this.database.get(sql, ...params)); - } - - all(sql: string, ...params: SqliteValue[]): Promise { - return this.observe(() => this.database.all(sql, ...params)); - } - - transaction(callback: (transaction: SqliteExecutor) => Promise): Promise { - return this.observe(() => this.database.transaction(callback)); - } - - /** Runs `operation` alone on the connection; see `NodeSqliteDatabase.exclusive`. */ - exclusive(operation: (executor: SqliteExecutor) => Promise): Promise { - if (this.closing !== undefined) return Promise.reject(closedError()); - return this.database.exclusive(operation); - } - - close(): Promise { - this.closing ??= this.closeOnce(); - return this.closing; - } - - private async closeOnce(): Promise { - let hookError: unknown; - let hookFailed = false; - try { - await this.database.exclusive((executor) => this.hooks.beforeClose(executor)); - } catch (error) { - hookFailed = true; - hookError = error; - } - try { - await this.database.close(); - } catch (closeError) { - if (hookFailed) throw new AggregateError([hookError, closeError], "Replication close hook and close failed"); - throw closeError; - } - if (hookFailed) throw hookError; - } - - // Delegating synchronously keeps call order: the operation enters the connection's queue before this returns. - private observe(operation: () => Promise): Promise { - if (this.closing !== undefined) return Promise.reject(closedError()); - return operation().finally(() => this.hooks.afterOperation()); - } -} - -export type { ReplicatedNodeSqliteDatabase }; - -const closedError = (): Error => new Error("Replicated SQLite database is closed"); - -/** Open a database file for replication. The file needs a WAL, so in-memory and temporary databases are refused. */ -export async function openReplicatedNodeSqliteDatabase( - path: string, - hooks: ReplicationHooks, - options: ReplicatedNodeSqliteDatabaseOptions = {}, -): Promise { - if (path === "" || path === ":memory:" || path.startsWith("file:")) { - throw new Error(`Replication needs a database file path, got ${JSON.stringify(path)}`); - } - const absolute = resolve(path); - const database = await openNodeSqliteDatabase(absolute, { - walAutoCheckpointPages: 0, - busyTimeoutMs: options.busyTimeoutMs, - }); - return new ReplicatedNodeSqliteDatabase(absolute, database, hooks); -} diff --git a/packages/chord/src/sqlite/replication/wal.ts b/packages/chord/src/sqlite/replication/wal.ts deleted file mode 100644 index aaa0dab1d..000000000 --- a/packages/chord/src/sqlite/replication/wal.ts +++ /dev/null @@ -1,155 +0,0 @@ -/** - * 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 deleted file mode 100644 index 188104be7..000000000 --- a/packages/chord/src/sqlite/replication/zstd.ts +++ /dev/null @@ -1,12 +0,0 @@ -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/node.test.ts b/packages/chord/test/sqlite/node.test.ts index d9edc14c8..11098d1c5 100644 --- a/packages/chord/test/sqlite/node.test.ts +++ b/packages/chord/test/sqlite/node.test.ts @@ -153,69 +153,6 @@ describe("Node SQLite driver", () => { await database.close(); }); - it("runs an exclusive operation alone, outside a transaction, after earlier calls", async () => { - const database = await openNodeSqliteDatabase(":memory:"); - await database.exec("CREATE TABLE exclusive_probe (value INTEGER)"); - const earlier = database.run("INSERT INTO exclusive_probe (value) VALUES (?)", 1); - let release!: () => void; - const gate = new Promise((resolve) => { - release = resolve; - }); - let handle!: SqliteExecutor; - const exclusive = database.exclusive(async (executor) => { - handle = executor; - const seen = await executor.all("SELECT value FROM exclusive_probe"); - // Not inside a transaction: the handle may begin and commit one itself. - await executor.exec("BEGIN; INSERT INTO exclusive_probe (value) VALUES (2); COMMIT"); - await gate; - return seen; - }); - let laterSettled = false; - const later = database.all("SELECT value FROM exclusive_probe ORDER BY value").finally(() => { - laterSettled = true; - }); - await earlier; - await new Promise((resolve) => setTimeout(resolve, 10)); - expect(laterSettled).toBe(false); - - release(); - await expect(exclusive).resolves.toEqual([{ value: 1 }]); - await expect(later).resolves.toEqual([{ value: 1 }, { value: 2 }]); - await expect(handle.exec("SELECT 1")).rejects.toThrow("SQLite exclusive handle is no longer active"); - await expect(database.exclusive(async () => Promise.reject(new Error("failed")))).rejects.toThrow("failed"); - expect(await database.get("SELECT count(*) AS count FROM exclusive_probe")).toEqual({ count: 2 }); - await database.close(); - }); - - it("rolls back a transaction an exclusive operation left open and rejects", async () => { - const database = await openNodeSqliteDatabase(":memory:"); - await database.exec("CREATE TABLE left_open (value INTEGER)"); - const leftOpen = "SQLite exclusive operation left a transaction open; it was rolled back"; - - await expect( - database.exclusive(async (executor) => { - await executor.exec("BEGIN; INSERT INTO left_open (value) VALUES (1)"); - return "returned without committing"; - }), - ).rejects.toThrow(leftOpen); - - const failure = new Error("failed mid-transaction"); - const rejected = database.exclusive(async (executor) => { - await executor.exec("BEGIN IMMEDIATE; INSERT INTO left_open (value) VALUES (2)"); - throw failure; - }); - await expect(rejected).rejects.toThrow(leftOpen); - await expect(rejected).rejects.toMatchObject({ cause: failure }); - - // Later calls run outside any leftover transaction: autocommit writes commit and transactions can begin. - await database.run("INSERT INTO left_open (value) VALUES (?)", 3); - await database.transaction(async (transaction) => { - await transaction.run("INSERT INTO left_open (value) VALUES (?)", 4); - }); - expect(await database.all("SELECT value FROM left_open ORDER BY value")).toEqual([{ value: 3 }, { value: 4 }]); - await database.close(); - }); - it("does not preserve the callback error when rollback itself fails", async () => { const database = await openNodeSqliteDatabase(":memory:"); await database.exec("CREATE TABLE rollback_probe (value INTEGER)"); diff --git a/packages/chord/test/sqlite/replication-checksum.test.ts b/packages/chord/test/sqlite/replication-checksum.test.ts deleted file mode 100644 index 2953a30f9..000000000 --- a/packages/chord/test/sqlite/replication-checksum.test.ts +++ /dev/null @@ -1,108 +0,0 @@ -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 deleted file mode 100644 index cddf02526..000000000 --- a/packages/chord/test/sqlite/replication-container.test.ts +++ /dev/null @@ -1,173 +0,0 @@ -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 deleted file mode 100644 index a098c6a43..000000000 --- a/packages/chord/test/sqlite/replication-keys.test.ts +++ /dev/null @@ -1,75 +0,0 @@ -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-node.test.ts b/packages/chord/test/sqlite/replication-node.test.ts deleted file mode 100644 index 5b333f7d7..000000000 --- a/packages/chord/test/sqlite/replication-node.test.ts +++ /dev/null @@ -1,142 +0,0 @@ -import { existsSync, mkdtempSync, readFileSync, rmSync, statSync } from "node:fs"; -import { tmpdir } from "node:os"; -import { join } from "node:path"; -import { afterEach, beforeEach, describe, expect, it } from "vitest"; -import type { SqliteExecutor } from "../../src/sqlite/index.ts"; -import * as replication from "../../src/sqlite/replication/node.ts"; -import { openReplicatedNodeSqliteDatabase, type ReplicationHooks } from "../../src/sqlite/replication/node.ts"; - -const WAL_HEADER_BYTES = 32; -const WAL_FRAME_HEADER_BYTES = 24; - -/** Committed frames in a WAL file, counted from its header without validating checksums. */ -function walCommitFrames(path: string): number { - const wal = readFileSync(path); - if (wal.length < WAL_HEADER_BYTES) return 0; - const pageSize = wal.readUInt32BE(8); - let commits = 0; - for (let offset = WAL_HEADER_BYTES; offset + WAL_FRAME_HEADER_BYTES + pageSize <= wal.length; ) { - if (wal.readUInt32BE(offset + 4) !== 0) commits++; - offset += WAL_FRAME_HEADER_BYTES + pageSize; - } - return commits; -} - -function recordingHooks(beforeClose: (executor: SqliteExecutor) => Promise = async () => {}) { - const hooks = { - operations: 0, - afterOperation() { - hooks.operations++; - }, - beforeClose, - } satisfies ReplicationHooks & { operations: number }; - return hooks; -} - -describe("replicated Node SQLite connection", () => { - let directory: string; - let path: string; - - beforeEach(() => { - directory = mkdtempSync(join(tmpdir(), "chord-replication-")); - path = join(directory, "replica.sqlite"); - }); - - afterEach(() => { - rmSync(directory, { recursive: true, force: true }); - }); - - it("can only be created by opening a file", () => { - // A constructor would accept connections with automatic checkpoints on, or ones the caller still writes through. - expect(Object.keys(replication).sort()).toEqual(["openReplicatedNodeSqliteDatabase"]); - }); - - it("refuses databases without a WAL file", async () => { - for (const target of ["", ":memory:", "file::memory:"]) { - await expect(openReplicatedNodeSqliteDatabase(target, recordingHooks())).rejects.toThrow( - "Replication needs a database file path", - ); - } - }); - - it("opens in WAL mode and never checkpoints on its own", async () => { - const database = await openReplicatedNodeSqliteDatabase(path, recordingHooks()); - expect(database.path).toBe(path); - expect(await database.get("PRAGMA journal_mode")).toEqual({ journal_mode: "wal" }); - expect(await database.get("PRAGMA wal_autocheckpoint")).toEqual({ wal_autocheckpoint: 0 }); - await database.exec("CREATE TABLE pages (value BLOB)"); - // SQLite's default would checkpoint once the WAL passes 1,000 pages; write well past that. - for (let batch = 0; batch < 12; batch++) { - await database.transaction(async (transaction) => { - for (let row = 0; row < 100; row++) - await transaction.run("INSERT INTO pages (value) VALUES (randomblob(4000))"); - }); - } - expect(walCommitFrames(`${path}-wal`)).toBe(13); - expect(statSync(`${path}-wal`).size).toBeGreaterThan(1_200 * 4_096); - await database.close(); - }); - - it("reports every operation, including failed ones and statements outside transactions", async () => { - const hooks = recordingHooks(); - const database = await openReplicatedNodeSqliteDatabase(path, hooks); - await database.exec("CREATE TABLE notes (value TEXT NOT NULL)"); - await database.run("INSERT INTO notes (value) VALUES (?)", "autocommit"); - await database.get("SELECT value FROM notes"); - await database.all("SELECT value FROM notes"); - await database.transaction(async (transaction) => { - await transaction.run("INSERT INTO notes (value) VALUES (?)", "transaction"); - }); - await expect(database.run("INSERT INTO notes (value) VALUES (NULL)")).rejects.toThrow(); - expect(hooks.operations).toBe(6); - await database.exclusive(async (executor) => { - await executor.exec("PRAGMA wal_checkpoint(TRUNCATE)"); - }); - expect(hooks.operations).toBe(6); - await database.close(); - }); - - it("runs the close hook after earlier operations, while the WAL holds every commit", async () => { - let framesAtClose = -1; - let rowsAtClose: unknown; - const database = await openReplicatedNodeSqliteDatabase( - path, - recordingHooks(async (executor) => { - framesAtClose = walCommitFrames(`${path}-wal`); - rowsAtClose = await executor.get("SELECT count(*) AS count FROM notes"); - }), - ); - await database.exec("CREATE TABLE notes (value TEXT)"); - // Called before close and still queued when close is called: these must commit before the hook runs. - const pending = [1, 2, 3].map((value) => database.run("INSERT INTO notes (value) VALUES (?)", String(value))); - const closed = database.close(); - expect(database.close()).toBe(closed); - await Promise.all(pending); - await closed; - - expect(framesAtClose).toBe(4); - expect(rowsAtClose).toEqual({ count: 3 }); - // SQLite checkpoints and removes the WAL when the last connection closes. - expect(existsSync(`${path}-wal`)).toBe(false); - await expect(database.run("INSERT INTO notes (value) VALUES (?)", "late")).rejects.toThrow( - "Replicated SQLite database is closed", - ); - await expect(database.exclusive(async () => {})).rejects.toThrow("Replicated SQLite database is closed"); - }); - - it("closes the connection even when the close hook fails", async () => { - const database = await openReplicatedNodeSqliteDatabase( - path, - recordingHooks(async () => { - throw new Error("spool failed"); - }), - ); - await database.exec("CREATE TABLE notes (value TEXT)"); - await expect(database.close()).rejects.toThrow("spool failed"); - expect(existsSync(`${path}-wal`)).toBe(false); - - const reopened = await openReplicatedNodeSqliteDatabase(path, recordingHooks()); - expect(await reopened.all("SELECT value FROM notes")).toEqual([]); - await reopened.close(); - }); -}); diff --git a/packages/chord/test/sqlite/replication-wal.test.ts b/packages/chord/test/sqlite/replication-wal.test.ts deleted file mode 100644 index a53daf1fe..000000000 --- a/packages/chord/test/sqlite/replication-wal.test.ts +++ /dev/null @@ -1,228 +0,0 @@ -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(); - }); -});