From 6a942694c47cc08cccbaf2e144b81d4fc2465262 Mon Sep 17 00:00:00 2001 From: Sal Rahman Date: Tue, 27 Oct 2026 18:39:42 -0700 Subject: [PATCH] ledger: a pure record parser under #readSegments (#117) #readSegments read the files and decided what their bytes meant in one loop: a torn or unterminated tail, a bad record with intact records after it, a seq that jumps, a duplicate delivery, and a record dated too far ahead. Those rules are what stop a half-synced folder from losing notes, and they could only be tested by writing real files into a temp folder. records.ts now holds the record format (encode, decode, LogEvent and Hlc, moved unchanged) and readRecords(buf, start, last, suspectAfter): a pure fold over a segment's lines that returns the records taken, the last seq, the offset just past the last good record, the latest record dated past suspectAfter, and the stall, if any, with its offset and reason. It is not exported from the package; ledger.ts re-exports encode, LogEvent and Hlc, so the public surface is the same. #readSegments keeps the listing, the reads and the move to the next segment, and applies what readRecords says to the cursor. The records taken are pushed in a loop rather than spread, since a full segment can hold more of them than a call takes arguments. The skew check uses the scan's own wall clock instead of calling Date.now() again per record. The format and every rule in multi-writer.md are unchanged, and the spec now says which rows of its table are readRecords. The ledger's existing tests pass unchanged. The new ones cover each row, a valid run cut at every byte (taken up to the last whole record, stalling at the cut, never past it), reading on from a cut taking every record once, and a bad line at every position. Co-Authored-By: Claude Opus 5.5 --- packages/ledger/ledger.ts | 88 +++++----------- packages/ledger/multi-writer.md | 2 + packages/ledger/records.test.ts | 178 ++++++++++++++++++++++++++++++++ packages/ledger/records.ts | 135 ++++++++++++++++++++++++ 4 files changed, 339 insertions(+), 64 deletions(-) create mode 100644 packages/ledger/records.test.ts create mode 100644 packages/ledger/records.ts 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(); +}