// PARTITION FORK HEAL — two co-owners signing on opposite sides of a partition.
//
// The presence election is per CONNECTED SET, so a partition elects a signer on each side, and both
// continue the chain from the same head: nextRev(head) is a pure function of the head, so they sign
// the SAME rev over DIFFERENT record sets. Measured 2026-10-13 against this code:
// - the frames are keyed __commit__/<rev>, so after the merge ONE overwrites the other (2 of 3 frames
// survive) and the losing fork leaves no trace in the shared log;
// - the next commit is diffed against the SURVIVING fork only, so it names that side's records and
// never the other's.
// The relay keeps whichever fork reached it first (it drops a #commit with rev <= the stored rev), and
// the reference bsky AppView applies a #commit's OPS and nothing else: #sync does not touch records,
// a `create` for a record it already holds is ignored, `update` upserts, `delete` removes. So a
// consumer that took the fork the log did NOT keep never learns the other side's posts.
//
// What healing requires, from that consumer's rules: both forks stay visible in the log, and the heal
// commit names EVERY path either fork touched since they split — as an update when the record exists in
// the merged state, as a delete when it does not.
import '../lib/bare-globals.js'
import test from 'brittle'
import b4a from 'b4a'
import { SwarmDevice } from '../lib/swarm-device.js'
import * as dagCbor from '@ipld/dag-cbor'
import { cidForBytesSync } from 'atproto-repo/lib/cid.js'
import { CID } from 'multiformats/cid'
import { commitLine } from 'atproto-room/lib/commit-line.js'
const DID = 'did:plc:partitionfork'
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 P = 'app.bsky.feed.post/'
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) }
}
const framesOf = async (sb) => Object.values(await sb.listNow('__commit__/'))
// The reference bsky AppView's record rules (packages/bsky data-plane subscription.ts + processor.ts,
// atproto 0bac60a): ops applied blindly; create over an indexed URI is a no-op; update upserts; delete
// removes (unknown = no-op). The relay in front drops any #commit whose rev <= the stored rev.
function appview () {
const records = new Map()
let rev = ''
return {
records,
commit (frame) {
if (frame.rev <= rev) return false
rev = frame.rev
for (const op of frame.ops || []) {
if (op.action === 'create') { if (!records.has(op.path)) records.set(op.path, op.cid) } else if (op.action === 'update') records.set(op.path, op.cid)
else if (op.action === 'delete') records.delete(op.path)
}
return true
}
}
}
// One partition round: common history, then each side writes and signs on its own replica.
async function partition ({ aWrites, bWrites }) {
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 common = await framesOf(sb0)
const sbA = new MemKV(sb0.m)
const sbB = new MemKV(sb0.m)
const A = new SwarmDevice(sbA, { did: DID, privateKey: KEY, self: 'A', firehose: true })
const B = new SwarmDevice(sbB, { did: DID, privateKey: KEY, self: 'B', firehose: true })
await aWrites(A)
await bWrites(B)
const hA = await A.sign()
const hB = await B.sign()
const forkA = (await framesOf(sbA)).find((f) => f.rev === hA.rev)
const forkB = (await framesOf(sbB)).find((f) => f.rev === hB.rev)
// Merge: the common history plus the keys each side actually WROTE. Copying a whole replica would
// clobber the other side's writes with untouched common values. A key written on both sides keeps
// one value, which is what the shared view does (the later-linearized write wins; here B's).
const sbM = new MemKV(sb0.m)
for (const side of [sbA, sbB]) {
for (const [k, v] of side.m) if (sb0.m.get(k) !== v) sbM.m.set(k, v)
}
const M = new SwarmDevice(sbM, { did: DID, privateKey: KEY, self: 'M', firehose: true })
const heal = await M.sign()
const healFrame = heal ? (await framesOf(sbM)).find((f) => f.rev === heal.rev) : null
// path -> record CID: a consumer must end with the same CONTENT, not just the same paths.
const merged = new Map((await M.repo.listRecords(null, { now: true })).map((r) => [r.collection + '/' + r.rkey, cidForBytesSync(dagCbor.encode(r.record)).toString()]))
return { common, forkA, forkB, hA, hB, sbM, M, healFrame, merged }
}
// A consumer that took fork `fork` at the relay, then the heal. Returns path -> record CID.
function consumerAfter (common, fork, healFrame) {
const av = appview()
for (const f of common.sort((a, b) => (a.rev < b.rev ? -1 : 1))) av.commit(f)
av.commit(fork)
if (healFrame) av.commit(healFrame)
return av.records
}
const same = (a, b) => a.size === b.size && [...a].every(([k, v]) => b.get(k) === v)
test('partition fork: both forks survive the merge in the shared log', async (t) => {
const r = await partition({
aWrites: (A) => A.putRecord('app.bsky.feed.post', 'ra', post('from A')),
bWrites: (B) => B.putRecord('app.bsky.feed.post', 'rb', post('from B'))
})
t.is(r.hA.rev, r.hB.rev, 'precondition: both sides sign nextRev(head) — the same rev')
t.ok(!r.hA.dataCid.equals(r.hB.dataCid), 'precondition: over different record sets')
const cids = new Set((await framesOf(r.sbM)).map((f) => f.commitCid))
t.ok(cids.has(r.forkA.commitCid), 'fork A\'s signed commit is still in the log')
t.ok(cids.has(r.forkB.commitCid), 'fork B\'s signed commit is still in the log')
})
test('partition fork: a consumer of EITHER fork converges to the merged record set after the heal', async (t) => {
const r = await partition({
aWrites: (A) => A.putRecord('app.bsky.feed.post', 'ra', post('from A')),
bWrites: (B) => B.putRecord('app.bsky.feed.post', 'rb', post('from B'))
})
t.ok(r.healFrame, 'the merged device signs a heal commit')
t.ok(r.merged.has(P + 'ra') && r.merged.has(P + 'rb'), 'precondition: the merged view holds both posts')
t.ok(same(consumerAfter(r.common, r.forkA, r.healFrame), r.merged), 'consumer that took fork A ends with the merged set')
t.ok(same(consumerAfter(r.common, r.forkB, r.healFrame), r.merged), 'consumer that took fork B ends with the merged set')
})
test('partition fork: a delete on one side and an edit on the other both reach every consumer', async (t) => {
const r = await partition({
aWrites: async (A) => { await A.deleteRecord('app.bsky.feed.post', 'base') },
bWrites: async (B) => { await B.putRecord('app.bsky.feed.post', 'shared', post('shared v2 from B')) }
})
t.ok(r.healFrame, 'the merged device signs a heal commit')
for (const fork of [r.forkA, r.forkB]) {
const got = consumerAfter(r.common, fork, r.healFrame)
t.ok(same(got, r.merged), `consumer of fork ${fork === r.forkA ? 'A' : 'B'} holds exactly the merged records (paths AND content)`)
}
t.ok(!r.merged.has(P + 'base'), 'precondition: the delete wins in the merged view')
})
test('partition fork: a heal is signed ONCE — the next sign with nothing new is a no-op', async (t) => {
const r = await partition({
aWrites: (A) => A.putRecord('app.bsky.feed.post', 'ra', post('from A')),
bWrites: (B) => B.putRecord('app.bsky.feed.post', 'rb', post('from B'))
})
t.ok(r.healFrame && (r.healFrame.heals || []).length === 1, 'the heal names the fork it heals')
const again = await r.M.sign()
t.is(again, null, 'no second heal and no empty commit')
const fresh = new SwarmDevice(r.sbM, { did: DID, privateKey: KEY, self: 'F', firehose: true })
t.is(await fresh.sign(), null, 'a device that restarts on the merged log does not heal again either')
})
test('partition fork: a LOSING branch of two commits still reaches every consumer', async (t) => {
// A signs twice, B three times, both from the same head: B has the higher rev and is the chain, so
// A's whole two-commit branch is the fork.
const r = await partition({
aWrites: async (A) => {
await A.putRecord('app.bsky.feed.post', 'ra1', post('A one'))
await A.sign()
await A.putRecord('app.bsky.feed.post', 'ra2', post('A two'))
},
bWrites: async (B) => {
await B.putRecord('app.bsky.feed.post', 'rb1', post('B one'))
await B.sign()
await B.putRecord('app.bsky.feed.post', 'rb2', post('B two'))
await B.sign()
await B.putRecord('app.bsky.feed.post', 'rb3', post('B three'))
}
})
t.ok(r.healFrame && (r.healFrame.heals || []).length === 2, 'the heal names both commits of the losing branch')
const commonCids = new Set(r.common.map((f) => f.commitCid))
const all = Object.values(await r.sbM.listNow('__commit__/'))
const branch = (re) => all.filter((f) => !commonCids.has(f.commitCid) && !(f.heals || []).length && (f.ops || []).some((o) => re.test(o.path)))
for (const [name, frames] of [['A (lost, 2 commits)', branch(/ra[12]$/)], ['B (chain, 3 commits)', branch(/rb[123]$/)]]) {
const av = appview()
for (const f of r.common.concat(frames).sort((a, b) => (a.rev < b.rev ? -1 : 1))) av.commit(f)
av.commit(r.healFrame)
t.ok(same(av.records, r.merged), `consumer of branch ${name} converges`)
}
})
test('partition fork: the device on the LOSING side of an equal-rev tie continues from the chain head', async (t) => {
// Two sides sign the same rev; the logs merge with NO heal yet, so the only thing above either fork is
// the other fork at the SAME rev. The device whose fork lost the tie must re-anchor to the chain head.
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.sign({ force: true })
const sbA = new MemKV(sb0.m); const sbB = new MemKV(sb0.m)
const A = new SwarmDevice(sbA, { did: DID, privateKey: KEY, self: 'A', firehose: true })
const B = new SwarmDevice(sbB, { did: DID, privateKey: KEY, self: 'B', firehose: true })
await A.putRecord('app.bsky.feed.post', 'ra', post('from A'))
await B.putRecord('app.bsky.feed.post', 'rb', post('from B'))
const hA = await A.sign(); const hB = await B.sign()
t.is(hA.rev, hB.rev, 'precondition: an equal-rev tie')
// Each side receives the other's writes; each device keeps its own head.
for (const [from, to] of [[sbA, sbB], [sbB, sbA]]) for (const [k, v] of from.m) if (sb0.m.get(k) !== v && !to.m.has(k)) to.m.set(k, v)
const { head } = commitLine(Object.values(await sbA.listNow('__commit__/')))
const [loser, loserHead] = head.commitCid === hA.commitCid.toString() ? [B, hB] : [A, hA]
await loser.putRecord('app.bsky.feed.post', 'later', post('after the merge'))
const h = await loser.sign()
t.ok(h, 'the losing side signs')
t.not(String(h.signedCommit.prev), loserHead.commitCid.toString(), 'it did not extend its own losing fork')
t.is(String(h.signedCommit.prev), head.commitCid, 'it continued from the chain head')
})