// repoDiffSince — com.atproto.sync.getRepo?since as a real PARTIAL CAR assembled from the firehose.
//
// The claim under test is NOT "a CAR comes back". It is that the CAR is genuinely a DIFF:
// - it OMITS the blocks the caller already has (proven by the tree being unwalkable on its own)
// - it COMPLETES the caller's prior state (proven by unioning with the old CAR and walking)
// - and it refuses, out loud, whenever completeness cannot be PROVEN from the firehose we hold.
// A test that only checked "the diff is smaller" would pass on an implementation that dropped blocks at
// random, and one that only checked "the records are there" would pass on a full CAR — which is exactly
// the accepted-and-ignored behaviour this replaces.
import test from 'brittle'
import { mkdtempSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import * as dagCbor from '@ipld/dag-cbor'
import { openCorestore } from '../lib/corestore.js'
import { RepoStore } from '../lib/repo-store.js'
import { RepoManager } from '../lib/repo-manager.js'
import { FirehoseSequencer } from '../lib/sequencer.js'
import { getRepoFromCommit } from '../lib/read-view.js'
import { repoDiffSince, collectFramesSince, unionDiffCar } from '../lib/repo-diff.js'
import { readCar, writeCar, listRecords, loadCommit } from 'atproto-repo/lib/repo.js'
import { commitCid } from 'atproto-repo/lib/commit.js'
import { cidForBytesSync } from 'atproto-repo/lib/cid.js'
const PRIV = new Uint8Array(32).fill(7)
const NOW = '2026-08-07T00:00:00.000Z'
// A rev BELOW every real one. The atproto TID alphabet is '234567abcdefghijklmnopqrstuvwxyz', so '2' is
// the LOWEST character, not 'a' — a sentinel of 'aaaa…' is mid-range and compares NEWER than a real
// '3k…' rev, which silently turns an "older than the repo" test into a "since is ahead of head" test.
const BEFORE_ALL = '2222222222222'
// A repo with a real firehose beside it: every applyWrite is sequenced, so the frames carry the genuine
// per-commit block CARs rather than anything this test made up.
async function newRepo (t) {
const dir = mkdtempSync(join(tmpdir(), 'repo-diff-'))
const store = openCorestore(dir)
const repo = await new RepoStore(store, { name: 'atproto-repo' }).ready()
let micros = 1_700_000_000_000_000n
const mgr = await new RepoManager(repo, { did: 'did:plc:diff', privateKey: PRIV, clock: () => (micros += 1_000_000n) }).init()
const seq = await new FirehoseSequencer(store.get({ name: 'firehose' })).ready()
t.teardown(async () => { await seq.close(); await repo.close(); await store.close(); rmSync(dir, { recursive: true, force: true }) })
let t0 = Date.parse(NOW)
const write = async (rkey, text) => {
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() } })
await seq.sequenceCommit(w, { time: new Date(t0).toISOString() })
return w
}
const head = () => mgr.signed || null
return { repo, mgr, seq, write, head, dir }
}
const frameReader = (seq) => ({ getFrame: (i) => seq.core.get(i), length: seq.core.length })
// The head block, as the front would supply it: the verified signed commit and its CID.
const headBlock = (w) => ({ headCommitCid: w.commitCid, headCommitBytes: dagCbor.encode(w.signedCommit), headRev: w.rev })
test('the diff OMITS untouched blocks — it cannot be walked on its own', async (t) => {
const r = await newRepo(t)
const w1 = await r.write('a1', 'first')
const w2 = await r.write('a2', 'second')
const w3 = await r.write('a3', 'third')
const d = await repoDiffSince({ ...frameReader(r.seq), since: w1.rev, ...headBlock(w3) })
t.ok(d.ok, 'a diff from a rev the firehose covers is served')
t.is(d.fromRev, w1.rev, 'anchored at the frame proving we hold that history')
t.is(d.toRev, w3.rev, 'and runs up to the head')
const full = await getRepoFromCommit(r.repo, w3.signedCommit)
const fullCar = await readCar(full.carBytes)
const diffCar = await readCar(d.carBytes)
// The property that makes it a diff rather than a small repo: the MST cannot be walked, because the
// nodes that did not change were deliberately left out. If this ever passes, the "diff" is a full CAR.
t.exception(() => listRecords(diffCar), 'the partial CAR is NOT self-contained — walking it throws on a missing block')
// And the record written BEFORE `since` is absent from it.
const firstRecordCid = [...fullCar.blocks.values()]
.find((b) => { try { const v = dagCbor.decode(b.bytes); return v && v.text === 'first' } catch { return false } })
t.absent(diffCar.blocks.has(firstRecordCid.cid.toString()), 'a record older than `since` is NOT in the diff')
// ⚠ A DIFF IS NOT NECESSARILY SMALLER, and asserting that it is was wrong. A diff spanning N commits
// carries every INTERMEDIATE MST root, which the full repo does not — on a small repo that can make it
// the larger of the two. The saving is real only once the untouched part of the tree dwarfs the churn.
const intermediate = [...diffCar.blocks.keys()].filter((k) => !fullCar.blocks.has(k))
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')
})
test('the diff COMPLETES the caller prior state — union walks to the full record set', async (t) => {
const r = await newRepo(t)
const w1 = await r.write('a1', 'first')
// What a caller holds at rev w1: the whole repo as of then.
const prior = await readCar((await getRepoFromCommit(r.repo, w1.signedCommit)).carBytes)
await r.write('a2', 'second')
const w3 = await r.write('a3', 'third')
const d = await repoDiffSince({ ...frameReader(r.seq), since: w1.rev, ...headBlock(w3) })
const diffCar = await readCar(d.carBytes)
// Apply: prior blocks + diff blocks, rooted at the NEW commit. This is exactly what a consumer with a
// block store does, and the only thing that proves the diff is sufficient rather than merely plausible.
const union = new Map([...prior.blocks, ...diffCar.blocks])
const merged = await readCar(await writeCar(w3.commitCid, [...union.values()]))
const recs = listRecords(merged)
const texts = recs.map((x) => x.value.text).sort()
t.alike(texts, ['first', 'second', 'third'], 'prior + diff walks to the COMPLETE record set')
t.ok(loadCommit(merged).cid.equals(w3.commitCid), 'rooted at the head commit')
})
test('since == head -> an EMPTY diff, which is a real answer and not an error', async (t) => {
const r = await newRepo(t)
await r.write('a1', 'first')
const w2 = await r.write('a2', 'second')
const d = await repoDiffSince({ ...frameReader(r.seq), since: w2.rev, ...headBlock(w2) })
t.ok(d.ok, 'served')
t.ok(d.empty, 'reported as empty')
const car = await readCar(d.carBytes)
t.is(car.blocks.size, 1, 'the CAR carries the commit block and nothing else')
t.ok(loadCommit(car).cid.equals(w2.commitCid), 'and is rooted at the current head')
})
test('a rev the firehose cannot reach is REFUSED, with a reason', async (t) => {
const r = await newRepo(t)
const w1 = await r.write('a1', 'first')
const w2 = await r.write('a2', 'second')
// A firehose that starts AFTER the requested rev: the first frame is hidden, so nothing proves we hold
// history back to `since` and nothing proves the frame we do have is genesis.
const truncated = { getFrame: (i) => r.seq.core.get(i + 1), length: r.seq.core.length - 1 }
const d = await repoDiffSince({ ...truncated, since: BEFORE_ALL, ...headBlock(w2) })
t.absent(d.ok, 'refused rather than served')
t.ok(/does not reach back/.test(d.reason), 'and says why: ' + d.reason)
t.absent(d.carBytes, 'no CAR is produced — the caller must fall back to the full repo')
t.ok(w1.rev, 'sanity: the hidden first commit really existed')
})
// `since` from before the repo existed is REFUSED, and that is the right answer.
//
// I expected this to be the genesis case and wrote it to assert coverage. It is not. RepoManager.init()
// mints an empty genesis commit and persists it as the head WITHOUT sequencing it, so the first frame's
// `since` points at a commit the firehose never carried. We therefore cannot prove our firehose reaches
// the repo's first write rather than having been truncated to look that way — and refusing costs the
// caller nothing, because "everything" is exactly what the full CAR already is.
test('a since from before the repo existed is refused, not guessed at', async (t) => {
const r = await newRepo(t)
const w1 = await r.write('a1', 'first')
const w2 = await r.write('a2', 'second')
const d = await repoDiffSince({ ...frameReader(r.seq), since: BEFORE_ALL, ...headBlock(w2) })
t.absent(d.ok, 'refused — the caller falls back to the full repo, which is what they asked for anyway')
t.ok(/does not reach back/.test(d.reason), 'and says why: ' + d.reason)
t.ok(w1.rev < w2.rev, 'sanity: revs are monotonic and lexicographically ordered')
})
// The coverage rule that DOES fire without an older frame present: the oldest frame we hold builds on a
// state at or before the caller's, so nothing sits between them.
test('coverage is proven from the oldest frame own `since`, not only from an older frame', async (t) => {
const r = await newRepo(t)
const w1 = await r.write('a1', 'first')
const w2 = await r.write('a2', 'second')
const w3 = await r.write('a3', 'third')
// Hide frame 0. The caller is at w1.rev; the oldest frame we still hold is w2, whose `since` IS w1.rev
// — so there is demonstrably no commit between the caller's state and our history.
const truncated = { getFrame: (i) => r.seq.core.get(i + 1), length: r.seq.core.length - 1 }
const d = await repoDiffSince({ ...truncated, since: w1.rev, ...headBlock(w3) })
t.ok(d.ok, 'served without ever seeing a frame at or older than `since`')
t.is(d.fromRev, w1.rev, 'anchored at the rev the oldest held frame builds on')
})
// Dedup is tested on unionDiffCar DIRECTLY, with frames built to share a block.
//
// The first attempt drove this through a real repo of four single-record commits and asserted the union
// was smaller than the sum of the frames. It passed nothing: with one MST node, each commit contributes
// a fresh commit block, a fresh root and a fresh record — three blocks, none shared — so the sum WAS the
// union and the assertion could only ever have measured itself. Sharing needs a multi-level tree and a
// particular key distribution, which would make the test a hostage to MST fanout. Constructing the
// overlap outright tests the mechanism instead of hoping the fixture exhibits it.
test('a block carried by two frames is unioned ONCE', async (t) => {
const mkBlock = (n) => { const bytes = dagCbor.encode({ n }); return { cid: cidForBytesSync(bytes), bytes } }
const shared = mkBlock('shared')
const onlyA = mkBlock('a')
const onlyB = mkBlock('b')
const head = mkBlock('head')
const frames = [
{ blocks: await writeCar(shared.cid, [shared, onlyA]) },
{ blocks: await writeCar(shared.cid, [shared, onlyB]) }
]
const u = await unionDiffCar({ frames, headCommitCid: head.cid, headCommitBytes: head.bytes })
const car = await readCar(u.carBytes)
t.is(u.blocks, 4, 'shared + a + b + head = 4, not 5 — the overlap collapsed')
t.is(car.blocks.size, u.blocks, 'the reported count matches the CAR')
t.ok(car.blocks.has(head.cid.toString()), 'the head commit block is present even though no frame carried it')
t.ok(car.roots[0].equals(head.cid), 'and it is the root')
})
test('a head block already present in a frame is not duplicated', async (t) => {
const r = await newRepo(t)
const w1 = await r.write('a1', 'first')
const w2 = await r.write('a2', 'second')
const d = await repoDiffSince({ ...frameReader(r.seq), since: w1.rev, ...headBlock(w2) })
const car = await readCar(d.carBytes)
t.is(car.blocks.size, d.blocks, 'the reported block count is the CAR count — the head was added idempotently')
t.ok(car.roots[0].equals(w2.commitCid), 'rooted at the verified head')
})
test('collectFramesSince scans backward and stops at the proof — not a full-history walk', async (t) => {
const r = await newRepo(t)
await r.write('a1', 'first')
await r.write('a2', 'second')
const w3 = await r.write('a3', 'third')
await r.write('a4', 'fourth')
let reads = 0
const counting = { getFrame: (i) => { reads++; return r.seq.core.get(i) }, length: r.seq.core.length }
const got = await collectFramesSince({ ...counting, since: w3.rev })
t.ok(got.covered, 'coverage proven')
t.is(got.frames.length, 1, 'exactly the one commit newer than that rev')
t.is(got.anchorRev, w3.rev, 'anchored at the frame that proved it')
t.is(reads, 2, 'read 2 frames, not all 4 — the scan stops at the proof')
})
test('a head ahead of the firehose tail is REFUSED — the diff could not be walked even with prior state', async (t) => {
const r = await newRepo(t)
const w1 = await r.write('a1', 'first')
const w2 = await r.write('a2', 'second')
// A front whose verified head is newer than anything the replicated firehose carries. Rooting a CAR at
// that commit would produce blocks that lead nowhere — worse than refusing, because it LOOKS valid.
const fakeHead = { ...w2.signedCommit, rev: 'zzzzzzzzzzzzz' }
const d = await repoDiffSince({
...frameReader(r.seq),
since: w1.rev,
headCommitCid: commitCid(fakeHead),
headCommitBytes: dagCbor.encode(fakeHead),
headRev: fakeHead.rev
})
t.absent(d.ok, 'refused')
t.ok(/ahead of the firehose tail/.test(d.reason), 'and says why: ' + d.reason)
})
// The head block is LOCATED in the firehose, never synthesised — with no headCommitBytes supplied.
//
// This is what lets a front ask for a diff without a dag-cbor dependency, and it is the stricter
// behaviour: a head the firehose does not carry becomes a refusal rather than a CAR rooted at a commit
// nobody can walk to.
test('the head block is found in the firehose without being handed in', async (t) => {
const r = await newRepo(t)
const w1 = await r.write('a1', 'first')
const w2 = await r.write('a2', 'second')
const d = await repoDiffSince({ ...frameReader(r.seq), since: w1.rev, headCommitCid: w2.commitCid, headRev: w2.rev })
t.ok(d.ok, 'served with no headCommitBytes given')
const car = await readCar(d.carBytes)
t.ok(car.roots[0].equals(w2.commitCid), 'rooted at the head')
t.ok(car.blocks.has(w2.commitCid.toString()), 'and the head block is present, sourced from the frames')
})
test('an EMPTY diff also locates its head block (no frames to take it from)', async (t) => {
const r = await newRepo(t)
await r.write('a1', 'first')
const w2 = await r.write('a2', 'second')
// since == head: no frames qualify, so the head block cannot come from the union — it is found by
// scanning back through the firehose instead.
const d = await repoDiffSince({ ...frameReader(r.seq), since: w2.rev, headCommitCid: w2.commitCid, headRev: w2.rev })
t.ok(d.ok && d.empty, 'served as an empty diff')
const car = await readCar(d.carBytes)
t.is(car.blocks.size, 1, 'commit block only')
t.ok(car.roots[0].equals(w2.commitCid), 'rooted at the head it located')
})
test('a head the firehose never carried is refused, not synthesised', async (t) => {
const r = await newRepo(t)
const w1 = await r.write('a1', 'first')
const w2 = await r.write('a2', 'second')
// A commit that verifies as far as this function is concerned but whose block is in no frame. Its rev
// is NOT ahead of the tail, so it gets past that guard and must be caught by the locate step.
const orphan = { ...w2.signedCommit, rev: w1.rev }
const d = await repoDiffSince({ ...frameReader(r.seq), since: w1.rev, headCommitCid: commitCid(orphan), headRev: w1.rev })
t.absent(d.ok, 'refused')
t.ok(/not carried by this firehose/.test(d.reason), 'and says why: ' + d.reason)
})
test('no since -> no diff (the caller must not get a partial CAR by accident)', async (t) => {
const r = await newRepo(t)
const w1 = await r.write('a1', 'first')
const d = await repoDiffSince({ ...frameReader(r.seq), since: null, ...headBlock(w1) })
t.absent(d.ok, 'a missing `since` is never treated as "from the beginning"')
})
// A from-genesis diff is refused BECAUSE IT IS BIGGER, not because it is wrong.
//
// ⛔ THIS BRANCH WAS UNREACHABLE IN FIXTURES AND ONLY A LIVE REPO EXPOSED IT. RepoManager.init() mints
// an unsequenced genesis commit, so a synthetic repo's first frame always carries a real `since` — every
// test above takes the other path. wj62's real firehose starts with a null `since` (its repo was
// re-authored), and on prod 2026-08-07 this branch served 202,262 bytes / 719 blocks where the full repo
// is 46,106 / 224. Same 168 records, 4.4x the bytes.
//
// Structural, not a fluke: the union from genesis holds every block the repo has EVER carried — every
// superseded MST node, every deleted record — so it is ALWAYS a superset of the current repo and can
// never beat the full CAR.
test('a from-genesis diff is refused — it is always a superset of the full repo', async (t) => {
const r = await newRepo(t)
const w1 = await r.write('a1', 'first')
const w2 = await r.write('a2', 'second')
// Synthesise what a real re-authored repo emits: an oldest frame declaring no prior state. Rewriting
// the frame body is the only way to reach this branch, precisely because the producer cannot make it.
const frames = []
for (let i = 0; i < r.seq.core.length; i++) frames.push(await r.seq.core.get(i))
const HDR = dagCbor.encode({ t: '#commit', op: 1 })
const patched = frames.map((f, i) => {
if (i !== 0) return f
const body = dagCbor.decode(f.subarray(HDR.length))
body.since = null
const bb = dagCbor.encode(body)
const out = new Uint8Array(HDR.length + bb.length)
out.set(HDR, 0); out.set(bb, HDR.length)
return out
})
const got = await collectFramesSince({ getFrame: (i) => patched[i], length: patched.length, since: BEFORE_ALL })
t.absent(got.covered, 'not treated as covered')
t.is(got.reason, 'from-genesis', 'and the reason is named, so the caller can say something specific')
const d = await repoDiffSince({ getFrame: (i) => patched[i], length: patched.length, since: BEFORE_ALL, headCommitCid: w2.commitCid, headRev: w2.rev })
t.absent(d.ok, 'refused')
t.ok(/superset of the full repo/.test(d.reason), 'explaining it is a SIZE decision, not a correctness one: ' + d.reason)
t.ok(w1.rev < w2.rev, 'sanity')
})