// projectRepo / projectFirehose DRIFT — the membrane's two projections must describe ONE state.
//
// serveMembrane.sync() writes two cores that a keyless consumer reads as a PAIR:
// - the membrane FIREHOSE gets every __commit__ frame, so the consumer's verified head is the
// NETWORK's latest signed commit (the log is shared + replicated).
// - the membrane REPO gets the record set, and the consumer rebuilds the MST from it and asserts
// it equals the verified head's data root (read-view getRepoFromCommit).
//
// So the load-bearing invariant is: the repo core must hold EXACTLY the record set the last
// projected frame commits to. projectRepo did not maintain it — it wrote `device.repo.listRecords()`,
// this NODE's live view, which is a different source from the frames. Three independent ways to break:
//
// 1. LAGGING view — a co-owner whose pointers/bytes have not caught up projects a SUBSET of what
// the head commits to. A non-signer never adopts the network head and only resolves bytes as a
// side effect of sign() (which returns at the election gate), so it can sit like this for days.
// Observed live: tabertron's device head stuck at s222 while the log head was s322 → every
// getRepo 500'd (hyper-social #14, incident 3).
// 2. DELETES never propagate — projectRepo only ever put(); it never del(). Once any delete is
// signed, the repo core keeps the dead record FOREVER, so the rebuilt MST can never again match
// the head. Unconditional and permanent, independent of any lag.
// 3. Running AHEAD — a view carrying pointers not yet signed into any frame projects a SUPERSET.
//
// The fix makes the repo projection a pure function of the SAME frames the firehose projection
// consumes (each frame carries the record blocks for its own create/update ops), verifies the
// result against the head frame's data root BEFORE writing anything, and advances both cores
// together or neither. That is the file's own stated premise — "__commit__ is rev-ordered +
// deterministic, so every device projects the SAME firehose" — extended to the repo half.
//
// Run: bare test/membrane-projection-drift.test.mjs
import '../lib/bare-globals.js'
import test from 'brittle'
import b4a from 'b4a'
import fs from 'bare-fs'
import * as dagCbor from '@ipld/dag-cbor'
import { secp256k1 } from '@noble/curves/secp256k1'
import { serveMembrane } from '../lib/serve-membrane.js'
import { buildCommitWrite } from '../lib/signer.js'
import { openCorestore } from 'atproto-store/lib/corestore.js'
import { getRepoFromCommit } from 'atproto-store/lib/read-view.js'
import { decodeCommitFrame } from 'atproto-store/lib/firehose.js'
import { readCar, loadCommit, listRecords as carListRecords } from 'atproto-repo/lib/repo.js'
import { verifyCommit } from 'atproto-repo/lib/commit.js'
const PRIV = new Uint8Array(b4a.from('9085d2bef69286a6cbb51623c8fa258629945cd55ca705cc4e66700396894e0c', 'hex'))
const PUB = secp256k1.getPublicKey(PRIV, true)
const DID = 'did:plc:membranedrift00000000000'
const POST = 'app.bsky.feed.post'
const tmp = () => { const d = '/tmp/ar-drift-' + Math.floor(Math.random() * 1e9); fs.mkdirSync(d, { recursive: true }); return d }
const rec = (text) => ({ $type: POST, text, createdAt: '2026-07-26T00:00:00.000Z' })
const A = { collection: POST, rkey: '3aaa', record: rec('a') }
const B = { collection: POST, rkey: '3bbb', record: rec('b') }
const path = (r) => r.collection + '/' + r.rkey
// serveMembrane only touches these few swarm members.
function fakeSwarm () {
return { on () {}, off () {}, keyPair: { publicKey: b4a.alloc(32, 7) }, join () { return { flushed: async () => {} } } }
}
// A device whose __commit__ log AND live record view we control independently — the whole point:
// the two sources can disagree, and the projection must not fabricate a pair out of them.
function device () {
const d = { _log: [], _view: [] }
d.repo = { listRecords: async () => d._view }
d.firehoseLog = async () => d._log
return d
}
// The __commit__ entry shape SwarmDevice.serializeEntry writes and serve-membrane's
// deserializeEntry consumes (binary as hex). Pinned here so this suite exercises the real contract.
const hex = (u8) => b4a.toString(b4a.from(u8), 'hex')
function serialize (write, seq) {
return {
seq,
time: new Date(0).toISOString(),
rev: write.rev,
since: write.since,
prev: write.signedCommit.prev ? write.signedCommit.prev.toString() : null,
prevData: write.prevData ? write.prevData.toString() : null,
commitCid: write.commitCid.toString(),
commit: hex(write.commitBlock.bytes),
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 })),
blocks: write.newBlocks.map((b) => ({ cid: b.cid.toString(), bytes: hex(b.bytes) }))
}
}
// Sign one frame continuing from `prevWrite` (null = genesis): before -> after record sets.
function frame (before, after, rev, prevWrite, seq) {
const w = buildCommitWrite(before, after, {
did: DID,
rev,
prev: prevWrite ? prevWrite.commitCid : null,
since: prevWrite ? prevWrite.rev : null,
privateKey: PRIV
})
return { write: w, entry: serialize(w, seq) }
}
// EXACTLY the keyless consumer read path (DidPeer.getRepo over a projecting front): take the head
// frame off the membrane firehose, verify its commit against the DID key, then rebuild the CAR from
// the membrane repo core. Throws iff the two projections disagree — the production 500.
async function consumerGetRepo (serving) {
const { body } = decodeCommitFrame(await serving.seq.core.get(serving.seq.core.length - 1))
const signed = loadCommit(await readCar(body.blocks)).value
if (!verifyCommit(signed, PUB)) throw new Error('head does not verify')
const { carBytes } = await getRepoFromCommit(serving.repo, signed)
return carListRecords(await readCar(carBytes)).map((r) => r.key).sort()
}
async function capture (t, fn) {
const lines = []
const orig = console.error
console.error = (s) => lines.push(String(s))
t.teardown(() => { console.error = orig })
try { await fn() } finally { console.error = orig }
return lines
}
const serve = async (dev) => serveMembrane(dev, { corestore: openCorestore(tmp()), swarm: fakeSwarm(), did: DID })
test('drift 1 — a LAGGING view must not be paired with the NETWORK head (tabertron s222 vs s322)', async (t) => {
const dev = device()
const f1 = frame([], [A], '3lll000000000', null, 1)
const f2 = frame([A], [A, B], '3lll000000001', f1.write, 2)
dev._log = [f1.entry, f2.entry]
// The node agrees on the POINTER set via the autobase but has not resolved B into its view — the
// exact state a non-signing co-owner sits in (bytes only resolve inside sign(), which it never reaches).
dev._view = [A]
const serving = await serve(dev)
t.teardown(() => serving.close())
await serving.sync()
const keys = await consumerGetRepo(serving)
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')
})
test('drift 2 — a signed DELETE must reach the repo core (projectRepo only ever put)', async (t) => {
const dev = device()
const f1 = frame([], [A, B], '3lll000000000', null, 1)
dev._log = [f1.entry]
dev._view = [A, B]
const serving = await serve(dev)
t.teardown(() => serving.close())
await serving.sync()
t.alike(await consumerGetRepo(serving), [path(A), path(B)].sort(), 'both records projected at the first head')
// B is deleted and the delete is SIGNED: the new head commits to {A} only.
const f2 = frame([A, B], [A], '3lll000000001', f1.write, 2)
dev._log = [f1.entry, f2.entry]
dev._view = [A]
await serving.sync()
t.alike(await consumerGetRepo(serving), [path(A)], 'the delete propagated — the repo core no longer serves the removed record')
})
test('drift 3 — a view running AHEAD of the log must not leak unsigned records into the repo core', async (t) => {
const dev = device()
const f1 = frame([], [A], '3lll000000000', null, 1)
dev._log = [f1.entry]
// B is written but NOT yet signed into any frame (the window between a write and the next sign).
dev._view = [A, B]
const serving = await serve(dev)
t.teardown(() => serving.close())
await serving.sync()
t.alike(await consumerGetRepo(serving), [path(A)], 'only the SIGNED record set is served — the unsigned write waits for its frame')
})
test('drift 4 — a core POLLUTED by the old live-view projection heals when the next frame arrives', async (t) => {
const dev = device()
const f1 = frame([], [A], '3lll000000000', null, 1)
dev._log = [f1.entry]
dev._view = [A]
const serving = await serve(dev)
t.teardown(() => serving.close())
// What the old projection left behind: a record no frame ever committed to (e.g. one deleted long
// ago, whose del() never happened). Every deployed node's core is in some version of this state.
await serving.repo.put(POST, '3zzz', rec('stale'))
const f2 = frame([A], [A, B], '3lll000000001', f1.write, 2)
dev._log = [f1.entry, f2.entry]
dev._view = [A, B]
await serving.sync()
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')
})
test('drift 4b — a polluted core heals at BOOT, with no new frame (the deployed-node migration)', async (t) => {
const dev = device()
const f1 = frame([], [A], '3lll000000000', null, 1)
dev._log = [f1.entry]
dev._view = [A]
// One corestore across two serveMembrane lifetimes = a restart onto the same cores on disk.
const corestore = openCorestore(tmp())
const before = await serveMembrane(dev, { corestore, swarm: fakeSwarm(), did: DID })
await before.sync()
await before.repo.put(POST, '3zzz', rec('stale'))
before.close()
// The log has NOT advanced, so the old `log.length === seq.length` guard skipped everything and the
// node kept 500ing — indefinitely for an idle DID. Verification is per-HEAD, not per-new-frame.
const after = await serveMembrane(dev, { corestore, swarm: fakeSwarm(), did: DID })
t.teardown(() => after.close())
t.alike(await consumerGetRepo(after), [path(A)], 'the initial sync repaired the pair without an operator step and without a new commit')
})
test('drift 5 — frames that cannot be replayed must advance NEITHER core, and say so', async (t) => {
const dev = device()
const f1 = frame([], [A], '3lll000000000', null, 1)
dev._log = [f1.entry]
dev._view = [A]
const serving = await serve(dev)
t.teardown(() => serving.close())
t.alike(await consumerGetRepo(serving), [path(A)], 'a consistent pair is being served at the first head')
// B's record block is missing from its frame (partial replication of the shared log): the frame's
// create op references a block we do not have, so its committed record set cannot be reconstructed.
const f2 = frame([A], [A, B], '3lll000000001', f1.write, 2)
const bCid = f2.write.ops.find((o) => o.path === path(B)).cid.toString()
f2.entry.blocks = f2.entry.blocks.filter((b) => b.cid !== bCid)
dev._log = [f1.entry, f2.entry]
dev._view = [A]
const logs = await capture(t, () => serving.sync())
t.is(serving.seq.core.length, 1, 'the firehose was NOT advanced onto a head the repo cannot match')
t.alike(await consumerGetRepo(serving), [path(A)], 'the previous CONSISTENT pair still serves — degraded, not a 500')
t.ok(logs.length >= 1, 'the refusal is reported, not silent')
t.ok(/projection/i.test(logs[0]), 'the message names the projection')
t.is(JSON.parse(logs[0]).did, DID, 'names the DID')
})
test('drift 6 — the anti-balloon guard still holds: no new frames, no writes to either core', async (t) => {
const dev = device()
const f1 = frame([], [A, B], '3lll000000000', null, 1)
dev._log = [f1.entry]
dev._view = [A, B]
const serving = await serve(dev)
t.teardown(() => serving.close())
await serving.sync()
const repoLen = serving.repo.core.length
const seqLen = serving.seq.core.length
for (let i = 0; i < 5; i++) await serving.sync()
t.is(serving.seq.core.length, seqLen, 'firehose core did not grow on idle syncs')
t.is(serving.repo.core.length, repoLen, 'repo core did not grow on idle syncs — the ~2s tick cannot balloon it')
})