// FRONT IDENTITY COLLIDES: the holder and every keyless seeder announce the SAME core key
// (hyper-social#16 P0 item 3).
//
// A front is keyed by firehoseKey, and the router kept ONE {conn, rpc} per key — so whoever announced LAST
// won the application-RPC slot. A seeder replicates the same cores and announces the same keys but serves
// no application RPC, so the front's write/session/blob relay could be pointed at a peer that answers
// `UNKNOWN_METHOD: Unknown method 'get-blob'` for everything. That is favela's getBlob failure and e2jev's
// intermittent images: the bytes were fine, the front was asking the wrong peer.
//
// A front must hold a SET of candidates and try them until one ANSWERS — while still surfacing a real
// error from a peer that DID answer, instead of masking it by retrying elsewhere.
// Run: node test/rpc-candidates.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 c from 'compact-encoding'
import ProtomuxRPC from 'protomux-rpc'
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, RPC_PROTOCOL } from '../index.js'
const assert = (c2, m) => { if (!c2) { 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 id = createAtprotoIdentityFromKey(new Uint8Array(32).fill(5), { handle: 'rpc.test', pdsUrl: 'https://pds.local' })
const DID = id.did
const testnet = await createTestnet(3)
const bootstrap = testnet.bootstrap
// ---- the HOLDER: owns the cores AND serves the application RPC ----
const hStore = openCorestore(newDir('holder-'))
const hRepo = await new RepoStore(hStore, { name: `atproto-repo:${DID}` }).ready()
let micros = 1_700_000_000_000_000n
const hMgr = await new RepoManager(hRepo, { did: DID, privateKey: id.privateKey, clock: () => (micros += 1_000_000n) }).init()
const hSeq = await new FirehoseSequencer(hStore.get({ name: 'firehose' })).ready()
const w = await hMgr.applyWrite({ action: 'create', collection: 'app.bsky.feed.post', rkey: '3aaaaaaaaaaa2', record: { $type: 'app.bsky.feed.post', text: 'hi', createdAt: iso, langs: ['en'] } })
await hSeq.sequenceCommit(w, { time: iso })
const holder = new DidPeer(hStore, { did: DID, verifyKey: id.publicKey })
await holder.provide({ repoKey: hRepo.core.key, firehoseKey: hSeq.core.key })
const hSwarm = new Hyperswarm({ bootstrap })
hSwarm.on('connection', (conn) => {
holder.attach(conn)
const rpc = new ProtomuxRPC(conn, { protocol: RPC_PROTOCOL, valueEncoding: c.json })
rpc.on('error', () => {})
rpc.respond('get-blob', (req) => ({ ok: true, from: 'holder', cid: req && req.cid }))
rpc.respond('boom', () => { throw new Error('the holder refuses this one') }) // a REAL answer, not a miss
})
await hSwarm.join(discoveryTopic(DID), { server: true, client: true }).flushed()
toClose.push(async () => { try { await hSwarm.destroy() } catch {}; try { await holder.close() } catch {}; try { await hStore.close() } catch {} })
// ---- the SEEDER: announces the SAME keys, opens the SAME RPC protocol, serves NO application method ----
const sStore = openCorestore(newDir('seeder-'))
const seeder = new DidPeer(sStore, { did: DID, verifyKey: id.publicKey })
await seeder.provide({ repoKey: hRepo.core.key, firehoseKey: hSeq.core.key }) // same core KEYS => same front
const sSwarm = new Hyperswarm({ bootstrap })
sSwarm.on('connection', (conn) => {
seeder.attach(conn)
const rpc = new ProtomuxRPC(conn, { protocol: RPC_PROTOCOL, valueEncoding: c.json })
rpc.on('error', () => {})
rpc.respond('seeder-only-noop', () => ({ ok: true })) // anything else -> UNKNOWN_METHOD
})
await sSwarm.join(discoveryTopic(DID), { server: true, client: true }).flushed()
toClose.push(async () => { try { await sSwarm.destroy() } catch {}; try { await seeder.close() } catch {}; try { await sStore.close() } catch {} })
// ---- the reader: the bridge's posture, rpc: true (the write/session/blob relay) ----
const store = openCorestore(newDir('reader-'))
const swarm = new Hyperswarm({ bootstrap })
const router = new DidPeerRouter(store, {
swarm,
join: (did) => { try { swarm.join(discoveryTopic(did), { server: true, client: true }); swarm.flush().catch(() => {}) } catch {} },
fronted: [DID],
resolveVerifyKey: async () => id.publicKey,
rpc: true
})
toClose.push(async () => { try { await swarm.destroy() } catch {}; try { await router.close() } catch {}; try { await store.close() } catch {} })
const front = await router.front(DID)
const gotBoth = await (async () => {
const end = Date.now() + 30000
while (Date.now() < end) { if (front._liveRpcs().length >= 2) return true; await sleep(250) }
return false
})()
{
const entries = router._byDid.get(DID) || new Map()
for (const [k, e] of entries) console.log(` entry ${k.slice(0, 8)}: conns=${e.conns.size} rpcs=${[...e.conns.values()].filter(Boolean).length}`)
console.log(` reader swarm conns=${swarm.connections.size} | front candidates=${front._liveRpcs().length} | fronts=${[...(router._fronts.get(DID) || new Map()).keys()].map((k) => k.slice(0, 8)).join(',')}`)
}
assert(gotBoth, `the front holds BOTH announcers as RPC candidates (got ${front._liveRpcs().length})`)
// Both orders must work. Whichever announced last used to win the single slot, so proving order-independence
// is proving the bug is gone — without needing to know which candidate is which.
const first = await front.getBlob('bafyfake')
assert(first && first.ok && first.from === 'holder', 'getBlob is answered by the HOLDER (candidate order as-found)')
front._rpcs = [...front._rpcs].reverse() // simulate the other announce order
const second = await front.getBlob('bafyfake')
assert(second && second.ok && second.from === 'holder', 'getBlob is answered by the HOLDER with the order REVERSED (a seeder no longer wins the slot)')
// Sticky: after falling through, the peer that answered is tried first next time.
const preferred = front.rpc
const third = await front.getBlob('bafyfake')
assert(third.from === 'holder' && front.rpc === preferred, 'the answering peer stays preferred (no repeated failed hops)')
// A peer that ANSWERS with a real error must surface it — not be retried away into someone else's miss.
let boomErr = null
try { await front._call('boom', {}) } catch (e) { boomErr = e }
assert(!!boomErr, 'a real error from the answering peer is surfaced')
assert(boomErr.code !== 'UNKNOWN_METHOD', `the surfaced error is the holder's, not a seeder miss (got ${boomErr.code})`)
// And a method NOBODY serves still fails as unknown, rather than hanging or reporting success.
let missErr = null
try { await front._call('nobody-serves-this', {}) } catch (e) { missErr = e }
assert(missErr && missErr.code === 'UNKNOWN_METHOD', 'a method no candidate serves still reports UNKNOWN_METHOD')
console.log('\nRPC CANDIDATES OK: the holder answers regardless of who announced last; real errors still surface')
for (const close of toClose) await close()
await testnet.destroy()
cleanupSync()