// DidPeer — the corrected, direction-free P2P primitive for an atproto DID's data (repo + firehose
// Hypercores), extracted from atproto-bridge as a shared module (audited vs keet + holepunch,
// 2026-07-13). A DID's data lives in a swarm keyed by discoveryTopic(did); every participant —
// signing device, availability relay/seedbox, read/HTTP bridge, favela client — is the SAME primitive.
// It is NOT a "front the origin by serverKey" client: it replicates the DID's cores from WHOEVER has
// them and serves the freshest commit that VERIFIES against the DID key.
//
// Corrections baked in (see atproto-bridge memory/feedback_p2p-no-directionality-event-driven):
// - ANNOUNCE, don't DERIVE / get-coords-PULL: on connect a holder PUSHES {repoKey,firehoseKey,lengths}
// over a protomux channel (one-mux-per-connection, keet model). No round-trip, no serverKey.
// - EVENT-DRIVEN head reconciliation: the served head re-derives on the firehose core's 'append'
// (fires on local append AND on replicated download) — never a data poll.
// - Auth mandatory: the DID topic is unauthenticated in ANY discovery scheme; trust comes from
// verifying the served repo's signed commit against the DID signing key (verifyKey).
//
// Transport-agnostic: the CONSUMER owns the swarm (joins discoveryTopic(did), and on each 'connection'
// calls didPeer.attach(conn)). The consumer also supplies verifyKey (resolve from the DID doc / PLC).
import b4a from 'b4a'
import c from 'compact-encoding'
import Protomux from 'protomux'
import ProtomuxRPC from 'protomux-rpc'
import { RepoStore } from 'atproto-store/lib/repo-store.js'
import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js'
import { decodeCommitFrame, syncFrame } from 'atproto-store/lib/firehose.js'
import { repoDiffSince } from 'atproto-store/lib/repo-diff.js'
import { getRepoFromCommit, getRecord as rvGetRecord, listRecords as rvListRecords } from 'atproto-store/lib/read-view.js'
import { readCar, loadCommit, decodeBlock, listRecords as carListRecords } from 'atproto-repo/lib/repo.js'
import { verifyCommit, commitCid } from 'atproto-repo/lib/commit.js'
const ANNOUNCE_PROTOCOL = 'atproto/did-announce/v1'
// The peer's application RPC (write/session/blob relay). Opened by DidPeerRouter ONE-per-connection when
// `{ rpc: true }` — the freshest front borrows it. MUST match hyper-proto-peer's serve RPC protocol.
export const RPC_PROTOCOL = 'hyper-proto-peer/rpc/v1'
const toKey = (k) => (b4a.isBuffer(k) ? k : b4a.from(k, 'hex'))
// Reject a promise if it doesn't settle within ms (bounds the router's staleness nudge on a degraded link).
export function withTimeout (p, ms) {
return new Promise((resolve, reject) => {
const t = setTimeout(() => reject(new Error('rpc timeout')), ms)
p.then((v) => { clearTimeout(t); resolve(v) }, (e) => { clearTimeout(t); reject(e) })
})
}
export class DidPeer {
// corestore: a Corestore (cores are global by key). did: the DID. verifyKey: its signing pubkey
// (Uint8Array) — REQUIRED to accept a served head. onHead: (head) => void on every NEW verified
// head. onError: optional (err) => void.
constructor (corestore, { did, verifyKey, handle = null, onHead = null, onError = null, noRepo = false } = {}) {
if (!corestore) throw new Error('DidPeer: corestore required')
if (!did) throw new Error('DidPeer: did required')
if (!verifyKey) throw new Error('DidPeer: verifyKey (DID signing pubkey) required — auth is mandatory')
this.corestore = corestore
this.did = did
this.verifyKey = verifyKey
this.handle = handle
this._onHead = onHead
this._onError = onError
// noRepo: follow the firehose ONLY (verified head + tail ops) and DO NOT open/replicate the repo core.
// Records come from elsewhere (the shared content-addressed OverlayRepoStore, co-located with the
// producer). This is the "one copy" consumer: no duplicate repo-core replica. getRepo/getRecord/
// listRecords return null/[]/throw (this.repo stays null) — the caller reads records from the store.
this._noRepo = !!noRepo
this.repo = null
this.seq = null
// Optional write/session/blob RPC to the peer (peer-signs-over-RPC). Set by DidPeerRouter from the
// per-connection ProtomuxRPC when the router runs with `{ rpc: true }`. Null for a read-only consumer
// (favela) — the keyless read surface below needs no RPC.
//
// A SET, not a slot. A front is keyed by firehoseKey, but the key HOLDER and every keyless SEEDER
// announce the SAME core key, so a single slot was won by whoever announced last — and a seeder answers
// `UNKNOWN_METHOD: Unknown method 'get-blob'` because it serves no application RPC at all. That is the
// favela getBlob failure and e2jev's intermittent images. `rpc` stays a property (assignment still
// works, and consumers still read it) but is now backed by the candidate list.
this._rpcs = []
this._repoKey = null
this._firehoseKey = null
// A LABELER DID's signed label core (hpp 0.3.128 membrane-labels:<did>), when its holder announces one.
// Plain hypercore of encoded com.atproto.label blocks; verifying each against the DID key is the
// consumer's job (the gateway lens), exactly as it verifies commits.
this._labelKey = null
this.labels = null
this._onLabels = null // set by DidPeerRouter: (peer) => void once a label core is open
this._channels = new Set()
this.head = null
this._opening = null
}
// Every _call below is reached from inside an HTTP request handler in atproto-bridge. Without a
// deadline an hpp that is alive-but-wedged hangs that request forever — no 504, no log, nothing
// to restart (hyper-social #47). protomux-rpc defaults timeout=-1 and only arms its timer when
// timeout > 0, so the value MUST be passed explicitly; there is no useful transport default.
//
// 30s covers a P2P round trip with a wide margin while still bounding a wedge. Bulk transfers
// (whole-CAR pull, blob upload) get their own budget at the call site — they are legitimately
// slow, and clamping them to the interactive default would break large repos and uploads.
static DEFAULT_RPC_TIMEOUT_MS = 30_000
static BULK_RPC_TIMEOUT_MS = 120_000
// ---- application RPC candidates ----
// `rpc` reads as the preferred LIVE candidate and assigning to it adds one (or clears the set with null),
// so every existing caller — `if (!this.rpc) throw`, `peer.rpc = entry.rpc`, `f.rpc && !f.rpc.closed` —
// keeps working unchanged against a set.
get rpc () { const live = this._liveRpcs(); return live.length ? live[0] : null }
set rpc (v) { if (v) this.addRpc(v); else this._rpcs = [] }
addRpc (rpc) { if (rpc && !this._rpcs.includes(rpc)) this._rpcs.push(rpc) }
_liveRpcs () { this._rpcs = this._rpcs.filter((r) => r && !r.closed); return this._rpcs }
// Try each candidate until one ANSWERS. Only a "wrong peer" / "dead link" failure moves on:
// - UNKNOWN_METHOD the peer serves no such method (a seeder, not the holder) -> try the next
// - CHANNEL_CLOSED/DESTROYED the link died -> try the next
// - TIMEOUT_EXCEEDED ambiguous: retried for READS only, never for a write (a request that may have
// been applied must not be replayed against another peer)
// Anything else is the peer ANSWERING with a real error (REQUEST_ERROR: "blob not found",
// "unauthorized"); surfacing that beats masking it behind a retry.
async _call (method, params, { readOnly = false, timeoutMs = DidPeer.DEFAULT_RPC_TIMEOUT_MS } = {}) {
const cands = this._liveRpcs()
if (!cands.length) throw new Error('DidPeer: no write RPC (read-only front)')
const nextOn = new Set(['UNKNOWN_METHOD', 'CHANNEL_CLOSED', 'CHANNEL_DESTROYED'])
if (readOnly) nextOn.add('TIMEOUT_EXCEEDED')
let last = null
for (let i = 0; i < cands.length; i++) {
const rpc = cands[i]
try {
// The deadline is DELEGATED to protomux-rpc, not raced locally: its _ontimeout also deletes
// the pending entry from `_requests` (protomux-rpc/index.js:161-168). A Promise.race wrapper
// would settle us and leak that entry for the life of the channel.
const res = await rpc.request(method, params, { timeout: timeoutMs })
// Sticky: the peer that answered goes to the front so later calls do not re-pay the failed hops.
if (i > 0) { this._rpcs = [rpc, ...this._rpcs.filter((r) => r !== rpc)] }
return res
} catch (e) {
last = e
if (!nextOn.has(e && e.code)) throw e
}
}
throw last
}
// This node HOLDS the cores (a signing device, a relay that already seeded, or a bootstrap). Open
// them so we can serve + announce. Idempotent.
async provide ({ repoKey, firehoseKey, labelKey = null } = {}) {
await this._openCores(repoKey, firehoseKey)
if (labelKey) await this.adoptLabelKey(labelKey)
}
// Open + fully seed a labeler's label core, follow its appends, and re-announce on each (a new label is
// news to downstream readers just like a new commit). Like adoptRepoKey: only on an established front,
// once. → true when it opened the core.
async adoptLabelKey (labelKey) {
if (this.labels || !labelKey || !this.seq || this._opening || this._labelOpening) return false
this._labelOpening = true
try {
this._labelKey = toKey(labelKey)
const core = this.corestore.get({ key: this._labelKey })
await core.ready()
try { core.download({ start: 0, end: -1 }) } catch {} // ONE live range: all + future appends
core.on('append', () => this._announceAll())
this.labels = core
this._announceAll()
if (this._onLabels) { try { this._onLabels(this) } catch {} }
return true
} finally { this._labelOpening = false }
}
// Re-open a repo core for a front that went back to PROJECTING after a stateless spell. The router
// drops the replica when a peer announces repoKey:null; without this the drop is ONE-WAY, so a peer
// that resumes projecting stays repo-less for the life of the process — permanently on the slower
// derive path. Observed: rolling projectRepoCore back to true did NOT heal the cached front.
async adoptRepoKey (repoKey) {
if (this._noRepo || this.repo || !repoKey) return false
// CRITICAL: during the INITIAL _openCores, this.repo is null while the open is in flight. Without
// this guard a second announce arriving mid-open races it and opens a COMPETING repo core for the
// same front — which deterministically broke the co-owned freshest-by-rev test (the ahead co-owner
// was never fronted). Only adopt for a front that is already established (seq open, nothing opening).
if (this._opening || !this.seq) return false
this._repoKey = toKey(repoKey)
this.repo = await new RepoStore(this.corestore, { key: this._repoKey }).ready()
try { this.repo.core.download({ start: 0, end: -1 }) } catch {}
this._derived = null // the derive cache is now redundant; the core is authoritative
this._derivedAt = null
return true
}
async _openCores (repoKey, firehoseKey) {
if (this.seq) return // firehose is opened in BOTH modes; the idempotency guard (repo may stay null under noRepo)
if (this._opening) return this._opening
this._opening = (async () => {
// repoKey is OPTIONAL now. A peer running the stateless virtual device (serve-membrane
// projectRepoCore:false) publishes no membrane-repo core at all — record bytes exist only as the
// content-addressed files, and the repo is DERIVED on demand via the get-repo-car RPC (see
// getRepo below). Requiring repoKey here is what previously made such a peer un-frontable.
this._repoKey = repoKey ? toKey(repoKey) : null
this._firehoseKey = toKey(firehoseKey)
// noRepo: skip the repo core entirely — never opened, so corestore.replicate never syncs it (no replica).
if (!this._noRepo && this._repoKey) this.repo = await new RepoStore(this.corestore, { key: this._repoKey }).ready()
this.seq = await new FirehoseSequencer(this.corestore.get({ key: this._firehoseKey })).ready()
// FULL seed so this node can serve the cores to others (download all + follow appends) — this is
// the "each peer seeds" torrent behaviour; a read-only consumer that does NOT want to seed can
// subclass/override, but seeding-by-default is the P2P-native contribution posture.
// ONE live download range per core: end:-1 downloads all current blocks AND follows future appends
// (verified), so a single call is sufficient. The prior form re-ran `core.download({0,-1})` inside a
// `core.on('append', …)` listener — every append minted a NEW, never-destroyed [0,-1] RangeRequest and
// added another append listener, an UNBOUNDED range/listener leak that a re-authoring or ballooning
// producer turned into runaway replication churn (the ~100MB/s "never settles" amplifier).
const seed = (core) => { try { core.download({ start: 0, end: -1 }) } catch {} }
if (this.repo) seed(this.repo.core)
seed(this.seq.core)
// EVENT-DRIVEN: re-derive the served head on every firehose append (local or replicated), and
// re-announce the new length to connected peers. No poll loop.
this.seq.core.on('append', () => { this._reconcile(); this._announceAll() })
await this._reconcile()
})()
try { await this._opening } finally { this._opening = null }
}
// Wire a swarm connection: replicate the corestore + open the announce channel. Symmetric — both
// sides call attach(); whoever HOLDS the cores announces, whoever lacks them learns from the announce.
attach (conn) {
try { this.corestore.replicate(conn) } catch (e) { this._fail(e) }
const mux = Protomux.from(conn)
const chan = mux.createChannel({
protocol: ANNOUNCE_PROTOCOL,
onopen: () => { if (this.seq) this._announceTo(chan) }, // ANNOUNCE-PUSH on open
onclose: () => this._channels.delete(chan)
})
if (!chan) return // already open on this mux
chan._announce = chan.addMessage({
encoding: c.json,
onmessage: (a) => this.onAnnounce(a).catch((e) => this._fail(e))
})
chan.open()
this._channels.add(chan)
if (this.seq) this._announceTo(chan)
}
// Does this peer currently hold (and can serve/seed) the DID's cores?
// We HOLD a DID once the firehose (the signed-commit trust anchor) is open. Previously this keyed
// off the repo core, so a stateless peer — which publishes no repo core — read as holding nothing.
get holds () { return !!this.seq }
// Send this peer's announce over a protomux message (used by DidPeerRouter's shared per-connection
// channel, where a single channel is multiplexed across many DIDs).
announceVia (message) {
if (!this.seq || !message) return
try {
message.send(this._announcement())
} catch (e) { this._fail(e) }
}
_announceTo (chan) {
if (!this.seq || !chan._announce) return
try {
chan._announce.send(this._announcement())
} catch (e) { this._fail(e) }
}
// One payload for both announce paths. labelKey/labelLength are carried by EVERY announcer that holds
// the core — a re-announcing seeder that dropped them would hide the labels from everyone fronting
// through it. null when there is no label core (the field exists, so a consumer can branch).
_announcement () {
return {
did: this.did,
repoKey: this._repoKey ? b4a.toString(this._repoKey, 'hex') : null,
firehoseKey: b4a.toString(this._firehoseKey, 'hex'),
repoLength: this.repo ? this.repo.core.length : null,
firehoseLength: this.seq.core.length,
labelKey: this.labels ? b4a.toString(this._labelKey, 'hex') : null,
labelLength: this.labels ? this.labels.length : null
}
}
_announceAll () { for (const chan of this._channels) this._announceTo(chan) }
async onAnnounce (a) {
// repoKey is OPTIONAL — a stateless peer announces { firehoseKey, repoKey: null }. The firehose is
// the load-bearing half (it carries the SIGNED commits, i.e. the trust anchor); the repo is either
// replicated, read from co-located files, or derived over RPC.
if (!a || a.did !== this.did || !a.firehoseKey) return
if (!this.seq) {
await this._openCores(a.repoKey, a.firehoseKey)
this._announceAll() // we can help seed now — tell peers we have them too (symmetric)
}
if (a.labelKey && !this.labels) await this.adoptLabelKey(a.labelKey)
// The announce is a HINT (peer says its firehose is at length N). Replication pushes the blocks;
// we react to the 'append' event to reconcile. One-shot update() nudges the fetch (event-triggered
// on announce, NOT a poll loop).
if (this.seq) this.seq.core.update().catch(() => {})
}
// The served head = the freshest signed commit that VERIFIES against the DID key. Re-derived on
// firehose 'append'. Rejects an unverifiable head (can't be tricked into serving a forged repo).
async _reconcile () {
if (!this.seq || this.seq.core.length === 0) return
try {
const { body } = decodeCommitFrame(await this.seq.core.get(this.seq.core.length - 1))
const signed = loadCommit(await readCar(body.blocks)).value
if (!verifyCommit(signed, this.verifyKey)) return
const rev = signed.rev
if (this.head && this.head.rev === rev) return // no change
this.head = { rev, cid: commitCid(signed).toString(), signed }
if (this._onHead) this._onHead(this.head)
} catch (e) { this._fail(e) }
}
_fail (e) { if (this._onError) this._onError(e) }
// ---- keyless read surface (from the replicated repo/firehose; no signing) ----
// Wait until a verified head exists (cores replicated + a commit that verifies against the DID key).
async ready ({ timeout = 20000 } = {}) {
if (this.head) return this.head
const end = Date.now() + timeout
while (Date.now() < end) {
if (this.head) return this.head
await new Promise((r) => setTimeout(r, 100))
}
return null
}
// com.atproto.sync.getRepo -> keyless signed CAR (records rebuilt from the replicated repo, signature
// from the verified firehose commit; getRepoFromCommit guards data-root integrity).
// `since` (com.atproto.sync.getRepo?since) — the PARTIAL CAR. Answered from the REPLICATED FIREHOSE,
// which every front already holds in both modes (_openCores opens `seq` whether or not there is a repo
// core), so this needs no RPC to the peer, no new storage and no change to hpp. Each #commit frame
// carries the blocks that were new at that commit; the union over the frames newer than `since` is the
// diff. See atproto-store/lib/repo-diff.js for the coverage proof and its refusals.
//
// The result ALWAYS declares what happened: `partial` and `sinceHonoured` are set explicitly, and a
// refusal falls back to the full CAR with a `reason`. A caller must never have to infer from the bytes
// which one it got — inferring is what "accepted and ignored" made everyone do.
async getRepo (opts) {
if (!this.head) throw new Error(`DidPeer: no verified head yet for ${this.did}`)
const since = opts && typeof opts === 'object' ? (opts.since || null) : null
if (since) {
const diff = await this._repoDiffSince(since)
if (diff) return diff
// fall through to the full CAR, with the refusal reported by the caller
}
if (this.repo) {
const full = await getRepoFromCommit(this.repo, this.head.signed)
return { ...full, partial: false, sinceHonoured: since ? false : null, since, reason: since ? this._sinceRefusal : null }
}
// STATELESS PEER: no membrane-repo core to rebuild from, because the peer derives the CAR from its
// record FILES instead of projecting them (serve-membrane projectRepoCore:false). Ask for it.
//
// The trust model does NOT relax. With a repo core we rebuild the MST locally and getRepoFromCommit
// asserts it matches the verified commit; here the CAR arrives whole, so we re-establish the same
// guarantee by verifying the commit INSIDE it against the DID key and refusing anything older than
// the head we already verified off the firehose. A peer that serves a forged or rolled-back repo is
// rejected exactly as before — we never serve bytes we have not authenticated.
if (!this.rpc) throw new Error(`DidPeer: ${this.did} has no repo core and no RPC to derive one`)
const r = await this._call('get-repo-car', {}, { readOnly: true, timeoutMs: DidPeer.BULK_RPC_TIMEOUT_MS })
if (!r || !r.car) throw new Error(`DidPeer: ${this.did} served no CAR`)
const carBytes = b4a.from(r.car, 'hex')
const signed = loadCommit(await readCar(carBytes)).value
if (!verifyCommit(signed, this.verifyKey)) throw new Error(`DidPeer: derived CAR commit does not verify for ${this.did}`)
if (signed.rev < this.head.rev) throw new Error(`DidPeer: derived CAR is older than the verified head (${signed.rev} < ${this.head.rev})`)
return { carBytes, commitCid: commitCid(signed), dataCid: signed.data, partial: false, sinceHonoured: since ? false : null, since, reason: since ? this._sinceRefusal : null }
}
// The partial CAR, or null when we cannot prove one is complete. Never throws: a coverage failure is a
// normal answer here (the caller serves the full repo instead), not an error, and turning it into one
// would take down a getRepo that has a perfectly good full answer available.
//
// We pass the head's CID — derived from the commit VERIFIED against the DID key — and let repo-diff
// LOCATE those bytes in the firehose rather than handing them over. That is deliberately stricter: a
// CAR rooted at a commit the firehose does not carry cannot be walked by anyone, so a head we cannot
// find is a refusal instead of a block we synthesise. It also keeps a dag-cbor dependency out of this
// package, which has none and is a symlinked sibling.
async _repoDiffSince (since) {
this._sinceRefusal = null
if (!this.seq) { this._sinceRefusal = 'no firehose replicated for this DID'; return null }
try {
const r = await repoDiffSince({
getFrame: (i) => this.seq.core.get(i),
length: this.seq.core.length,
since,
headCommitCid: commitCid(this.head.signed),
headRev: this.head.rev
})
if (!r || !r.ok) { this._sinceRefusal = (r && r.reason) || 'could not assemble a diff'; return null }
return {
carBytes: r.carBytes,
commitCid: commitCid(this.head.signed),
dataCid: this.head.signed.data,
partial: true,
sinceHonoured: true,
since,
fromRev: r.fromRev,
toRev: r.toRev,
empty: !!r.empty,
blocks: r.blocks
}
} catch (e) {
this._sinceRefusal = 'diff assembly failed: ' + ((e && e.message) || String(e))
return null
}
}
// The record set for a peer with NO repo core, derived from the CAR getRepo() already fetches and
// VERIFIES. Cached against the head rev, so N reads at one head cost ONE derive and the cache
// invalidates the moment a new commit lands — no staleness, no per-read CAR fetch.
//
// Without this, getRecord/listRecords/collections returned null/[] whenever this.repo was null. That
// was tolerable when the only repo-less mode was noRepo (favela, which reads records from the shared
// store instead), but it silently broke com.atproto.repo.getRecord the moment a peer went STATELESS —
// observed in production as RecordNotFound for a profile that was present and correctly served by
// getRepo. The trust story is unchanged: these read out of the same verified CAR.
async _derivedRecords () {
if (!this.head) return []
// Derive ONLY when it can actually work. A front with no repo core and no RPC (one still mid-open,
// or a plain seeder) previously returned an empty list from these readers; making it THROW instead
// changed observable behaviour for every such caller — it broke the co-owned freshest-by-rev test,
// where a poller reading a not-yet-open front started erroring instead of reporting "nothing yet".
if (!this.rpc) return []
if (this._derivedAt === this.head.rev && this._derived) return this._derived
const { carBytes } = await this.getRepo()
const car = await readCar(carBytes)
// listRecords already yields the record CID (the MST value), so nothing is recomputed and no
// dag-cbor dependency is pulled in here — the CAR is the single source for key, cid and value.
this._derived = carListRecords(car).map((r) => ({ key: r.key, value: r.value, cid: r.cid.toString() }))
this._derivedAt = this.head.rev
return this._derived
}
// com.atproto.repo.getRecord
async getRecord (collection, rkey) {
if (this.repo) return rvGetRecord(this.repo, collection, rkey, { did: this.did })
if (this._noRepo) return null // records live in the consumer's own store, by design
const key = collection + '/' + rkey
const hit = (await this._derivedRecords()).find((r) => r.key === key)
return hit ? { uri: `at://${this.did}/${key}`, cid: hit.cid, value: hit.value } : null
}
// com.atproto.repo.listRecords
// Returns { records, cursor } per the lexicon. `limit`/`reverse`/`cursor` were dropped on the floor
// here as well as in read-view, so a caller's paging parameters never reached anything that honoured
// them. Both derived and repo-backed paths now page identically — a client must not get different
// semantics depending on which internal path answered.
async listRecords (collection, { limit = 50, reverse = false, cursor = null } = {}) {
if (this.repo) return rvListRecords(this.repo, collection, { did: this.did, limit, reverse, cursor })
if (this._noRepo) return { records: [], cursor: undefined }
const n = Math.min(100, Math.max(1, Number(limit) || 50))
const prefix = collection + '/'
const all = (await this._derivedRecords())
.filter((r) => r.key.startsWith(prefix))
.map((r) => ({ rkey: r.key.slice(prefix.length), uri: `at://${this.did}/${r.key}`, cid: r.cid, value: r.value }))
if (reverse) all.reverse()
let start = 0
if (cursor) {
const i = all.findIndex((r) => r.rkey === cursor)
start = i >= 0 ? i + 1 : all.length
}
const page = all.slice(start, start + n)
const more = start + page.length < all.length
const last = page.length ? page[page.length - 1].rkey : null
return { records: page.map(({ uri, cid, value }) => ({ uri, cid, value })), cursor: more && last ? last : undefined }
}
// com.atproto.sync.getLatestCommit -> { cid, rev } from the verified head (keyless).
getLatestCommit () {
if (!this.head) return null
return { cid: this.head.cid, rev: this.head.rev }
}
// com.atproto.sync.subscribeRepos -> the peer's own signed firehose frames (cursor=0 backfills).
subscribeRepos (cursor = 0, opts) {
if (!this.seq) throw new Error(`DidPeer: cores not open for ${this.did}`)
return this.seq.subscribe(cursor, opts)
}
// com.atproto.sync.getRepoStatus -> a fronted peer we replicate + verified is active.
getRepoStatus () { return { did: this.did, active: true, status: 'active' } }
// com.atproto.repo.describeRepo -> { handle, did, didDoc, collections, handleIsCorrect }. Keyless:
// did/handle come from the announce, collections from the replicated repo, didDoc a best-effort PLC
// fetch (null on failure). doFetch: (url, opts) -> { status, body:Buffer } (the server's wrapper shape).
async describeRepo ({ doFetch } = {}) {
const collections = await this.collections()
let didDoc = null
try {
if (doFetch) {
const r = await doFetch(`https://plc.directory/${this.did}`, { method: 'GET' })
if (r && r.status === 200 && r.body) didDoc = JSON.parse(r.body.toString())
}
} catch {}
return { handle: this.handle || null, did: this.did, didDoc, collections, handleIsCorrect: !!this.handle }
}
// ---- write / session / blob relay (peer-signs-over-RPC) — the PEER holds the key + applies + signs;
// this consumer never holds a key. Requires the router to run with `{ rpc: true }` (this.rpc set). The
// signed result flows back on the peer's firehose (which this DidPeer already replicates + reconciles).
applyWrite (op) { return this._call('apply-write', op) }
applyWrites (op) { return this._call('apply-writes', op) }
createSession ({ identifier, password } = {}) { return this._call('create-session', { identifier, password }) }
refreshSession (refreshJwt) { return this._call('refresh-session', { refreshJwt }) }
getServiceAuth ({ aud, lxm, token } = {}) { return this._call('get-service-auth', { aud, lxm, token }, { readOnly: true }) }
getPreferences ({ token } = {}) { return this._call('get-preferences', { token }, { readOnly: true }) }
putPreferences ({ preferences, token } = {}) { return this._call('put-preferences', { preferences, token }) }
getBlob (cid) { return this._call('get-blob', { cid }, { readOnly: true }) }
// The peer's published blob inventory (com.atproto.sync.listBlobs). Read-only, same gate as getBlob.
// Returns { cids, cursor }. A peer too old to answer replies not-supported rather than erroring, so a
// mixed-version fleet degrades to "orphans unchecked" instead of failing the whole call.
listBlobs ({ limit, cursor } = {}) { return this._call('list-blobs', { limit, cursor }, { readOnly: true }) }
uploadBlob ({ bytes, mimeType, token } = {}) { return this._call('upload-blob', { bytes, mimeType, token }, { timeoutMs: DidPeer.BULK_RPC_TIMEOUT_MS }) }
// Decode a #commit frame BODY (from firehose.decodeFrame) into normalized ops with the changed record
// INLINE — read from the frame's OWN CAR (body.blocks) keyed by op.cid, so it is race-free (does NOT
// depend on the repo core having caught up). Returns [{ action, collection, rkey, record? }]; delete
// ops omit record; create/update carry the decoded record (null if the block was absent). This keeps
// the atproto CAR codec in the P2P-read layer — consumers (favela) get plain ops, no atproto-repo dep.
static async commitOps (body) {
if (!body || !Array.isArray(body.ops)) return []
const needsCar = body.ops.some(o => (o.action === 'create' || o.action === 'update') && o.cid)
let car = null
if (needsCar && body.blocks) { try { car = await readCar(body.blocks) } catch { car = null } }
const out = []
for (const op of body.ops) {
const path = op.path || ''
const i = path.indexOf('/')
if (i <= 0) continue
const collection = path.slice(0, i)
const rkey = path.slice(i + 1)
if (op.action === 'delete') { out.push({ action: 'delete', collection, rkey }); continue }
if (op.action === 'create' || op.action === 'update') {
let record = null
if (op.cid && car) { try { record = decodeBlock(car, op.cid) } catch { record = null } }
out.push({ action: op.action, collection, rkey, record })
}
}
return out
}
// Distinct collection NSIDs present in the replicated repo (keys are "<collection>/<rkey>").
async collections () {
const set = new Set()
const rows = this.repo ? await this.repo.records() : (this._noRepo ? [] : await this._derivedRecords())
for (const r of rows) {
const i = r.key.indexOf('/')
if (i > 0) set.add(r.key.slice(0, i))
}
return [...set]
}
status () {
return {
did: this.did,
hasCores: !!this.repo,
repoLength: this.repo ? this.repo.core.length : 0,
firehoseLength: this.seq ? this.seq.core.length : 0,
head: this.head ? { rev: this.head.rev, cid: this.head.cid } : null,
channels: this._channels.size
}
}
async close () {
for (const chan of this._channels) { try { chan.close() } catch {} }
this._channels.clear()
if (this.seq) { try { await this.seq.close() } catch {} }
if (this.repo) { try { await this.repo.close() } catch {} }
if (this.labels) { try { await this.labels.close() } catch {} }
}
}
// Multi-DID, co-owned-aware router over ONE shared corestore + swarm — the CANONICAL consumer primitive
// (unifies atproto-bridge's PeerRouter/PeerFront + favela's earlier single-front DidPeerRouter, per
// EXTRACTION-PLAN Phase 1). Fixes the per-DidPeer attach() channel collision (Protomux allows one channel
// per (protocol,id) per mux) by opening ONE announce channel per connection and dispatching by DID.
//
// Beyond that, it absorbs PeerRouter's co-owned handling: a DID can have MANY self-custody holders (wj62 =
// notary + co-owner; each announces the SAME did with its OWN membrane core keys — distinct keys, identical
// deterministic content, differing only in replication lag). We track a SET of fronts per DID (one per
// firehoseKey) and serve reads from the FRESHEST by commit REV (not frame count — a lagging co-owner can
// out-FRAME a fresher one; the wj62 prod symptom). Optional layers, all off by default (favela's read-only
// posture is `new DidPeerRouter(store)`):
// swarm — router OWNS it: auto-wires onconnection + a pull-only refresh/failover timer. Omit to drive
// onconnection() yourself and join topics yourself (favela / tests).
// join — (did)=>void topic-join fn (bridge passes `d=>swarm.join(discoveryTopic(d),{...})`); keeps
// atproto-identity OUT of this module's runtime deps.
// fronted — DIDs to allow-list immediately (bridge). resolveVerifyKey — async (did)=>pubkey fallback.
// relay — FleetRelay: fold the FRESHEST co-owner's firehose per DID (single source) + failover + #sync.
// rpc — open ONE protomux-rpc per connection (the write/session/blob relay); the front borrows it.
// handles/onHandles — persist handle->did resolution across restarts.
export class DidPeerRouter {
constructor (corestore, {
swarm = null, fronted = [], relay = null, resolveVerifyKey = null, rpc = false, join = null,
refreshMs = 10000, refreshTimeoutMs = 5000, staleFrontMs = 300000, continuityScan = 500,
handles = {}, onHandles = null
} = {}) {
this.corestore = corestore
this.swarm = swarm
this.relay = relay
this._resolveVerifyKey = resolveVerifyKey
this._useRpc = !!rpc
this._join = join
this._refreshMs = refreshMs
this._refreshTimeoutMs = refreshTimeoutMs
// Age-out ceiling: a front whose peer has not announced for this long is dropped — but ONLY while a
// live, verified sibling remains to serve the DID (see _ageOutFronts). 0 disables age-out.
this._staleFrontMs = staleFrontMs
// How many frames back the continuity walk will look for the head we already serve. 0 disables the
// continuity gate entirely (freshest-by-rev only — the pre-0.2.8 behaviour).
this._continuityScan = continuityScan
this._peers = new Map() // did -> primary DidPeer (backward-compat add()/get())
this._fronts = new Map() // did -> Map<fhKeyHex, DidPeer> (co-owned multi-front)
this._byDid = new Map() // did -> Map<fhKeyHex, { coords, conns: Map<conn, rpc|null> }>
// MANY peers announce one (did, firehoseKey): the holder + every seeder
this._opts = new Map() // did -> { verifyKey, handle, onHead, onError } from add()
this._verifyKeys = new Map() // did -> Promise<pubkey>
this._fronted = new Set(fronted)
this._opening = new Map() // "did|fhKeyHex" -> in-flight open promise
this._waiters = new Map() // did -> Set<fn> woken when ANY co-owner of that did lands
this._lastAppend = new Map() // did -> ts of last firehose 'append' (event-driven liveness)
this._lastLive = new Map() // "did|fhKeyHex" -> ts of that co-owner's last announce (age-out input)
this._denied = new Map() // did -> Set<fhKeyHex> evicted by an operator; a re-announce must NOT resurrect
this._servedHead = new Map() // did -> { rev, cid, fhKeyHex } the head reads currently resolve to
this._continuity = new Map() // "candCid|servedCid" -> bool (does the candidate chain from what we serve?)
this._forkRejected = new Map() // "did|fhKeyHex" -> count of times this front was rejected as off-line
this._relaySourceKey = new Map() // did -> firehoseKeyHex currently folded into the relay
this._handleToDid = new Map(Object.entries(handles || {}))
this._onHandles = onHandles
this._channels = new Set()
if (this.swarm) {
this._onconnbound = (conn) => this.onconnection(conn)
this.swarm.on('connection', this._onconnbound)
}
for (const did of this._fronted) this._joinDid(did)
if (this._refreshMs > 0 && (this.swarm || this.relay)) {
this._refreshTimer = setInterval(() => { this._refreshFronts().catch(() => {}) }, this._refreshMs)
if (this._refreshTimer && this._refreshTimer.unref) this._refreshTimer.unref()
}
}
get (did) { return this._peers.get(did) || null }
peers () { return [...this._peers.values()] }
isFronted (did) { return !!did && this._fronted.has(did) }
list () { return [...this._fronted] }
resolveHandle (handle) { return this._handleToDid.get(handle) || null }
_isCared (did) { return this._fronted.has(did) || this._peers.has(did) || this._opts.has(did) }
_joinDid (did) { if (this._join) { try { this._join(did) } catch {} } }
// A label core opening is not a firehose event, so a consumer serving labels (the gateway) has no other
// way to learn that a front gained one — the common case is the labeler upgrading while we run, which
// the router learns of from an announce on an already-open front. → unsubscribe().
onLabels (fn) {
if (!this._labelListeners) this._labelListeners = new Set()
this._labelListeners.add(fn)
return () => this._labelListeners.delete(fn)
}
_tellLabels (did, peer) {
if (!this._labelListeners) return
for (const fn of this._labelListeners) { try { fn(did, peer) } catch {} }
}
_wake (did) { const ws = this._waiters.get(did); if (ws) { this._waiters.delete(did); for (const fn of ws) fn() } }
_setHandle (handle, did) {
if (!handle || !did || this._handleToDid.get(handle) === did) return
this._handleToDid.set(handle, did)
if (this._onHandles) { try { this._onHandles(Object.fromEntries(this._handleToDid)) } catch {} }
}
// Cached DID->verifyKey. Per-DID add({verifyKey}) wins; else the injected resolver (PLC).
_verifyKey (did) {
const o = this._opts.get(did) || {}
if (o.verifyKey) return Promise.resolve(o.verifyKey)
if (!this._verifyKeys.has(did)) {
if (!this._resolveVerifyKey) return Promise.reject(new Error(`DidPeerRouter: no verifyKey for ${did} (pass add(did,{verifyKey}) or {resolveVerifyKey})`))
this._verifyKeys.set(did, Promise.resolve(this._resolveVerifyKey(did)))
}
return this._verifyKeys.get(did)
}
// Register a DID. With a verifyKey (favela) it eagerly creates + returns the primary DidPeer (its cores
// load from the announce); without one (bridge allow-set) it just allow-lists + returns `this` (chainable,
// verifyKey resolved lazily at front time). Idempotent.
add (did, opts = {}) {
this._fronted.add(did)
if (opts.verifyKey || opts.onHead || opts.onError || opts.handle || opts.noRepo) {
this._opts.set(did, { ...(this._opts.get(did) || {}), ...opts })
}
this._joinDid(did)
const existing = this._peers.get(did)
if (existing) return existing
const o = this._opts.get(did) || {}
if (!o.verifyKey) return this // bridge allow-set add (verifyKey lazy); return value unused/chainable
const p = new DidPeer(this.corestore, { did, verifyKey: o.verifyKey, handle: o.handle || null, onHead: o.onHead || null, onError: o.onError || null, noRepo: o.noRepo })
this._peers.set(did, p)
for (const chan of this._channels) p.announceVia(chan._announce) // (re-add) announce if already held
return p
}
// Wire ONE swarm connection: replicate the corestore ONCE, open ONE announce channel (+ one write RPC
// when `{ rpc: true }`), and route each received announce by DID -> (did, firehoseKey) tracking.
onconnection (conn) {
try { this.corestore.replicate(conn) } catch {}
const mux = Protomux.from(conn)
let rpc = null
if (this._useRpc) {
try { rpc = new ProtomuxRPC(conn, { protocol: RPC_PROTOCOL, valueEncoding: c.json }); rpc.on('error', () => {}) } catch {}
}
const chan = mux.createChannel({
protocol: ANNOUNCE_PROTOCOL,
onopen: () => this._announceHeld(chan),
onclose: () => this._channels.delete(chan)
})
if (!chan) return // already open on this mux
chan._announce = chan.addMessage({
encoding: c.json,
onmessage: (a) => { this._onAnnounce(a, rpc, conn).catch(() => {}) }
})
chan.open()
this._channels.add(chan)
this._announceHeld(chan)
conn.on('close', () => this._onConnClose(conn))
}
// Learn a (did, firehoseKey) co-owner from a PUSHED announce; open its front (best-effort). A DIFFERENT
// firehoseKey under the same DID is a DISTINCT co-owner (tracked, never evicting the other).
async _onAnnounce (a, rpc, conn) {
// repoKey is OPTIONAL — same contract as DidPeer.onAnnounce. A peer running the stateless virtual
// device (serve-membrane projectRepoCore:false) publishes no membrane-repo core and announces
// { firehoseKey, repoKey: null }; its repo is DERIVED over the get-repo-car RPC. Requiring repoKey
// HERE silently dropped that announce before it ever reached the DidPeer, so the peer went live:false
// and was never fronted — the fix in DidPeer.onAnnounce alone was not enough, because the router
// gates first. The firehose key is the only thing genuinely required: it is the trust anchor.
if (!a || !a.did || !a.firehoseKey) return
const did = a.did
const fhKeyHex = a.firehoseKey
if (a.handle) this._setHandle(a.handle, did)
if (!this._isCared(did)) return
// An operator-EVICTED co-owner must stay evicted. Without this, dropFront({forget:true}) is undone by
// the offender's very next announce (peers re-announce on connect AND on every append), which is the
// difference between an eviction an operator can rely on and one that lasts milliseconds.
if (this._isDenied(did, fhKeyHex)) return
this._lastLive.set(did + '|' + fhKeyHex, Date.now())
if (!this._byDid.has(did)) this._byDid.set(did, new Map())
// MANY peers announce the SAME (did, firehoseKey): the key holder AND every keyless seeder. Keep them
// ALL — a single {conn, rpc} slot was overwritten by whoever announced last, so the front's application
// RPC could end up pointed at a seeder that answers UNKNOWN_METHOD for every method.
const entries = this._byDid.get(did)
if (!entries.has(fhKeyHex)) entries.set(fhKeyHex, { coords: a, conns: new Map() })
const entry = entries.get(fhKeyHex)
entry.coords = a
entry.conns.set(conn, rpc || null)
const existing = this._fronts.get(did) && this._fronts.get(did).get(fhKeyHex)
if (existing) { // refresh write rpc + handle; nudge fetch (cores re-replicate via corestore regardless)
if (this._useRpc && rpc) existing.addRpc(rpc)
if (a.handle) existing.handle = a.handle
// The peer went STATELESS while we still hold a repo-core replica (it used to project, now it
// announces repoKey:null). That replica is frozen at whatever it last replicated, so once the head
// advances getRepoFromCommit's data-root assertion fails — safe, but permanently broken. Drop it so
// getRepo takes the derive-over-RPC path instead. The firehose (the trust anchor) is untouched.
if (!a.repoKey && existing.repo) {
const stale = existing.repo
existing.repo = null
existing._repoKey = null
try { await stale.close() } catch {}
} else if (a.repoKey && !existing.repo) {
// ...and the reverse: the peer resumed projecting, so take the core back. Without this the drop
// above is one-way and a rolled-back flip never heals.
existing.adoptRepoKey(a.repoKey).catch(() => {})
}
if (a.labelKey && !existing.labels) existing.adoptLabelKey(a.labelKey).catch(() => {})
if (existing.seq) existing.seq.core.update().catch(() => {})
this._wake(did)
return
}
this._wake(did)
// Add THIS announcer's rpc once the front exists. An announce that arrives while the front is still
// OPENING takes the dedup path (it awaits the in-flight promise), and that open captured only the
// candidates present when it started — so without this the second announcer's rpc is silently dropped
// and the candidate set stays stuck at one, which is the single-slot bug wearing a different hat.
this._ensureFront(did, fhKeyHex)
.then((f) => { if (f && this._useRpc && rpc) f.addRpc(rpc) })
.catch(() => {})
}
// Open (once, dedup-guarded) the front for ONE co-owner (did, firehoseKey). Reuses the primary DidPeer
// (from add()) for its FIRST co-owner; a co-owner with a different firehoseKey gets its own DidPeer.
async _ensureFront (did, fhKeyHex) {
if (this._isDenied(did, fhKeyHex)) return null // operator-evicted; re-front only after allowFront()
const set = this._fronts.get(did)
if (set && set.has(fhKeyHex)) return set.get(fhKeyHex)
const openKey = did + '|' + fhKeyHex
const inflight = this._opening.get(openKey)
if (inflight) return inflight
const entry = this._byDid.get(did) && this._byDid.get(did).get(fhKeyHex)
if (!entry) return null
const p = (async () => {
const verifyKey = await this._verifyKey(did)
const o = this._opts.get(did) || {}
let peer = this._peers.get(did)
// Reuse the primary only while it is UNBOUND. Test BINDING (_firehoseKey), not `holds`: `holds`
// (=> !!seq) flips true only once _openCores COMPLETES, whereas _firehoseKey is assigned
// synchronously inside provide(). A second co-owner announcing while the primary's FIRST provide()
// was still in flight therefore read holds=false, reused the primary, and its provide() returned the
// first co-owner's in-flight open — so _fronts held two keys ALIASING ONE DidPeer bound to one
// co-owner. The co-owner SET silently collapsed to a single front, freshest-by-rev had nothing to
// choose between, and which co-owner got served was a coin flip (co-owned.test.mjs flaked on exactly
// that; diagnose() counted map ENTRIES so it still reported 2). There is no await between this check
// and provide() below, so reading _firehoseKey here is race-free.
if (peer && peer._firehoseKey) {
const boundFh = b4a.toString(peer._firehoseKey, 'hex')
if (boundFh !== fhKeyHex) peer = null // primary already bound to a different co-owner -> new front
}
if (!peer) peer = new DidPeer(this.corestore, { did, verifyKey, handle: entry.coords.handle || o.handle || null, onHead: o.onHead || null, onError: o.onError || null, noRepo: o.noRepo })
if (entry.coords.handle) peer.handle = entry.coords.handle
// EVERY peer announcing this (did, firehoseKey) is a candidate; _call tries them until one answers.
if (this._useRpc) for (const r of entry.conns.values()) if (r) peer.addRpc(r)
peer._onLabels = (p) => this._tellLabels(did, p)
await peer.provide({ repoKey: entry.coords.repoKey, firehoseKey: entry.coords.firehoseKey, labelKey: entry.coords.labelKey || null })
// Re-sync: more peers may have announced this same (did, firehoseKey) during the open above.
if (this._useRpc) for (const r of entry.conns.values()) if (r) peer.addRpc(r)
if (!this._fronts.has(did)) this._fronts.set(did, new Map())
this._fronts.get(did).set(fhKeyHex, peer)
if (!this._peers.has(did)) this._peers.set(did, peer) // first front becomes the primary (get())
if (peer.seq && peer.seq.core && !peer._appendWired) {
peer._appendWired = true
this._lastAppend.set(did, Date.now())
peer.seq.core.on('append', () => { this._lastAppend.set(did, Date.now()) })
}
if (peer.handle) this._setHandle(peer.handle, did)
this._wake(did)
this._pickRelaySource(did).catch(() => {})
return peer
})().catch(() => null)
this._opening.set(openKey, p)
try { return await p } finally { this._opening.delete(openKey) }
}
// The freshest VERIFIED front for a DID = the co-owner whose head has the highest REV (a head is only set
// after its commit verifies against the DID key). Prefer a CURRENTLY-announced (live) front; fall back to
// any verified front if none are announced (all co-owners momentarily disconnected). Revs are TID strings
// (lexicographic == chronological). Never regresses to a lagging co-owner once a fresher one is open.
_freshestFront (did) {
const set = this._fronts.get(did)
if (!set || !set.size) return null
const live = this._byDid.get(did) || new Map()
const scored = []
for (const [fhKeyHex, f] of set) {
if (!f || !f.head || !f.head.rev) continue // only verified fronts
scored.push({ f, rev: f.head.rev, live: live.has(fhKeyHex) })
}
if (!scored.length) return null
const pickMaxRev = (arr) => arr.reduce((b, x) => (!b || x.rev > b.rev ? x : b), null)
const liveOnes = scored.filter((x) => x.live)
const chosen = pickMaxRev(liveOnes) || pickMaxRev(scored)
return chosen ? chosen.f : null
}
// ---- continuity ---------------------------------------------------------------------------------
// freshest-by-rev alone trusts a rev with NO continuity check, so a FORKED line whose rev sorts higher
// simply wins. That is how loser-club's public getLatestCommit served a rival 1-record head
// (3mrnruepxss22) over the authoritative 139-record one (3mrfgnyiurk22) until the bridge was bounced: a
// rev REGRESSION was guarded against, a higher rev on a DIFFERENT LINE was not.
//
// Real co-owners are not a different line. `atproto-store/lib/tid.js:nextRev` makes the chain a pure
// function of the previous rev, so two co-owners continuing from the same head over the same finalized
// record set produce BYTE-IDENTICAL commits — a lagging co-owner's firehose is a strict PREFIX of a
// fresher one's. So "chains from the head we serve" is decidable: walk the candidate's own frames back
// and look for the exact commit we are serving.
//
// Enforced only while the INCUMBENT IS STILL LIVE. A peer that legitimately rebuilds its cores (storage
// repair, rebuild-did) goes away and comes back announcing new keys on a new line, and refusing that
// would strand the DID on dead cores — the case test/peer-router.test.js 'REBUILDS its cores' covers.
// Live incumbent + a rival line => fork, keep serving the incumbent. No incumbent (gone, or evicted by
// an operator) => the new line is the only truth on offer, take it.
async _chainsFromServed (front, served) {
if (!front || !front.head) return false
if (front.head.rev === served.rev) return front.head.cid === served.cid // same rev MUST be same commit
if (front.head.rev < served.rev) return false // regression (pre-existing policy)
const ck = front.head.cid + '|' + served.cid
if (this._continuity.has(ck)) return this._continuity.get(ck)
const verdict = await this._scanForServed(front, served) // 'same' | 'fork' | 'unknown'
if (verdict !== 'unknown') this._continuity.set(ck, verdict === 'same') // never cache an inconclusive read
return verdict === 'same'
}
// Walk the candidate's firehose backwards looking for the served commit. Bounded by continuityScan.
// Reads are wait:false — a block we do not hold yet is INCONCLUSIVE, not evidence of a fork (the verdict
// is left uncached so the next read re-decides once replication catches up).
async _scanForServed (front, served) {
const core = front.seq && front.seq.core
if (!core || !core.length) return 'unknown'
const limit = Math.min(core.length, this._continuityScan)
for (let i = 1; i <= limit; i++) {
let body = null
try {
const raw = await core.get(core.length - i, { wait: false })
if (!raw) return 'unknown' // not replicated locally — cannot conclude
const d = decodeCommitFrame(raw)
body = d && d.body
} catch { return 'unknown' }
if (!body || !body.rev) continue // #sync/#identity/#account frames carry no commit
if (body.rev === served.rev) return (body.commit && body.commit.toString() === served.cid) ? 'same' : 'fork'
if (body.rev < served.rev) return 'fork' // walked PAST it without meeting it -> different line
}
return limit < core.length ? 'unknown' : 'fork' // exhausted the window vs exhausted the core
}
// The freshest front that also CHAINS from what we already serve. Returns the incumbent when every
// fresher candidate is on a rival line. Records what it picks as the new incumbent.
async _freshestChained (did) {
const set = this._fronts.get(did)
if (!set || !set.size) return null
if (!this._continuityScan) return this._freshestFront(did) // gate disabled
const live = this._byDid.get(did) || new Map()
const scored = []
for (const [fhKeyHex, f] of set) {
if (!f || !f.head || !f.head.rev) continue
scored.push({ fhKeyHex, f, rev: f.head.rev, live: live.has(fhKeyHex) })
}
if (!scored.length) return null
const pool = scored.some((x) => x.live) ? scored.filter((x) => x.live) : scored
pool.sort((a, b) => (a.rev < b.rev ? 1 : a.rev > b.rev ? -1 : 0)) // highest rev first
const served = this._servedHead.get(did)
const incumbentLive = !!(served && live.has(served.fhKeyHex) && set.has(served.fhKeyHex))
for (const cand of pool) {
if (served && incumbentLive && cand.fhKeyHex !== served.fhKeyHex) {
if (!(await this._chainsFromServed(cand.f, served))) {
const k = did + '|' + cand.fhKeyHex
this._forkRejected.set(k, (this._forkRejected.get(k) || 0) + 1)
continue
}
}
this._servedHead.set(did, { rev: cand.rev, cid: cand.f.head.cid, fhKeyHex: cand.fhKeyHex })
return cand.f
}
return (served && set.get(served.fhKeyHex)) || null // every rival was a fork: hold the line
}
// Get the FRESHEST verified front for a fronted DID: ensure fronts for every identified co-owner, then
// return the highest-rev one that chains from what we serve. This is what every read
// (getRepo/listRecords/…) resolves against — a lagging co-owner never freezes the served head.
async front (did, { timeout } = {}) {
if (!this._isCared(did)) throw new Error(`DidPeerRouter: ${did} is not a fronted DID`)
this._joinDid(did)
if (!this._byDid.has(did) || this._byDid.get(did).size === 0) {
// Every co-owner is momentarily away. _freshestFront already documents a fallback to any VERIFIED
// front for exactly this case, but it was unreachable from here: _waitForDid blocked first, so a read
// during a blip stalled for the full timeout and then threw, even though a verified head was cached.
// (It is also what makes age-out's "never drop the last front" worth anything.) Cores keep
// replicating over the next connection, so the cached head is the freshest thing that exists.
const cached = await this._freshestChained(did)
if (cached) return cached
await this._waitForDid(did, timeout || 20000)
}
const entries = this._byDid.get(did) || new Map()
await Promise.all([...entries.keys()].map((fh) => this._ensureFront(did, fh).catch(() => null)))
// A front is opened once provide() kicks off replication, but its head is only set once the firehose
// commit REPLICATES + verifies (event-driven, on 'append'). For a single front() call (no retry — a
// real getRepo request), wait for AT LEAST ONE co-owner to reach a verified head, then serve the
// freshest among those ready. A slow co-owner never delays serving from a ready one (freshest-of-ready).
let best = await this._freshestChained(did)
if (!best) {
const end = Date.now() + (timeout || 20000)
while (!best && Date.now() < end) {
await new Promise((r) => setTimeout(r, 100))
best = await this._freshestChained(did)
}
}
if (!best) throw new Error(`DidPeerRouter: no frontable co-owner for ${did}`)
return best
}
// Resolve that AT LEAST ONE co-owner of a DID is identified — immediately if any is connected, else wait.
_waitForDid (did, timeout) {
const e = this._byDid.get(did)
if (e && e.size > 0) return Promise.resolve(e)
return new Promise((resolve, reject) => {
const fn = () => {
clearTimeout(timer)
const e2 = this._byDid.get(did)
e2 && e2.size > 0 ? resolve(e2) : reject(new Error(`DidPeerRouter: ${did} identified but entry missing`))
}
const timer = setTimeout(() => {
const ws = this._waiters.get(did); if (ws) ws.delete(fn)
reject(new Error(`DidPeerRouter: timeout finding ${did}`))
}, timeout)
if (!this._waiters.has(did)) this._waiters.set(did, new Set())
this._waiters.get(did).add(fn)
})
}
// A peer disconnected: drop its (did, firehoseKey) identified-connection entries so _onAnnounce re-runs
// on reconnect. Do NOT close the cached front — its cores keep replicating over any remaining/next conn.
// (Keeping it is right for a transient reconnect and WRONG forever: see _ageOutFronts + dropFront.)
_onConnClose (conn) {
for (const [did, entries] of this._byDid) {
// Drop only THIS connection's candidate. A (did, firehoseKey) announced by several peers stays
// identified while any of them is still connected — one seeder leaving must not un-identify the DID.
for (const [fhKeyHex, entry] of entries) {
entry.conns.delete(conn)
if (entry.conns.size === 0) entries.delete(fhKeyHex)
}
if (entries.size === 0) this._byDid.delete(did)
}
}
// ---- eviction ----------------------------------------------------------------------------------
// A cached front used to live for the life of the PROCESS: _onConnClose deliberately keeps it, and
// _ensureFront returns the existing entry on every later announce. So the only way to stop serving a
// wrong head (a peer that stopped announcing, or one announcing a FORKED line) was to restart the
// bridge — unacceptable for a production read surface. These are the surgical alternatives.
_isDenied (did, fhKeyHex) { const s = this._denied.get(did); return !!(s && s.has(fhKeyHex)) }
// Evict one co-owner's front (or every front of a DID with no firehoseKey). `forget: true` also
// DENIES that firehose key, so the peer's next announce cannot resurrect it — required whenever the
// reason for evicting is the content the front is serving, because the offender keeps announcing.
// Never denies when dropping ALL of a DID's fronts implicitly (that is forgetDid's job, and denying a
// whole DID would make it permanently unservable).
async dropFront (did, { firehoseKey = null, forget = false } = {}) {
const set = this._fronts.get(did)
const targets = firehoseKey ? [firehoseKey] : (set ? [...set.keys()] : [])
const dropped = []
for (const fhKeyHex of targets) {
const f = set && set.get(fhKeyHex)
if (set) set.delete(fhKeyHex)
const entries = this._byDid.get(did)
if (entries) { entries.delete(fhKeyHex); if (entries.size === 0) this._byDid.delete(did) }
this._lastLive.delete(did + '|' + fhKeyHex)
if (forget && firehoseKey) {
if (!this._denied.has(did)) this._denied.set(did, new Set())
this._denied.get(did).add(fhKeyHex)
}
// Stop folding an evicted front into the aggregate relay before it can emit another frame.
if (this._relaySourceKey.get(did) === fhKeyHex) {
this._relaySourceKey.delete(did)
if (this.relay) { try { this.relay.removeSource(fhKeyHex) } catch {} }
}
// Re-point the primary (get()/announce-held) at a surviving front before closing this one.
if (this._peers.get(did) === f) {
const survivor = set && set.size ? [...set.values()][0] : null
if (survivor) this._peers.set(did, survivor); else this._peers.delete(did)
}
// If the evicted front was the INCUMBENT, clear it: the operator's drop is an explicit decision that
// this line is wrong, so a surviving rival must be allowed to take over instead of being held off by
// the continuity gate as a "fork" of the head we just threw away.
const served = this._servedHead.get(did)
if (served && served.fhKeyHex === fhKeyHex) this._servedHead.delete(did)
this._forkRejected.delete(did + '|' + fhKeyHex)
if (f) { dropped.push(fhKeyHex); try { await f.close() } catch {} }
}
if (set && set.size === 0) this._fronts.delete(did)
// A surviving co-owner must take over the relay slot immediately, not at the next refresh tick.
this._pickRelaySource(did).catch(() => {})
return { did, dropped, forgotten: forget && firehoseKey ? [firehoseKey] : [] }
}
// Un-deny a previously forgotten front (omit firehoseKey to clear every denial for the DID) so the
// peer's next announce re-fronts it. An eviction with no undo is an outage waiting to happen.
allowFront (did, firehoseKey = null) {
const s = this._denied.get(did)
if (!s) return { did, allowed: [] }
const allowed = firehoseKey ? (s.delete(firehoseKey) ? [firehoseKey] : []) : [...s]
if (!firehoseKey) s.clear()
if (s.size === 0) this._denied.delete(did)
return { did, allowed }
}
// Forget EVERYTHING cached for a DID — every co-owner front, its identified connections and its
// resolved verify key — so the next request re-fronts from scratch. The DID stays allow-listed (it must
// still be servable) and nothing is denied: this is the "re-derive, don't restart" lever. To evict a
// specific misbehaving co-owner and keep it out, use dropFront({ firehoseKey, forget: true }).
async forgetDid (did) {
const r = await this.dropFront(did)
this._byDid.delete(did)
this._verifyKeys.delete(did)
this._lastAppend.delete(did)
this._relaySourceKey.delete(did)
this._servedHead.delete(did) // no incumbent => the continuity gate starts clean on the next front()
return { did, dropped: r.dropped }
}
// Age out fronts whose peer has stopped announcing. Conservative by construction: a front is only
// dropped while a LIVE, VERIFIED sibling remains to serve the DID, so age-out can never be the reason a
// DID becomes unservable — the single-holder case (the overwhelmingly common one) is never touched.
async _ageOutFronts (did) {
if (!this._staleFrontMs) return []
const set = this._fronts.get(did)
if (!set || set.size < 2) return []
const live = this._byDid.get(did) || new Map()
const hasLiveVerified = [...set].some(([k, f]) => live.has(k) && f && f.head)
if (!hasLiveVerified) return []
const now = Date.now()
const stale = [...set.keys()].filter((k) => !live.has(k) && now - (this._lastLive.get(did + '|' + k) || now) > this._staleFrontMs)
const dropped = []
for (const k of stale) {
if (this._fronts.get(did) && this._fronts.get(did).size < 2) break // never drop the last front
const r = await this.dropFront(did, { firehoseKey: k })
dropped.push(...r.dropped)
}
return dropped
}
// Fold the FRESHEST co-owner's firehose into the aggregate relay (single source per DID — two co-owners
// are the SAME frames under different sourceIds, so folding both duplicates every #commit). On a source
// change (fresher co-owner won, or the folded one stalled) swap removeSource->addSource + emit #sync.
async _pickRelaySource (did) {
if (!this.relay) return
// The relay must fold the SAME line the read surface serves — folding a fork's frames would publish it
// to every downstream subscriber, which is worse than serving it on one read.
const best = await this._freshestChained(did)
if (!best || !best.seq || !best.seq.core) return
const key = b4a.toString(best.seq.core.key, 'hex')
if (this._relaySourceKey.get(did) === key) return
const old = this._relaySourceKey.get(did)
this._relaySourceKey.set(did, key)
;(async () => {
if (old) { try { this.relay.removeSource(old) } catch {} }
try { await this.relay.addSource(key, best.seq.core) } catch {}
try { await this._emitSync(did, best) } catch {}
})().catch(() => {})
}
// `force`: the relay dedupes frames by key, and an automatic #sync is keyed by (did, rev) so a re-anchor
// never re-emits the same head. An OPERATOR re-announce must not be deduplicated: on 2026-10-13 bsky's
// relay held loser-club at a newer July rev and dropped every commit; the fix is one #sync at our head,
// and the (did, rev) key — already used by a #sync on 2026-07-26 — made emit-sync a silent no-op. So a
// forced emission's key carries its time: each operator request is its own announcement.
async _emitSync (did, front, { force = false } = {}) {
if (!this.relay) return { did, emitted: false, reason: 'no relay' }
const signed = front.head && front.head.signed
if (!signed) return { did, emitted: false, reason: 'no commit' }
const time = new Date().toISOString()
const frameBytes = await syncFrame({ seq: 0, did, signedCommit: signed, time })
const key = force ? `sync:${did}:${signed.rev}:${time}` : `sync:${did}:${signed.rev}`
const emitted = await this.relay.emit(key, frameBytes)
return { did, rev: signed.rev, emitted, forced: !!force }
}
// Public control-plane trigger (the operator's `emit-sync`): FORCE a #sync re-baseline for a fronted DID
// at its freshest front, even if that head was #synced before.
async emitSync (did) { const f = await this.front(did); return this._emitSync(did, f, { force: true }) }
// Pull-only staleness ceiling + relay failover (owns-swarm/relay mode). NEVER destroys a connection —
// cores live-follow via the standing download; this just nudges a front that saw no 'append' for a window
// and re-picks the relay source if the folded co-owner stalled while another advanced.
async _refreshFronts () {
for (const did of [...this._fronts.keys()]) { // snapshot: _ageOutFronts mutates _fronts
const set = this._fronts.get(did)
if (!set) continue
for (const [, f] of set) {
if (!f || !f.seq || !f.seq.core) continue
if (Date.now() - (this._lastAppend.get(did) || 0) < this._refreshMs) continue
try { await withTimeout(f.seq.core.update({ wait: true }), this._refreshTimeoutMs) } catch {}
}
try { await this._ageOutFronts(did) } catch {}
this._pickRelaySource(did).catch(() => {})
}
}
diagnose () {
const fronts = []
const now = Date.now()
for (const [did, set] of this._fronts) {
for (const [fhKeyHex, f] of set) {
const live = (this._byDid.get(did) || new Map()).has(fhKeyHex)
fronts.push({
did,
handle: f.handle || null,
repoKey: f._repoKey ? b4a.toString(f._repoKey, 'hex') : null,
firehoseKey: fhKeyHex,
firehoseLen: (f.seq && f.seq.core) ? f.seq.core.length : null,
rev: f.head ? f.head.rev : null,
rpcConnected: !!(f.rpc && !f.rpc.closed),
relaySource: this._relaySourceKey.get(did) === fhKeyHex,
live,
// How long this co-owner has been silent, and whether age-out would take it — so an operator can
// SEE an evictable front instead of inferring one.
silentMs: live ? 0 : now - (this._lastLive.get(did + '|' + fhKeyHex) || now),
// Is this the head reads resolve to, and has it been refused as a rival line? A front sitting at
// a HIGHER rev than the served one with forkRejected > 0 is the loser-club signature.
served: (this._servedHead.get(did) || {}).fhKeyHex === fhKeyHex,
forkRejected: this._forkRejected.get(did + '|' + fhKeyHex) || 0
})
}
}
const denied = {}
for (const [did, s] of this._denied) denied[did] = [...s]
return {
fronted: [...this._fronted],
cachedFronts: fronts.length,
// Distinct front OBJECTS. cachedFronts counts map entries, which is why an aliased set (one DidPeer
// filed under N keys) read as healthy right up to 0.2.6.
distinctFronts: new Set([].concat(...[...this._fronts.values()].map((s) => [...s.values()]))).size,
fronts,
denied,
staleFrontMs: this._staleFrontMs,
// did -> ms timestamp of the last firehose append seen from any co-owner: the liveness signal the
// refresh timer already uses, now visible to an operator (the gateway's metrics).
lastAppendAt: Object.fromEntries(this._lastAppend),
relay: this.relay ? { sources: this.relay.sourceCount ? this.relay.sourceCount() : null } : null
}
}
// Announce every HELD DID on a channel (symmetric seeding: tell the peer which cores we can serve).
_announceHeld (chan) {
for (const p of this._peers.values()) { if (p && p.holds) p.announceVia(chan._announce) }
}
async close () {
if (this._refreshTimer) { try { clearInterval(this._refreshTimer) } catch {} this._refreshTimer = null }
if (this.swarm && this._onconnbound) { try { this.swarm.off('connection', this._onconnbound) } catch {} }
for (const chan of this._channels) { try { chan.close() } catch {} }
this._channels.clear()
const closed = new Set()
for (const set of this._fronts.values()) { for (const f of set.values()) { if (closed.has(f)) continue; closed.add(f); try { await f.close() } catch {} } }
for (const p of this._peers.values()) { if (closed.has(p)) continue; closed.add(p); try { await p.close() } catch {} }
this._fronts.clear(); this._peers.clear(); this._byDid.clear()
}
}
export { ANNOUNCE_PROTOCOL }