// src/gen-codec.ts // // Binary codecs for the per-file payloads of a persistent index generation // (stage 5): the store image and every derived-index image. Each file is a // self-describing envelope — 4-byte magic, u32 format version, payload, and a // trailing crc32 over everything before it — so a torn or mismatched write is // detected at load and only ever costs that one file (a corrupt store image // rejects the whole generation; a corrupt index image rebuilds that index // from the loaded store; neither touches the authoritative snapshot/WAL). // // Writers stream to disk in ~1 MiB writev batches with a running crc (large // store images never sit wholly in RAM twice); readers load the whole file — // the payload becomes in-memory state anyway — and verify the crc up front. // // Key encoding pitfall: MiniDb's canonical key strings are BINARY strings // (each char code == one utf8 byte of the key). They are always written as // raw bytes (u16 length + Buffer.from(k, 'binary')); writing them as utf8 // would corrupt every non-ASCII key. Genuine text (index names, fields, // scalar keys, group/order strings, terms) is written as utf8. // // Internal to the package — NOT re-exported from the root entry point. import fs from 'node:fs/promises'; import fsSync from 'node:fs'; import type { FileHandle } from 'node:fs/promises'; import { crc32 } from './crc32.ts'; import type { ValueRef } from './store.js'; /** Thrown by every reader on a malformed/truncated/crc-mismatched generation * file. Distinct from CorruptFrameError so the loader can route precisely. */ export class GenerationCorruptError extends Error { readonly code = 'GENERATION_CORRUPT'; constructor(message: string) { super(message); this.name = 'GenerationCorruptError'; } } // ---- byte-level writer/reader ---------------------------------------------- /** Growable record encoder; one generation file record must fit one chunk. */ class ByteWriter { buf: Buffer; off = 0; constructor(sizeHint = 64) { this.buf = Buffer.allocUnsafe(sizeHint); } ensure(n: number): void { if (this.off + n <= this.buf.length) return; let cap = this.buf.length * 2; while (cap < this.off + n) cap *= 2; const next = Buffer.allocUnsafe(cap); this.buf.copy(next, 0, 0, this.off); this.buf = next; } u8(v: number): void { this.ensure(1); this.buf.writeUInt8(v, this.off); this.off += 1; } u16(v: number): void { this.ensure(2); this.buf.writeUInt16LE(v, this.off); this.off += 2; } u32(v: number): void { this.ensure(4); this.buf.writeUInt32LE(v >>> 0, this.off); this.off += 4; } u64(v: number): void { this.ensure(8); this.buf.writeBigUInt64LE(BigInt(v), this.off); this.off += 8; } i64(v: number): void { this.ensure(8); this.buf.writeBigInt64LE(BigInt(v), this.off); this.off += 8; } f64(v: number): void { this.ensure(8); this.buf.writeDoubleLE(v, this.off); this.off += 8; } bytes(b: Buffer): void { this.ensure(b.length); b.copy(this.buf, this.off); this.off += b.length; } /** A canonical (binary) key string: u16 byte length + raw bytes. */ key(kstr: string): void { const b = Buffer.from(kstr, 'binary'); this.u16(b.length); this.bytes(b); } /** A genuine text string: u32 utf8 byte length + utf8 bytes. */ text(s: string): void { const b = Buffer.from(s, 'utf8'); this.u32(b.length); this.bytes(b); } /** A term (bounded by the text index's uint16 limit): u16 + utf8. */ term(s: string): void { const b = Buffer.from(s, 'utf8'); this.u16(b.length); this.bytes(b); } } export class ByteReader { off = 0; constructor(readonly buf: Buffer) {} private need(n: number): void { if (this.off + n > this.buf.length) throw new GenerationCorruptError('generation file truncated'); } u8(): number { this.need(1); const v = this.buf.readUInt8(this.off); this.off += 1; return v; } u16(): number { this.need(2); const v = this.buf.readUInt16LE(this.off); this.off += 2; return v; } u32(): number { this.need(4); const v = this.buf.readUInt32LE(this.off); this.off += 4; return v; } u64(): number { this.need(8); const v = Number(this.buf.readBigUInt64LE(this.off)); this.off += 8; return v; } i64(): number { this.need(8); const v = Number(this.buf.readBigInt64LE(this.off)); this.off += 8; return v; } f64(): number { this.need(8); const v = this.buf.readDoubleLE(this.off); this.off += 8; return v; } bytes(n: number): Buffer { this.need(n); const b = this.buf.subarray(this.off, this.off + n); this.off += n; return b; } key(): string { const n = this.u16(); return this.bytes(n).toString('binary'); } text(): string { const n = this.u32(); return this.bytes(n).toString('utf8'); } term(): string { const n = this.u16(); return this.bytes(n).toString('utf8'); } get done(): boolean { return this.off === this.buf.length; } } // ---- file envelope + streaming writer -------------------------------------- const FLUSH_BYTES = 1 << 20; /** Streaming generation-file writer: envelope header, ~1 MiB writev batches, * running crc32, fsync on finish. The crc/bytes it reports feed the * manifest's per-file integrity records. */ export class GenFileWriter { private readonly chunks: Buffer[] = []; private queued = 0; private crc = 0; bytes = 0; private readonly rec = new ByteWriter(256); private constructor( private readonly fh: FileHandle, magic: string, version: number, ) { const w = new ByteWriter(8); for (let i = 0; i < 4; i++) w.u8(magic.charCodeAt(i)); w.u32(version); const head = w.buf.subarray(0, w.off); this.chunks.push(Buffer.from(head)); this.queued = head.length; } static async open(path: string, magic: string, version: number): Promise { if (magic.length !== 4) throw new RangeError('generation file magic must be 4 chars'); const fh = await fs.open(path, 'w'); try { return new GenFileWriter(fh, magic, version); } catch (e) { await fh.close().catch(() => {}); throw e; } } /** Encode one record with `encode(w)` and queue it for the next batch. */ async writeRecord(encode: (w: ByteWriter) => void): Promise { const w = this.rec; w.off = 0; encode(w); const b = Buffer.from(w.buf.subarray(0, w.off)); this.chunks.push(b); this.queued += b.length; if (this.queued >= FLUSH_BYTES) await this.flush(); } private async flush(): Promise { if (this.chunks.length === 0) return; const bufs = this.chunks.splice(0, this.chunks.length); this.queued = 0; for (const b of bufs) this.crc = crc32(b, this.crc); // Index-based consumption walk: a flush can carry tens of thousands of // tiny record buffers (the text docs image averages ~14 B/record), and a // shift()-per-buffer loop is O(n²) on the main thread — measured 20+s of // self time across one 1M-doc generation publish, surfacing as the // multi-second event-loop stalls at the publish tail. let idx = 0; let off = 0; while (idx < bufs.length) { const toWrite = off > 0 ? [bufs[idx]!.subarray(off), ...bufs.slice(idx + 1)] : idx === 0 ? bufs : bufs.slice(idx); const { bytesWritten } = await this.fh.writev(toWrite); if (bytesWritten === 0) throw new Error('generation file writev made no progress (short write)'); this.bytes += bytesWritten; let rem = bytesWritten; while (rem > 0 && idx < bufs.length) { const left = bufs[idx]!.length - off; if (rem < left) { off += rem; rem = 0; } else { rem -= left; idx++; off = 0; } } } } /** Flush, append the crc trailer, fsync, close. Returns the manifest's * per-file integrity record ({ bytes, crc32 }). */ async finish(): Promise<{ bytes: number; crc32: number }> { try { await this.flush(); const trailer = Buffer.allocUnsafe(4); trailer.writeUInt32LE(this.crc >>> 0, 0); let written = 0; while (written < 4) { const { bytesWritten } = await this.fh.write(trailer, written); if (bytesWritten === 0) throw new Error('generation file write made no progress (short write)'); written += bytesWritten; } this.bytes += 4; await this.fh.sync(); return { bytes: this.bytes, crc32: this.crc >>> 0 }; } finally { await this.fh.close().catch(() => {}); } } /** Abort without finishing: close (the caller removes the tmp file). */ async abort(): Promise { await this.fh.close().catch(() => {}); } } /** A verified generation file: the payload reader plus the whole-file * integrity record (byte length + crc32) for cross-checking against the * manifest. */ export interface VerifiedGenerationFile { payload: ByteReader; bytes: number; crc32: number; } /** Read + verify one whole generation file: magic, version, and the crc * trailer. Returns the payload reader and the computed integrity record. */ export async function readGenerationFile(path: string, magic: string, version: number): Promise { let buf: Buffer; try { buf = await fs.readFile(path); } catch (e) { throw new GenerationCorruptError(`generation file unreadable: ${(e as NodeJS.ErrnoException).code ?? String(e)}`); } return parseGenerationBuffer(buf, magic, version); } /** Pure-buffer variant of readGenerationFile (tests, in-memory verification). */ export function parseGenerationBuffer(buf: Buffer, magic: string, version: number): VerifiedGenerationFile { if (buf.length < 8 + 4) throw new GenerationCorruptError('generation file too short'); for (let i = 0; i < 4; i++) { if (buf.readUInt8(i) !== magic.charCodeAt(i)) throw new GenerationCorruptError(`bad magic (want ${magic})`); } if (buf.readUInt32LE(4) !== version) throw new GenerationCorruptError(`unsupported file version (want ${version})`); const stored = buf.readUInt32LE(buf.length - 4); const calc = crc32(buf.subarray(0, buf.length - 4)); if (stored !== calc) throw new GenerationCorruptError('generation file crc mismatch'); return { payload: new ByteReader(buf.subarray(8, buf.length - 4)), bytes: buf.length, crc32: stored }; } /** Read + verify a generation file AND cross-check it against the manifest's * integrity record — a file swapped in from another generation (or a * manifest from another build) is caught here. */ export async function readGenerationFileChecked( path: string, magic: string, version: number, expected: { bytes: number; crc32: number }, ): Promise { const f = await readGenerationFile(path, magic, version); if (f.bytes !== expected.bytes || f.crc32 !== expected.crc32) { throw new GenerationCorruptError('generation file does not match manifest record'); } return f.payload; } /** Sliced variant of readGenerationFileChecked (stage 6): the file is read * and crc'd in 1 MiB slices with event-loop yields, so verifying a large * dictionary/docs image never blocks the loop for the whole pass. The * payload still lands in RAM wholesale (it becomes in-memory state), only * the read + verification is sliced. */ export async function readGenerationFileCheckedAsync( path: string, magic: string, version: number, expected: { bytes: number; crc32: number }, ): Promise { let buf: Buffer; try { buf = await fs.readFile(path); } catch (e) { throw new GenerationCorruptError(`generation file unreadable: ${(e as NodeJS.ErrnoException).code ?? String(e)}`); } if (buf.length < 8 + 4) throw new GenerationCorruptError('generation file too short'); for (let i = 0; i < 4; i++) { if (buf.readUInt8(i) !== magic.charCodeAt(i)) throw new GenerationCorruptError(`bad magic (want ${magic})`); } if (buf.readUInt32LE(4) !== version) throw new GenerationCorruptError(`unsupported file version (want ${version})`); const stored = buf.readUInt32LE(buf.length - 4); let crc = 0; const SLICE = 1 << 20; for (let pos = 0; pos < buf.length - 4; pos += SLICE) { crc = crc32(buf.subarray(pos, Math.min(pos + SLICE, buf.length - 4)), crc); await new Promise((r) => setImmediate(r)); } if (stored !== crc) throw new GenerationCorruptError('generation file crc mismatch'); if (buf.length !== expected.bytes || stored !== expected.crc32) { throw new GenerationCorruptError('generation file does not match manifest record'); } return new ByteReader(buf.subarray(8, buf.length - 4)); } const yieldToLoop = (): Promise => new Promise((r) => setImmediate(r)); /** Verify a raw file against the manifest's integrity record by streaming * its bytes (bounded memory — used for the postings files, which have no * envelope of their own and are otherwise only verified lazily, per-record, * at search time). One sequential read: cheap insurance that a corrupt base * is discarded and rebuilt at OPEN, not discovered mid-query. */ export function verifyFileIntegritySync(path: string, expected: { bytes: number; crc32: number }): void { const fd = fsSync.openSync(path, 'r'); try { const st = fsSync.fstatSync(fd); if (st.size !== expected.bytes) throw new GenerationCorruptError('file size does not match manifest record'); let crc = 0; const buf = Buffer.allocUnsafe(1 << 16); let pos = 0; while (pos < st.size) { const n = fsSync.readSync(fd, buf, 0, Math.min(buf.length, st.size - pos), pos); if (n === 0) throw new GenerationCorruptError('file shrank during integrity check'); crc = crc32(buf.subarray(0, n), crc); pos += n; } if ((crc >>> 0) !== expected.crc32) throw new GenerationCorruptError('file crc does not match manifest record'); } finally { fsSync.closeSync(fd); } } /** The read/verify chunk size of verifyFileIntegrityAsync: one crc32 slice per * chunk, then a yield — a large postings file verifies in bounded ~ms * slices instead of one synchronous pass. */ const VERIFY_CHUNK_BYTES = 1 << 20; /** Async sliced variant of verifyFileIntegritySync (the open-time main-thread * path): bounded 1 MiB positioned reads, a crc32 slice and an event-loop * yield per chunk, so verifying a large postings file never blocks the loop * for the whole pass. Identical error semantics. */ export async function verifyFileIntegrityAsync(path: string, expected: { bytes: number; crc32: number }): Promise { const fh = await fs.open(path, 'r'); try { const st = await fh.stat(); if (st.size !== expected.bytes) throw new GenerationCorruptError('file size does not match manifest record'); let crc = 0; const buf = Buffer.allocUnsafe(Math.min(VERIFY_CHUNK_BYTES, Math.max(st.size, 1))); let pos = 0; while (pos < st.size) { const { bytesRead } = await fh.read(buf, 0, Math.min(buf.length, st.size - pos), pos); if (bytesRead === 0) throw new GenerationCorruptError('file shrank during integrity check'); crc = crc32(buf.subarray(0, bytesRead), crc); pos += bytesRead; await yieldToLoop(); } if ((crc >>> 0) !== expected.crc32) throw new GenerationCorruptError('file crc does not match manifest record'); } finally { await fh.close().catch(() => {}); } } // ---- store image ------------------------------------------------------------ const STORE_MAGIC = 'MDGS'; // v4: the dt header fields are u32 (column count, metaLen, column-name // length) — the v3 u8/u16 widths threw a RangeError mid-build on extreme // shapes (many/wide dt columns), and every later build would fail the same // deterministic way. v3: binary dt (no JSON per record). v2: `{"dt":...}` // meta JSON. Older files are rejected as a whole (unsupported version), // never silently misread. export const STORE_VERSION = 4; /** One record of the store image: the exact data needed to rebuild a * StoreRecord. `ref` is a memory ref (inline value bytes) or a disk ref into * the generation's snapshot / the anchored WAL. `metaBytes` is the record's * Store-side dt accounting value (byte length of the `{"dt":...}` meta JSON, * 0 for none) — precomputed at build time so the load never re-stringifies. */ export interface StoreImageRecord { kstr: string; ref: ValueRef; expireAt: number; dt: Record | null; /** Read side: the record's Store-side dt accounting value (0 for none). * Ignored on write (recomputed from `dt`). */ metaBytes?: number; } const TAG_INLINE = 0; const TAG_SNAPSHOT_LOC = 1; const TAG_WAL_LOC = 2; /** The Store's dt accounting value for one record (mirrors Store.metaBytes: * byte length of the canonical `{"dt":...}` meta JSON, 0 for none). */ function dtMetaBytes(dt: Record | null): number { return dt ? Buffer.byteLength(JSON.stringify({ dt }), 'utf8') : 0; } /** Stream the store image to `path`. `records` must yield live records in * ascending canonical-key order (the load path bulk-builds the ordered index * from file order). Returns the manifest file info + the record count. */ export async function writeStoreImage( path: string, records: Iterable, ): Promise<{ bytes: number; crc32: number; count: number }> { const w = await GenFileWriter.open(path, STORE_MAGIC, STORE_VERSION); let count = 0; try { for (const r of records) { await w.writeRecord((b) => { b.key(r.kstr); b.i64(r.expireAt); const cols = r.dt ? Object.entries(r.dt) : []; b.u32(dtMetaBytes(r.dt)); b.u32(cols.length); for (const [name, ms] of cols) { const nb = Buffer.from(name, 'utf8'); b.u32(nb.length); b.bytes(nb); b.f64(ms); } if (r.ref.kind === 'memory') { b.u8(TAG_INLINE); b.u32(r.ref.value.length); b.bytes(r.ref.value); } else { b.u8(r.ref.loc.file === 'snapshot' ? TAG_SNAPSHOT_LOC : TAG_WAL_LOC); b.u64(r.ref.loc.off); b.u32(r.ref.loc.len); } }); count++; } const info = await w.finish(); return { ...info, count }; } catch (e) { await w.abort(); throw e; } } /** Parse a store image payload, yielding records in file (sorted) order. * Values are COPIED out of the shared file buffer (the store must own its * memory refs) and metaBytes carries the exact Store accounting hint. */ export function* readStoreImage(r: ByteReader): Generator { while (!r.done) { const kstr = r.key(); const expireAt = r.i64(); const metaBytes = r.u32(); const colCount = r.u32(); let dt: Record | null = null; if (colCount > 0) { dt = {}; for (let i = 0; i < colCount; i++) { const nameLen = r.u32(); const name = r.bytes(nameLen).toString('utf8'); dt[name] = r.f64(); } } const tag = r.u8(); let ref: ValueRef; if (tag === TAG_INLINE) { const len = r.u32(); ref = { kind: 'memory', value: Buffer.from(r.bytes(len)) }; } else if (tag === TAG_SNAPSHOT_LOC || tag === TAG_WAL_LOC) { const off = r.u64(); const len = r.u32(); ref = { kind: 'disk', loc: { file: tag === TAG_SNAPSHOT_LOC ? 'snapshot' : 'wal', off, len } }; } else { throw new GenerationCorruptError(`store image: unknown value tag ${tag}`); } yield { kstr, ref, expireAt, dt, metaBytes }; } } // ---- dt index image --------------------------------------------------------- const DT_MAGIC = 'MDGD'; const DT_VERSION = 1; export interface DtImageColumn { name: string; /** (ms, key) pairs sorted ascending — the skiplist's natural order. */ entries: { ms: number; key: string }[]; } export async function writeDtIndexImage(path: string, cols: DtImageColumn[]): Promise<{ bytes: number; crc32: number }> { const w = await GenFileWriter.open(path, DT_MAGIC, DT_VERSION); try { await w.writeRecord((b) => b.u32(cols.length)); for (const c of cols) { await w.writeRecord((b) => { b.text(c.name); b.u64(c.entries.length); }); for (const e of c.entries) { await w.writeRecord((b) => { b.f64(e.ms); b.key(e.key); }); } } return await w.finish(); } catch (e) { await w.abort(); throw e; } } export function readDtIndexImage(r: ByteReader): DtImageColumn[] { const colCount = r.u32(); const cols: DtImageColumn[] = []; for (let i = 0; i < colCount; i++) { const name = r.text(); const n = r.u64(); const entries: DtImageColumn['entries'] = []; for (let j = 0; j < n; j++) entries.push({ ms: r.f64(), key: r.key() }); cols.push({ name, entries }); } if (!r.done) throw new GenerationCorruptError('dt index image: trailing bytes'); return cols; } // ---- secondary index image -------------------------------------------------- const SECONDARY_MAGIC = 'MDSI'; const SECONDARY_VERSION = 1; export interface SecondaryImageIndex { name: string; field: string; type: 'equality' | 'range'; unique: boolean; sparse: boolean; /** Equality state: (scalarKey -> pks). Null for range indexes. */ equality: { scalarKey: string; pks: string[] }[] | null; /** Range state: (value, pk) pairs sorted ascending. Null for equality. */ range: { value: number; pk: string }[] | null; } export async function writeSecondaryIndexImage( path: string, indexes: SecondaryImageIndex[], ): Promise<{ bytes: number; crc32: number }> { const w = await GenFileWriter.open(path, SECONDARY_MAGIC, SECONDARY_VERSION); try { await w.writeRecord((b) => b.u32(indexes.length)); for (const idx of indexes) { await w.writeRecord((b) => { b.text(idx.name); b.text(idx.field); b.u8(idx.type === 'range' ? 2 : 1); b.u8((idx.unique ? 1 : 0) | (idx.sparse ? 2 : 0)); }); if (idx.type === 'equality') { const values = idx.equality ?? []; await w.writeRecord((b) => b.u64(values.length)); for (const v of values) { await w.writeRecord((b) => { b.text(v.scalarKey); b.u64(v.pks.length); }); for (const pk of v.pks) await w.writeRecord((b) => b.key(pk)); } } else { const entries = idx.range ?? []; await w.writeRecord((b) => b.u64(entries.length)); for (const e of entries) { await w.writeRecord((b) => { b.f64(e.value); b.key(e.pk); }); } } } return await w.finish(); } catch (e) { await w.abort(); throw e; } } export function readSecondaryIndexImage(r: ByteReader): SecondaryImageIndex[] { const count = r.u32(); const out: SecondaryImageIndex[] = []; for (let i = 0; i < count; i++) { const name = r.text(); const field = r.text(); const typeTag = r.u8(); const flags = r.u8(); const type = typeTag === 2 ? 'range' : typeTag === 1 ? 'equality' : null; if (type === null) throw new GenerationCorruptError(`secondary image: unknown index type ${typeTag}`); let equality: SecondaryImageIndex['equality'] = null; let range: SecondaryImageIndex['range'] = null; if (type === 'equality') { equality = []; const valueCount = r.u64(); for (let v = 0; v < valueCount; v++) { const scalarKey = r.text(); const pkCount = r.u64(); const pks: string[] = []; for (let p = 0; p < pkCount; p++) pks.push(r.key()); equality.push({ scalarKey, pks }); } } else { range = []; const n = r.u64(); for (let j = 0; j < n; j++) range.push({ value: r.f64(), pk: r.key() }); } out.push({ name, field, type, unique: (flags & 1) !== 0, sparse: (flags & 2) !== 0, equality, range }); } if (!r.done) throw new GenerationCorruptError('secondary index image: trailing bytes'); return out; } /** Sliced variant of readSecondaryIndexImage (stage 6): identical result, * with event-loop yields every `yieldEvery` parsed entries so a large image * never parses in one synchronous run. */ export async function readSecondaryIndexImageAsync(r: ByteReader, yieldEvery = 32768): Promise { const count = r.u32(); const out: SecondaryImageIndex[] = []; let n = 0; const tick = async (): Promise => { if (++n % yieldEvery === 0) await yieldToLoop(); }; for (let i = 0; i < count; i++) { const name = r.text(); const field = r.text(); const typeTag = r.u8(); const flags = r.u8(); const type = typeTag === 2 ? 'range' : typeTag === 1 ? 'equality' : null; if (type === null) throw new GenerationCorruptError(`secondary image: unknown index type ${typeTag}`); let equality: SecondaryImageIndex['equality'] = null; let range: SecondaryImageIndex['range'] = null; if (type === 'equality') { equality = []; const valueCount = r.u64(); for (let v = 0; v < valueCount; v++) { const scalarKey = r.text(); const pkCount = r.u64(); const pks: string[] = []; for (let p = 0; p < pkCount; p++) { pks.push(r.key()); await tick(); } equality.push({ scalarKey, pks }); } } else { range = []; const m = r.u64(); for (let j = 0; j < m; j++) { range.push({ value: r.f64(), pk: r.key() }); await tick(); } } out.push({ name, field, type, unique: (flags & 1) !== 0, sparse: (flags & 2) !== 0, equality, range }); } if (!r.done) throw new GenerationCorruptError('secondary index image: trailing bytes'); return out; } // ---- compound index image --------------------------------------------------- const COMPOUND_MAGIC = 'MDCI'; const COMPOUND_VERSION = 1; export type CompoundImageGroupValue = number | string | boolean | null; export interface CompoundImageIndex { name: string; groupBy: string; orderBy: string; orderType: 'number' | 'string'; groups: { group: CompoundImageGroupValue; entries: { order: number | string; pk: string }[] }[]; } const GTAG_NUMBER = 1; const GTAG_STRING = 2; const GTAG_FALSE = 3; const GTAG_TRUE = 4; const GTAG_NULL = 5; function writeGroupValue(b: ByteWriter, v: CompoundImageGroupValue): void { if (v === null) { b.u8(GTAG_NULL); } else if (typeof v === 'number') { b.u8(GTAG_NUMBER); b.f64(v); } else if (typeof v === 'string') { b.u8(GTAG_STRING); b.text(v); } else if (v === false) { b.u8(GTAG_FALSE); } else { b.u8(GTAG_TRUE); } } function readGroupValue(r: ByteReader): CompoundImageGroupValue { const tag = r.u8(); if (tag === GTAG_NUMBER) return r.f64(); if (tag === GTAG_STRING) return r.text(); if (tag === GTAG_FALSE) return false; if (tag === GTAG_TRUE) return true; if (tag === GTAG_NULL) return null; throw new GenerationCorruptError(`compound image: unknown group tag ${tag}`); } export async function writeCompoundIndexImage( path: string, indexes: CompoundImageIndex[], ): Promise<{ bytes: number; crc32: number }> { const w = await GenFileWriter.open(path, COMPOUND_MAGIC, COMPOUND_VERSION); try { await w.writeRecord((b) => b.u32(indexes.length)); for (const idx of indexes) { await w.writeRecord((b) => { b.text(idx.name); b.text(idx.groupBy); b.text(idx.orderBy); b.u8(idx.orderType === 'string' ? 2 : 1); b.u64(idx.groups.length); }); for (const g of idx.groups) { await w.writeRecord((b) => { writeGroupValue(b, g.group); b.u64(g.entries.length); }); for (const e of g.entries) { await w.writeRecord((b) => { if (idx.orderType === 'string') b.text(String(e.order)); else b.f64(Number(e.order)); b.key(e.pk); }); } } } return await w.finish(); } catch (e) { await w.abort(); throw e; } } export function readCompoundIndexImage(r: ByteReader): CompoundImageIndex[] { const count = r.u32(); const out: CompoundImageIndex[] = []; for (let i = 0; i < count; i++) { const name = r.text(); const groupBy = r.text(); const orderBy = r.text(); const ot = r.u8(); const orderType = ot === 2 ? 'string' : ot === 1 ? 'number' : null; if (orderType === null) throw new GenerationCorruptError(`compound image: unknown order type ${ot}`); const groupCount = r.u64(); const groups: CompoundImageIndex['groups'] = []; for (let g = 0; g < groupCount; g++) { const group = readGroupValue(r); const n = r.u64(); const entries: { order: number | string; pk: string }[] = []; for (let j = 0; j < n; j++) { const order = orderType === 'string' ? r.text() : r.f64(); entries.push({ order, pk: r.key() }); } groups.push({ group, entries }); } out.push({ name, groupBy, orderBy, orderType, groups }); } if (!r.done) throw new GenerationCorruptError('compound index image: trailing bytes'); return out; } /** Sliced variant of readCompoundIndexImage (stage 6): identical result, with * event-loop yields every `yieldEvery` parsed entries. */ export async function readCompoundIndexImageAsync(r: ByteReader, yieldEvery = 32768): Promise { const count = r.u32(); const out: CompoundImageIndex[] = []; let n = 0; const tick = async (): Promise => { if (++n % yieldEvery === 0) await yieldToLoop(); }; for (let i = 0; i < count; i++) { const name = r.text(); const groupBy = r.text(); const orderBy = r.text(); const ot = r.u8(); const orderType = ot === 2 ? 'string' : ot === 1 ? 'number' : null; if (orderType === null) throw new GenerationCorruptError(`compound image: unknown order type ${ot}`); const groupCount = r.u64(); const groups: CompoundImageIndex['groups'] = []; for (let g = 0; g < groupCount; g++) { const group = readGroupValue(r); const m = r.u64(); const entries: { order: number | string; pk: string }[] = []; for (let j = 0; j < m; j++) { const order = orderType === 'string' ? r.text() : r.f64(); entries.push({ order, pk: r.key() }); await tick(); } groups.push({ group, entries }); } out.push({ name, groupBy, orderBy, orderType, groups }); } if (!r.done) throw new GenerationCorruptError('compound index image: trailing bytes'); return out; } // ---- text index images ------------------------------------------------------ const TEXT_DICT_MAGIC = 'MDTD'; const TEXT_DICT_VERSION = 1; const TEXT_DOCS_MAGIC = 'MDTC'; const TEXT_DOCS_VERSION = 1; export interface TextDictImageEntry { term: string; off: number; len: number; df: number; } export async function writeTextDictionaryImage( path: string, entries: Iterable, ): Promise<{ bytes: number; crc32: number }> { const w = await GenFileWriter.open(path, TEXT_DICT_MAGIC, TEXT_DICT_VERSION); try { for (const e of entries) { await w.writeRecord((b) => { b.term(e.term); b.u64(e.off); b.u32(e.len); b.u32(e.df); }); } return await w.finish(); } catch (e) { await w.abort(); throw e; } } export function readTextDictionaryImage(r: ByteReader): TextDictImageEntry[] { const out: TextDictImageEntry[] = []; while (!r.done) out.push({ term: r.term(), off: r.u64(), len: r.u32(), df: r.u32() }); return out; } /** Sliced variant (stage 6): identical result, with event-loop yields every * `yieldEvery` entries so a million-term dictionary never parses in one * synchronous run. */ export async function readTextDictionaryImageAsync(r: ByteReader, yieldEvery = 65536): Promise { const out: TextDictImageEntry[] = []; let n = 0; while (!r.done) { out.push({ term: r.term(), off: r.u64(), len: r.u32(), df: r.u32() }); if (++n % yieldEvery === 0) await new Promise((res) => setImmediate(res)); } return out; } /** The per-doc table plus the write-buffer state of a text index: docID -> * key (undefined = hole), docID -> token count, the live-doc count N, the * tombstoned docIDs, and the in-memory delta (term -> (docID -> freq)). * Serializing the delta + tombstones makes the loaded index EXACTLY equal to * the live one at seal time — including writes that landed while the * generation was being built. */ export interface TextDocsImage { keys: (string | undefined)[]; docLens: (number | undefined)[]; liveCount: number; removed: number[]; delta: { term: string; docs: { docID: number; freq: number }[] }[]; } export async function writeTextDocsImage(path: string, image: TextDocsImage): Promise<{ bytes: number; crc32: number }> { const w = await GenFileWriter.open(path, TEXT_DOCS_MAGIC, TEXT_DOCS_VERSION); try { await w.writeRecord((b) => { b.u64(image.keys.length); b.u64(image.liveCount); b.u64(image.removed.length); b.u64(image.delta.length); }); for (let i = 0; i < image.keys.length; i++) { const k = image.keys[i]; await w.writeRecord((b) => { if (k === undefined) { b.u8(0); } else { b.u8(1); b.key(k); } b.u32(image.docLens[i] ?? 0); }); } for (const id of image.removed) await w.writeRecord((b) => b.u32(id)); for (const d of image.delta) { await w.writeRecord((b) => { b.term(d.term); b.u64(d.docs.length); }); for (const doc of d.docs) { await w.writeRecord((b) => { b.u32(doc.docID); b.u32(doc.freq); }); } } return await w.finish(); } catch (e) { await w.abort(); throw e; } } export function readTextDocsImage(r: ByteReader): TextDocsImage { const docCount = r.u64(); const liveCount = r.u64(); const removedCount = r.u64(); const deltaCount = r.u64(); const keys: (string | undefined)[] = []; const docLens: (number | undefined)[] = []; for (let i = 0; i < docCount; i++) { const present = r.u8(); if (present === 1) { keys.push(r.key()); docLens.push(r.u32()); } else if (present === 0) { keys.push(undefined); const len = r.u32(); docLens.push(len === 0 ? undefined : len); } else { throw new GenerationCorruptError(`text docs image: unknown presence tag ${present}`); } } const removed: number[] = []; for (let i = 0; i < removedCount; i++) removed.push(r.u32()); const delta: TextDocsImage['delta'] = []; for (let i = 0; i < deltaCount; i++) { const term = r.term(); const n = r.u64(); const docs: { docID: number; freq: number }[] = []; for (let j = 0; j < n; j++) docs.push({ docID: r.u32(), freq: r.u32() }); delta.push({ term, docs }); } if (!r.done) throw new GenerationCorruptError('text docs image: trailing bytes'); return { keys, docLens, liveCount, removed, delta }; } /** Sliced variant of readTextDocsImage (stage 6): identical result, with * event-loop yields every `yieldEvery` parsed entries so a large doc table * or delta never parses in one synchronous run. */ export async function readTextDocsImageAsync(r: ByteReader, yieldEvery = 32768): Promise { const docCount = r.u64(); const liveCount = r.u64(); const removedCount = r.u64(); const deltaCount = r.u64(); let n = 0; const tick = async (): Promise => { if (++n % yieldEvery === 0) await yieldToLoop(); }; const keys: (string | undefined)[] = []; const docLens: (number | undefined)[] = []; for (let i = 0; i < docCount; i++) { const present = r.u8(); if (present === 1) { keys.push(r.key()); docLens.push(r.u32()); } else if (present === 0) { keys.push(undefined); const len = r.u32(); docLens.push(len === 0 ? undefined : len); } else { throw new GenerationCorruptError(`text docs image: unknown presence tag ${present}`); } await tick(); } const removed: number[] = []; for (let i = 0; i < removedCount; i++) { removed.push(r.u32()); await tick(); } const delta: TextDocsImage['delta'] = []; for (let i = 0; i < deltaCount; i++) { const term = r.term(); const m = r.u64(); const docs: { docID: number; freq: number }[] = []; for (let j = 0; j < m; j++) { docs.push({ docID: r.u32(), freq: r.u32() }); await tick(); } delta.push({ term, docs }); } if (!r.done) throw new GenerationCorruptError('text docs image: trailing bytes'); return { keys, docLens, liveCount, removed, delta }; }