// PRE-FLIGHT for the two prod fold bugs that only surfaced on live bsky (swarm posts arrived as
// #sync-only, never #commit). Both live in FleetRelay.addSource — the SEAM a fronted peer's firehose
// crosses to reach the aggregate subscribeRepos. Neither repo's unit suite exercised a REPLICATED
// (remote) source, so both slipped through. These run fully offline (testnet), no prod.
//
// Bug A: a freshly-opened REPLICATED core reads length=0 until update(); 'append' never fires for
// frames that existed BEFORE we connected — so a peer's backlog never folds. (the update fix)
// Bug B: FleetRelay dedups by (sourceId, index); reusing one sourceId across DIFFERENT cores (a
// peer whose firehose core changed, e.g. single-tenant -> swarm) collides indices 0..n and
// drops the new core. peer-router must key by core.key. (the per-core sourceId fix)
import test from 'brittle'
import { mkdtempSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import b4a from 'b4a'
import Hyperswarm from 'hyperswarm'
import createTestnet from '@hyperswarm/testnet'
import { openCorestore } from 'atproto-store/lib/corestore.js'
import { accountFrame } from 'atproto-store/lib/firehose.js'
import { FleetRelay } from '../lib/fleet-relay.js'
import { serveReplication, connectReplica } from '../lib/fleet-replicate.js'
const T = '2026-07-05T00:00:00.000Z'
const frame = (did) => accountFrame({ seq: 0, did, active: true, time: T })
async function collect (relay, cursor = 0, n = 20) {
const ac = new AbortController()
const it = relay.subscribe(cursor, { signal: ac.signal })
const out = []
try { for (let k = 0; k < n; k++) { const p = it.next(); const r = await Promise.race([p, new Promise((res) => setTimeout(() => res({ done: true }), 500))]); if (r.done || !r.value) break; out.push(r.value.seq) } } finally { ac.abort() }
return out
}
// Seam coverage — the fold must ingest a peer's backlog from a REPLICATED (remote-by-key) firehose,
// the exact integration neither repo's unit suite exercised. NOTE: this passes with OR without the
// defensive update() (the replicated core syncs its length passively here), so it does NOT by itself
// isolate that change — the load-bearing prod fix is the per-core dedup in test B. Kept because the
// missing coverage was the *seam* itself; a regression that broke replicated folding would trip here.
test('preflight A: the fold ingests a REPLICATED source over the swarm (seam coverage)', async (t) => {
const testnet = await createTestnet(3)
const srcDir = mkdtempSync(join(tmpdir(), 'fold-src-'))
const brDir = mkdtempSync(join(tmpdir(), 'fold-br-'))
const aggDir = mkdtempSync(join(tmpdir(), 'fold-agg-'))
const srcStore = openCorestore(srcDir); const brStore = openCorestore(brDir); const aggStore = openCorestore(aggDir)
// A source firehose Hypercore with 3 frames that ALREADY EXIST before the bridge ever connects.
const srcCore = srcStore.get({ name: 'firehose' })
await srcCore.ready()
for (let i = 0; i < 3; i++) await srcCore.append(frame('did:plc:peer'))
const fkey = srcCore.key
const srcSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap })
const brSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap })
const serve = serveReplication(srcStore, srcSwarm); await serve.ready()
await connectReplica(brStore, brSwarm, serve.serverKey)
// The bridge opens the peer's firehose core BY KEY — remote, length 0 until update() syncs it.
const remote = brStore.get({ key: fkey })
await remote.ready()
const relay = await new FleetRelay(aggStore.get({ name: 'agg', valueEncoding: 'json' })).ready()
t.teardown(async () => {
serve.close(); await srcSwarm.destroy(); await brSwarm.destroy(); await relay.close()
await srcStore.close(); await brStore.close(); await aggStore.close(); await testnet.destroy()
for (const d of [srcDir, brDir, aggDir]) rmSync(d, { recursive: true, force: true })
})
await relay.addSource(b4a.toString(fkey, 'hex'), remote)
const out = await collect(relay)
t.is(out.length, 3, 'the replicated backlog folded into the aggregate (0 without the update() fix — it would see length 0 and rely on append, which never fires for pre-existing frames)')
})
// Bug B — reusing a sourceId across different cores collides; keying by core.key folds both.
test('preflight B: fold source id must be per-core, not per-did (the dedup-collision prod bug)', async (t) => {
const dir = mkdtempSync(join(tmpdir(), 'fold-dedup-'))
const store = openCorestore(dir)
const mk = async (name) => { const c = store.get({ name }); await c.ready(); for (let i = 0; i < 3; i++) await c.append(frame('did:plc:same')); return c }
const core1 = await mk('fh-old') // e.g. the single-tenant firehose
const core2 = await mk('fh-new') // e.g. the swarm membrane firehose (SAME did, different core)
const relay = await new FleetRelay(store.get({ name: 'agg', valueEncoding: 'json' })).ready()
t.teardown(async () => { await relay.close(); await store.close(); rmSync(dir, { recursive: true, force: true }) })
// WRONG (the prod bug): both keyed by the did → core2's indices 0..2 collide with core1's → dropped.
await relay.addSource('did:plc:same', core1)
await relay.addSource('did:plc:same', core2)
t.is((await collect(relay)).length, 3, 'reusing the did as sourceId drops the second core entirely (only 3 of 6 fold) — the collision peer-router hit on the single-tenant→swarm migration')
// RIGHT (the fix): keyed by core.key → distinct sources → both fold.
const relay2 = await new FleetRelay(store.get({ name: 'agg2', valueEncoding: 'json' })).ready()
t.teardown(async () => { await relay2.close() })
await relay2.addSource(b4a.toString(core1.key, 'hex'), core1)
await relay2.addSource(b4a.toString(core2.key, 'hex'), core2)
t.is((await collect(relay2)).length, 6, 'keyed by core.key, both cores fold (6 frames) — what the fix guarantees')
})