ledger: a pure record parser under #readSegments (#117)

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<T = unknown> = {
-  /** 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<T = unknown> = Pick<LogEvent<T>, "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");
-
-/** `<crc32 of the JSON, 8 hex> <JSON>\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);
     }
   }
 

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.

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",
+    });
+  }
+});