From 5de705fcf47083ac95a8962ffc534e389d199b56 Mon Sep 17 00:00:00 2001 From: William Stein Date: Thu, 22 Oct 2026 08:39:19 -0700 Subject: [PATCH 1/3] compactHistory: retention policies that keep chosen versions exactly A pure, deterministic compaction of a history: keepAll, keepSince, uniform, tiered, latestOnly and union. Heads and their merge bases are always kept; every kept version is a diff from its nearest kept ancestors (merges record their merged value), roots become snapshots, dropped patches are squashed into their kept descendants with meta.squashed authorship, and untouched records are kept as they were (idempotent). Never larger than the input; an operation budget (diffs and size diffed) returns the input unchanged when exceeded. Property and fuzz tests over random concurrent text and db histories check: kept versions exact, newest value unchanged, valid and deterministic history, idempotence, authorship, monotone size; plus a scale test and pnpm bench:compact. README documents the record format a host must store. --- PLAN.md | 9 + README.md | 23 ++ examples/bench-compact-history.ts | 116 +++++++ package.json | 3 +- src/compact-history.test.ts | 470 ++++++++++++++++++++++++++ src/compact-history.ts | 540 ++++++++++++++++++++++++++++++ src/index.ts | 1 + 7 files changed, 1161 insertions(+), 1 deletion(-) create mode 100644 examples/bench-compact-history.ts create mode 100644 src/compact-history.test.ts create mode 100644 src/compact-history.ts diff --git a/PLAN.md b/PLAN.md index 880ad98..20fe01e 100644 --- a/PLAN.md +++ b/PLAN.md @@ -1,5 +1,14 @@ # Patchflow Plans +## (done) History compaction + +`compactHistory` (src/compact-history.ts, README "History compaction"): retention policies (keepAll, keepSince, uniform, tiered, latestOnly, union) that keep chosen versions exactly, with property and fuzz tests and `pnpm bench:compact`. + +Follow-ups: + +- [ ] Diffs of large texts: with `@cocalc/diff-match-patch` 1.1.0 and `DIFF_WORK_LIMIT` 25M, a one-line edit to a text of about 800 KB or more gives a patch that replaces the whole text (charges like `2 * (len1 + len2)` before the common prefix and suffix are removed). This affects every commit to such a text, and makes compacted diffs over many edits of a large text whole-text replacements. +- [ ] A compacted diff larger than the value itself could store the value instead (a record with `isSnapshot`, `snapshot` and `parents`), once hosts store such records. + ## (done) Optimize patch application for the immer based document type Currently the implementation of src/db-document-immer.ts is not optimal. Our goal is to make it much more efficient in terms of speed and memory. diff --git a/README.md b/README.md index 7882652..61bd2c5 100644 --- a/README.md +++ b/README.md @@ -61,6 +61,29 @@ A patch that merges concurrent heads (a merge commit, with several parents) is a Merge commits written before this (no `mergeParent` and no `hash`) were made by versions that applied every patch in time order (patchflow 0.8 and earlier); their values are computed the same way, from the patches their parents descend from, so the values of such a history do not change either. +### History compaction + +`compactHistory({ patches, codec, policy, now, budget })` returns a smaller history that keeps chosen versions exactly (see [src/compact-history.ts](./src/compact-history.ts)). It is pure and deterministic: storing the result is up to the host. A policy chooses the versions to keep: + +| Policy | Keeps | +| ------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------- | +| `{ kind: "keepAll" }` | everything (returns the input) | +| `{ kind: "keepSince", since }` | the version current at `since` and every version after it | +| `{ kind: "uniform", interval }` | the newest version of every `interval` (aligned to the epoch) | +| `{ kind: "tiered", tiers }` | backup style: every version younger than the first tier, then the newest of each tier's `interval` for versions at least `olderThan` old | +| `{ kind: "latestOnly" }` | only the newest version | +| `{ kind: "union", policies }` | every version any of them keeps | + +The heads, and the merge bases their merged value is computed from, are always kept, so the newest value is unchanged, also with concurrent heads. Every kept version gets its nearest kept ancestors as parents and a diff from their exact value, so its value is exactly what it was; a merge records its merged value (`mergeParent`, `mergePatch`). A kept version without kept ancestors becomes a snapshot, so the result starts with one. A patch that absorbs dropped patches records whose edits it holds in `meta.squashed` (`users` in order, `start`, `end`, `count`). Patches that neither change parents nor absorb anything are kept as they were, so compacting again with the same policy changes nothing. The result is never larger than the input: if it would be, or the `budget` (diffs, and total size of the versions diffed) runs out, the input is returned unchanged (`stats.identity`, `stats.overBudget`). The codec must provide `merge3`, and the history must be complete (starting at a root or a snapshot). The property and fuzz tests in [src/compact-history.test.ts](./src/compact-history.test.ts) check all of this on random concurrent histories of texts and databases; `pnpm bench:compact` measures it on a long history. + +**Storing a compacted history.** The result replaces the whole history: write it atomically, as a new history, in its order. Every record is a `Patch` as it is given; nothing refers to the records that are dropped. + +- The first record (and any other record with no parents) is a snapshot: `isSnapshot: true`, the full value in `snapshot`, `parents: []`, no `patch`, and `hash`. It is the start of the new history, like the first snapshot of a document; it has no older history to load (no `seqInfo`). +- Every other record is a patch with `parents` (all earlier in the result), `patch`, `hash`, and for a merge `mergeParent` and `mergePatch`. Its `time` is the original patch id, so ids stay unique and stable, and `userId` and `wall` are those of the original patch. +- `meta.squashed` (and any other `meta`) must be stored with the record. +- Snapshot records of the old history are not in the result (a kept patch loses the snapshot text a store attached to it). A host makes snapshots of the new history as for any history, e.g. one of the newest version right away, so that clients do not load it in full. +- A rare repaired record has `isSnapshot` and `snapshot` with `parents` (its value is stored, not diffed): store it like any patch with a snapshot. + ## How Patchflow compares | Approach | Examples | What is exchanged | History | Merging | diff --git a/examples/bench-compact-history.ts b/examples/bench-compact-history.ts new file mode 100644 index 0000000..db33014 --- /dev/null +++ b/examples/bench-compact-history.ts @@ -0,0 +1,116 @@ +/* +Benchmark compactHistory on a long linear history of a large text. + + pnpm bench:compact [patches] [kilobytes] + +Builds a history of `patches` one-line edits (one a minute) to a text of +about `kilobytes` KB that starts as a snapshot, with a snapshot every +`every` patches as CoCalc stores them (0: none), then compacts it with +several policies, reporting time, memory, versions and bytes. + + pnpm bench:compact [patches] [kilobytes] [every] +*/ + +import { compactHistory, type CompactionPolicy } from "../src/compact-history"; +import { mergeStrings3 } from "../src/merge3"; +import { encodePatchId } from "../src/patch-id"; +import { StringDocument } from "../src/string-document"; +import type { DocCodec, Patch } from "../src/types"; + +const MINUTE = 60_000; +const HOUR = 60 * MINUTE; +const DAY = 24 * HOUR; + +const codec: DocCodec = { + fromString: (s) => new StringDocument(s), + toString: (d) => d.toString(), + applyPatch: (d, p) => d.applyPatch(p), + applyPatchBatch: (d, ps) => d.applyPatchBatch(ps), + makePatch: (a, b) => a.makePatch(b), + merge3: (base, a, b, ancestors) => + new StringDocument( + mergeStrings3({ + base: base.toString(), + a: a.toString(), + b: b.toString(), + ancestors: ancestors?.map((d) => d.toString()), + }), + ), +}; + +const n = Number(process.argv[2] ?? 50_000); +// (Up to about 700 KB: from about 800 KB, @cocalc/diff-match-patch 1.1.0's +// work budget runs out on any edit and patches replace the whole text.) +const kb = Number(process.argv[3] ?? 500); +const every = Number(process.argv[4] ?? 300); +const T0 = 19_675 * DAY; + +let t = Date.now(); +const lines = Array.from({ length: Math.ceil((kb * 1024) / 64) }, (_, i) => + `line ${i} ${"x".repeat(56)}`.slice(0, 63), +); +let prev = codec.fromString(lines.join("\n")); +const patches: Patch[] = [ + { + time: encodePatchId(T0, "u0"), + parents: [], + userId: 0, + isSnapshot: true, + snapshot: prev.toString(), + }, +]; +for (let i = 1; i < n; i++) { + const k = (i * 7919) % lines.length; + lines[k] = `${lines[k].slice(0, 40)} edit ${i}`; + const next = codec.fromString(lines.join("\n")); + // As a store would deliver them (JSON): a generated diff of a whole text + // is made of slices of it, which would keep every version alive. + patches.push( + JSON.parse( + JSON.stringify({ + time: encodePatchId(T0 + i * MINUTE, `u${i % 7}`), + parents: [patches[i - 1].time], + userId: i % 7, + patch: codec.makePatch(prev, next), + ...(every > 0 && i % every === 0 ? { isSnapshot: true, snapshot: next.toString() } : {}), + }), + ), + ); + prev = next; +} +const input = patches; +console.log(`built ${n} patches on ${kb} KB (snapshot every ${every}) in ${Date.now() - t} ms`); + +const now = T0 + n * MINUTE; +const policies: [string, CompactionPolicy][] = [ + ["latestOnly", { kind: "latestOnly" }], + ["uniform 1 day", { kind: "uniform", interval: DAY }], + [ + "tiered", + { + kind: "tiered", + tiers: [ + { olderThan: 2 * DAY, interval: HOUR }, + { olderThan: 7 * DAY, interval: DAY }, + { olderThan: 30 * DAY, interval: 7 * DAY }, + ], + }, + ], + ["keepSince 1 day", { kind: "keepSince", since: now - DAY }], +]; +for (const [name, policy] of policies) { + global.gc?.(); + const heap0 = process.memoryUsage().heapUsed; + t = Date.now(); + const { stats, patches: out } = compactHistory({ patches: input, codec, policy, now }); + const sizes = out.map((p) => JSON.stringify(p).length).sort((a, b) => b - a); + global.gc?.(); + const ms = Date.now() - t; + const heap = Math.round((process.memoryUsage().heapUsed - heap0) / 1e6); // retained + console.log( + `${name.padEnd(16)} ${String(ms).padStart(7)} ms heap +${heap} MB ` + + `versions ${stats.versionsBefore} -> ${stats.versionsAfter} ` + + `bytes ${stats.bytesBefore} -> ${stats.bytesAfter} diffs ${stats.diffs} ` + + `largest ${sizes.slice(0, 3).join(",")} over 100k ${sizes.filter((x) => x > 100_000).length}`, + ); +} diff --git a/package.json b/package.json index 991bd86..f261676 100644 --- a/package.json +++ b/package.json @@ -54,7 +54,8 @@ "example:db-immer:client": "pnpm build && node dist/esm/examples/db-immer-session.js --role=client --host=127.0.0.1", "example:db-immer:server:ts": "node --loader ts-node/esm --experimental-specifier-resolution=node examples/db-immer-session.ts --role=server", "example:db-immer:client:ts": "node --loader ts-node/esm --experimental-specifier-resolution=node examples/db-immer-session.ts --role=client --host=127.0.0.1", - "bench:immer": "node --loader ts-node/esm --experimental-specifier-resolution=node examples/bench-immer-apply.ts" + "bench:immer": "node --loader ts-node/esm --experimental-specifier-resolution=node examples/bench-immer-apply.ts", + "bench:compact": "node --expose-gc --loader ts-node/esm --experimental-specifier-resolution=node examples/bench-compact-history.ts" }, "dependencies": { "@cocalc/diff-match-patch": "^1.1.0", diff --git a/src/compact-history.test.ts b/src/compact-history.test.ts new file mode 100644 index 0000000..14f5f6c --- /dev/null +++ b/src/compact-history.test.ts @@ -0,0 +1,470 @@ +import { describe, expect, it } from "vitest"; +import { compactHistory, encodedSize, selectVersions, squashedInfo } from "./compact-history"; +import type { CompactionPolicy } from "./compact-history"; +import { createDbCodec } from "./db-document-immutable"; +import { mergeStrings3 } from "./merge3"; +import { PatchGraph } from "./patch-graph"; +import { encodePatchId, type PatchId } from "./patch-id"; +import { StringDocument } from "./string-document"; +import type { DocCodec, Document, Patch } from "./types"; + +const HOUR = 3_600_000; +const DAY = 24 * HOUR; +// a day boundary (intervals are aligned to the epoch) +const T0 = 19_675 * DAY; + +const stringCodec: DocCodec = { + fromString: (s) => new StringDocument(s), + toString: (d) => d.toString(), + applyPatch: (d, p) => d.applyPatch(p), + applyPatchBatch: (d, ps) => d.applyPatchBatch(ps), + makePatch: (a, b) => a.makePatch(b), + merge3: (base, a, b, ancestors) => + new StringDocument( + mergeStrings3({ + base: base.toString(), + a: a.toString(), + b: b.toString(), + ancestors: ancestors?.map((d) => d.toString()), + }), + ), +}; + +// A db codec with a simple record-level three-way merge (the later side wins +// a record both sides changed). +const plainDb = createDbCodec({ primaryKeys: ["id"] }); +const records = (d: Document): Map => { + const m = new Map(); + for (const line of plainDb.toString(d).split("\n")) { + if (!line) continue; + m.set(String(JSON.parse(line).id), line); + } + return m; +}; +const fromRecords = (m: Map): Document => + plainDb.fromString( + [...m.entries()] + .sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0)) + .map(([, line]) => line) + .join("\n"), + ); +const dbCodec: DocCodec = { + ...plainDb, + merge3: (base, a, b) => { + const [mb, ma, mbb] = [records(base), records(a), records(b)]; + const out = new Map(); + for (const id of new Set([...mb.keys(), ...ma.keys(), ...mbb.keys()])) { + const [x, y, z] = [mb.get(id), ma.get(id), mbb.get(id)]; + const v = y === x ? z : z === x ? y : z; + if (v != null) out.set(id, v); + } + return fromRecords(out); + }, +}; + +function rng(seed: number) { + let a = seed >>> 0; + return (n: number) => { + a = (a + 0x6d2b79f5) >>> 0; + let t = a; + t = Math.imul(t ^ (t >>> 15), t | 1); + t ^= t + Math.imul(t ^ (t >>> 7), t | 61); + return Math.floor((((t ^ (t >>> 14)) >>> 0) / 4294967296) * n); + }; +} + +const WORDS = ["alpha", "beta", "gamma", "delta", "eps", "zeta", "eta", "theta"]; + +function editString(r: (n: number) => number, s: string): string { + const lines = s.length ? s.split("\n") : []; + const i = r(lines.length + 1); + switch (r(3)) { + case 0: + lines.splice(i, 0, `${WORDS[r(WORDS.length)]} ${r(1000)}`); + break; + case 1: + if (lines.length > 0) lines.splice(Math.min(i, lines.length - 1), 1); + else lines.push("start"); + break; + default: + if (lines.length > 0) lines[Math.min(i, lines.length - 1)] += ` ${WORDS[r(WORDS.length)]}`; + else lines.push("start"); + } + return lines.join("\n"); +} + +function editDb(r: (n: number) => number, d: Document): Document { + const m = records(d); + const id = `k${r(8)}`; + if (r(4) === 0) m.delete(id); + else m.set(id, JSON.stringify({ id, v: `${WORDS[r(WORDS.length)]}${r(100)}` })); + return fromRecords(m); +} + +// A random history of `n` patches by up to 3 clients editing concurrently: +// each commits on the heads it knows, sometimes catching up on all heads (a +// merge commit when there are several), over about `days` days. +function randomHistory(seed: number, codec: DocCodec, n: number, days = 30): Patch[] { + const r = rng(seed); + const g = new PatchGraph({ codec }); + const out: Patch[] = []; + const clients = 1 + r(3); + const known: PatchId[][] = Array.from({ length: clients }, () => []); + let ms = T0; + for (let i = 0; i < n; i++) { + ms += 1 + r(Math.floor((days * DAY) / n) * 2); + const c = r(clients); + if (r(3) === 0 && out.length > 0) known[c] = g.getHeads(); + const parents = known[c]; + const from = parents.length > 0 ? g.exactValueOf(parents)! : codec.fromString(""); + const to = + codec === dbCodec ? editDb(r, from) : codec.fromString(editString(r, codec.toString(from))); + const time = encodePatchId(ms, `client${c}`); + const p: Patch = { + time, + parents, + userId: c, + patch: codec.makePatch(from, to), + hash: g.hashOf(to), + }; + if (parents.length > 1 && r(2) === 0) { + p.mergeParent = parents[0]; + p.mergePatch = codec.makePatch(g.exactValueOf([parents[0]])!, from); + } + g.add([p]); + out.push(p); + known[c] = [time]; + } + return out; +} + +function same(codec: DocCodec, a: Document, b: Document): boolean { + return a.isEqual(b) || codec.toString(a) === codec.toString(b); +} + +function users(patches: Patch[]): Set { + const out = new Set(); + for (const p of patches) { + if (p.userId != null) out.add(p.userId); + for (const u of squashedInfo(p)?.users ?? []) out.add(u); + } + return out; +} + +// Every invariant of compactHistory for one history and policy. +function checkInvariants(codec: DocCodec, patches: Patch[], policy: CompactionPolicy, now: number) { + const result = compactHistory({ patches, codec, policy, now }); + const old = new PatchGraph({ codec }); + old.add(patches); + const fresh = new PatchGraph({ codec }); + fresh.add(result.patches); + + // 1. kept versions are exact + for (const t of result.versions) { + expect(same(codec, fresh.version(t), old.version(t))).toBe(true); + } + // 2. the newest value is unchanged, and every head is kept + expect(same(codec, fresh.value(), old.value())).toBe(true); + for (const h of old.getHeads()) expect(result.versions).toContain(h); + // the policy's versions are kept + for (const t of selectVersions(policy, old.versions(), now)) { + expect(result.versions).toContain(t); + } + // 3. a valid history: unique ids in time order, parents before children, + // a snapshot first, exact values without more history + const times = result.patches.map((p) => p.time); + expect(new Set(times).size).toBe(times.length); + expect([...times].sort()).toEqual(times); + expect(times).toEqual(result.versions); + const seen = new Set(); + for (const p of result.patches) { + for (const parent of p.parents ?? []) { + if (!result.stats.identity) expect(seen.has(parent)).toBe(true); + } + seen.add(p.time); + } + if (!result.stats.identity) { + expect(result.patches[0].isSnapshot).toBe(true); + expect(fresh.needsMoreHistory()).toBe(false); + } + // deterministic + const again = compactHistory({ patches, codec, policy, now }); + expect(JSON.stringify(again)).toBe(JSON.stringify(result)); + // 4. idempotent + const twice = compactHistory({ patches: result.patches, codec, policy, now }); + expect(JSON.stringify(twice.patches)).toBe(JSON.stringify(result.patches)); + // 5. authorship survives + for (const u of users(patches)) expect(users(result.patches).has(u)).toBe(true); + // 7. never larger + const before = old.versions().reduce((n, t) => n + encodedSize(old.getPatch(t)), 0); + const after = result.patches.reduce((n, p) => n + encodedSize(p), 0); + expect(result.stats.bytesBefore).toBe(before); + expect(result.stats.bytesAfter).toBe(after); + expect(after).toBeLessThanOrEqual(before); + return result; +} + +function randomPolicy(r: (n: number) => number, now: number): CompactionPolicy { + const pick = (): CompactionPolicy => { + switch (r(6)) { + case 0: + return { kind: "keepAll" }; + case 1: + return { kind: "keepSince", since: now - r(40) * DAY }; + case 2: + return { kind: "uniform", interval: [HOUR, 6 * HOUR, DAY, 7 * DAY][r(4)] }; + case 3: + return { + kind: "tiered", + tiers: [ + { olderThan: 2 * DAY, interval: HOUR }, + { olderThan: 7 * DAY, interval: DAY }, + { olderThan: 21 * DAY, interval: 7 * DAY }, + ], + }; + case 4: + return { kind: "latestOnly" }; + default: + return { kind: "union", policies: [pick(), pick()] }; + } + }; + return pick(); +} + +describe("selectVersions", () => { + const t = (ms: number, c = "a") => encodePatchId(ms, c); + const times = [t(T0), t(T0 + HOUR), t(T0 + 2 * HOUR), t(T0 + DAY), t(T0 + DAY + 1)]; + + it("keepSince keeps the version current at the time and those after", () => { + expect([...selectVersions({ kind: "keepSince", since: T0 + 90 * 60_000 }, times)]).toEqual([ + times[2], + times[3], + times[4], + times[1], + ]); + }); + + it("uniform keeps the newest of each interval", () => { + const kept = selectVersions({ kind: "uniform", interval: DAY }, times); + expect([...kept].sort()).toEqual([times[2], times[4]].sort()); + }); + + it("tiered keeps everything young and thins out older versions", () => { + const now = T0 + DAY + 2; + const kept = selectVersions( + { kind: "tiered", tiers: [{ olderThan: 12 * HOUR, interval: DAY }] }, + times, + now, + ); + expect([...kept].sort()).toEqual([times[2], times[3], times[4]].sort()); + }); + + it("latestOnly selects nothing (the heads are added), keepAll everything", () => { + expect(selectVersions({ kind: "latestOnly" }, times).size).toBe(0); + expect(selectVersions({ kind: "keepAll" }, times).size).toBe(times.length); + }); + + it("rejects a non-positive interval", () => { + expect(() => selectVersions({ kind: "uniform", interval: 0 }, times)).toThrow(); + }); +}); + +describe("compactHistory on a linear text history", () => { + // a single-branch history by three users, one line an hour + const g = new PatchGraph({ codec: stringCodec }); + const linear: Patch[] = []; + let value = ""; + for (let i = 0; i < 20; i++) { + const next = `${value}line ${i}\n`; + const time = encodePatchId(T0 + i * HOUR, `u${i % 3}`); + const p: Patch = { + time, + parents: linear.length ? [linear[linear.length - 1].time] : [], + userId: i % 3, + patch: stringCodec.makePatch(stringCodec.fromString(value), stringCodec.fromString(next)), + }; + g.add([p]); + linear.push(p); + value = next; + } + + it("latestOnly keeps one snapshot with everyone's authorship", () => { + const result = checkInvariants(stringCodec, linear, { kind: "latestOnly" }, T0 + DAY); + expect(result.patches).toHaveLength(1); + expect(result.patches[0].isSnapshot).toBe(true); + expect(result.patches[0].snapshot).toBe(value); + const info = squashedInfo(result.patches[0])!; + expect(info.users).toEqual([0, 1, 2]); + expect(info.start).toBe(linear[0].time); + expect(info.end).toBe(linear[19].time); + expect(info.count).toBe(20); + expect(result.stats.bytesAfter).toBeLessThan(result.stats.bytesBefore); + }); + + it("keepAll is the identity", () => { + const result = checkInvariants(stringCodec, linear, { kind: "keepAll" }, T0 + DAY); + expect(result.stats.identity).toBe(true); + expect(result.patches).toEqual(linear); + }); + + it("keepSince keeps recent patches as they were", () => { + const since = T0 + 15 * HOUR; + const result = checkInvariants(stringCodec, linear, { kind: "keepSince", since }, T0 + DAY); + expect(result.versions).toEqual(linear.slice(14).map((p) => p.time)); + // the first kept version becomes the snapshot; the rest are unchanged + expect(result.patches[0].isSnapshot).toBe(true); + expect(result.patches.slice(1)).toEqual(linear.slice(15)); + }); + + it("uniform squashes each interval into its newest patch", () => { + const result = checkInvariants( + stringCodec, + linear, + { kind: "uniform", interval: 4 * HOUR }, + T0 + DAY, + ); + expect(result.versions).toHaveLength(5); + const counts = result.patches.map((p) => squashedInfo(p)?.count ?? 1); + expect(counts.reduce((a, b) => a + b, 0)).toBe(20); + }); + + it("compacts a history that starts at a snapshot", () => { + const g = new PatchGraph({ codec: stringCodec }); + g.add(linear); + const start: Patch = { + ...linear[4], + isSnapshot: true, + snapshot: g.version(linear[4].time).toString(), + }; + const result = checkInvariants( + stringCodec, + [start, ...linear.slice(5)], + { kind: "latestOnly" }, + T0, + ); + expect(result.patches).toHaveLength(1); + expect(result.patches[0].snapshot).toBe(value); + }); + + it("refuses an incomplete history", () => { + expect(() => + compactHistory({ + patches: linear.slice(1), + codec: stringCodec, + policy: { kind: "latestOnly" }, + }), + ).toThrow(/incomplete/); + const gap = [linear[0], ...linear.slice(2)]; + expect(() => + compactHistory({ patches: gap, codec: stringCodec, policy: { kind: "latestOnly" } }), + ).toThrow(/incomplete/); + }); +}); + +describe("compactHistory budget", () => { + const patches = randomHistory(7, stringCodec, 40); + const policy: CompactionPolicy = { kind: "uniform", interval: DAY }; + const now = T0 + 40 * DAY; + const free = compactHistory({ patches, codec: stringCodec, policy, now }); + + it("counts the work", () => { + expect(free.stats.diffs).toBeGreaterThan(0); + expect(free.stats.diffSize).toBeGreaterThan(0); + expect(free.stats.overBudget).toBe(false); + }); + + it("returns the input unchanged when the budget runs out", () => { + for (const budget of [ + { maxDiffs: free.stats.diffs - 1 }, + { maxDiffSize: free.stats.diffSize - 1 }, + ]) { + const result = compactHistory({ patches, codec: stringCodec, policy, now, budget }); + expect(result.stats.overBudget).toBe(true); + expect(result.stats.identity).toBe(true); + expect(result.stats.diffs).toBeLessThanOrEqual(free.stats.diffs); + const old = new PatchGraph({ codec: stringCodec }); + old.add(patches); + expect(result.patches).toEqual(old.versions().map((t) => old.getPatch(t))); + } + }); + + it("gives the same result within budget", () => { + const budget = { maxDiffs: free.stats.diffs, maxDiffSize: free.stats.diffSize }; + const result = compactHistory({ patches, codec: stringCodec, policy, now, budget }); + expect(JSON.stringify(result)).toBe(JSON.stringify(free)); + }); +}); + +describe("compactHistory fuzz: concurrent histories", () => { + for (const [name, codec] of [ + ["text", stringCodec], + ["db", dbCodec], + ] as const) { + it(`keeps every invariant for random ${name} histories and policies`, () => { + let compacted = 0; + for (let seed = 1; seed <= 120; seed++) { + const r = rng(seed * 7919); + const patches = randomHistory(seed, codec, 5 + r(60)); + const now = T0 + 40 * DAY; + const policy = randomPolicy(r, now); + const result = checkInvariants(codec, patches, policy, now); + if (!result.stats.identity) compacted++; + } + // most of them actually compact + expect(compacted).toBeGreaterThan(60); + }); + } + + it("keeps several concurrent heads, merging to the same value", () => { + for (let seed = 200; seed < 260; seed++) { + const patches = randomHistory(seed, stringCodec, 30); + const old = new PatchGraph({ codec: stringCodec }); + old.add(patches); + if (old.getHeads().length < 2) continue; + const result = checkInvariants(stringCodec, patches, { kind: "latestOnly" }, T0); + const fresh = new PatchGraph({ codec: stringCodec }); + fresh.add(result.patches); + expect(fresh.getHeads()).toEqual(old.getHeads()); + } + }); +}); + +describe("compactHistory scale", () => { + it("compacts 10k patches of a large text quickly", () => { + const codec = stringCodec; + const patches: Patch[] = []; + let value = Array.from({ length: 2000 }, (_, i) => `line ${i} ${"x".repeat(40)}`).join("\n"); + let prev = codec.fromString(""); + let doc = codec.fromString(value); + const g0 = encodePatchId(T0, "u"); + patches.push({ time: g0, parents: [], userId: 0, isSnapshot: true, snapshot: value }); + prev = doc; + for (let i = 1; i < 10_000; i++) { + const lines = value.split("\n"); + lines[i % lines.length] += ` e${i}`; + value = lines.join("\n"); + doc = codec.fromString(value); + patches.push({ + time: encodePatchId(T0 + i * 60_000, `u${i % 5}`), + parents: [patches[i - 1].time], + userId: i % 5, + patch: codec.makePatch(prev, doc), + }); + prev = doc; + } + const start = Date.now(); + const result = compactHistory({ + patches, + codec, + policy: { kind: "uniform", interval: DAY }, + now: T0 + 10_000 * 60_000, + }); + const elapsed = Date.now() - start; + expect(result.versions.length).toBeLessThan(10); + expect(result.stats.bytesAfter).toBeLessThan(result.stats.bytesBefore); + const fresh = new PatchGraph({ codec }); + fresh.add(result.patches); + expect(fresh.value().toString()).toBe(value); + expect(elapsed).toBeLessThan(30_000); + }, 60_000); +}); diff --git a/src/compact-history.ts b/src/compact-history.ts new file mode 100644 index 0000000..bf1c16a --- /dev/null +++ b/src/compact-history.ts @@ -0,0 +1,540 @@ +/* +History compaction: a smaller history that keeps chosen versions exactly. + +TimeTravel is revision control for some documents (code, papers) and disaster +recovery for others (chats, whose newest version already holds everything), +so a host can offer retention strategies instead of only deleting history. +compactHistory is the pure, deterministic part: a history and a policy in, a +new history out. Storing it (atomically, as a new history) is the host's job. + +How it works. A policy chooses versions (patch times) to keep; the heads, and +the merge bases the value of several heads is merged from, are always kept. +Every kept version becomes a patch whose parents are its nearest kept +ancestors, with a diff from their exact value in the old history, so its value +is exactly its old value. A kept version with several parents records its +merged value (Patch.mergeParent/mergePatch), so its value never depends on how +a later version merges. A kept version without kept ancestors becomes a +snapshot. Dropped patches are squashed into their nearest kept descendants, +which record whose edits they absorbed (meta.squashed). A kept patch whose +parents do not change and that absorbs nothing is kept as it was (minus the +snapshot text a store may have attached to it), which makes compaction +idempotent and leaves untouched history byte for byte. + +Guarantees (see compact-history.test.ts): +- every kept version has exactly its old value; +- the newest value (of all heads) is unchanged; +- the result is a valid history, the same for the same input; +- compacting the result again with the same policy changes nothing; +- the users whose edits were dropped are recorded on the patches absorbing them; +- the result is never larger than the input (else the input is returned). +*/ + +import { PatchGraph } from "./patch-graph"; +import { comparePatchId, decodePatchId, type PatchId } from "./patch-id"; +import type { DocCodec, Document, JSONValue, Patch } from "./types"; + +// Which versions to keep (besides the heads and their merge bases, which are +// always kept). Times are milliseconds; a version's time is that of its patch +// id (decodePatchId). +export type CompactionPolicy = + // Everything: the result is the input. + | { kind: "keepAll" } + // The version current at `since` and every version after it. + | { kind: "keepSince"; since: number } + // The newest version of every `interval` (aligned to the epoch). + | { kind: "uniform"; interval: number } + // Backup style: versions younger than the first tier's olderThan are all + // kept; a version at least `olderThan` old (relative to `now`), for the + // largest such tier, is kept if it is the newest of its `interval` (every + // version if interval <= 0). Example: all for 7 days, hourly for 4 weeks, + // daily for a year, weekly after: + // [{ olderThan: 7*DAY, interval: HOUR }, + // { olderThan: 35*DAY, interval: DAY }, + // { olderThan: 365*DAY, interval: 7*DAY }] + | { kind: "tiered"; tiers: { olderThan: number; interval: number }[] } + // Only the newest version. + | { kind: "latestOnly" } + // Every version any of the policies keeps. + | { kind: "union"; policies: CompactionPolicy[] }; + +// Recorded in Patch.meta.squashed on a patch that absorbed dropped patches: +// the users whose edits it holds (in the order of their first edit, +// including its own author), its first and last absorbed times, and how +// many patches of the original history it stands for. +export interface SquashedInfo { + users: number[]; + start: PatchId; + end: PatchId; + count: number; +} + +// Limits on the work compaction may do. Diffs dominate its cost (each is a +// diff of two whole versions), so they are counted, not timed: the same input +// and budget always give the same result. Over budget, the input is returned +// unchanged (stats.overBudget), before any diff that would exceed it. +export interface CompactionBudget { + // Diffs computed (one per rewritten patch, a few more per merge commit). + maxDiffs?: number; + // Total size of the versions diffed (both sides of every diff; characters + // of a text, the codec's size estimate for other documents). + maxDiffSize?: number; +} + +export interface CompactionStats { + versionsBefore: number; + versionsAfter: number; + // Encoded size (see encodedSize) of all records. + bytesBefore: number; + bytesAfter: number; + // The input was returned unchanged: the policy keeps everything, the + // compacted history would not have been smaller, or the budget ran out. + identity: boolean; + // The budget ran out (the input was returned unchanged). + overBudget: boolean; + // Work done: diffs computed and the total size of the versions diffed. + diffs: number; + diffSize: number; +} + +export interface CompactionResult { + // The new history in time order: the first record is a snapshot (unless + // the input is returned unchanged), and every parent precedes its children. + patches: Patch[]; + // The kept versions (times of the input), in time order. + versions: PatchId[]; + stats: CompactionStats; +} + +// Size of a record as a host would store it (UTF-8 JSON). +export function encodedSize(patch: Patch): number { + return utf8Length(JSON.stringify(patch) ?? ""); +} + +function utf8Length(s: string): number { + let n = 0; + for (let i = 0; i < s.length; i++) { + const c = s.charCodeAt(i); + if (c < 0x80) n += 1; + else if (c < 0x800) n += 2; + else if (c >= 0xd800 && c <= 0xdbff && i + 1 < s.length) { + n += 4; + i++; + } else n += 3; + } + return n; +} + +// Parents considered for a merge commit's recorded merged value, as Session +// does (see Session.recordMerge). +const MAX_MERGE_PARENT_CANDIDATES = 2; + +function patchSize(patch: unknown): number { + if (patch instanceof Uint8Array) return patch.byteLength; + return JSON.stringify(patch)?.length ?? 0; +} + +function timeMs(time: PatchId): number { + return decodePatchId(time).timeMs; +} + +// The versions a policy keeps, of `times` (sorted, oldest first). +export function selectVersions( + policy: CompactionPolicy, + times: PatchId[], + now: number = Date.now(), +): Set { + const keep = new Set(); + const newestPer = (key: (t: PatchId) => string | undefined) => { + // times are sorted, so the last one seen per key is the newest + const newest = new Map(); + for (const t of times) { + const k = key(t); + if (k === undefined) keep.add(t); + else newest.set(k, t); + } + for (const t of newest.values()) keep.add(t); + }; + switch (policy.kind) { + case "keepAll": + for (const t of times) keep.add(t); + break; + case "latestOnly": + break; + case "keepSince": { + let current: PatchId | undefined; + for (const t of times) { + if (timeMs(t) >= policy.since) keep.add(t); + else current = t; + } + if (current !== undefined) keep.add(current); + break; + } + case "uniform": { + const interval = positive(policy.interval, "uniform interval"); + newestPer((t) => `${Math.floor(timeMs(t) / interval)}`); + break; + } + case "tiered": { + const tiers = [...policy.tiers].sort((a, b) => a.olderThan - b.olderThan); + newestPer((t) => { + const age = now - timeMs(t); + let tier = -1; + for (let i = 0; i < tiers.length; i++) if (age >= tiers[i].olderThan) tier = i; + if (tier < 0 || !(tiers[tier].interval > 0)) return undefined; + return `${tier}:${Math.floor(timeMs(t) / tiers[tier].interval)}`; + }); + break; + } + case "union": + for (const p of policy.policies) for (const t of selectVersions(p, times, now)) keep.add(t); + break; + default: + throw new Error(`unknown compaction policy: ${JSON.stringify(policy)}`); + } + return keep; +} + +function positive(x: number, what: string): number { + if (!(Number.isFinite(x) && x > 0)) throw new Error(`${what} must be positive, got ${x}`); + return x; +} + +export function compactHistory(opts: { + patches: Patch[]; + // The document type (must provide merge3: values must be exact). + codec: DocCodec; + policy: CompactionPolicy; + // For age based policies (tiered); default Date.now(). + now?: number; + budget?: CompactionBudget; +}): CompactionResult { + const { codec, policy } = opts; + const work = new Work(codec, opts.budget); + const now = opts.now ?? Date.now(); + if (codec.merge3 == null) { + throw new Error("compactHistory: the codec must provide merge3 (exact values)"); + } + const old = new PatchGraph({ codec }); + old.add(opts.patches); + if (old.needsMoreHistory() || old.valueUnavailable()) { + throw new Error( + "compactHistory: the history is incomplete (a patch depends on one that is missing)", + ); + } + const times = old.versions(); + const records = times.map((t) => old.getPatch(t)); + const bytesBefore = records.reduce((n, p) => n + encodedSize(p), 0); + const identity = (overBudget = false): CompactionResult => ({ + patches: records, + versions: times, + stats: { + versionsBefore: times.length, + versionsAfter: times.length, + bytesBefore, + bytesAfter: bytesBefore, + identity: true, + overBudget, + diffs: work.diffs, + diffSize: work.diffSize, + }, + }); + if (times.length === 0 || policy.kind === "keepAll") return identity(); + + const has = (t: PatchId) => { + try { + old.getPatch(t); + return true; + } catch { + return false; + } + }; + const parentsOf = (t: PatchId) => (old.getPatch(t).parents ?? []).filter(has); + + const keep = selectVersions(policy, times, now); + const heads = old.getHeads(); + for (const h of heads) keep.add(h); + for (const t of mergeBases(heads, parentsOf)) keep.add(t); + if (keep.size === times.length) return identity(); + + const kept = times.filter((t) => keep.has(t)); + let fresh = new PatchGraph({ codec }); + const out: Patch[] = []; + let bytesAfter = 0; + for (const t of kept) { + // Nearest kept ancestors, and the dropped patches only t builds on. + const frontier = new Set(); + const absorbed: PatchId[] = []; + const seen = new Set([t]); + const stack = [...parentsOf(t)]; + while (stack.length > 0) { + const u = stack.pop()!; + if (seen.has(u)) continue; + seen.add(u); + if (keep.has(u)) { + frontier.add(u); + continue; + } + absorbed.push(u); + stack.push(...parentsOf(u)); + } + const parents = maximal([...frontier], parentsOf).sort(comparePatchId); + const original = old.getPatch(t); + const value = old.version(t); + let record: Patch; + try { + record = buildRecord({ + codec, + work, + old, + original, + parents, + absorbed: absorbed.sort(comparePatchId), + value, + }); + } catch (err) { + if (err instanceof OverBudget) return identity(true); + throw err; + } + fresh.add([record]); + if (!same(codec, fresh.version(t), value)) { + // The diff did not reproduce the value (it should): store the value. + record = { ...withoutPatch(record), isSnapshot: true, snapshot: codec.toString(value) }; + fresh = new PatchGraph({ codec }); + fresh.add([...out, record]); + if (!same(codec, fresh.version(t), value)) { + throw new Error(`compactHistory: cannot reproduce the value of ${t}`); + } + } + out.push(record); + bytesAfter += encodedSize(record); + // Already no smaller than the input: stop (the input is the result). + if (bytesAfter > bytesBefore) return identity(); + } + + const result = new PatchGraph({ codec }); + result.add(out); + if (!same(codec, result.value(), old.value()) || result.needsMoreHistory()) { + // The heads did not merge to the same value (e.g. independent roots, + // which a history starting with snapshots cannot merge from empty). + return identity(); + } + return { + patches: out, + versions: kept, + stats: { + versionsBefore: times.length, + versionsAfter: kept.length, + bytesBefore, + bytesAfter, + identity: false, + overBudget: false, + diffs: work.diffs, + diffSize: work.diffSize, + }, + }; +} + +function same(codec: DocCodec, a: Document, b: Document): boolean { + return a.isEqual(b) || codec.toString(a) === codec.toString(b); +} + +function withoutPatch(p: Patch): Patch { + const { patch: _p, mergeParent: _mp, mergePatch: _mpatch, ...rest } = p; + void _p; + void _mp; + void _mpatch; + return rest; +} + +// A record as stored, without the snapshot a store attached to it (a host +// writes its own snapshots of the new history). +function withoutSnapshot(p: Patch): Patch { + const { isSnapshot: _i, snapshot: _s, seqInfo: _q, ...rest } = p; + void _i; + void _s; + void _q; + return rest; +} + +class OverBudget extends Error {} + +// Diffs done so far, against the budget. +class Work { + diffs = 0; + diffSize = 0; + constructor( + private codec: DocCodec, + private budget: CompactionBudget = {}, + ) {} + + makePatch(a: Document, b: Document): unknown { + const size = this.size(a) + this.size(b); + const { maxDiffs, maxDiffSize } = this.budget; + if (maxDiffs != null && this.diffs + 1 > maxDiffs) throw new OverBudget(); + if (maxDiffSize != null && this.diffSize + size > maxDiffSize) throw new OverBudget(); + this.diffs += 1; + this.diffSize += size; + return this.codec.makePatch(a, b); + } + + private size(doc: Document): number { + const n = doc.size?.() ?? doc.count?.(); + return Number.isFinite(n) && n! >= 0 ? n! : this.codec.toString(doc).length; + } +} + +function buildRecord(opts: { + codec: DocCodec; + work: Work; + old: PatchGraph; + original: Patch; + parents: PatchId[]; + absorbed: PatchId[]; + value: Document; +}): Patch { + const { codec, work, old, original, parents, absorbed, value } = opts; + const originalParents = [...(original.parents ?? [])].sort(comparePatchId); + const sameParents = + originalParents.length === parents.length && originalParents.every((p, i) => p === parents[i]); + // Unchanged: same parents, nothing absorbed, and a value that does not + // depend on how its parents merge. + if (absorbed.length === 0 && sameParents) { + // a root snapshot without a patch, as compaction writes them + if (parents.length === 0 && original.isSnapshot && original.snapshot != null) { + if (original.patch == null) return original; + } + if ( + parents.length === 1 || + (parents.length > 1 && original.mergeParent != null && original.mergePatch != null) + ) { + if (original.patch != null) return withoutSnapshot(original); + } + } + const meta = squashMeta(old, original, absorbed); + const base: Patch = { + time: original.time, + parents, + ...(original.wall != null ? { wall: original.wall } : {}), + ...(original.userId != null ? { userId: original.userId } : {}), + ...(original.file ? { file: true } : {}), + hash: old.hashOf(value), + ...(meta != null ? { meta } : {}), + }; + if (parents.length === 0) { + return { ...base, isSnapshot: true, snapshot: codec.toString(value) }; + } + if (parents.length === 1) { + const from = old.version(parents[0]); + return { ...base, patch: work.makePatch(from, value) }; + } + // A merge commit: record its merged value as a diff from one parent. + const merged = old.exactValueOf(parents); + if (merged == null) throw new Error(`compactHistory: cannot merge ${parents.join(",")}`); + let best: { parent: PatchId; patch: unknown; size: number } | undefined; + for (const parent of parents.slice(0, MAX_MERGE_PARENT_CANDIDATES)) { + const patch = work.makePatch(old.version(parent), merged); + const size = patchSize(patch); + if (best == null || size < best.size) best = { parent, patch, size }; + } + return { + ...base, + patch: work.makePatch(merged, value), + mergeParent: best!.parent, + mergePatch: best!.patch, + }; +} + +// The authorship of a kept patch that absorbs dropped ones (Patch.meta.squashed). +function squashMeta( + old: PatchGraph, + original: Patch, + absorbed: PatchId[], +): { [key: string]: JSONValue } | undefined { + if (absorbed.length === 0) return original.meta; + const users: number[] = []; + const addUsers = (p: Patch) => { + const info = squashedInfo(p); + for (const u of info?.users ?? (p.userId != null ? [p.userId] : [])) { + if (!users.includes(u)) users.push(u); + } + }; + let start = original.time; + let count = squashedInfo(original)?.count ?? 1; + for (const t of absorbed) { + const p = old.getPatch(t); + addUsers(p); + const info = squashedInfo(p); + count += info?.count ?? 1; + const first = info?.start ?? t; + if (comparePatchId(first, start) < 0) start = first; + } + addUsers(original); + const ownStart = squashedInfo(original)?.start; + if (ownStart != null && comparePatchId(ownStart, start) < 0) start = ownStart; + const squashed: SquashedInfo = { users, start, end: original.time, count }; + return { ...original.meta, squashed: squashed as unknown as JSONValue }; +} + +export function squashedInfo(p: Patch): SquashedInfo | undefined { + const s = p.meta?.squashed as unknown as SquashedInfo | undefined; + return s != null && Array.isArray(s.users) ? s : undefined; +} + +// The maximal elements of `times` (those not an ancestor of another). +function maximal(times: PatchId[], parentsOf: (t: PatchId) => PatchId[]): PatchId[] { + if (times.length <= 1) return times; + const set = new Set(times); + const below = new Set(); + for (const t of times) { + const stack = [...parentsOf(t)]; + const seen = new Set(); + // Patches are older than their children, so nothing older than the + // oldest candidate can be one. + const oldest = [...set].sort(comparePatchId)[0]; + while (stack.length > 0) { + const u = stack.pop()!; + if (seen.has(u) || comparePatchId(u, oldest) < 0) continue; + seen.add(u); + if (set.has(u)) below.add(u); + stack.push(...parentsOf(u)); + } + } + return times.filter((t) => !below.has(t)); +} + +function ancestorsOf(times: PatchId[], parentsOf: (t: PatchId) => PatchId[]): Set { + const seen = new Set(); + const stack = [...times]; + while (stack.length > 0) { + const t = stack.pop()!; + if (seen.has(t)) continue; + seen.add(t); + stack.push(...parentsOf(t)); + } + return seen; +} + +// Every patch the merged value of `heads` is computed from, as PatchGraph +// merges them (heads in time order, each into the merge of those before, from +// their maximal common ancestors, themselves merged the same way). Kept, they +// give the compacted history the same merges and so the same newest value. +function mergeBases(heads: PatchId[], parentsOf: (t: PatchId) => PatchId[]): Set { + const out = new Set(); + const visit = (set: PatchId[]) => { + const sorted = [...new Set(set)].sort(comparePatchId); + if (sorted.length < 2) return; + let acc = [sorted[0]]; + for (const t of sorted.slice(1)) { + const a = ancestorsOf(acc, parentsOf); + const b = ancestorsOf([t], parentsOf); + const common = [...a].filter((u) => b.has(u)); + const bases = maximal(common, parentsOf); + const fresh = bases.filter((u) => !out.has(u)); + for (const u of bases) out.add(u); + if (fresh.length > 0) visit(bases); + acc = [...acc, t]; + } + }; + visit(heads); + return out; +} diff --git a/src/index.ts b/src/index.ts index 115f83e..dbe2f96 100644 --- a/src/index.ts +++ b/src/index.ts @@ -25,3 +25,4 @@ export type { } from "./db-document-immer"; export * as dbDocumentImmer from "./db-document-immer"; export * from "./client-id"; +export * from "./compact-history"; From 0c190a66f3a62df83b9c75b8c5fd08aafca27e5b Mon Sep 17 00:00:00 2001 From: William Stein Date: Thu, 22 Oct 2026 08:49:42 -0700 Subject: [PATCH 2/3] compactHistory: no old stream position on kept records; follow every set of merge bases @review (#32): - A kept record reused unchanged (e.g. the root snapshot) kept the old history's seqInfo (and source): the new history's first record then claimed an old stream position. Reused records drop seqInfo and source. - mergeBases recursed only when a base was new, so a set of bases whose members were all seen before (in other sets) was not itself merged from its own bases, and those were not kept (the final check then returned the input). Recurse once per set of bases. Regression tests for both (each fails before). README: an identity result means write nothing; no record carries seqInfo or source. --- README.md | 2 +- src/compact-history.test.ts | 75 +++++++++++++++++++++++++++++++++++++ src/compact-history.ts | 23 ++++++++++-- 3 files changed, 95 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 61bd2c5..f298c26 100644 --- a/README.md +++ b/README.md @@ -76,7 +76,7 @@ Merge commits written before this (no `mergeParent` and no `hash`) were made by The heads, and the merge bases their merged value is computed from, are always kept, so the newest value is unchanged, also with concurrent heads. Every kept version gets its nearest kept ancestors as parents and a diff from their exact value, so its value is exactly what it was; a merge records its merged value (`mergeParent`, `mergePatch`). A kept version without kept ancestors becomes a snapshot, so the result starts with one. A patch that absorbs dropped patches records whose edits it holds in `meta.squashed` (`users` in order, `start`, `end`, `count`). Patches that neither change parents nor absorb anything are kept as they were, so compacting again with the same policy changes nothing. The result is never larger than the input: if it would be, or the `budget` (diffs, and total size of the versions diffed) runs out, the input is returned unchanged (`stats.identity`, `stats.overBudget`). The codec must provide `merge3`, and the history must be complete (starting at a root or a snapshot). The property and fuzz tests in [src/compact-history.test.ts](./src/compact-history.test.ts) check all of this on random concurrent histories of texts and databases; `pnpm bench:compact` measures it on a long history. -**Storing a compacted history.** The result replaces the whole history: write it atomically, as a new history, in its order. Every record is a `Patch` as it is given; nothing refers to the records that are dropped. +**Storing a compacted history.** The result replaces the whole history: write it atomically, as a new history, in its order. If `stats.identity` is true, the result is the input as it was stored: write nothing. Every record is a `Patch` as it is given; nothing refers to the records that are dropped, and no record carries a stream position (`seqInfo`) or transport (`source`) of the old history. - The first record (and any other record with no parents) is a snapshot: `isSnapshot: true`, the full value in `snapshot`, `parents: []`, no `patch`, and `hash`. It is the start of the new history, like the first snapshot of a document; it has no older history to load (no `seqInfo`). - Every other record is a patch with `parents` (all earlier in the result), `patch`, `hash`, and for a merge `mergeParent` and `mergePatch`. Its `time` is the original patch id, so ids stay unique and stable, and `userId` and `wall` are those of the original patch. diff --git a/src/compact-history.test.ts b/src/compact-history.test.ts index 14f5f6c..148956b 100644 --- a/src/compact-history.test.ts +++ b/src/compact-history.test.ts @@ -415,6 +415,81 @@ describe("compactHistory fuzz: concurrent histories", () => { }); } + it("a kept root snapshot starts the new history: no old stream position", () => { + const codec = stringCodec; + const g = new PatchGraph({ codec }); + const t = (i: number) => encodePatchId(T0 + i * HOUR, `u${i}`); + const root: Patch = { + time: t(0), + parents: [], + userId: 0, + isSnapshot: true, + snapshot: "a\n", + seqInfo: { seq: 500, prevSeq: 400 }, + source: "persist", + }; + g.add([root]); + const commit = (i: number, parent: number, text: string): Patch => { + const from = g.exactValueOf([t(parent)])!; + const to = codec.fromString(text); + const p: Patch = { + time: t(i), + parents: [t(parent)], + userId: i, + patch: codec.makePatch(from, to), + hash: g.hashOf(to), + }; + g.add([p]); + return p; + }; + // two branches from the root; the first has an edit that is dropped + const patches = [ + root, + commit(1, 0, "a\nb\n"), + commit(2, 0, "c\na\n"), + commit(3, 1, "a\nb\nd\n"), + ]; + const result = checkInvariants(codec, patches, { kind: "latestOnly" }, T0); + expect(result.stats.identity).toBe(false); + expect(result.versions).toEqual([t(0), t(2), t(3)]); + expect(result.patches[0].time).toBe(t(0)); + expect(result.patches[0].isSnapshot).toBe(true); + expect(result.patches[0].seqInfo).toBeUndefined(); + expect(result.patches[0].source).toBeUndefined(); + }); + + it("follows every set of merge bases, also one of bases seen before", () => { + // (@review) heads 4, 5, 6, 7 merge from bases 2 and 3, which merge from + // base 1: 1 must be kept, 0 need not be. + const codec = stringCodec; + const parentsOf = [[], [0], [1], [1], [3], [2, 3], [1, 2], [1, 2, 3]]; + const t = (i: number) => encodePatchId(T0 + i * HOUR, `u${i}`); + const g = new PatchGraph({ codec }); + const patches: Patch[] = []; + parentsOf.forEach((ps, i) => { + const parents = ps.map(t); + const from = parents.length ? g.exactValueOf(parents)! : codec.fromString(""); + const to = codec.fromString(`${from.toString()}line ${i}\n`); + const p: Patch = { + time: t(i), + parents, + userId: i % 3, + patch: codec.makePatch(from, to), + hash: g.hashOf(to), + }; + if (parents.length > 1) { + p.mergeParent = parents[0]; + p.mergePatch = codec.makePatch(g.exactValueOf([parents[0]])!, from); + } + g.add([p]); + patches.push(p); + }); + expect(g.getHeads()).toEqual([4, 5, 6, 7].map(t)); + const result = checkInvariants(codec, patches, { kind: "latestOnly" }, T0); + expect(result.stats.identity).toBe(false); + expect(result.versions).toEqual([1, 2, 3, 4, 5, 6, 7].map(t)); + }); + it("keeps several concurrent heads, merging to the same value", () => { for (let seed = 200; seed < 260; seed++) { const patches = randomHistory(seed, stringCodec, 30); diff --git a/src/compact-history.ts b/src/compact-history.ts index bf1c16a..d8b62e9 100644 --- a/src/compact-history.ts +++ b/src/compact-history.ts @@ -356,6 +356,16 @@ function withoutSnapshot(p: Patch): Patch { return rest; } +// Without where a record was stored or came from: the stream position of the +// old history (seqInfo) and the transport (source). The new history is +// written as a new stream. +function withoutProvenance(p: Patch): Patch { + const { seqInfo: _q, source: _s, ...rest } = p; + void _q; + void _s; + return rest; +} + class OverBudget extends Error {} // Diffs done so far, against the budget. @@ -401,13 +411,13 @@ function buildRecord(opts: { if (absorbed.length === 0 && sameParents) { // a root snapshot without a patch, as compaction writes them if (parents.length === 0 && original.isSnapshot && original.snapshot != null) { - if (original.patch == null) return original; + if (original.patch == null) return withoutProvenance(original); } if ( parents.length === 1 || (parents.length > 1 && original.mergeParent != null && original.mergePatch != null) ) { - if (original.patch != null) return withoutSnapshot(original); + if (original.patch != null) return withoutProvenance(withoutSnapshot(original)); } } const meta = squashMeta(old, original, absorbed); @@ -520,18 +530,23 @@ function ancestorsOf(times: PatchId[], parentsOf: (t: PatchId) => PatchId[]): Se // give the compacted history the same merges and so the same newest value. function mergeBases(heads: PatchId[], parentsOf: (t: PatchId) => PatchId[]): Set { const out = new Set(); + // Sets already merged: each set of bases is itself merged from its own + // bases, also when every member was seen before (in another set). + const visited = new Set(); const visit = (set: PatchId[]) => { const sorted = [...new Set(set)].sort(comparePatchId); if (sorted.length < 2) return; + const key = sorted.join(","); + if (visited.has(key)) return; + visited.add(key); let acc = [sorted[0]]; for (const t of sorted.slice(1)) { const a = ancestorsOf(acc, parentsOf); const b = ancestorsOf([t], parentsOf); const common = [...a].filter((u) => b.has(u)); const bases = maximal(common, parentsOf); - const fresh = bases.filter((u) => !out.has(u)); for (const u of bases) out.add(u); - if (fresh.length > 0) visit(bases); + visit(bases); acc = [...acc, t]; } }; From 0b382e582331fb17cca4fa20d418ce57fd266b84 Mon Sep 17 00:00:00 2001 From: William Stein Date: Thu, 22 Oct 2026 08:52:58 -0700 Subject: [PATCH 3/3] README: hash is on rewritten records (kept ones keep theirs); a kept root snapshot is carried over @review (#32): the record contract claimed a hash on every record and no old snapshots, but records kept as they were keep their own fields, and a kept root snapshot starts the new history. --- README.md | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index f298c26..f270e83 100644 --- a/README.md +++ b/README.md @@ -78,10 +78,11 @@ The heads, and the merge bases their merged value is computed from, are always k **Storing a compacted history.** The result replaces the whole history: write it atomically, as a new history, in its order. If `stats.identity` is true, the result is the input as it was stored: write nothing. Every record is a `Patch` as it is given; nothing refers to the records that are dropped, and no record carries a stream position (`seqInfo`) or transport (`source`) of the old history. -- The first record (and any other record with no parents) is a snapshot: `isSnapshot: true`, the full value in `snapshot`, `parents: []`, no `patch`, and `hash`. It is the start of the new history, like the first snapshot of a document; it has no older history to load (no `seqInfo`). -- Every other record is a patch with `parents` (all earlier in the result), `patch`, `hash`, and for a merge `mergeParent` and `mergePatch`. Its `time` is the original patch id, so ids stay unique and stable, and `userId` and `wall` are those of the original patch. +- The first record (and any other record with no parents) is a snapshot: `isSnapshot: true`, the full value in `snapshot`, `parents: []` and no `patch`. It is the start of the new history, like the first snapshot of a document; it has no older history to load (no `seqInfo`). +- Every other record is a patch with `parents` (all earlier in the result) and `patch`, and for a merge `mergeParent` and `mergePatch`. Its `time` is the original patch id, so ids stay unique and stable, and `userId` and `wall` are those of the original patch. +- Records compaction rewrites (a new snapshot, or a patch whose parents changed or that absorbs dropped patches) carry `hash`. A record kept as it was keeps its own fields, so it has `hash` only if it had one (a patch written before hashes, or marked `inexact`, has none); store `hash` when present. - `meta.squashed` (and any other `meta`) must be stored with the record. -- Snapshot records of the old history are not in the result (a kept patch loses the snapshot text a store attached to it). A host makes snapshots of the new history as for any history, e.g. one of the newest version right away, so that clients do not load it in full. +- Snapshots of the old history are not carried over, except a root snapshot that is kept as the start of the new history: a kept patch loses the snapshot text a store attached to it. A host makes snapshots of the new history as for any history, e.g. one of the newest version right away, so that clients do not load it in full. - A rare repaired record has `isSnapshot` and `snapshot` with `parents` (its value is stored, not diffed): store it like any patch with a snapshot. ## How Patchflow compares