// PARTITION FORK × MEMBRANE — each side's membrane projected its own fork; after the merge and the heal
// both must serve the merged repo with the heal as the firehose head, and neither may refuse.
//
// serveMembrane projects the chain BY POSITION (firehose append from seq.length) and rebuilds the repo
// core by replaying ops. The fork that loses has the lower rev, and every commit adds 1µs, so a losing
// branch is never longer than the winner's: a membrane that projected the losing fork is never AHEAD of
// the chain. When the heal arrives its ops, applied to that fork's state, must reproduce the merged root.
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 { cidForBytesSync } from 'atproto-repo/lib/cid.js'
import { serveMembrane } from 'atproto-room/lib/serve-membrane.js'
import { decodeFrame } from 'atproto-store/lib/firehose.js'
import { openCorestore } from 'atproto-store/lib/corestore.js'
import { commitLine } from 'atproto-room/lib/commit-line.js'
import { SwarmDevice } from '../lib/swarm-device.js'
const DID = 'did:plc:partitionmembrane'
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' })
const tmp = () => { const d = '/tmp/hpp-pfm-' + Math.floor(Math.random() * 1e9); fs.mkdirSync(d, { recursive: true }); return d }
class MemKV {
constructor (m) { this.m = new Map(m || []) }
async set (k, v) { this.m.set(k, v) }
async get (k) { return this.m.has(k) ? this.m.get(k) : null }
async getNow (k) { return this.get(k) }
async del (k) { this.m.delete(k) }
async list (prefix = '') { const o = {}; for (const [k, v] of this.m) if (k.startsWith(prefix)) o[k] = v; return o }
async listNow (prefix = '') { return this.list(prefix) }
}
function fakeSwarm () {
return { on () {}, off () {}, keyPair: { publicKey: b4a.alloc(32, 7) }, join () { return { flushed: async () => {} } } }
}
async function quietErrors (fn) {
const lines = []
const orig = console.error
console.error = (s) => lines.push(String(s))
try { await fn() } finally { console.error = orig }
return lines
}
const recordCids = (rows) => new Map(rows.map((r) => [r.key, cidForBytesSync(dagCbor.encode(r.record)).toString()]))
async function scenario (t, { aSigns, bSigns }) {
const sb0 = new MemKV()
const S = new SwarmDevice(sb0, { did: DID, privateKey: KEY, self: 'S', firehose: true })
await S.putRecord('app.bsky.feed.post', 'base', post('base'))
await S.putRecord('app.bsky.feed.post', 'shared', post('shared v1'))
await S.sign({ force: true })
const side = async (name, signs) => {
const sb = new MemKV(sb0.m)
const dev = new SwarmDevice(sb, { did: DID, privateKey: KEY, self: name, firehose: true })
for (let i = 1; i <= signs; i++) { await dev.putRecord('app.bsky.feed.post', 'r' + name + i, post(name + ' ' + i)); await dev.sign() }
const membrane = await serveMembrane(dev, { corestore: openCorestore(tmp()), swarm: fakeSwarm(), did: DID })
return { sb, dev, membrane }
}
const A = await side('a', aSigns)
const B = await side('b', bSigns)
if (aSigns === 2 && bSigns === 2) await A.dev.putRecord('app.bsky.feed.post', 'shared', post('shared v2 from A'))
if (aSigns === 2 && bSigns === 2) await A.dev.sign()
// Merge: each replica receives the other's writes (keys the other side changed from the common log).
const aOnly = [...A.sb.m].filter(([k, v]) => sb0.m.get(k) !== v)
const bOnly = [...B.sb.m].filter(([k, v]) => sb0.m.get(k) !== v)
for (const [k, v] of bOnly) A.sb.m.set(k, v)
for (const [k, v] of aOnly) B.sb.m.set(k, v)
// The heal is signed on A's replica and replicates to B's.
const heal = await A.dev.sign()
t.ok(heal, 'A signs the heal')
for (const [k, v] of A.sb.m) if (k.startsWith('__commit__/') && !B.sb.m.has(k)) B.sb.m.set(k, v)
const { line } = commitLine(Object.values(await A.sb.listNow('__commit__/')))
const merged = recordCids((await A.dev.repo.listRecords(null, { now: true })).map((r) => ({ key: r.collection + '/' + r.rkey, record: r.record })))
for (const [name, s] of [['A', A], ['B', B]]) {
const errors = await quietErrors(() => s.membrane.sync())
t.absent(errors.find((l) => /REFUSED|STALE/.test(l)), `membrane ${name}: no refusal or stale warning`)
const repo = recordCids(await s.membrane.repo.records())
t.ok(repo.size === merged.size && [...merged].every(([k, v]) => repo.get(k) === v), `membrane ${name}: repo core holds the merged records`)
const last = decodeFrame(await s.membrane.seq.core.get(s.membrane.seq.core.length - 1))
t.is(last.body.commit.toString(), heal.commitCid.toString(), `membrane ${name}: firehose head is the heal`)
t.ok(s.membrane.seq.core.length <= line.length, `membrane ${name}: firehose never longer than the chain`)
}
}
test('membranes on both sides of a one-commit fork converge on the heal', async (t) => {
await scenario(t, { aSigns: 1, bSigns: 1 })
})
test('membranes converge when one side lost a two-commit branch', async (t) => {
await scenario(t, { aSigns: 3, bSigns: 2 })
await scenario(t, { aSigns: 2, bSigns: 3 })
})
test('membranes converge with a concurrent edit of one record', async (t) => {
await scenario(t, { aSigns: 2, bSigns: 2 })
})