mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
fix(chord): remove WAL page replication
Reverts48c8f3228and4759f09fb. 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.
This commit is contained in:
Armin Ronacher
parent
4759f09fbc
commit
89e309581b
16 changed files with 7 additions and 1799 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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<string, StatementSync>, scope: HandleScope, kind: string) {
|
||||
constructor(database: DatabaseSync, statements: Map<string, StatementSync>, scope: TransactionScope) {
|
||||
super(database, statements);
|
||||
this.scope = scope;
|
||||
this.kind = kind;
|
||||
}
|
||||
|
||||
protected async runOperation<T>(operation: () => T): Promise<T> {
|
||||
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<T>(operation: (executor: SqliteExecutor) => Promise<T>): Promise<T> {
|
||||
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<void> {
|
||||
return this.access.run(() => {
|
||||
if (this.closed) return;
|
||||
|
|
|
|||
|
|
@ -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<bigint> {
|
||||
const input = new Uint8Array(4 + page.length);
|
||||
new DataView(input.buffer).setUint32(0, pageNo);
|
||||
input.set(page, 4);
|
||||
const digest = await crypto.subtle.digest("SHA-256", input);
|
||||
return new DataView(digest).getBigUint64(0);
|
||||
}
|
||||
|
||||
/** Hash several pages, keyed like the input. */
|
||||
export async function hashPages(pages: ReadonlyMap<number, Uint8Array>): Promise<Map<number, bigint>> {
|
||||
const entries = [...pages];
|
||||
const hashes = await Promise.all(entries.map(([pageNo, page]) => pageHash(pageNo, page)));
|
||||
return new Map(entries.map(([pageNo], index) => [pageNo, hashes[index] ?? 0n]));
|
||||
}
|
||||
|
||||
/** A checksum as written in headers: 16 lowercase hex digits. */
|
||||
export function formatChecksum(checksum: bigint): string {
|
||||
return checksum.toString(16).padStart(16, "0");
|
||||
}
|
||||
|
||||
/** Parse a checksum written by `formatChecksum`. */
|
||||
export function parseChecksum(text: string): bigint {
|
||||
if (!/^[0-9a-f]{16}$/.test(text)) throw new Error(`Invalid database checksum ${JSON.stringify(text)}`);
|
||||
return BigInt(`0x${text}`);
|
||||
}
|
||||
|
||||
/**
|
||||
* The hash of every page of one database, and their XOR. This is the writer's sidecar and a restoring reader's
|
||||
* running state; both update it from the pages of each segment they produce or apply.
|
||||
*/
|
||||
export class PageHashes {
|
||||
private hashes: BigUint64Array;
|
||||
private count: number;
|
||||
private xor: bigint;
|
||||
|
||||
private constructor(hashes: BigUint64Array, count: number) {
|
||||
this.hashes = hashes;
|
||||
this.count = count;
|
||||
let xor = 0n;
|
||||
for (let index = 0; index < count; index++) xor ^= hashes[index] ?? 0n;
|
||||
this.xor = xor;
|
||||
}
|
||||
|
||||
/** The hashes of an empty database. */
|
||||
static empty(): PageHashes {
|
||||
return new PageHashes(new BigUint64Array(0), 0);
|
||||
}
|
||||
|
||||
/** Hash every page of a complete database file. */
|
||||
static async fromDatabase(bytes: Uint8Array, pageSize: number): Promise<PageHashes> {
|
||||
if (bytes.length % pageSize !== 0) {
|
||||
throw new Error(`Database of ${bytes.length} bytes is not a whole number of ${pageSize}-byte pages`);
|
||||
}
|
||||
const pages = new Map<number, Uint8Array>();
|
||||
for (let offset = 0; offset < bytes.length; offset += pageSize) {
|
||||
pages.set(offset / pageSize + 1, bytes.subarray(offset, offset + pageSize));
|
||||
}
|
||||
const hashes = PageHashes.empty();
|
||||
hashes.apply(await hashPages(pages), pages.size);
|
||||
return hashes;
|
||||
}
|
||||
|
||||
/** Read a sidecar written by `encode`: one little-endian u64 per page, page `p` at offset `(p - 1) * 8`. */
|
||||
static decode(bytes: Uint8Array): PageHashes {
|
||||
if (bytes.length % 8 !== 0) throw new Error(`Page hash file of ${bytes.length} bytes is not a multiple of 8`);
|
||||
const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength);
|
||||
const count = bytes.length / 8;
|
||||
const hashes = new BigUint64Array(count);
|
||||
for (let index = 0; index < count; index++) hashes[index] = view.getBigUint64(index * 8, true);
|
||||
return new PageHashes(hashes, count);
|
||||
}
|
||||
|
||||
/** Database size in pages. */
|
||||
get pageCount(): number {
|
||||
return this.count;
|
||||
}
|
||||
|
||||
/** The database checksum. */
|
||||
get checksum(): bigint {
|
||||
return this.xor;
|
||||
}
|
||||
|
||||
/** The hash of page `pageNo`, or `undefined` beyond the database. */
|
||||
hash(pageNo: number): bigint | undefined {
|
||||
return pageNo >= 1 && pageNo <= this.count ? this.hashes[pageNo - 1] : undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* Apply the pages of one segment: set the hash of every changed page, then resize to `dbPageCount`. Pages past
|
||||
* `dbPageCount` must not be present. When the database grows, every new page must be among `pages`; SQLite writes
|
||||
* every page it adds, and a page this database has never seen has no known contents to hash. A violation throws
|
||||
* and leaves the hashes unchanged.
|
||||
*/
|
||||
apply(pages: ReadonlyMap<number, bigint>, dbPageCount: number): void {
|
||||
if (!Number.isSafeInteger(dbPageCount) || dbPageCount < 0) throw new Error(`Invalid page count ${dbPageCount}`);
|
||||
for (const pageNo of pages.keys()) {
|
||||
if (!Number.isSafeInteger(pageNo) || pageNo < 1 || pageNo > dbPageCount) {
|
||||
throw new Error(`Page ${pageNo} is outside a database of ${dbPageCount} pages`);
|
||||
}
|
||||
}
|
||||
for (let pageNo = this.count + 1; pageNo <= dbPageCount; pageNo++) {
|
||||
if (!pages.has(pageNo)) {
|
||||
throw new Error(`Database grew to ${dbPageCount} pages without page ${pageNo}`);
|
||||
}
|
||||
}
|
||||
|
||||
for (let pageNo = dbPageCount + 1; pageNo <= this.count; pageNo++) this.xor ^= this.hashes[pageNo - 1] ?? 0n;
|
||||
if (dbPageCount > this.hashes.length) {
|
||||
const grown = new BigUint64Array(Math.max(dbPageCount, this.hashes.length * 2));
|
||||
grown.set(this.hashes.subarray(0, this.count));
|
||||
this.hashes = grown;
|
||||
} else if (dbPageCount < this.count) {
|
||||
this.hashes.fill(0n, dbPageCount, this.count);
|
||||
}
|
||||
const previousCount = this.count;
|
||||
this.count = dbPageCount;
|
||||
for (const [pageNo, hash] of pages) {
|
||||
if (pageNo <= previousCount) this.xor ^= this.hashes[pageNo - 1] ?? 0n;
|
||||
this.hashes[pageNo - 1] = hash;
|
||||
this.xor ^= hash;
|
||||
}
|
||||
}
|
||||
|
||||
/** Write the sidecar format read by `decode`. */
|
||||
encode(): Uint8Array {
|
||||
const bytes = new Uint8Array(this.count * 8);
|
||||
const view = new DataView(bytes.buffer);
|
||||
for (let index = 0; index < this.count; index++) view.setBigUint64(index * 8, this.hashes[index] ?? 0n, true);
|
||||
return bytes;
|
||||
}
|
||||
|
||||
/** An independent copy. */
|
||||
clone(): PageHashes {
|
||||
return new PageHashes(this.hashes.slice(0, this.count), this.count);
|
||||
}
|
||||
}
|
||||
|
|
@ -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<SegmentHeader, "pageCount" | "compression">
|
||||
| Omit<BaseHeader, "pageCount" | "compression">;
|
||||
|
||||
/** Where one page's blob lies in an encoded container. */
|
||||
export type PageIndexEntry = { readonly pageNo: number; readonly offset: number; readonly length: number };
|
||||
|
||||
/** A container read from its complete bytes. */
|
||||
export type DecodedContainer = {
|
||||
readonly header: ContainerHeader;
|
||||
readonly index: readonly PageIndexEntry[];
|
||||
/** Decompress every page, by page number. Throws if `compression` does not match the header. */
|
||||
pages(compression: PageCompression): Map<number, Uint8Array>;
|
||||
};
|
||||
|
||||
/** Thrown for bytes that are not a valid container. */
|
||||
export class ReplicationFormatError extends Error {
|
||||
constructor(message: string) {
|
||||
super(message);
|
||||
this.name = "ReplicationFormatError";
|
||||
}
|
||||
}
|
||||
|
||||
/** Encode a segment or base. Pages may be given in any order; each must be exactly `pageSize` bytes. */
|
||||
export function encodeContainer(
|
||||
input: ContainerHeaderInput,
|
||||
pages: ReadonlyMap<number, Uint8Array>,
|
||||
compression: PageCompression,
|
||||
): Uint8Array {
|
||||
const header = validateHeader({ ...input, pageCount: pages.size, compression: compression.name });
|
||||
const pageNumbers = [...pages.keys()].sort((a, b) => a - b);
|
||||
if (header.kind === "base" && pageNumbers.some((pageNo, index) => pageNo !== index + 1)) {
|
||||
throw new ReplicationFormatError(`A base must contain every page from 1 to ${header.dbPageCount}`);
|
||||
}
|
||||
const blobs: Uint8Array[] = [];
|
||||
for (const pageNo of pageNumbers) {
|
||||
const page = pages.get(pageNo);
|
||||
if (page?.length !== header.pageSize) {
|
||||
throw new ReplicationFormatError(`Page ${pageNo} is ${page?.length} bytes, not ${header.pageSize}`);
|
||||
}
|
||||
if (pageNo > header.dbPageCount) {
|
||||
throw new ReplicationFormatError(`Page ${pageNo} is outside a database of ${header.dbPageCount} pages`);
|
||||
}
|
||||
blobs.push(compression.compress(page));
|
||||
}
|
||||
|
||||
const headerBytes = new TextEncoder().encode(JSON.stringify(canonicalHeader(header)));
|
||||
if (headerBytes.length > MAX_HEADER_BYTES) throw new ReplicationFormatError("Container header is too large");
|
||||
const blobBytes = blobs.reduce((total, blob) => total + blob.length, 0);
|
||||
const indexOffset = PREFIX_BYTES + headerBytes.length + blobBytes;
|
||||
const total = indexOffset + pageNumbers.length * INDEX_ENTRY_BYTES + FOOTER_BYTES;
|
||||
const bytes = new Uint8Array(total);
|
||||
const view = new DataView(bytes.buffer);
|
||||
bytes.set(MAGIC, 0);
|
||||
view.setUint32(MAGIC_BYTES, headerBytes.length, true);
|
||||
bytes.set(headerBytes, PREFIX_BYTES);
|
||||
|
||||
let offset = PREFIX_BYTES + headerBytes.length;
|
||||
let entry = indexOffset;
|
||||
pageNumbers.forEach((pageNo, index) => {
|
||||
const blob = blobs[index] ?? new Uint8Array();
|
||||
bytes.set(blob, offset);
|
||||
view.setUint32(entry, pageNo, true);
|
||||
view.setBigUint64(entry + 4, BigInt(offset), true);
|
||||
view.setUint32(entry + 12, blob.length, true);
|
||||
offset += blob.length;
|
||||
entry += INDEX_ENTRY_BYTES;
|
||||
});
|
||||
view.setBigUint64(total - FOOTER_BYTES, BigInt(indexOffset), true);
|
||||
view.setUint32(total - 4, crc32(bytes.subarray(0, total - 4)), true);
|
||||
return bytes;
|
||||
}
|
||||
|
||||
/**
|
||||
* Read and validate the header from the first bytes of a container. `prefix` may be the whole object or just its
|
||||
* beginning; it must contain at least the magic, the length and the header. This is what a store uses to validate an
|
||||
* upload before accepting it.
|
||||
*/
|
||||
export function decodeContainerHeader(prefix: Uint8Array): ContainerHeader {
|
||||
if (prefix.length < PREFIX_BYTES || !MAGIC.every((byte, index) => prefix[index] === byte)) {
|
||||
throw new ReplicationFormatError("Not a replication container");
|
||||
}
|
||||
const headerLength = new DataView(prefix.buffer, prefix.byteOffset, prefix.byteLength).getUint32(MAGIC_BYTES, true);
|
||||
if (headerLength > MAX_HEADER_BYTES) throw new ReplicationFormatError("Container header is too large");
|
||||
if (prefix.length < PREFIX_BYTES + headerLength) throw new ReplicationFormatError("Container header is truncated");
|
||||
let json: unknown;
|
||||
try {
|
||||
json = JSON.parse(
|
||||
new TextDecoder("utf-8", { fatal: true }).decode(prefix.subarray(PREFIX_BYTES, PREFIX_BYTES + headerLength)),
|
||||
);
|
||||
} catch {
|
||||
throw new ReplicationFormatError("Container header is not valid JSON");
|
||||
}
|
||||
return validateHeader(json);
|
||||
}
|
||||
|
||||
/** Decode a complete container, checking its CRC, header and index. Pages are decompressed on request. */
|
||||
export function decodeContainer(bytes: Uint8Array): DecodedContainer {
|
||||
if (bytes.length < PREFIX_BYTES + FOOTER_BYTES) throw new ReplicationFormatError("Container is truncated");
|
||||
const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength);
|
||||
if (crc32(bytes.subarray(0, bytes.length - 4)) !== view.getUint32(bytes.length - 4, true)) {
|
||||
throw new ReplicationFormatError("Container CRC does not match");
|
||||
}
|
||||
const header = decodeContainerHeader(bytes);
|
||||
const blobsStart = PREFIX_BYTES + view.getUint32(MAGIC_BYTES, true);
|
||||
const indexOffset = Number(view.getBigUint64(bytes.length - FOOTER_BYTES, true));
|
||||
const index = decodeIndex(bytes, indexOffset, bytes.length - FOOTER_BYTES, header.pageCount, blobsStart);
|
||||
if (header.kind === "base" && index.some((entry, position) => entry.pageNo !== position + 1)) {
|
||||
throw new ReplicationFormatError(`A base must contain every page from 1 to ${header.dbPageCount}`);
|
||||
}
|
||||
if (index.some((entry) => entry.pageNo > header.dbPageCount)) {
|
||||
throw new ReplicationFormatError("Container has a page outside the database");
|
||||
}
|
||||
return {
|
||||
header,
|
||||
index,
|
||||
pages(compression) {
|
||||
if (compression.name !== header.compression) {
|
||||
throw new ReplicationFormatError(`Container uses ${header.compression}, not ${compression.name}`);
|
||||
}
|
||||
const pages = new Map<number, Uint8Array>();
|
||||
for (const entry of index) {
|
||||
const page = compression.decompress(
|
||||
bytes.subarray(entry.offset, entry.offset + entry.length),
|
||||
header.pageSize,
|
||||
);
|
||||
if (page.length !== header.pageSize) {
|
||||
throw new ReplicationFormatError(`Page ${entry.pageNo} decompressed to ${page.length} bytes`);
|
||||
}
|
||||
pages.set(entry.pageNo, page);
|
||||
}
|
||||
return pages;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function decodeIndex(
|
||||
bytes: Uint8Array,
|
||||
indexOffset: number,
|
||||
indexEnd: number,
|
||||
pageCount: number,
|
||||
blobsStart: number,
|
||||
): PageIndexEntry[] {
|
||||
if (indexOffset < blobsStart || indexEnd - indexOffset !== pageCount * INDEX_ENTRY_BYTES) {
|
||||
throw new ReplicationFormatError("Container index does not match its page count");
|
||||
}
|
||||
const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength);
|
||||
const index: PageIndexEntry[] = [];
|
||||
let expectedOffset = blobsStart;
|
||||
for (let entry = indexOffset; entry < indexEnd; entry += INDEX_ENTRY_BYTES) {
|
||||
const pageNo = view.getUint32(entry, true);
|
||||
const offset = Number(view.getBigUint64(entry + 4, true));
|
||||
const length = view.getUint32(entry + 12, true);
|
||||
const previous = index.at(-1);
|
||||
if (pageNo === 0 || (previous !== undefined && pageNo <= previous.pageNo)) {
|
||||
throw new ReplicationFormatError("Container index is not in ascending page order");
|
||||
}
|
||||
if (offset !== expectedOffset || length === 0) {
|
||||
throw new ReplicationFormatError("Container index does not describe contiguous page blobs");
|
||||
}
|
||||
expectedOffset = offset + length;
|
||||
index.push({ pageNo, offset, length });
|
||||
}
|
||||
if (expectedOffset !== indexOffset) throw new ReplicationFormatError("Container page blobs do not end at the index");
|
||||
return index;
|
||||
}
|
||||
|
||||
function canonicalHeader(header: ContainerHeader): Record<string, unknown> {
|
||||
const common = { version: header.version, kind: header.kind, replicaId: header.replicaId, epoch: header.epoch };
|
||||
const sizes = { pageSize: header.pageSize, dbPageCount: header.dbPageCount };
|
||||
const tail = { pageCount: header.pageCount, compression: header.compression };
|
||||
return header.kind === "segment"
|
||||
? {
|
||||
...common,
|
||||
fromSeq: header.fromSeq,
|
||||
toSeq: header.toSeq,
|
||||
...sizes,
|
||||
preChecksum: header.preChecksum,
|
||||
postChecksum: header.postChecksum,
|
||||
...tail,
|
||||
}
|
||||
: { ...common, seq: header.seq, ...sizes, checksum: header.checksum, ...tail };
|
||||
}
|
||||
|
||||
const REPLICA_ID = /^[A-Za-z0-9._-]{1,128}$/;
|
||||
const CHECKSUM = /^[0-9a-f]{16}$/;
|
||||
|
||||
/** Check a parsed header against the format rules. Unknown fields are ignored and dropped. */
|
||||
export function validateHeader(value: unknown): ContainerHeader {
|
||||
if (typeof value !== "object" || value === null || Array.isArray(value)) {
|
||||
throw new ReplicationFormatError("Container header is not an object");
|
||||
}
|
||||
const field = (name: string): unknown => (value as Record<string, unknown>)[name];
|
||||
const integer = (name: string, minimum: number): number => {
|
||||
const number = field(name);
|
||||
if (typeof number !== "number" || !Number.isSafeInteger(number) || number < minimum) {
|
||||
throw new ReplicationFormatError(`Container header field ${name} must be an integer of at least ${minimum}`);
|
||||
}
|
||||
return number;
|
||||
};
|
||||
const checksum = (name: string): string => {
|
||||
const text = field(name);
|
||||
if (typeof text !== "string" || !CHECKSUM.test(text)) {
|
||||
throw new ReplicationFormatError(`Container header field ${name} must be 16 lowercase hex digits`);
|
||||
}
|
||||
return text;
|
||||
};
|
||||
|
||||
if (field("version") !== 1) throw new ReplicationFormatError(`Unsupported container version ${field("version")}`);
|
||||
const replicaId = field("replicaId");
|
||||
if (typeof replicaId !== "string" || !REPLICA_ID.test(replicaId)) {
|
||||
throw new ReplicationFormatError("Container header field replicaId is not a valid replica id");
|
||||
}
|
||||
const compression = field("compression");
|
||||
if (compression !== "zstd" && compression !== "none") {
|
||||
throw new ReplicationFormatError(`Unsupported page compression ${JSON.stringify(compression)}`);
|
||||
}
|
||||
const pageSize = integer("pageSize", 512);
|
||||
if (pageSize > 65536 || (pageSize & (pageSize - 1)) !== 0) {
|
||||
throw new ReplicationFormatError(`Invalid page size ${pageSize}`);
|
||||
}
|
||||
const common: CommonHeader = {
|
||||
version: 1,
|
||||
replicaId,
|
||||
epoch: integer("epoch", 1),
|
||||
pageSize,
|
||||
dbPageCount: integer("dbPageCount", 0),
|
||||
pageCount: integer("pageCount", 0),
|
||||
compression,
|
||||
};
|
||||
if (common.pageCount > common.dbPageCount) {
|
||||
throw new ReplicationFormatError("Container has more pages than the database");
|
||||
}
|
||||
|
||||
const kind = field("kind");
|
||||
if (kind === "segment") {
|
||||
const fromSeq = integer("fromSeq", 1);
|
||||
const toSeq = integer("toSeq", fromSeq);
|
||||
if (common.pageCount < 1) throw new ReplicationFormatError("A segment must contain at least one page");
|
||||
return {
|
||||
...common,
|
||||
kind,
|
||||
fromSeq,
|
||||
toSeq,
|
||||
preChecksum: checksum("preChecksum"),
|
||||
postChecksum: checksum("postChecksum"),
|
||||
};
|
||||
}
|
||||
if (kind === "base") {
|
||||
if (common.pageCount !== common.dbPageCount) {
|
||||
throw new ReplicationFormatError("A base must contain every page of the database");
|
||||
}
|
||||
return { ...common, kind, seq: integer("seq", 0), checksum: checksum("checksum") };
|
||||
}
|
||||
throw new ReplicationFormatError(`Unknown container kind ${JSON.stringify(kind)}`);
|
||||
}
|
||||
|
||||
let crcTable: Uint32Array | undefined;
|
||||
|
||||
/** IEEE CRC-32, as in zlib. */
|
||||
export function crc32(bytes: Uint8Array): number {
|
||||
if (crcTable === undefined) {
|
||||
crcTable = new Uint32Array(256);
|
||||
for (let n = 0; n < 256; n++) {
|
||||
let c = n;
|
||||
for (let bit = 0; bit < 8; bit++) c = c & 1 ? 0xedb88320 ^ (c >>> 1) : c >>> 1;
|
||||
crcTable[n] = c >>> 0;
|
||||
}
|
||||
}
|
||||
let crc = 0xffffffff;
|
||||
for (const byte of bytes) crc = (crcTable[(crc ^ byte) & 0xff] ?? 0) ^ (crc >>> 8);
|
||||
return (crc ^ 0xffffffff) >>> 0;
|
||||
}
|
||||
|
|
@ -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";
|
||||
|
|
@ -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<Record<keyof StartMarker, unknown>> | null;
|
||||
const valid = (field: unknown): field is number =>
|
||||
typeof field === "number" && Number.isSafeInteger(field) && field >= 1;
|
||||
if (typeof marker !== "object" || marker === null || marker.version !== 1) {
|
||||
throw new Error("Start marker has an unsupported version");
|
||||
}
|
||||
if (!valid(marker.epoch) || !valid(marker.startSeq)) throw new Error("Start marker has an invalid epoch or seq");
|
||||
return { version: 1, epoch: marker.epoch, startSeq: marker.startSeq };
|
||||
}
|
||||
|
|
@ -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<void>;
|
||||
}
|
||||
|
||||
/** 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<void> | undefined;
|
||||
|
||||
constructor(path: string, database: NodeSqliteDatabase, hooks: ReplicationHooks) {
|
||||
this.path = path;
|
||||
this.database = database;
|
||||
this.hooks = hooks;
|
||||
}
|
||||
|
||||
exec(sql: string): Promise<void> {
|
||||
return this.observe(() => this.database.exec(sql));
|
||||
}
|
||||
|
||||
run(sql: string, ...params: SqliteValue[]): Promise<void> {
|
||||
return this.observe(() => this.database.run(sql, ...params));
|
||||
}
|
||||
|
||||
get<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T | undefined> {
|
||||
return this.observe(() => this.database.get<T>(sql, ...params));
|
||||
}
|
||||
|
||||
all<T extends object>(sql: string, ...params: SqliteValue[]): Promise<T[]> {
|
||||
return this.observe(() => this.database.all<T>(sql, ...params));
|
||||
}
|
||||
|
||||
transaction<T>(callback: (transaction: SqliteExecutor) => Promise<T>): Promise<T> {
|
||||
return this.observe(() => this.database.transaction(callback));
|
||||
}
|
||||
|
||||
/** Runs `operation` alone on the connection; see `NodeSqliteDatabase.exclusive`. */
|
||||
exclusive<T>(operation: (executor: SqliteExecutor) => Promise<T>): Promise<T> {
|
||||
if (this.closing !== undefined) return Promise.reject(closedError());
|
||||
return this.database.exclusive(operation);
|
||||
}
|
||||
|
||||
close(): Promise<void> {
|
||||
this.closing ??= this.closeOnce();
|
||||
return this.closing;
|
||||
}
|
||||
|
||||
private async closeOnce(): Promise<void> {
|
||||
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<T>(operation: () => Promise<T>): Promise<T> {
|
||||
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<ReplicatedNodeSqliteDatabase> {
|
||||
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);
|
||||
}
|
||||
|
|
@ -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<number, Uint8Array>;
|
||||
/** Database size in pages after the last transaction, or 0 when there was none. */
|
||||
readonly dbPageCount: number;
|
||||
/** The position just after the last commit frame, or the start position when there was none. */
|
||||
readonly end: WalPosition;
|
||||
};
|
||||
|
||||
/**
|
||||
* Parse a WAL header. Returns `undefined` when `bytes` is shorter than a header or the header is not valid (wrong
|
||||
* magic, version, page size or checksum), which is also how an empty or freshly truncated WAL looks.
|
||||
*/
|
||||
export function parseWalHeader(bytes: Uint8Array): WalHeader | undefined {
|
||||
if (bytes.length < WAL_HEADER_BYTES) return undefined;
|
||||
const view = new DataView(bytes.buffer, bytes.byteOffset, WAL_HEADER_BYTES);
|
||||
const magic = view.getUint32(0);
|
||||
if ((magic & ~1) !== WAL_MAGIC || view.getUint32(4) !== WAL_FORMAT_VERSION) return undefined;
|
||||
const pageSize = view.getUint32(8);
|
||||
if (pageSize < 512 || pageSize > 65536 || (pageSize & (pageSize - 1)) !== 0) return undefined;
|
||||
const bigEndianChecksums = (magic & 1) === 1;
|
||||
const [checksum1, checksum2] = walChecksum(bytes.subarray(0, 24), bigEndianChecksums, 0, 0);
|
||||
if (checksum1 !== view.getUint32(24) || checksum2 !== view.getUint32(28)) return undefined;
|
||||
return {
|
||||
bigEndianChecksums,
|
||||
pageSize,
|
||||
checkpointSequence: view.getUint32(12),
|
||||
salt1: view.getUint32(16),
|
||||
salt2: view.getUint32(20),
|
||||
checksum1,
|
||||
checksum2,
|
||||
};
|
||||
}
|
||||
|
||||
/** The position of the first frame under `header`. */
|
||||
export function walStart(header: WalHeader): WalPosition {
|
||||
return { offset: WAL_HEADER_BYTES, checksum1: header.checksum1, checksum2: header.checksum2 };
|
||||
}
|
||||
|
||||
/**
|
||||
* Scan committed transactions from `from` to the last valid commit frame.
|
||||
*
|
||||
* `bytes` holds the WAL file from offset `from.offset` onwards, usually to its current end. Frames after the last
|
||||
* valid commit frame are ignored; a later scan from `end` reads them once their transaction has committed.
|
||||
*
|
||||
* Pages are kept by their newest version, and pages beyond the final database size are dropped, which is exactly how
|
||||
* SQLite itself reads the log: the newest frame of each page wins, and the size from the last commit applies.
|
||||
*/
|
||||
export function scanWal(header: WalHeader, bytes: Uint8Array, from: WalPosition): WalScan {
|
||||
const frameBytes = WAL_FRAME_HEADER_BYTES + header.pageSize;
|
||||
const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength);
|
||||
const committed = new Map<number, Uint8Array>();
|
||||
const pending = new Map<number, Uint8Array>();
|
||||
let transactions = 0;
|
||||
let dbPageCount = 0;
|
||||
let end = from;
|
||||
let checksum1 = from.checksum1;
|
||||
let checksum2 = from.checksum2;
|
||||
|
||||
for (let local = 0; local + frameBytes <= bytes.length; local += frameBytes) {
|
||||
const pageNo = view.getUint32(local);
|
||||
const commitSize = view.getUint32(local + 4);
|
||||
if (pageNo === 0 || view.getUint32(local + 8) !== header.salt1 || view.getUint32(local + 12) !== header.salt2) {
|
||||
break;
|
||||
}
|
||||
const page = bytes.subarray(local + WAL_FRAME_HEADER_BYTES, local + frameBytes);
|
||||
[checksum1, checksum2] = walChecksum(
|
||||
bytes.subarray(local, local + 8),
|
||||
header.bigEndianChecksums,
|
||||
checksum1,
|
||||
checksum2,
|
||||
);
|
||||
[checksum1, checksum2] = walChecksum(page, header.bigEndianChecksums, checksum1, checksum2);
|
||||
if (checksum1 !== view.getUint32(local + 16) || checksum2 !== view.getUint32(local + 20)) break;
|
||||
|
||||
pending.set(pageNo, page);
|
||||
if (commitSize === 0) continue;
|
||||
for (const [number, data] of pending) committed.set(number, data);
|
||||
pending.clear();
|
||||
transactions++;
|
||||
dbPageCount = commitSize;
|
||||
end = { offset: from.offset + local + frameBytes, checksum1, checksum2 };
|
||||
}
|
||||
|
||||
for (const pageNo of committed.keys()) {
|
||||
if (pageNo > dbPageCount) committed.delete(pageNo);
|
||||
}
|
||||
return { transactions, pages: committed, dbPageCount, end };
|
||||
}
|
||||
|
||||
/**
|
||||
* SQLite's WAL checksum: Fibonacci-weighted sums over pairs of 32-bit words, continued from `s1` and `s2`. `bytes` must
|
||||
* be a multiple of 8 long, which headers, frame headers and pages all are.
|
||||
*/
|
||||
function walChecksum(bytes: Uint8Array, bigEndian: boolean, s1: number, s2: number): [number, number] {
|
||||
const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength);
|
||||
const littleEndian = !bigEndian;
|
||||
for (let offset = 0; offset < bytes.length; offset += 8) {
|
||||
s1 = (s1 + view.getUint32(offset, littleEndian) + s2) >>> 0;
|
||||
s2 = (s2 + view.getUint32(offset + 4, littleEndian) + s1) >>> 0;
|
||||
}
|
||||
return [s1, s2];
|
||||
}
|
||||
|
|
@ -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 }),
|
||||
};
|
||||
|
|
@ -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<void>((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)");
|
||||
|
|
|
|||
|
|
@ -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");
|
||||
});
|
||||
});
|
||||
|
|
@ -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, unknown>, string][] = [
|
||||
[{ ...valid, version: 2 }, "Unsupported container version 2"],
|
||||
[{ ...valid, kind: "snapshot" }, "Unknown container kind"],
|
||||
[{ ...valid, replicaId: "has/slash" }, "not a valid replica id"],
|
||||
[{ ...valid, replicaId: "" }, "not a valid replica id"],
|
||||
[{ ...valid, epoch: 0 }, "epoch must be an integer of at least 1"],
|
||||
[{ ...valid, fromSeq: 1.5 }, "fromSeq must be an integer"],
|
||||
[{ ...valid, toSeq: 120 }, "toSeq must be an integer of at least 121"],
|
||||
[{ ...valid, pageSize: 3000 }, "Invalid page size 3000"],
|
||||
[{ ...valid, pageSize: 131072 }, "Invalid page size 131072"],
|
||||
[{ ...valid, preChecksum: "ABCDEF0123456789" }, "preChecksum must be 16 lowercase hex digits"],
|
||||
[{ ...valid, compression: "gzip" }, "Unsupported page compression"],
|
||||
[{ ...valid, pageCount: 700 }, "more pages than the database"],
|
||||
[{ ...valid, toSeq: Number.MAX_SAFE_INTEGER + 2 }, "toSeq must be an integer"],
|
||||
];
|
||||
for (const [header, message] of invalid) {
|
||||
expect(() => validateHeader(header), message).toThrow(message);
|
||||
}
|
||||
expect(() => validateHeader(null)).toThrow("not an object");
|
||||
expect(() => validateHeader([])).toThrow("not an object");
|
||||
});
|
||||
|
||||
it("computes the same CRC-32 as zlib", () => {
|
||||
for (const bytes of [new Uint8Array(), new TextEncoder().encode("PISYNC01"), sqliteLikePage(3)]) {
|
||||
expect(crc32(bytes)).toBe(zlibCrc32(bytes));
|
||||
}
|
||||
});
|
||||
});
|
||||
|
|
@ -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<string>();
|
||||
const violations: string[] = [];
|
||||
const visit = async (file: string): Promise<void> => {
|
||||
if (visited.has(file)) return;
|
||||
visited.add(file);
|
||||
const source = await readFile(file, "utf8");
|
||||
for (const [, specifier = ""] of source.matchAll(/(?:import|export)[^;]*?from\s*["']([^"']+)["']/gu)) {
|
||||
if (specifier.startsWith(".")) await visit(resolve(dirname(file), specifier));
|
||||
else violations.push(`${file}: ${specifier}`);
|
||||
}
|
||||
};
|
||||
await visit(resolve(directory, "index.ts"));
|
||||
expect(violations).toEqual([]);
|
||||
expect(visited.size).toBeGreaterThan(4);
|
||||
});
|
||||
});
|
||||
|
|
@ -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<void> = 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();
|
||||
});
|
||||
});
|
||||
|
|
@ -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<number, Uint8Array>, pageSize: number, dbPageCount: number) {
|
||||
const result = new Uint8Array(dbPageCount * pageSize);
|
||||
result.set(image.subarray(0, Math.min(image.length, result.length)));
|
||||
for (const [pageNo, page] of pages) result.set(page, (pageNo - 1) * pageSize);
|
||||
return result;
|
||||
}
|
||||
|
||||
function header(wal: Uint8Array): WalHeader {
|
||||
const parsed = parseWalHeader(wal);
|
||||
if (parsed === undefined) throw new Error("WAL has no valid header");
|
||||
return parsed;
|
||||
}
|
||||
|
||||
function scanFrom(wal: Uint8Array, from: WalPosition) {
|
||||
return scanWal(header(wal), wal.subarray(from.offset), from);
|
||||
}
|
||||
|
||||
/** A WAL built by hand, for byte orders and shapes SQLite on this machine does not produce. */
|
||||
function syntheticWal(
|
||||
bigEndian: boolean,
|
||||
pageSize: number,
|
||||
frames: readonly { pageNo: number; commitSize: number; fill: number }[],
|
||||
) {
|
||||
const checksum = (bytes: Uint8Array, s: [number, number]): [number, number] => {
|
||||
const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength);
|
||||
let [s1, s2] = s;
|
||||
for (let offset = 0; offset < bytes.length; offset += 8) {
|
||||
s1 = (s1 + view.getUint32(offset, !bigEndian) + s2) >>> 0;
|
||||
s2 = (s2 + view.getUint32(offset + 4, !bigEndian) + s1) >>> 0;
|
||||
}
|
||||
return [s1, s2];
|
||||
};
|
||||
const bytes = new Uint8Array(WAL_HEADER_BYTES + frames.length * (WAL_FRAME_HEADER_BYTES + pageSize));
|
||||
const view = new DataView(bytes.buffer);
|
||||
view.setUint32(0, bigEndian ? 0x377f0683 : 0x377f0682);
|
||||
view.setUint32(4, 3007000);
|
||||
view.setUint32(8, pageSize);
|
||||
view.setUint32(12, 7);
|
||||
view.setUint32(16, 0x11111111);
|
||||
view.setUint32(20, 0x22222222);
|
||||
let s = checksum(bytes.subarray(0, 24), [0, 0]);
|
||||
view.setUint32(24, s[0]);
|
||||
view.setUint32(28, s[1]);
|
||||
frames.forEach((frame, index) => {
|
||||
const offset = WAL_HEADER_BYTES + index * (WAL_FRAME_HEADER_BYTES + pageSize);
|
||||
view.setUint32(offset, frame.pageNo);
|
||||
view.setUint32(offset + 4, frame.commitSize);
|
||||
view.setUint32(offset + 8, 0x11111111);
|
||||
view.setUint32(offset + 12, 0x22222222);
|
||||
bytes.fill(frame.fill, offset + WAL_FRAME_HEADER_BYTES, offset + WAL_FRAME_HEADER_BYTES + pageSize);
|
||||
s = checksum(bytes.subarray(offset, offset + 8), s);
|
||||
s = checksum(bytes.subarray(offset + WAL_FRAME_HEADER_BYTES, offset + WAL_FRAME_HEADER_BYTES + pageSize), s);
|
||||
view.setUint32(offset + 16, s[0]);
|
||||
view.setUint32(offset + 20, s[1]);
|
||||
});
|
||||
return bytes;
|
||||
}
|
||||
|
||||
describe("WAL reader", () => {
|
||||
let directory: string;
|
||||
let path: string;
|
||||
let database: DatabaseSync;
|
||||
|
||||
beforeEach(() => {
|
||||
directory = mkdtempSync(join(tmpdir(), "chord-wal-"));
|
||||
path = join(directory, "replica.sqlite");
|
||||
database = new DatabaseSync(path);
|
||||
database.exec("PRAGMA journal_mode = WAL; PRAGMA wal_autocheckpoint = 0");
|
||||
database.exec("CREATE TABLE rows (id INTEGER PRIMARY KEY, value BLOB)");
|
||||
database.exec("PRAGMA wal_checkpoint(TRUNCATE)");
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
if (database.isOpen) database.close();
|
||||
rmSync(directory, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
const insert = (count: number, size = 300) => {
|
||||
database.exec("BEGIN");
|
||||
for (let row = 0; row < count; row++)
|
||||
database.prepare("INSERT INTO rows (value) VALUES (randomblob(?))").run(size);
|
||||
database.exec("COMMIT");
|
||||
};
|
||||
const readWal = () => new Uint8Array(readFileSync(`${path}-wal`));
|
||||
|
||||
it("treats an empty or missing header as no WAL", () => {
|
||||
expect(parseWalHeader(new Uint8Array())).toBeUndefined();
|
||||
expect(parseWalHeader(new Uint8Array(32))).toBeUndefined();
|
||||
insert(1);
|
||||
const wal = readWal();
|
||||
expect(parseWalHeader(wal)).toMatchObject({ pageSize: 4096, bigEndianChecksums: false });
|
||||
const corrupted = wal.slice(0, 32);
|
||||
corrupted[17] = (corrupted[17] ?? 0) ^ 1;
|
||||
expect(parseWalHeader(corrupted)).toBeUndefined();
|
||||
});
|
||||
|
||||
it("reproduces the checkpointed database from the pages it scans", () => {
|
||||
const before = new Uint8Array(readFileSync(path));
|
||||
insert(40);
|
||||
database.exec("UPDATE rows SET value = randomblob(900) WHERE id % 3 = 0");
|
||||
insert(5, 5000);
|
||||
database.exec("DELETE FROM rows WHERE id % 2 = 0");
|
||||
|
||||
const wal = readWal();
|
||||
const parsed = header(wal);
|
||||
const scan = scanWal(parsed, wal.subarray(WAL_HEADER_BYTES), walStart(parsed));
|
||||
expect(scan.transactions).toBe(4);
|
||||
expect(scan.end.offset).toBe(wal.length);
|
||||
|
||||
database.exec("PRAGMA wal_checkpoint(TRUNCATE)");
|
||||
const after = new Uint8Array(readFileSync(path));
|
||||
expect(scan.dbPageCount * parsed.pageSize).toBe(after.length);
|
||||
expect(Buffer.from(applyPages(before, scan.pages, parsed.pageSize, scan.dbPageCount))).toEqual(
|
||||
Buffer.from(after),
|
||||
);
|
||||
});
|
||||
|
||||
it("drops pages beyond the final size when the database shrinks", () => {
|
||||
insert(30, 4000);
|
||||
database.exec("PRAGMA wal_checkpoint(TRUNCATE)");
|
||||
const before = new Uint8Array(readFileSync(path));
|
||||
database.exec("DELETE FROM rows; VACUUM");
|
||||
const wal = readWal();
|
||||
const scan = scanFrom(wal, walStart(header(wal)));
|
||||
expect(scan.dbPageCount * 4096).toBeLessThan(before.length);
|
||||
expect([...scan.pages.keys()].every((pageNo) => pageNo <= scan.dbPageCount)).toBe(true);
|
||||
database.exec("PRAGMA wal_checkpoint(TRUNCATE)");
|
||||
expect(Buffer.from(applyPages(before, scan.pages, 4096, scan.dbPageCount))).toEqual(readFileSync(path));
|
||||
});
|
||||
|
||||
it("continues from where the previous scan ended", () => {
|
||||
insert(3);
|
||||
insert(3);
|
||||
const first = readWal();
|
||||
const firstScan = scanFrom(first, walStart(header(first)));
|
||||
expect(firstScan.transactions).toBe(2);
|
||||
|
||||
insert(3);
|
||||
const second = readWal();
|
||||
const secondScan = scanFrom(second, firstScan.end);
|
||||
expect(secondScan.transactions).toBe(1);
|
||||
expect(secondScan.end.offset).toBe(second.length);
|
||||
|
||||
const nothing = scanFrom(second, secondScan.end);
|
||||
expect(nothing).toMatchObject({ transactions: 0, dbPageCount: 0, end: secondScan.end });
|
||||
expect(nothing.pages.size).toBe(0);
|
||||
});
|
||||
|
||||
it("stops at the last valid commit frame before a torn or uncommitted tail", () => {
|
||||
insert(3);
|
||||
const committed = readWal().length;
|
||||
insert(20, 2000);
|
||||
const wal = readWal();
|
||||
const frameBytes = WAL_FRAME_HEADER_BYTES + 4096;
|
||||
const start = walStart(header(wal));
|
||||
|
||||
// A transaction cut short: its frames without the commit frame, then half a frame.
|
||||
for (const length of [wal.length - frameBytes, wal.length - frameBytes / 2, committed + frameBytes]) {
|
||||
const scan = scanFrom(wal.subarray(0, length), start);
|
||||
expect(scan.transactions).toBe(1);
|
||||
expect(scan.end.offset).toBe(committed);
|
||||
}
|
||||
|
||||
// A flipped byte in the second transaction's last page fails its checksum.
|
||||
const corrupted = wal.slice();
|
||||
const last = wal.length - 1;
|
||||
corrupted[last] = (corrupted[last] ?? 0) ^ 0xff;
|
||||
expect(scanFrom(corrupted, start).end.offset).toBe(committed);
|
||||
});
|
||||
|
||||
it("ignores frames left over from before the WAL restarted", () => {
|
||||
insert(20, 2000);
|
||||
insert(20, 2000);
|
||||
const longWal = statSync(`${path}-wal`).size;
|
||||
database.exec("PRAGMA wal_checkpoint(RESTART)");
|
||||
insert(1);
|
||||
const wal = readWal();
|
||||
// SQLite rewrote the WAL from the start with new salts; the old frames are still in the file.
|
||||
expect(wal.length).toBe(longWal);
|
||||
const scan = scanFrom(wal, walStart(header(wal)));
|
||||
expect(scan.transactions).toBe(1);
|
||||
expect(scan.end.offset).toBeLessThan(wal.length);
|
||||
});
|
||||
|
||||
it("validates big-endian and little-endian checksums", () => {
|
||||
const frames = [
|
||||
{ pageNo: 1, commitSize: 0, fill: 1 },
|
||||
{ pageNo: 2, commitSize: 2, fill: 2 },
|
||||
{ pageNo: 2, commitSize: 0, fill: 3 },
|
||||
{ pageNo: 3, commitSize: 3, fill: 4 },
|
||||
{ pageNo: 1, commitSize: 0, fill: 5 },
|
||||
];
|
||||
for (const bigEndian of [true, false]) {
|
||||
const wal = syntheticWal(bigEndian, 512, frames);
|
||||
const parsed = header(wal);
|
||||
expect(parsed).toMatchObject({ bigEndianChecksums: bigEndian, checkpointSequence: 7, pageSize: 512 });
|
||||
const scan = scanWal(parsed, wal.subarray(WAL_HEADER_BYTES), walStart(parsed));
|
||||
expect(scan.transactions).toBe(2);
|
||||
expect(scan.dbPageCount).toBe(3);
|
||||
expect([...scan.pages].map(([pageNo, page]) => [pageNo, page[0]])).toEqual([
|
||||
[1, 1],
|
||||
[2, 3],
|
||||
[3, 4],
|
||||
]);
|
||||
}
|
||||
// The same bytes read with the other byte order do not validate.
|
||||
const wal = syntheticWal(true, 512, frames);
|
||||
const view = new DataView(wal.buffer);
|
||||
view.setUint32(0, 0x377f0682);
|
||||
expect(parseWalHeader(wal)).toBeUndefined();
|
||||
});
|
||||
});
|
||||
Loading…
Add table
Add a link
Reference in a new issue