// MULTI-PARTITION × RELAY — a consumer fed by a rev-guarded relay must converge after ANY sequence of
// partitions, not just one.
//
// Found by the autobee signer spike (2026-10-13, seeded fuzz): after repeated partitions every node
// agreed, yet the consumer did not. The relay accepts whatever arrives with a higher rev, so a consumer
// HOPS between branches — it took two commits of one losing branch, then two of another, dropped both
// heals of the first fork (an equal rev had arrived first), then took a line commit. Its state matched
// no branch. 0.3.125's heal modelled each consumer as sitting on ONE fork and diffed from that fork's
// state, and it counted a fork as healed once a commit NAMED it — though no consumer may ever have
// received that commit. The signer cannot know what any consumer received.
//
// Deterministic version of the fuzz: seeded random partitions over MemKV replicas (one per device), a
// deterministic merge (the globally latest write of a key wins — a valid linearization), every device
// signs every step, every device's membrane feeds one consumer: the relay drops rev <= stored; the
// reference AppView applies ops blindly, ignores a create for a record it holds, upserts on update.
import '../lib/bare-globals.js'
import test from 'brittle'
import b4a from 'b4a'
import * as dagCbor from '@ipld/dag-cbor'
import { cidForBytesSync } from 'atproto-repo/lib/cid.js'
import { commitLine } from 'atproto-room/lib/commit-line.js'
import { SwarmDevice } from '../lib/swarm-device.js'
import process from 'bare-process'
// Wider runs without editing the file: MPR_SEEDS=150 MPR_STEPS=24 bare test/multi-partition-relay.test.mjs
const SEEDS = Number(process.env.MPR_SEEDS || 30)
const STEPS = Number(process.env.MPR_STEPS || 12)
const DID = 'did:plc:multipartition'
const KEY = (() => { const k = b4a.alloc(32); k[31] = 7; return k })()
const post = (text) => ({ $type: 'app.bsky.feed.post', text, createdAt: '2026-10-13T00:00:00.000Z' })
let clock = 0 // global write order: the merge keeps the latest write of a key
class MemKV {
constructor () { this.m = new Map() } // key -> { v, at }
async set (k, v) { this.m.set(k, { v, at: ++clock }) }
async get (k) { const e = this.m.get(k); return e ? e.v : null }
async getNow (k) { return this.get(k) }
async del (k) { this.m.delete(k) }
async list (prefix = '') { const o = {}; for (const [k, e] of this.m) if (k.startsWith(prefix)) o[k] = e.v; return o }
async listNow (prefix = '') { return this.list(prefix) }
}
// Make every member of a group hold the union of the group's writes (latest write of each key wins).
function sync (group) {
const best = new Map()
for (const n of group) for (const [k, e] of n.sb.m) { const b = best.get(k); if (!b || e.at > b.at) best.set(k, e) }
for (const n of group) for (const [k, e] of best) n.sb.m.set(k, e)
}
class Consumer {
constructor () { this.rev = ''; this.records = new Map(); this.seen = new Set() }
offer (f) {
if (this.seen.has(f.commitCid)) return
this.seen.add(f.commitCid)
if (f.rev <= this.rev) return
this.rev = f.rev
for (const op of f.ops || []) {
if (op.action === 'create') { if (!this.records.has(op.path)) this.records.set(op.path, op.cid) } else if (op.action === 'update') this.records.set(op.path, op.cid)
else if (op.action === 'delete') this.records.delete(op.path)
}
}
}
const cidsOf = async (dev) => new Map((await dev.repo.listRecords(null, { now: true })).map((r) => [r.collection + '/' + r.rkey, cidForBytesSync(dagCbor.encode(r.record)).toString()]))
const same = (a, b) => a.size === b.size && [...a].every(([k, v]) => b.get(k) === v)
async function run (seed, steps = 12, maxCommitOps = 200) {
let s = seed >>> 0
const rnd = () => { s = (s * 1664525 + 1013904223) >>> 0; return s / 4294967296 }
const pick = (arr) => arr[Math.floor(rnd() * arr.length)]
const all = ['A', 'B', 'C'].map((name) => { const sb = new MemKV(); return { name, sb, dev: new SwarmDevice(sb, { did: DID, privateKey: KEY, self: name, firehose: true, maxCommitOps }) } })
const consumer = new Consumer()
const signAll = async (group) => { for (const n of group) { await n.dev.sign(); for (const f of await n.dev.firehoseLog()) consumer.offer(f) } }
let groups = [all]
let next = 0
const live = new Set()
for (let step = 0; step < steps; step++) {
if (rnd() < 0.4) {
if (rnd() < 0.35) groups = [all]
else { const lone = pick(all); groups = [[lone], all.filter((x) => x !== lone)] }
}
for (const g of groups) {
for (let k = 0; k < 1 + Math.floor(rnd() * 2); k++) {
const dev = pick(g).dev
const r = rnd()
if (r < 0.6 || live.size === 0) { const rkey = 'p' + (next++); live.add(rkey); await dev.putRecord('app.bsky.feed.post', rkey, post(seed + ' ' + step + ' ' + rkey)) } else if (r < 0.85) { await dev.putRecord('app.bsky.feed.post', pick([...live]), post('edit ' + seed + ' ' + step)) } else { const rkey = pick([...live]); live.delete(rkey); await dev.deleteRecord('app.bsky.feed.post', rkey) }
}
sync(g)
}
for (const g of groups) { await signAll(g); sync(g) }
}
sync(all)
for (let i = 0; i < 40; i++) { await signAll(all); sync(all) }
const merged = await cidsOf(all[0].dev)
const heads = new Set()
let nodesAgree = true
for (const n of all) { heads.add(commitLine(await n.dev.firehoseFrames()).head.commitCid); if (!same(await cidsOf(n.dev), merged)) nodesAgree = false }
return { nodesAgree, oneHead: heads.size === 1, consumerOk: same(consumer.records, merged) }
}
test('a relay-fed consumer converges after ANY sequence of partitions (seeded)', async (t) => {
const bad = []
for (let seed = 1; seed <= SEEDS; seed++) {
const r = await run(seed, STEPS)
if (!r.nodesAgree || !r.oneHead || !r.consumerOk) bad.push(seed + ':' + JSON.stringify(r))
}
t.alike(bad, [], 'every seed: nodes agree, one head, consumer == merged records')
})
test('the same, with heals SPLIT: a 5-op cap forces partial heals that interleave with partitions ', async (t) => {
const bad = []
for (let seed = 1; seed <= SEEDS; seed++) {
const r = await run(seed, STEPS, 5)
if (!r.nodesAgree || !r.oneHead || !r.consumerOk) bad.push(seed + ':' + JSON.stringify(r))
}
t.alike(bad, [], 'every seed converges with split heals')
})