| … | … | @@ -0,0 +1,248 @@ |
| 1 | // repoDiffSince — com.atproto.sync.getRepo?since as a real PARTIAL CAR assembled from the firehose. |
| 2 | // |
| 3 | // The claim under test is NOT "a CAR comes back". It is that the CAR is genuinely a DIFF: |
| 4 | // - it OMITS the blocks the caller already has (proven by the tree being unwalkable on its own) |
| 5 | // - it COMPLETES the caller's prior state (proven by unioning with the old CAR and walking) |
| 6 | // - and it refuses, out loud, whenever completeness cannot be PROVEN from the firehose we hold. |
| 7 | // A test that only checked "the diff is smaller" would pass on an implementation that dropped blocks at |
| 8 | // random, and one that only checked "the records are there" would pass on a full CAR — which is exactly |
| 9 | // the accepted-and-ignored behaviour this replaces. |
| 10 | |
| 11 | import test from 'brittle' |
| 12 | import { mkdtempSync, rmSync } from 'node:fs' |
| 13 | import { tmpdir } from 'node:os' |
| 14 | import { join } from 'node:path' |
| 15 | import * as dagCbor from '@ipld/dag-cbor' |
| 16 | import { openCorestore } from '../lib/corestore.js' |
| 17 | import { RepoStore } from '../lib/repo-store.js' |
| 18 | import { RepoManager } from '../lib/repo-manager.js' |
| 19 | import { FirehoseSequencer } from '../lib/sequencer.js' |
| 20 | import { getRepoFromCommit } from '../lib/read-view.js' |
| 21 | import { repoDiffSince, collectFramesSince, unionDiffCar } from '../lib/repo-diff.js' |
| 22 | import { readCar, writeCar, listRecords, loadCommit } from 'atproto-repo/lib/repo.js' |
| 23 | import { commitCid } from 'atproto-repo/lib/commit.js' |
| 24 | import { cidForBytesSync } from 'atproto-repo/lib/cid.js' |
| 25 | |
| 26 | const PRIV = new Uint8Array(32).fill(7) |
| 27 | const NOW = '2026-08-07T00:00:00.000Z' |
| 28 | // A rev BELOW every real one. The atproto TID alphabet is '234567abcdefghijklmnopqrstuvwxyz', so '2' is |
| 29 | // the LOWEST character, not 'a' — a sentinel of 'aaaa…' is mid-range and compares NEWER than a real |
| 30 | // '3k…' rev, which silently turns an "older than the repo" test into a "since is ahead of head" test. |
| 31 | const BEFORE_ALL = '2222222222222' |
| 32 | |
| 33 | // A repo with a real firehose beside it: every applyWrite is sequenced, so the frames carry the genuine |
| 34 | // per-commit block CARs rather than anything this test made up. |
| 35 | async function newRepo (t) { |
| 36 | const dir = mkdtempSync(join(tmpdir(), 'repo-diff-')) |
| 37 | const store = openCorestore(dir) |
| 38 | const repo = await new RepoStore(store, { name: 'atproto-repo' }).ready() |
| 39 | let micros = 1_700_000_000_000_000n |
| 40 | const mgr = await new RepoManager(repo, { did: 'did:plc:diff', privateKey: PRIV, clock: () => (micros += 1_000_000n) }).init() |
| 41 | const seq = await new FirehoseSequencer(store.get({ name: 'firehose' })).ready() |
| 42 | t.teardown(async () => { await seq.close(); await repo.close(); await store.close(); rmSync(dir, { recursive: true, force: true }) }) |
| 43 | |
| 44 | let t0 = Date.parse(NOW) |
| 45 | const write = async (rkey, text) => { |
| 46 | const w = await mgr.applyWrite({ action: 'create', collection: 'app.bsky.feed.post', rkey, record: { $type: 'app.bsky.feed.post', text, createdAt: new Date(t0 += 1000).toISOString() } }) |
| 47 | await seq.sequenceCommit(w, { time: new Date(t0).toISOString() }) |
| 48 | return w |
| 49 | } |
| 50 | const head = () => mgr.signed || null |
| 51 | return { repo, mgr, seq, write, head, dir } |
| 52 | } |
| 53 | |
| 54 | const frameReader = (seq) => ({ getFrame: (i) => seq.core.get(i), length: seq.core.length }) |
| 55 | |
| 56 | // The head block, as the front would supply it: the verified signed commit and its CID. |
| 57 | const headBlock = (w) => ({ headCommitCid: w.commitCid, headCommitBytes: dagCbor.encode(w.signedCommit), headRev: w.rev }) |
| 58 | |
| 59 | test('the diff OMITS untouched blocks — it cannot be walked on its own', async (t) => { |
| 60 | const r = await newRepo(t) |
| 61 | const w1 = await r.write('a1', 'first') |
| 62 | const w2 = await r.write('a2', 'second') |
| 63 | const w3 = await r.write('a3', 'third') |
| 64 | |
| 65 | const d = await repoDiffSince({ ...frameReader(r.seq), since: w1.rev, ...headBlock(w3) }) |
| 66 | t.ok(d.ok, 'a diff from a rev the firehose covers is served') |
| 67 | t.is(d.fromRev, w1.rev, 'anchored at the frame proving we hold that history') |
| 68 | t.is(d.toRev, w3.rev, 'and runs up to the head') |
| 69 | |
| 70 | const full = await getRepoFromCommit(r.repo, w3.signedCommit) |
| 71 | const fullCar = await readCar(full.carBytes) |
| 72 | const diffCar = await readCar(d.carBytes) |
| 73 | |
| 74 | // The property that makes it a diff rather than a small repo: the MST cannot be walked, because the |
| 75 | // nodes that did not change were deliberately left out. If this ever passes, the "diff" is a full CAR. |
| 76 | t.exception(() => listRecords(diffCar), 'the partial CAR is NOT self-contained — walking it throws on a missing block') |
| 77 | |
| 78 | // And the record written BEFORE `since` is absent from it. |
| 79 | const firstRecordCid = [...fullCar.blocks.values()] |
| 80 | .find((b) => { try { const v = dagCbor.decode(b.bytes); return v && v.text === 'first' } catch { return false } }) |
| 81 | t.absent(diffCar.blocks.has(firstRecordCid.cid.toString()), 'a record older than `since` is NOT in the diff') |
| 82 | |
| 83 | // ⚠ A DIFF IS NOT NECESSARILY SMALLER, and asserting that it is was wrong. A diff spanning N commits |
| 84 | // carries every INTERMEDIATE MST root, which the full repo does not — on a small repo that can make it |
| 85 | // the larger of the two. The saving is real only once the untouched part of the tree dwarfs the churn. |
| 86 | const intermediate = [...diffCar.blocks.keys()].filter((k) => !fullCar.blocks.has(k)) |
| 87 | t.ok(intermediate.length > 0, 'the diff carries intermediate blocks the full repo does not — this is why it can be BIGGER on a small repo') |
| 88 | }) |
| 89 | |
| 90 | test('the diff COMPLETES the caller prior state — union walks to the full record set', async (t) => { |
| 91 | const r = await newRepo(t) |
| 92 | const w1 = await r.write('a1', 'first') |
| 93 | // What a caller holds at rev w1: the whole repo as of then. |
| 94 | const prior = await readCar((await getRepoFromCommit(r.repo, w1.signedCommit)).carBytes) |
| 95 | |
| 96 | await r.write('a2', 'second') |
| 97 | const w3 = await r.write('a3', 'third') |
| 98 | |
| 99 | const d = await repoDiffSince({ ...frameReader(r.seq), since: w1.rev, ...headBlock(w3) }) |
| 100 | const diffCar = await readCar(d.carBytes) |
| 101 | |
| 102 | // Apply: prior blocks + diff blocks, rooted at the NEW commit. This is exactly what a consumer with a |
| 103 | // block store does, and the only thing that proves the diff is sufficient rather than merely plausible. |
| 104 | const union = new Map([...prior.blocks, ...diffCar.blocks]) |
| 105 | const merged = await readCar(await writeCar(w3.commitCid, [...union.values()])) |
| 106 | const recs = listRecords(merged) |
| 107 | const texts = recs.map((x) => x.value.text).sort() |
| 108 | t.alike(texts, ['first', 'second', 'third'], 'prior + diff walks to the COMPLETE record set') |
| 109 | t.ok(loadCommit(merged).cid.equals(w3.commitCid), 'rooted at the head commit') |
| 110 | }) |
| 111 | |
| 112 | test('since == head -> an EMPTY diff, which is a real answer and not an error', async (t) => { |
| 113 | const r = await newRepo(t) |
| 114 | await r.write('a1', 'first') |
| 115 | const w2 = await r.write('a2', 'second') |
| 116 | |
| 117 | const d = await repoDiffSince({ ...frameReader(r.seq), since: w2.rev, ...headBlock(w2) }) |
| 118 | t.ok(d.ok, 'served') |
| 119 | t.ok(d.empty, 'reported as empty') |
| 120 | const car = await readCar(d.carBytes) |
| 121 | t.is(car.blocks.size, 1, 'the CAR carries the commit block and nothing else') |
| 122 | t.ok(loadCommit(car).cid.equals(w2.commitCid), 'and is rooted at the current head') |
| 123 | }) |
| 124 | |
| 125 | test('a rev the firehose cannot reach is REFUSED, with a reason', async (t) => { |
| 126 | const r = await newRepo(t) |
| 127 | const w1 = await r.write('a1', 'first') |
| 128 | const w2 = await r.write('a2', 'second') |
| 129 | |
| 130 | // A firehose that starts AFTER the requested rev: the first frame is hidden, so nothing proves we hold |
| 131 | // history back to `since` and nothing proves the frame we do have is genesis. |
| 132 | const truncated = { getFrame: (i) => r.seq.core.get(i + 1), length: r.seq.core.length - 1 } |
| 133 | const d = await repoDiffSince({ ...truncated, since: BEFORE_ALL, ...headBlock(w2) }) |
| 134 | t.absent(d.ok, 'refused rather than served') |
| 135 | t.ok(/does not reach back/.test(d.reason), 'and says why: ' + d.reason) |
| 136 | t.absent(d.carBytes, 'no CAR is produced — the caller must fall back to the full repo') |
| 137 | t.ok(w1.rev, 'sanity: the hidden first commit really existed') |
| 138 | }) |
| 139 | |
| 140 | // `since` from before the repo existed is REFUSED, and that is the right answer. |
| 141 | // |
| 142 | // I expected this to be the genesis case and wrote it to assert coverage. It is not. RepoManager.init() |
| 143 | // mints an empty genesis commit and persists it as the head WITHOUT sequencing it, so the first frame's |
| 144 | // `since` points at a commit the firehose never carried. We therefore cannot prove our firehose reaches |
| 145 | // the repo's first write rather than having been truncated to look that way — and refusing costs the |
| 146 | // caller nothing, because "everything" is exactly what the full CAR already is. |
| 147 | test('a since from before the repo existed is refused, not guessed at', async (t) => { |
| 148 | const r = await newRepo(t) |
| 149 | const w1 = await r.write('a1', 'first') |
| 150 | const w2 = await r.write('a2', 'second') |
| 151 | const d = await repoDiffSince({ ...frameReader(r.seq), since: BEFORE_ALL, ...headBlock(w2) }) |
| 152 | t.absent(d.ok, 'refused — the caller falls back to the full repo, which is what they asked for anyway') |
| 153 | t.ok(/does not reach back/.test(d.reason), 'and says why: ' + d.reason) |
| 154 | t.ok(w1.rev < w2.rev, 'sanity: revs are monotonic and lexicographically ordered') |
| 155 | }) |
| 156 | |
| 157 | // The coverage rule that DOES fire without an older frame present: the oldest frame we hold builds on a |
| 158 | // state at or before the caller's, so nothing sits between them. |
| 159 | test('coverage is proven from the oldest frame own `since`, not only from an older frame', async (t) => { |
| 160 | const r = await newRepo(t) |
| 161 | const w1 = await r.write('a1', 'first') |
| 162 | const w2 = await r.write('a2', 'second') |
| 163 | const w3 = await r.write('a3', 'third') |
| 164 | // Hide frame 0. The caller is at w1.rev; the oldest frame we still hold is w2, whose `since` IS w1.rev |
| 165 | // — so there is demonstrably no commit between the caller's state and our history. |
| 166 | const truncated = { getFrame: (i) => r.seq.core.get(i + 1), length: r.seq.core.length - 1 } |
| 167 | const d = await repoDiffSince({ ...truncated, since: w1.rev, ...headBlock(w3) }) |
| 168 | t.ok(d.ok, 'served without ever seeing a frame at or older than `since`') |
| 169 | t.is(d.fromRev, w1.rev, 'anchored at the rev the oldest held frame builds on') |
| 170 | }) |
| 171 | |
| 172 | // Dedup is tested on unionDiffCar DIRECTLY, with frames built to share a block. |
| 173 | // |
| 174 | // The first attempt drove this through a real repo of four single-record commits and asserted the union |
| 175 | // was smaller than the sum of the frames. It passed nothing: with one MST node, each commit contributes |
| 176 | // a fresh commit block, a fresh root and a fresh record — three blocks, none shared — so the sum WAS the |
| 177 | // union and the assertion could only ever have measured itself. Sharing needs a multi-level tree and a |
| 178 | // particular key distribution, which would make the test a hostage to MST fanout. Constructing the |
| 179 | // overlap outright tests the mechanism instead of hoping the fixture exhibits it. |
| 180 | test('a block carried by two frames is unioned ONCE', async (t) => { |
| 181 | const mkBlock = (n) => { const bytes = dagCbor.encode({ n }); return { cid: cidForBytesSync(bytes), bytes } } |
| 182 | const shared = mkBlock('shared') |
| 183 | const onlyA = mkBlock('a') |
| 184 | const onlyB = mkBlock('b') |
| 185 | const head = mkBlock('head') |
| 186 | |
| 187 | const frames = [ |
| 188 | { blocks: await writeCar(shared.cid, [shared, onlyA]) }, |
| 189 | { blocks: await writeCar(shared.cid, [shared, onlyB]) } |
| 190 | ] |
| 191 | const u = await unionDiffCar({ frames, headCommitCid: head.cid, headCommitBytes: head.bytes }) |
| 192 | const car = await readCar(u.carBytes) |
| 193 | t.is(u.blocks, 4, 'shared + a + b + head = 4, not 5 — the overlap collapsed') |
| 194 | t.is(car.blocks.size, u.blocks, 'the reported count matches the CAR') |
| 195 | t.ok(car.blocks.has(head.cid.toString()), 'the head commit block is present even though no frame carried it') |
| 196 | t.ok(car.roots[0].equals(head.cid), 'and it is the root') |
| 197 | }) |
| 198 | |
| 199 | test('a head block already present in a frame is not duplicated', async (t) => { |
| 200 | const r = await newRepo(t) |
| 201 | const w1 = await r.write('a1', 'first') |
| 202 | const w2 = await r.write('a2', 'second') |
| 203 | const d = await repoDiffSince({ ...frameReader(r.seq), since: w1.rev, ...headBlock(w2) }) |
| 204 | const car = await readCar(d.carBytes) |
| 205 | t.is(car.blocks.size, d.blocks, 'the reported block count is the CAR count — the head was added idempotently') |
| 206 | t.ok(car.roots[0].equals(w2.commitCid), 'rooted at the verified head') |
| 207 | }) |
| 208 | |
| 209 | test('collectFramesSince scans backward and stops at the proof — not a full-history walk', async (t) => { |
| 210 | const r = await newRepo(t) |
| 211 | await r.write('a1', 'first') |
| 212 | await r.write('a2', 'second') |
| 213 | const w3 = await r.write('a3', 'third') |
| 214 | await r.write('a4', 'fourth') |
| 215 | |
| 216 | let reads = 0 |
| 217 | const counting = { getFrame: (i) => { reads++; return r.seq.core.get(i) }, length: r.seq.core.length } |
| 218 | const got = await collectFramesSince({ ...counting, since: w3.rev }) |
| 219 | t.ok(got.covered, 'coverage proven') |
| 220 | t.is(got.frames.length, 1, 'exactly the one commit newer than that rev') |
| 221 | t.is(got.anchorRev, w3.rev, 'anchored at the frame that proved it') |
| 222 | t.is(reads, 2, 'read 2 frames, not all 4 — the scan stops at the proof') |
| 223 | }) |
| 224 | |
| 225 | test('a head ahead of the firehose tail is REFUSED — the diff could not be walked even with prior state', async (t) => { |
| 226 | const r = await newRepo(t) |
| 227 | const w1 = await r.write('a1', 'first') |
| 228 | const w2 = await r.write('a2', 'second') |
| 229 | // A front whose verified head is newer than anything the replicated firehose carries. Rooting a CAR at |
| 230 | // that commit would produce blocks that lead nowhere — worse than refusing, because it LOOKS valid. |
| 231 | const fakeHead = { ...w2.signedCommit, rev: 'zzzzzzzzzzzzz' } |
| 232 | const d = await repoDiffSince({ |
| 233 | ...frameReader(r.seq), |
| 234 | since: w1.rev, |
| 235 | headCommitCid: commitCid(fakeHead), |
| 236 | headCommitBytes: dagCbor.encode(fakeHead), |
| 237 | headRev: fakeHead.rev |
| 238 | }) |
| 239 | t.absent(d.ok, 'refused') |
| 240 | t.ok(/ahead of the firehose tail/.test(d.reason), 'and says why: ' + d.reason) |
| 241 | }) |
| 242 | |
| 243 | test('no since -> no diff (the caller must not get a partial CAR by accident)', async (t) => { |
| 244 | const r = await newRepo(t) |
| 245 | const w1 = await r.write('a1', 'first') |
| 246 | const d = await repoDiffSince({ ...frameReader(r.seq), since: null, ...headBlock(w1) }) |
| 247 | t.absent(d.ok, 'a missing `since` is never treated as "from the beginning"') |
| 248 | }) |