// CONTINUITY: a higher rev on a DIFFERENT LINE must not win the read surface (hyper-social#16 P0).
//
// loser-club: admitting a co-owner flipped the creator role, the new node authored a RIVAL GENESIS
// (1 record), and because its rev sorted HIGHER than the authoritative 139-record head, the bridge's
// freshest-by-rev logic served the fork from the public getLatestCommit until the process was bounced.
// A rev REGRESSION was already guarded against; a higher rev on a different line was not.
//
// The fixture models co-owners the way the system actually produces them. atproto-store/lib/tid.js
// `nextRev` makes the chain a pure function of the previous rev, so co-owners continuing from the same
// head over the same record set emit BYTE-IDENTICAL commits — a lagging co-owner's firehose is a strict
// PREFIX of a fresher one's. So: ONE canonical line, each co-owner replaying the same records into its own
// repo and the same canonical FRAMES into its own firehose, truncated at its own lag. The fork is a
// genuinely independent line whose rev sorts higher (a later clock base).
// Run: node test/fork.test.mjs
import { mkdtempSync, rmSync } from 'node:fs'
import { join } from 'node:path'
import { tmpdir } from 'node:os'
import createTestnet from '@hyperswarm/testnet'
import Hyperswarm from 'hyperswarm'
import { RepoStore } from 'atproto-store/lib/repo-store.js'
import { RepoManager } from 'atproto-store/lib/repo-manager.js'
import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js'
import { openCorestore } from 'atproto-store/lib/corestore.js'
import { createAtprotoIdentityFromKey } from 'atproto-identity/lib/atproto-identity.js'
import { discoveryTopic } from 'atproto-identity/lib/discovery.js'
import { DidPeer, DidPeerRouter } from '../index.js'
const assert = (c, m) => { if (!c) { console.error('FAIL:', m); cleanupSync(); process.exit(1) } console.log(' ok -', m) }
const iso = '2026-07-27T00:00:00.000Z'
const dirs = []
const toClose = []
function newDir (p) { const d = mkdtempSync(join(tmpdir(), p)); dirs.push(d); return d }
function cleanupSync () { for (const d of dirs) { try { rmSync(d, { recursive: true, force: true }) } catch {} } }
const sleep = (ms) => new Promise((r) => setTimeout(r, ms))
const post = (rk, text) => ({ action: 'create', collection: 'app.bsky.feed.post', rkey: rk, record: { $type: 'app.bsky.feed.post', text, createdAt: iso, langs: ['en'] } })
const id = createAtprotoIdentityFromKey(new Uint8Array(32).fill(3), { handle: 'fork.test', pdsUrl: 'https://pds.local' })
const DID = id.did
const testnet = await createTestnet(3)
const bootstrap = testnet.bootstrap
// ---- the canonical line: ONE writer, N commits. Its FRAMES are what every co-owner replicates. ----
const OPS = [post('3aaaaaaaaaaa2', 'one'), post('3bbbbbbbbbbb2', 'two'), post('3ccccccccccc2', 'three'), post('3ddddddddddd2', 'four')]
const canonStore = openCorestore(newDir('canon-'))
const canonRepo = await new RepoStore(canonStore, { name: `atproto-repo:${DID}` }).ready()
let canonMicros = 1_700_000_000_000_000n
const canonMgr = await new RepoManager(canonRepo, { did: DID, privateKey: id.privateKey, clock: () => (canonMicros += 1_000_000n) }).init()
const canonSeq = await new FirehoseSequencer(canonStore.get({ name: 'canon-firehose' })).ready()
const canonRevs = []
for (const op of OPS) { const w = await canonMgr.applyWrite(op); await canonSeq.sequenceCommit(w, { time: iso }); canonRevs.push(w.rev) }
const canonFrames = []
for (let i = 0; i < canonSeq.core.length; i++) canonFrames.push(await canonSeq.core.get(i))
toClose.push(async () => { try { await canonSeq.close() } catch {}; try { await canonRepo.close() } catch {}; try { await canonStore.close() } catch {} })
console.log('canonical line revs:', canonRevs.join(' '))
// A co-owner AT LAG k: its own repo holding the first k records, its own firehose holding the first k
// CANONICAL frames (identical bytes => identical commits => a real prefix of the same line).
async function coOwner (k, tag) {
const store = openCorestore(newDir('co-' + tag + '-'))
const repo = await new RepoStore(store, { name: `atproto-repo:${DID}` }).ready()
let micros = 1_700_000_000_000_000n
const mgr = await new RepoManager(repo, { did: DID, privateKey: id.privateKey, clock: () => (micros += 1_000_000n) }).init()
for (let i = 0; i < k; i++) await mgr.applyWrite(OPS[i]) // same records => same MST => same data root
const seq = await new FirehoseSequencer(store.get({ name: 'firehose' })).ready()
for (let i = 0; i < k; i++) await seq.append(canonFrames[i]) // the canonical commits, verbatim
return finishHolder(store, repo, seq, tag)
}
// A FORK: an independent line (different record, later clock base => its rev sorts HIGHER).
async function forkOwner (tag) {
const store = openCorestore(newDir('co-' + tag + '-'))
const repo = await new RepoStore(store, { name: `atproto-repo:${DID}` }).ready()
let micros = 1_900_000_000_000_000n // far ahead => higher TID => wins freshest-by-rev
const mgr = await new RepoManager(repo, { did: DID, privateKey: id.privateKey, clock: () => (micros += 1_000_000n) }).init()
const seq = await new FirehoseSequencer(store.get({ name: 'firehose' })).ready()
const w = await mgr.applyWrite(post('3zzzzzzzzzzz2', 'rival genesis'))
await seq.sequenceCommit(w, { time: iso })
console.log('fork rev:', w.rev, '(canonical head:', canonRevs[canonRevs.length - 1] + ')')
return finishHolder(store, repo, seq, tag)
}
async function finishHolder (store, repo, seq, tag) {
const peer = new DidPeer(store, { did: DID, verifyKey: id.publicKey })
await peer.provide({ repoKey: repo.core.key, firehoseKey: seq.core.key })
const swarm = new Hyperswarm({ bootstrap })
swarm.on('connection', (c) => peer.attach(c))
// server AND client, as a real peer joins: a holder that appears LATE must be able to find the reader
// (which is already announcing) rather than waiting for the reader's next lookup.
await swarm.join(discoveryTopic(DID), { server: true, client: true }).flushed()
toClose.push(async () => { try { await swarm.destroy() } catch {}; try { await peer.close() } catch {}; try { await store.close() } catch {} })
return { peer, swarm, fhKey: seq.core.key.toString('hex'), tag, rev: peer.head.rev }
}
const behind = await coOwner(2, 'behind') // same line, lagging
const ahead = await coOwner(4, 'ahead') // same line, current
assert(ahead.rev === canonRevs[3] && behind.rev === canonRevs[1], 'co-owners sit on the SAME line at different lag')
// Reader in the bridge's posture (owns-swarm), continuity gate on (default).
const store = openCorestore(newDir('reader-'))
const swarm = new Hyperswarm({ bootstrap })
const router = new DidPeerRouter(store, {
swarm,
// exactly as atproto-bridge/pds/bin/pds-serve.js joins: server AND client
join: (did) => { try { swarm.join(discoveryTopic(did), { server: true, client: true }); swarm.flush().catch(() => {}) } catch {} },
fronted: [DID],
resolveVerifyKey: async () => id.publicKey
})
toClose.push(async () => { try { await swarm.destroy() } catch {}; try { await router.close() } catch {}; try { await store.close() } catch {} })
const servedRev = async () => { const f = await router.front(DID).catch(() => null); const lc = f && f.getLatestCommit(); return lc ? lc.rev : null }
const servedRecs = async () => { const f = await router.front(DID).catch(() => null); return f ? (((await f.listRecords('app.bsky.feed.post').catch(() => null)) || {}).records || []).length : -1 }
async function waitRev (want, ms = 20000) { const end = Date.now() + ms; let last = null; while (Date.now() < end) { last = await servedRev(); if (last === want) return last; await sleep(250) } return last }
// 1. Same line: the fresher co-owner wins, continuity permits it (its frames CONTAIN the lagging head).
assert(await waitRev(canonRevs[3]) === canonRevs[3], 'on ONE line the freshest co-owner is served (continuity permits the upgrade)')
assert(await servedRecs() === 4, 'serving all 4 records of the canonical head')
// 2. The fork appears, live, with a HIGHER rev. It must NOT take the read surface.
const fork = await forkOwner('fork')
// Poll until the rival is actually FRONTED + verified, so the assertions below test the gate rather than
// a rival that simply hadn't arrived yet.
const forkArrived = await (async () => {
const end = Date.now() + 30000
while (Date.now() < end) {
await servedRev() // drive front() so the gate runs
const row = router.diagnose().fronts.find((f) => f.firehoseKey === fork.fhKey)
if (row && row.rev) return true
await sleep(250)
}
return false
})()
assert(forkArrived, 'the rival co-owner was fronted + verified (the gate is being exercised)')
const afterFork = await servedRev()
assert(afterFork === canonRevs[3], `a higher-rev rival LINE does not win the read surface (still ${afterFork})`)
assert(await servedRecs() === 4, 'the 4-record authoritative head is still served, not the 1-record rival')
const diag = router.diagnose()
console.log(' fronts:', diag.fronts.map((f) => `${f.firehoseKey.slice(0, 6)} rev=${f.rev} live=${f.live} served=${f.served} forkRejected=${f.forkRejected}`).join(' | '))
console.log(' looking for fork fh', fork.fhKey.slice(0, 6))
const forkRow = diag.fronts.find((f) => f.firehoseKey === fork.fhKey)
assert(!!forkRow, 'the rival is still FRONTED (it is tracked, not ignored)')
assert(forkRow.rev > canonRevs[3], 'the rival really does carry the higher rev')
assert(forkRow.forkRejected > 0, 'diagnose NAMES the rejection (forkRejected > 0) — an operator can see the disagreement')
assert(diag.fronts.find((f) => f.served).rev === canonRevs[3], 'diagnose reports the authoritative front as the served one')
// The gateway's metrics show how long ago each DID's firehose last advanced — the 'is this peer alive'
// signal. It existed (_lastAppend) and nothing could read it.
assert(typeof diag.lastAppendAt[DID] === 'number' && diag.lastAppendAt[DID] > 0, 'diagnose reports when the DID\'s firehose last appended')
// 3. The gate is not a permanent lock-out: with the whole canonical line gone (the rebuild case — a peer
// that rebuilt its cores comes back on a new line), the surviving line is served.
await ahead.swarm.destroy()
await behind.swarm.destroy()
const afterLineGone = await waitRev(fork.rev)
assert(afterLineGone === fork.rev, `once the incumbent line is gone the new line IS served (${afterLineGone})`)
console.log('\nFORK OK: a higher rev on a rival line loses to the served line while that line is live')
for (const close of toClose) await close()
await testnet.destroy()
cleanupSync()