compactHistory: retention policies that keep chosen versions exactly

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.

README.md

@@ -61,6 +61,30 @@ 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. 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: []` 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.
+- 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
 
 | Approach                     | Examples                                                            | What is exchanged                      | History                                 | Merging                                        |

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

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",