| … | … | @@ -0,0 +1,254 @@ |
| 1 | // projectRepo / projectFirehose DRIFT — the membrane's two projections must describe ONE state. |
| 2 | // |
| 3 | // serveMembrane.sync() writes two cores that a keyless consumer reads as a PAIR: |
| 4 | // - the membrane FIREHOSE gets every __commit__ frame, so the consumer's verified head is the |
| 5 | // NETWORK's latest signed commit (the log is shared + replicated). |
| 6 | // - the membrane REPO gets the record set, and the consumer rebuilds the MST from it and asserts |
| 7 | // it equals the verified head's data root (read-view getRepoFromCommit). |
| 8 | // |
| 9 | // So the load-bearing invariant is: the repo core must hold EXACTLY the record set the last |
| 10 | // projected frame commits to. projectRepo did not maintain it — it wrote `device.repo.listRecords()`, |
| 11 | // this NODE's live view, which is a different source from the frames. Three independent ways to break: |
| 12 | // |
| 13 | // 1. LAGGING view — a co-owner whose pointers/bytes have not caught up projects a SUBSET of what |
| 14 | // the head commits to. A non-signer never adopts the network head and only resolves bytes as a |
| 15 | // side effect of sign() (which returns at the election gate), so it can sit like this for days. |
| 16 | // Observed live: tabertron's device head stuck at s222 while the log head was s322 → every |
| 17 | // getRepo 500'd (hyper-social #14, incident 3). |
| 18 | // 2. DELETES never propagate — projectRepo only ever put(); it never del(). Once any delete is |
| 19 | // signed, the repo core keeps the dead record FOREVER, so the rebuilt MST can never again match |
| 20 | // the head. Unconditional and permanent, independent of any lag. |
| 21 | // 3. Running AHEAD — a view carrying pointers not yet signed into any frame projects a SUPERSET. |
| 22 | // |
| 23 | // The fix makes the repo projection a pure function of the SAME frames the firehose projection |
| 24 | // consumes (each frame carries the record blocks for its own create/update ops), verifies the |
| 25 | // result against the head frame's data root BEFORE writing anything, and advances both cores |
| 26 | // together or neither. That is the file's own stated premise — "__commit__ is rev-ordered + |
| 27 | // deterministic, so every device projects the SAME firehose" — extended to the repo half. |
| 28 | // |
| 29 | // Run: bare test/membrane-projection-drift.test.mjs |
| 30 | import '../lib/bare-globals.js' |
| 31 | import test from 'brittle' |
| 32 | import b4a from 'b4a' |
| 33 | import fs from 'bare-fs' |
| 34 | import * as dagCbor from '@ipld/dag-cbor' |
| 35 | import { secp256k1 } from '@noble/curves/secp256k1' |
| 36 | import { serveMembrane } from '../lib/serve-membrane.js' |
| 37 | import { buildCommitWrite } from '../lib/signer.js' |
| 38 | import { openCorestore } from 'atproto-store/lib/corestore.js' |
| 39 | import { getRepoFromCommit } from 'atproto-store/lib/read-view.js' |
| 40 | import { decodeCommitFrame } from 'atproto-store/lib/firehose.js' |
| 41 | import { readCar, loadCommit, listRecords as carListRecords } from 'atproto-repo/lib/repo.js' |
| 42 | import { verifyCommit } from 'atproto-repo/lib/commit.js' |
| 43 | |
| 44 | const PRIV = new Uint8Array(b4a.from('9085d2bef69286a6cbb51623c8fa258629945cd55ca705cc4e66700396894e0c', 'hex')) |
| 45 | const PUB = secp256k1.getPublicKey(PRIV, true) |
| 46 | const DID = 'did:plc:membranedrift00000000000' |
| 47 | const POST = 'app.bsky.feed.post' |
| 48 | const tmp = () => { const d = '/tmp/ar-drift-' + Math.floor(Math.random() * 1e9); fs.mkdirSync(d, { recursive: true }); return d } |
| 49 | |
| 50 | const rec = (text) => ({ $type: POST, text, createdAt: '2026-07-26T00:00:00.000Z' }) |
| 51 | const A = { collection: POST, rkey: '3aaa', record: rec('a') } |
| 52 | const B = { collection: POST, rkey: '3bbb', record: rec('b') } |
| 53 | const path = (r) => r.collection + '/' + r.rkey |
| 54 | |
| 55 | // serveMembrane only touches these few swarm members. |
| 56 | function fakeSwarm () { |
| 57 | return { on () {}, off () {}, keyPair: { publicKey: b4a.alloc(32, 7) }, join () { return { flushed: async () => {} } } } |
| 58 | } |
| 59 | |
| 60 | // A device whose __commit__ log AND live record view we control independently — the whole point: |
| 61 | // the two sources can disagree, and the projection must not fabricate a pair out of them. |
| 62 | function device () { |
| 63 | const d = { _log: [], _view: [] } |
| 64 | d.repo = { listRecords: async () => d._view } |
| 65 | d.firehoseLog = async () => d._log |
| 66 | return d |
| 67 | } |
| 68 | |
| 69 | // The __commit__ entry shape SwarmDevice.serializeEntry writes and serve-membrane's |
| 70 | // deserializeEntry consumes (binary as hex). Pinned here so this suite exercises the real contract. |
| 71 | const hex = (u8) => b4a.toString(b4a.from(u8), 'hex') |
| 72 | function serialize (write, seq) { |
| 73 | return { |
| 74 | seq, |
| 75 | time: new Date(0).toISOString(), |
| 76 | rev: write.rev, |
| 77 | since: write.since, |
| 78 | prev: write.signedCommit.prev ? write.signedCommit.prev.toString() : null, |
| 79 | prevData: write.prevData ? write.prevData.toString() : null, |
| 80 | commitCid: write.commitCid.toString(), |
| 81 | commit: hex(write.commitBlock.bytes), |
| 82 | ops: write.ops.map((o) => ({ action: o.action, path: o.path, cid: o.cid ? o.cid.toString() : null, prev: o.prev ? o.prev.toString() : null })), |
| 83 | blocks: write.newBlocks.map((b) => ({ cid: b.cid.toString(), bytes: hex(b.bytes) })) |
| 84 | } |
| 85 | } |
| 86 | |
| 87 | // Sign one frame continuing from `prevWrite` (null = genesis): before -> after record sets. |
| 88 | function frame (before, after, rev, prevWrite, seq) { |
| 89 | const w = buildCommitWrite(before, after, { |
| 90 | did: DID, |
| 91 | rev, |
| 92 | prev: prevWrite ? prevWrite.commitCid : null, |
| 93 | since: prevWrite ? prevWrite.rev : null, |
| 94 | privateKey: PRIV |
| 95 | }) |
| 96 | return { write: w, entry: serialize(w, seq) } |
| 97 | } |
| 98 | |
| 99 | // EXACTLY the keyless consumer read path (DidPeer.getRepo over a projecting front): take the head |
| 100 | // frame off the membrane firehose, verify its commit against the DID key, then rebuild the CAR from |
| 101 | // the membrane repo core. Throws iff the two projections disagree — the production 500. |
| 102 | async function consumerGetRepo (serving) { |
| 103 | const { body } = decodeCommitFrame(await serving.seq.core.get(serving.seq.core.length - 1)) |
| 104 | const signed = loadCommit(await readCar(body.blocks)).value |
| 105 | if (!verifyCommit(signed, PUB)) throw new Error('head does not verify') |
| 106 | const { carBytes } = await getRepoFromCommit(serving.repo, signed) |
| 107 | return carListRecords(await readCar(carBytes)).map((r) => r.key).sort() |
| 108 | } |
| 109 | |
| 110 | async function capture (t, fn) { |
| 111 | const lines = [] |
| 112 | const orig = console.error |
| 113 | console.error = (s) => lines.push(String(s)) |
| 114 | t.teardown(() => { console.error = orig }) |
| 115 | try { await fn() } finally { console.error = orig } |
| 116 | return lines |
| 117 | } |
| 118 | |
| 119 | const serve = async (dev) => serveMembrane(dev, { corestore: openCorestore(tmp()), swarm: fakeSwarm(), did: DID }) |
| 120 | |
| 121 | test('drift 1 — a LAGGING view must not be paired with the NETWORK head (tabertron s222 vs s322)', async (t) => { |
| 122 | const dev = device() |
| 123 | const f1 = frame([], [A], '3lll000000000', null, 1) |
| 124 | const f2 = frame([A], [A, B], '3lll000000001', f1.write, 2) |
| 125 | dev._log = [f1.entry, f2.entry] |
| 126 | // The node agrees on the POINTER set via the autobase but has not resolved B into its view — the |
| 127 | // exact state a non-signing co-owner sits in (bytes only resolve inside sign(), which it never reaches). |
| 128 | dev._view = [A] |
| 129 | |
| 130 | const serving = await serve(dev) |
| 131 | t.teardown(() => serving.close()) |
| 132 | await serving.sync() |
| 133 | |
| 134 | const keys = await consumerGetRepo(serving) |
| 135 | t.alike(keys, [path(A), path(B)].sort(), 'the repo core holds exactly what the projected head commits to — derived from the frames, not from the lagging view') |
| 136 | }) |
| 137 | |
| 138 | test('drift 2 — a signed DELETE must reach the repo core (projectRepo only ever put)', async (t) => { |
| 139 | const dev = device() |
| 140 | const f1 = frame([], [A, B], '3lll000000000', null, 1) |
| 141 | dev._log = [f1.entry] |
| 142 | dev._view = [A, B] |
| 143 | |
| 144 | const serving = await serve(dev) |
| 145 | t.teardown(() => serving.close()) |
| 146 | await serving.sync() |
| 147 | t.alike(await consumerGetRepo(serving), [path(A), path(B)].sort(), 'both records projected at the first head') |
| 148 | |
| 149 | // B is deleted and the delete is SIGNED: the new head commits to {A} only. |
| 150 | const f2 = frame([A, B], [A], '3lll000000001', f1.write, 2) |
| 151 | dev._log = [f1.entry, f2.entry] |
| 152 | dev._view = [A] |
| 153 | await serving.sync() |
| 154 | |
| 155 | t.alike(await consumerGetRepo(serving), [path(A)], 'the delete propagated — the repo core no longer serves the removed record') |
| 156 | }) |
| 157 | |
| 158 | test('drift 3 — a view running AHEAD of the log must not leak unsigned records into the repo core', async (t) => { |
| 159 | const dev = device() |
| 160 | const f1 = frame([], [A], '3lll000000000', null, 1) |
| 161 | dev._log = [f1.entry] |
| 162 | // B is written but NOT yet signed into any frame (the window between a write and the next sign). |
| 163 | dev._view = [A, B] |
| 164 | |
| 165 | const serving = await serve(dev) |
| 166 | t.teardown(() => serving.close()) |
| 167 | await serving.sync() |
| 168 | |
| 169 | t.alike(await consumerGetRepo(serving), [path(A)], 'only the SIGNED record set is served — the unsigned write waits for its frame') |
| 170 | }) |
| 171 | |
| 172 | test('drift 4 — a core POLLUTED by the old live-view projection heals when the next frame arrives', async (t) => { |
| 173 | const dev = device() |
| 174 | const f1 = frame([], [A], '3lll000000000', null, 1) |
| 175 | dev._log = [f1.entry] |
| 176 | dev._view = [A] |
| 177 | |
| 178 | const serving = await serve(dev) |
| 179 | t.teardown(() => serving.close()) |
| 180 | // What the old projection left behind: a record no frame ever committed to (e.g. one deleted long |
| 181 | // ago, whose del() never happened). Every deployed node's core is in some version of this state. |
| 182 | await serving.repo.put(POST, '3zzz', rec('stale')) |
| 183 | |
| 184 | const f2 = frame([A], [A, B], '3lll000000001', f1.write, 2) |
| 185 | dev._log = [f1.entry, f2.entry] |
| 186 | dev._view = [A, B] |
| 187 | await serving.sync() |
| 188 | |
| 189 | t.alike(await consumerGetRepo(serving), [path(A), path(B)].sort(), 'the stale record was reconciled AWAY — the core is made to match the head, not merely added to') |
| 190 | }) |
| 191 | |
| 192 | test('drift 4b — a polluted core heals at BOOT, with no new frame (the deployed-node migration)', async (t) => { |
| 193 | const dev = device() |
| 194 | const f1 = frame([], [A], '3lll000000000', null, 1) |
| 195 | dev._log = [f1.entry] |
| 196 | dev._view = [A] |
| 197 | |
| 198 | // One corestore across two serveMembrane lifetimes = a restart onto the same cores on disk. |
| 199 | const corestore = openCorestore(tmp()) |
| 200 | const before = await serveMembrane(dev, { corestore, swarm: fakeSwarm(), did: DID }) |
| 201 | await before.sync() |
| 202 | await before.repo.put(POST, '3zzz', rec('stale')) |
| 203 | before.close() |
| 204 | |
| 205 | // The log has NOT advanced, so the old `log.length === seq.length` guard skipped everything and the |
| 206 | // node kept 500ing — indefinitely for an idle DID. Verification is per-HEAD, not per-new-frame. |
| 207 | const after = await serveMembrane(dev, { corestore, swarm: fakeSwarm(), did: DID }) |
| 208 | t.teardown(() => after.close()) |
| 209 | t.alike(await consumerGetRepo(after), [path(A)], 'the initial sync repaired the pair without an operator step and without a new commit') |
| 210 | }) |
| 211 | |
| 212 | test('drift 5 — frames that cannot be replayed must advance NEITHER core, and say so', async (t) => { |
| 213 | const dev = device() |
| 214 | const f1 = frame([], [A], '3lll000000000', null, 1) |
| 215 | dev._log = [f1.entry] |
| 216 | dev._view = [A] |
| 217 | |
| 218 | const serving = await serve(dev) |
| 219 | t.teardown(() => serving.close()) |
| 220 | t.alike(await consumerGetRepo(serving), [path(A)], 'a consistent pair is being served at the first head') |
| 221 | |
| 222 | // B's record block is missing from its frame (partial replication of the shared log): the frame's |
| 223 | // create op references a block we do not have, so its committed record set cannot be reconstructed. |
| 224 | const f2 = frame([A], [A, B], '3lll000000001', f1.write, 2) |
| 225 | const bCid = f2.write.ops.find((o) => o.path === path(B)).cid.toString() |
| 226 | f2.entry.blocks = f2.entry.blocks.filter((b) => b.cid !== bCid) |
| 227 | dev._log = [f1.entry, f2.entry] |
| 228 | dev._view = [A] |
| 229 | |
| 230 | const logs = await capture(t, () => serving.sync()) |
| 231 | |
| 232 | t.is(serving.seq.core.length, 1, 'the firehose was NOT advanced onto a head the repo cannot match') |
| 233 | t.alike(await consumerGetRepo(serving), [path(A)], 'the previous CONSISTENT pair still serves — degraded, not a 500') |
| 234 | t.ok(logs.length >= 1, 'the refusal is reported, not silent') |
| 235 | t.ok(/projection/i.test(logs[0]), 'the message names the projection') |
| 236 | t.is(JSON.parse(logs[0]).did, DID, 'names the DID') |
| 237 | }) |
| 238 | |
| 239 | test('drift 6 — the anti-balloon guard still holds: no new frames, no writes to either core', async (t) => { |
| 240 | const dev = device() |
| 241 | const f1 = frame([], [A, B], '3lll000000000', null, 1) |
| 242 | dev._log = [f1.entry] |
| 243 | dev._view = [A, B] |
| 244 | |
| 245 | const serving = await serve(dev) |
| 246 | t.teardown(() => serving.close()) |
| 247 | await serving.sync() |
| 248 | const repoLen = serving.repo.core.length |
| 249 | const seqLen = serving.seq.core.length |
| 250 | |
| 251 | for (let i = 0; i < 5; i++) await serving.sync() |
| 252 | t.is(serving.seq.core.length, seqLen, 'firehose core did not grow on idle syncs') |
| 253 | t.is(serving.repo.core.length, repoLen, 'repo core did not grow on idle syncs — the ~2s tick cannot balloon it') |
| 254 | }) |