diff --git a/packages/ledger/ledger.ts b/packages/ledger/ledger.ts index 453db4b..2bcc69a 100644 --- a/packages/ledger/ledger.ts +++ b/packages/ledger/ledger.ts @@ -5,26 +5,14 @@ import { createHash, randomUUID } from "node:crypto"; import { execFileSync } from "node:child_process"; import { tmpdir, userInfo } from "node:os"; import { DatabaseSync } from "node:sqlite"; -import { crc32 } from "node:zlib"; - -// On-disk format 1. Must stay byte-compatible with existing ledgers. - -export type Hlc = { l: number; c: number }; - -export type LogEvent = { - /** 1, 2, 3… gap-free within one writer directory. Detects damage; does not order. */ - seq: number; - /** The writer directory (one writer at a time). Named `device` for format compatibility. */ - device: string; - /** Orders the merged log (ties broken by `device`). */ - hlc: Hlc; - id: string; - type: string; - v: number; - /** Wall clock, for display only. */ - at: string; - payload: T; -}; +import { decode, encode, readRecords } from "./records.ts"; +import type { Hlc, LogEvent } from "./records.ts"; + +// On-disk format 1. Must stay byte-compatible with existing ledgers. The +// record format and what a segment's bytes mean are `records.ts`. + +export { encode }; +export type { Hlc, LogEvent }; /** What consumers see. Fold order already encodes `hlc`, so a reducer that overwrites gets last-writer-wins. */ export type Event = Pick, "type" | "v" | "at" | "payload">; @@ -112,27 +100,6 @@ const toEvent = ({ type, v, at, payload }: LogEvent): Event => Object.freeze({ t const segmentName = (n: number) => `${String(n).padStart(16, "0")}.log`; const pad2 = (n: number) => String(n).padStart(2, "0"); const localDay = (d: Date) => `${d.getFullYear()}-${pad2(d.getMonth() + 1)}-${pad2(d.getDate())}`; -const hex8 = (s: string) => crc32(s).toString(16).padStart(8, "0"); - -/** ` \n` */ -export const encode = (e: LogEvent) => { - const json = JSON.stringify(e); - return `${hex8(json)} ${json}\n`; -}; - -function decode(line: string): LogEvent | undefined { - if (line[8] !== " ") return; - const json = line.slice(9); - if (hex8(json) !== line.slice(0, 8)) return; - let e: any; - try { - e = JSON.parse(json); - } catch { - return; - } - return typeof e?.seq === "number" && typeof e.device === "string" && - typeof e.hlc?.l === "number" && typeof e.hlc.c === "number" ? e : undefined; -} function parseMarker(text: string): { read: number; write: number } | undefined { const lines = text.split("\n"); @@ -636,7 +603,7 @@ export class Ledger { async #readWriter(device: string, c: Cursor, at: Instant, out: LogEvent[]) { const seen: Seen = { list: "", tail: "", mtime: undefined }; try { - await this.#readSegments(join(this.#root, device), c, seen, out); + await this.#readSegments(join(this.#root, device), c, seen, at, out); } finally { c.checkedAt = at.mono; const key = `${seen.list}|${seen.tail}`; @@ -646,7 +613,13 @@ export class Ledger { } } - async #readSegments(dir: string, c: Cursor, seen: Seen, out: LogEvent[]) { + /** + * Lists `dir` and reads its segments from the cursor, one after the next, + * until one stops short or the next is not there. What the bytes of each + * mean is `readRecords`; this reads them, and moves the cursor by what it + * says. + */ + async #readSegments(dir: string, c: Cursor, seen: Seen, at: Instant, out: LogEvent[]) { c.ignored = []; let entries: Dirent[]; try { @@ -691,29 +664,16 @@ export class Ledger { Object.assign(c, { state: "unavailable", detail: String(err?.message ?? err) }); return; } - let i = 0; - let firstBad = -1; - for (let nl; (nl = buf.indexOf(10, i)) >= 0; ) { - const at = start + i; - const e = decode(buf.toString("utf8", i, nl)); - i = nl + 1; - if (!e) { - if (firstBad < 0) firstBad = at; - continue; - } - if (firstBad >= 0) return stall(seg, firstBad, "intact records follow it"); - if (e.seq > c.last + 1) return stall(seg, at, `jumps from seq ${c.last} to ${e.seq}`); - if (e.seq === c.last + 1) { - c.last = e.seq; - out.push(e); - if (e.hlc.l > Date.now() + MAX_CLOCK_SKEW_MS) c.suspect = Math.max(c.suspect ?? 0, e.hlc.l); - } // else seq <= last: duplicate delivery, skip but count its bytes + const read = readRecords(buf, start, c.last, at.wall + MAX_CLOCK_SKEW_MS); + // A loop, not a spread: a full segment holds more records than a call takes arguments. + for (const e of read.accepted) out.push(e); + c.last = read.last; + if (read.suspect !== undefined) c.suspect = Math.max(c.suspect ?? 0, read.suspect); + if (read.end !== undefined) { c.seg = seg; - c.offset = start + i; + c.offset = read.end; } - // A bad or unterminated last record: may still be syncing, or the session crashed mid-write. - if (firstBad >= 0 || i < buf.length) - return stall(seg, firstBad >= 0 ? firstBad : start + i, "torn tail (may be from a crash)"); + if (read.stall) return stall(seg, read.stall.at, read.stall.why); } } diff --git a/packages/ledger/multi-writer.md b/packages/ledger/multi-writer.md index 89ecc5e..1340178 100644 --- a/packages/ledger/multi-writer.md +++ b/packages/ledger/multi-writer.md @@ -230,6 +230,8 @@ So **every directory is foreign**, even the one this ledger's writer is appendin | The directory or a segment cannot be listed or read | `unavailable` | | A `seq` at or below the cursor | Skipped, not a stop | +What a segment's bytes say (every row above but the missing segment and `unavailable`, which are about files) is `readRecords` in `records.ts`: pure, with a test for each row, and one that cuts a valid run at every byte. Change a rule there and here together. + Folding past a gap is the one thing worse than stopping at it: an event whose causal predecessor is missing builds a state that is quietly wrong, and quietly wrong is the failure this codebase is built to avoid. > **A foreign log can never prevent the log from opening.** One stalled directory costs that directory's tail. Every other directory still folds. diff --git a/packages/ledger/records.test.ts b/packages/ledger/records.test.ts new file mode 100644 index 0000000..c9fabf9 --- /dev/null +++ b/packages/ledger/records.test.ts @@ -0,0 +1,178 @@ +/** Run with: npm test */ +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { encode, readRecords } from "./records.ts"; +import type { LogEvent } from "./records.ts"; + +// Composition root for these tests: records built by hand, and a fixed time +// past which a record is dated too far ahead. + +const AHEAD = 10_000; + +const rec = (seq: number, l = 1_000, payload: unknown = `r${seq}`): LogEvent => ({ + seq, + device: "w", + hlc: { l, c: 0 }, + id: `w-${seq}`, + type: "t", + v: 1, + at: new Date(l).toISOString(), + payload, +}); +const bytes = (...lines: string[]) => Buffer.from(lines.join("")); +const lines = (...es: LogEvent[]) => es.map(encode); +const len = (s: string) => Buffer.byteLength(s); +const seqs = (es: LogEvent[]) => es.map((e) => e.seq); + +test("a clean run is taken whole, and the next read starts past it", () => { + const all = lines(rec(1), rec(2), rec(3)); + const read = readRecords(bytes(...all), 0, 0, AHEAD); + assert.deepEqual(seqs(read.accepted), [1, 2, 3]); + assert.equal(read.last, 3); + assert.equal(read.end, len(all.join(""))); + assert.equal(read.stall, undefined); +}); + +test("nothing to read is nothing taken, no stall, and the cursor stays", () => { + assert.deepEqual(readRecords(Buffer.alloc(0), 120, 7, AHEAD), { + accepted: [], + last: 7, + end: undefined, + suspect: undefined, + }); +}); + +test("a duplicate delivery is skipped, but read past", () => { + const all = lines(rec(1), rec(2), rec(2), rec(3)); + const read = readRecords(bytes(...all), 0, 1, AHEAD); + assert.deepEqual(seqs(read.accepted), [2, 3]); + assert.equal(read.end, len(all.join(""))); + // Only duplicates: nothing taken, and still past them. + const dups = readRecords(bytes(...lines(rec(1), rec(2))), 0, 2, AHEAD); + assert.deepEqual(dups.accepted, []); + assert.equal(dups.last, 2); + assert.equal(dups.end, len(lines(rec(1), rec(2)).join(""))); +}); + +test("a seq that jumps stops reading at it", () => { + const [one, three] = lines(rec(1), rec(3)); + const read = readRecords(bytes(one, three), 0, 0, AHEAD); + assert.deepEqual(seqs(read.accepted), [1]); + assert.equal(read.end, len(one)); + assert.deepEqual(read.stall, { at: len(one), why: "jumps from seq 1 to 3" }); + // From the first record on: nothing taken, and the cursor stays. + assert.deepEqual(readRecords(bytes(three), 0, 0, AHEAD), { + accepted: [], + last: 0, + end: undefined, + suspect: undefined, + stall: { at: 0, why: "jumps from seq 0 to 3" }, + }); +}); + +test("a bad record with intact records after it stops reading at the bad one", () => { + const [one, two] = lines(rec(1), rec(2)); + const read = readRecords(bytes(one, "not a record\n", two), 0, 0, AHEAD); + assert.deepEqual(seqs(read.accepted), [1]); + assert.equal(read.end, len(one)); + assert.deepEqual(read.stall, { at: len(one), why: "intact records follow it" }); +}); + +test("a record whose checksum does not match is a bad record", () => { + const [one, two, three] = lines(rec(1), rec(2), rec(3)); + const flipped = two.replace('"r2"', '"r9"'); + const read = readRecords(bytes(one, flipped, three), 0, 0, AHEAD); + assert.deepEqual(seqs(read.accepted), [1]); + assert.deepEqual(read.stall, { at: len(one), why: "intact records follow it" }); +}); + +test("a bad or unterminated last record is a torn tail, at its start", () => { + const [one, two] = lines(rec(1), rec(2)); + const torn = { at: len(one), why: "torn tail (may be from a crash)" }; + // Bad, with its newline. + assert.deepEqual(readRecords(bytes(one, "garbage\n"), 0, 0, AHEAD).stall, torn); + // Good so far, but the newline has not arrived. + const cut = readRecords(bytes(one, two.slice(0, -1)), 0, 0, AHEAD); + assert.deepEqual(seqs(cut.accepted), [1]); + assert.equal(cut.end, len(one)); + assert.deepEqual(cut.stall, torn); + // Several bad lines at the end: the first of them. + assert.deepEqual(readRecords(bytes(one, "x\n", "y\n"), 0, 0, AHEAD).stall, torn); +}); + +test("offsets are in the segment, from where the read started", () => { + const [four, five] = lines(rec(4), rec(5)); + const start = 1234; + const read = readRecords(bytes(four, "bad\n", five), start, 3, AHEAD); + assert.equal(read.end, start + len(four)); + assert.deepEqual(read.stall, { at: start + len(four), why: "intact records follow it" }); +}); + +test("a record dated too far ahead is taken, and the latest such is reported", () => { + const read = readRecords( + bytes(...lines(rec(1, AHEAD + 5), rec(2, 500), rec(3, AHEAD + 9), rec(4, AHEAD))), + 0, + 0, + AHEAD, + ); + assert.deepEqual(seqs(read.accepted), [1, 2, 3, 4]); + assert.equal(read.suspect, AHEAD + 9); + // A duplicate is not taken, so it is not reported either. + assert.equal(readRecords(bytes(...lines(rec(1, AHEAD + 5))), 0, 1, AHEAD).suspect, undefined); +}); + +/** + * Records with text that is more than one byte a character, so a cut lands in + * the middle of one as well as between them. + */ +const run = lines( + rec(1, 1_000, "plain"), + rec(2, 1_001, "café ☕ — naïve"), + rec(3, 1_002, { nested: ["日本語", "🙂🙃"] }), + rec(4, 1_003, ""), + rec(5, 1_004, "end"), +); +const whole = bytes(...run); +/** The byte offset at which each record's line ends, newline included. */ +const ends = run.map((_, k) => len(run.slice(0, k + 1).join(""))); + +test("cut at any byte, a run is taken up to the last whole record and stops at the cut", () => { + for (let cut = 0; cut <= whole.length; cut++) { + const read = readRecords(whole.subarray(0, cut), 0, 0, AHEAD); + const whole_ = ends.filter((e) => e <= cut); + const upTo = whole_.at(-1); + assert.deepEqual(seqs(read.accepted), whole_.map((_, k) => k + 1), `cut at ${cut}`); + assert.equal(read.end, upTo, `cut at ${cut}`); + // Never a record past the cut, and a stall exactly when a record is cut. + assert.ok((read.end ?? 0) <= cut, `cut at ${cut}`); + assert.deepEqual( + read.stall, + cut === (upTo ?? 0) ? undefined : { at: upTo ?? 0, why: "torn tail (may be from a crash)" }, + `cut at ${cut}`, + ); + } +}); + +test("read up to a cut and then on from where it stopped, a run is taken once, whole", () => { + for (let cut = 0; cut <= whole.length; cut++) { + const first = readRecords(whole.subarray(0, cut), 0, 0, AHEAD); + const from = first.end ?? 0; + const rest = readRecords(whole.subarray(from), from, first.last, AHEAD); + assert.deepEqual(seqs([...first.accepted, ...rest.accepted]), [1, 2, 3, 4, 5], `cut at ${cut}`); + // Where the cursor ends up: past the run, whichever read got it there. + assert.equal(rest.end ?? first.end, whole.length, `cut at ${cut}`); + assert.equal(rest.stall, undefined, `cut at ${cut}`); + } +}); + +test("a bad line anywhere in a run takes what is before it and nothing after", () => { + for (let k = 0; k <= run.length; k++) { + const read = readRecords(bytes(...run.slice(0, k), "bad\n", ...run.slice(k)), 0, 0, AHEAD); + const before = k === 0 ? 0 : ends[k - 1]; + assert.deepEqual(seqs(read.accepted), run.slice(0, k).map((_, j) => j + 1), `bad at ${k}`); + assert.deepEqual(read.stall, { + at: before, + why: k === run.length ? "torn tail (may be from a crash)" : "intact records follow it", + }); + } +}); diff --git a/packages/ledger/records.ts b/packages/ledger/records.ts new file mode 100644 index 0000000..76e4723 --- /dev/null +++ b/packages/ledger/records.ts @@ -0,0 +1,135 @@ +/** + * The record format, and the rules for what a run of a segment's bytes + * means: which records are taken, where reading stops, and why. + * + * Pure: no file, no clock. `ledger.ts` reads the bytes and keeps the cursor; + * this decides what they say. Not exported from the package, which is + * `ledger.ts`; that re-exports what is public (`encode`, `LogEvent`, `Hlc`). + * + * The rules are the table under "Reading foreign logs" in `multi-writer.md`, + * and must not change without it. + */ +import { crc32 } from "node:zlib"; + +// On-disk format 1. Must stay byte-compatible with existing ledgers. + +export type Hlc = { l: number; c: number }; + +export type LogEvent = { + /** 1, 2, 3… gap-free within one writer directory. Detects damage; does not order. */ + seq: number; + /** The writer directory (one writer at a time). Named `device` for format compatibility. */ + device: string; + /** Orders the merged log (ties broken by `device`). */ + hlc: Hlc; + id: string; + type: string; + v: number; + /** Wall clock, for display only. */ + at: string; + payload: T; +}; + +const hex8 = (s: string) => crc32(s).toString(16).padStart(8, "0"); + +/** ` \n` */ +export const encode = (e: LogEvent) => { + const json = JSON.stringify(e); + return `${hex8(json)} ${json}\n`; +}; + +/** One line, without its newline, as a record; undefined for one that is not. */ +export function decode(line: string): LogEvent | undefined { + if (line[8] !== " ") return; + const json = line.slice(9); + if (hex8(json) !== line.slice(0, 8)) return; + let e: any; + try { + e = JSON.parse(json); + } catch { + return; + } + return typeof e?.seq === "number" && typeof e.device === "string" && + typeof e.hlc?.l === "number" && typeof e.hlc.c === "number" ? e : undefined; +} + +/** What one read of a segment's bytes found. */ +export type RecordsRead = { + /** The records taken, in order: each the `seq` after the one before. */ + accepted: LogEvent[]; + /** The `seq` of the last record taken, or the `last` read from if none was. */ + last: number; + /** + * The byte offset in the segment just past the last good record, taken or + * skipped as a duplicate, where the next read resumes. Undefined when there + * was none: the cursor stays where it was. + */ + end?: number; + /** The latest `hlc.l` among the records taken that are dated past `suspectAfter`. */ + suspect?: number; + /** + * Where in the segment reading stopped short of the end of the bytes, and + * why. Undefined when it read them all, and the next segment is next. + */ + stall?: { at: number; why: string }; +}; + +/** + * Reads the records in `buf`, which is a segment's bytes from offset `start`, + * after the record whose `seq` is `last`. + * + * A left fold over the newline-ended lines, stopping at the first that says + * stop. A record with the next `seq` is taken. One at or below `last` is a + * duplicate delivery: skipped, but past it. One past the next `seq` stops + * reading at it: folding past a gap builds a state that is quietly wrong. A + * line that is not a record is remembered: a record after it stops reading at + * it, and so does the end of the bytes, as does a last line with no newline. + * Nothing after a stop is taken, so a stall costs the rest of this directory + * and nothing else. + * + * `suspectAfter` is the time past which a record is dated too far ahead to + * trust (the scan's wall clock plus the allowed skew). Such a record is still + * taken; it is only reported. + * + * Offsets in the result are in the segment, not in `buf`. Reading the bytes + * up to `end`, then the rest from `end` with the `last` the first read gave, + * takes the same records as reading them all at once. + */ +export function readRecords( + buf: Buffer, + start: number, + last: number, + suspectAfter: number, +): RecordsRead { + const accepted: LogEvent[] = []; + let end: number | undefined; + let suspect: number | undefined; + const done = (stall?: { at: number; why: string }): RecordsRead => + stall + ? { accepted, last, end, suspect, stall } + : { accepted, last, end, suspect }; + + let i = 0; + let firstBad = -1; + for (let nl; (nl = buf.indexOf(10, i)) >= 0; ) { + const at = start + i; + const e = decode(buf.toString("utf8", i, nl)); + i = nl + 1; + if (!e) { + if (firstBad < 0) firstBad = at; + continue; + } + if (firstBad >= 0) return done({ at: firstBad, why: "intact records follow it" }); + if (e.seq > last + 1) return done({ at, why: `jumps from seq ${last} to ${e.seq}` }); + if (e.seq === last + 1) { + last = e.seq; + accepted.push(e); + if (e.hlc.l > suspectAfter) suspect = Math.max(suspect ?? 0, e.hlc.l); + } // else seq <= last: duplicate delivery, skip but count its bytes + end = start + i; + } + // A bad or unterminated last record: may still be syncing, or the session crashed mid-write. + if (firstBad >= 0 || i < buf.length) + return done({ at: firstBad >= 0 ? firstBad : start + i, why: "torn tail (may be from a crash)" }); + return done(); +}