| … | … | @@ -6,6 +6,14 @@ |
| 6 | 6 | // AUTHENTICATES it against the DID's signing key (resolved from the DID doc) before serving — so |
| 7 | 7 | // the bridge holds no key and can't be tricked into fronting a squatter. `isFronted(did)` lets |
| 8 | 8 | // server-bare route each request to the right home. |
| 9 | // |
| 10 | // CO-OWNED DIDs (multiple self-custody peers of the SAME did — e.g. wj62 = stortron notary + tabertron |
| 11 | // co-owner): every co-owner runs serve-membrane, so every co-owner announces the SAME did with its OWN |
| 12 | // membrane core keys (each device projects the shared, deterministic __commit__ log into its own named |
| 13 | // firehose/repo cores → distinct keys, IDENTICAL content, differing only in replication lag). We therefore |
| 14 | // front ALL of a did's co-owners (a SET of fronts per did) and serve each read from the FRESHEST |
| 15 | // (max firehose length). This replaces the old "front whichever announced first and stick" model, which |
| 16 | // could lock onto a lagging/stale co-owner and serve a frozen head until a bridge restart. |
| 9 | 17 | |
| 10 | 18 | import b4a from 'b4a' |
| 11 | 19 | import c from 'compact-encoding' |
| … | … | @@ -37,8 +45,12 @@ export class PeerRouter { |
| 37 | 45 | this.swarm = swarm |
| 38 | 46 | this.corestore = corestore |
| 39 | 47 | this._resolveVerifyKey = resolveVerifyKey // async (did) -> secp256k1 pubkey |
| 48 | this._verifyKeys = new Map() // did -> Promise<pubkey> (resolve once, reuse for every co-owner front) |
| 40 | 49 | this._fronted = new Set(fronted) // DIDs this bridge fronts (vs local Pds accounts) |
| 41 | | this._fronts = new Map() // did -> PeerFront (verified, cached) |
| 50 | // did -> Map<firehoseKeyHex, PeerFront>. A SET of fronts per did (one per co-owner membrane). Reads |
| 51 | // pick the freshest; the map is keyed by firehose core key so each co-owner folds/serves distinctly. |
| 52 | this._fronts = new Map() |
| 53 | this._opening = new Map() // "did|fhKeyHex" -> in-flight open Promise (concurrent callers AWAIT it) |
| 42 | 54 | // handle -> did. PERSISTED (loaded from `handles`, saved via onHandles) so a bridge restart does NOT |
| 43 | 55 | // drop handle resolution — an OFFLINE peer (served only by the blind-relay, e.g. ashatron) cannot |
| 44 | 56 | // re-teach its handle at boot, so an in-memory-only map left it unresolvable (com.atproto.identity. |
| … | … | @@ -46,158 +58,170 @@ export class PeerRouter { |
| 46 | 58 | this._handleToDid = new Map(Object.entries(handles || {})) |
| 47 | 59 | this._onHandles = onHandles // (obj) => void — persist the handle map when it grows |
| 48 | 60 | this.relay = relay // optional FleetRelay: aggregates fronted peers' firehoses |
| 49 | | // Stall-backstop config (see _refreshFronts): the front's cores are REMOTE replicas kept live by a |
| 50 | | // STANDING download (PeerFront.fromEntry) that flips `downloading:true`, so `corestore.replicate` |
| 51 | | // re-attaches them direction-free on every reconnect and the download actively pulls. This timer is |
| 52 | | // ONLY a pull-only staleness ceiling — a bounded update({wait}) nudge when a front has seen no 'append' |
| 53 | | // for a window. It does NOT destroy connections: that earlier watchdog fought hyperswarm's own 5s |
| 54 | | // keepalive + topic-rejoin reconnect (and force-destroying an inbound conn is worse — no auto-requeue), |
| 55 | | // and it masked the real bug (the missing standing download). Set refreshMs<=0 to disable (tests). |
| 61 | this._relaySourceKey = new Map() // did -> firehoseKeyHex currently folded into the relay (single source per did — avoids duplicate frames) |
| 56 | 62 | this._refreshMs = refreshMs |
| 57 | 63 | this._refreshTimeoutMs = refreshTimeoutMs |
| 58 | 64 | this._staleStrikeLimit = staleStrikeLimit // retained for back-compat; no longer drives a destroy |
| 59 | | this._staleStrikes = new Map() // did -> consecutive failed nudges (diagnose() visibility only) |
| 60 | | this._lastAppend = new Map() // did -> ts of the firehose's last 'append' (event-driven liveness) |
| 65 | this._staleStrikes = new Map() // "did|fhKeyHex" -> consecutive failed nudges (diagnose() visibility only) |
| 66 | this._lastAppend = new Map() // did -> ts of ANY co-owner firehose's last 'append' (event-driven liveness) |
| 61 | 67 | |
| 62 | | // The membrane is ONE edge fronting MANY DIDs (rooms) over ONE swarm — the keet model: keet keeps |
| 63 | | // one rootCorestore + one swarm, joins each room as a topic, and on EVERY connection does exactly |
| 64 | | // one rootCorestore.replicate(conn), letting hypercore mux all rooms' cores over that stream |
| 65 | | // (keet-core index.js:1239). We mirror that: on each connection we (1) replicate the corestore |
| 66 | | // (all fronted DIDs' cores mux over it) AND (2) establish exactly ONE ProtomuxRPC and learn which |
| 67 | | // DID this connection serves from the peer's PUSHED announce, caching did -> { conn, rpc, coords }. |
| 68 | | // front() is then a cache lookup. Probing per-front for coords would open duplicate RPC channels on |
| 69 | | // shared streams (DUPLICATE_CHANNEL, a boot crash-loop); one-rpc-per-connection + push can't. |
| 70 | | this._byDid = new Map() // did -> { conn, rpc, coords } (identified connection) |
| 68 | // ONE edge fronting MANY DIDs over ONE swarm (keet model): on EVERY connection, replicate the |
| 69 | // corestore (all cores mux over it) + open ONE ProtomuxRPC and learn which did+cores this connection |
| 70 | // serves from the peer's PUSHED announce. did -> Map<firehoseKeyHex, { conn, rpc, coords }>: multiple |
| 71 | // co-owner connections per did coexist (was one). |
| 72 | this._byDid = new Map() |
| 71 | 73 | this._identified = new WeakSet() // connections we've already opened an identify rpc on |
| 72 | | this._waiters = new Map() // did -> Set<fn> woken when that DID's connection lands |
| 74 | this._waiters = new Map() // did -> Set<fn> woken when ANY co-owner of that did lands |
| 73 | 75 | |
| 74 | 76 | this._onconnection = (conn) => { |
| 75 | 77 | this.corestore.replicate(conn) // keet pattern: mux ALL fronted DIDs' cores over this stream |
| 76 | | this._identify(conn) // + one rpc, learn DID from the announce, cache by DID |
| 78 | this._identify(conn) // + one rpc, learn did+cores from the announce, track by (did, firehoseKey) |
| 77 | 79 | const rk = conn.remotePublicKey ? b4a.toString(conn.remotePublicKey, 'hex') : null |
| 78 | 80 | console.log(JSON.stringify({ msg: 'swarm connection', peer: rk, connections: this.swarm.connections ? this.swarm.connections.size : null })) |
| 79 | 81 | conn.on('close', () => { this._onConnClose(conn); console.log(JSON.stringify({ msg: 'swarm disconnection', peer: rk, connections: this.swarm.connections ? this.swarm.connections.size : null })) }) |
| 80 | 82 | } |
| 81 | 83 | this.swarm.on('connection', this._onconnection) |
| 82 | 84 | |
| 83 | | // Join every fronted DID's topic up front (like keet activateRoom joins each room's topic on the |
| 84 | | // shared swarm) so peers connect + get identified in the background, before any request arrives. |
| 85 | 85 | for (const did of this._fronted) this._joinDid(did) |
| 86 | 86 | |
| 87 | | // Start the freshness watchdog (0/negative refreshMs disables it — used by tests that drive |
| 88 | | // refresh manually). unref() so it never keeps the process alive on its own. |
| 89 | 87 | if (this._refreshMs > 0) { |
| 90 | 88 | this._refreshTimer = setInterval(() => { this._refreshFronts().catch(() => {}) }, this._refreshMs) |
| 91 | 89 | if (this._refreshTimer && this._refreshTimer.unref) this._refreshTimer.unref() |
| 92 | 90 | } |
| 93 | 91 | } |
| 94 | 92 | |
| 95 | | // Keep every cached front live-following its peer over a possibly-degraded link. For each front: |
| 96 | | // force `seq.core.update({wait:true})` (a round-trip pull that recovers an idle-but-alive |
| 97 | | // connection where passive replication has gone quiet). If that repeatedly fails (a zombie |
| 98 | | // connection: swarm counts it, but the stream is dead), destroy the borrowed connection so |
| 99 | | // hyperswarm re-establishes a FRESH one — `_onConnClose` drops the `_byDid` entry, the reconnect |
| 100 | | // re-runs `_identify` (refreshing the front's rpc/conn) and `corestore.replicate(newConn)` |
| 101 | | // re-attaches the cores → passive live-follow resumes. A fresh connection is empirically the only |
| 102 | | // thing that recovers a stalled front (offline repro: a clean reconnect self-heals; prod over a |
| 103 | | // flaky link does not, until re-connected). Best-effort + fully guarded — never throws. |
| 104 | | async _refreshFronts () { |
| 105 | | for (const [did, f] of this._fronts) { |
| 93 | // Cached DID→verifyKey (resolved once, reused when opening a front for each co-owner of the did). |
| 94 | _verifyKey (did) { |
| 95 | if (!this._verifyKeys.has(did)) this._verifyKeys.set(did, Promise.resolve(this._resolveVerifyKey(did))) |
| 96 | return this._verifyKeys.get(did) |
| 97 | } |
| 98 | |
| 99 | // The freshest OPEN front for a did = the co-owner membrane with the longest replicated firehose. |
| 100 | // PREFER a front whose (did, firehoseKey) is CURRENTLY announced (live in _byDid): a peer that REBUILT |
| 101 | // its cores (or a stale membrane a daemon abandoned) leaves a frozen front behind whose OLD key is no |
| 102 | // longer announced — and it can tie the new core on raw length (both 1 commit), so a pure length pick |
| 103 | // would keep serving the dead one. Among live co-owners take the longest firehose; fall back to any open |
| 104 | // front only if none are currently announced (all co-owners momentarily disconnected → serve last-known). |
| 105 | _freshestFront (did) { |
| 106 | const set = this._fronts.get(did) |
| 107 | if (!set || !set.size) return null |
| 108 | const live = this._byDid.get(did) || new Map() |
| 109 | let best = null; let bestLive = null |
| 110 | for (const [fhKeyHex, f] of set) { |
| 106 | 111 | if (!f || !f.seq || !f.seq.core) continue |
| 107 | | // EVENT-INFORMED: if the firehose appended within the last window the front is demonstrably live |
| 108 | | // (PeerFront's standing per-core download is delivering) — skip the nudge entirely. |
| 109 | | if (Date.now() - (this._lastAppend.get(did) || 0) < this._refreshMs) { this._staleStrikes.set(did, 0); continue } |
| 110 | | // PULL-ONLY stall backstop (staleness ceiling): a bounded update({wait}) that pulls from whatever peer |
| 111 | | // is currently attached. It NEVER destroys the connection. The old strike-then-`f._conn.destroy()` |
| 112 | | // escalation was an anti-pattern: it fought hyperswarm's own 5s keepalive + topic-rejoin reconnect |
| 113 | | // (and force-destroying an INBOUND connection is strictly worse — hyperswarm does not auto-requeue |
| 114 | | // those), AND it masked the real bug — the front had no standing download, so `corestore.replicate` |
| 115 | | // never re-attached the core to a reconnected stream (`_shouldReplicate` needs `downloading:true`). |
| 116 | | // PeerFront.fromEntry's standing download now makes recovery direction-free + event-driven; this |
| 117 | | // backstop only nudges a quiet-but-alive link. Strikes are tracked for `diagnose()` visibility only. |
| 118 | | try { await withTimeout(f.seq.core.update({ wait: true }), this._refreshTimeoutMs); this._staleStrikes.set(did, 0) } |
| 119 | | catch { this._staleStrikes.set(did, (this._staleStrikes.get(did) || 0) + 1) } |
| 112 | if (!best || f.seq.core.length > best.seq.core.length) best = f |
| 113 | if (live.has(fhKeyHex) && (!bestLive || f.seq.core.length > bestLive.seq.core.length)) bestLive = f |
| 114 | } |
| 115 | return bestLive || best |
| 116 | } |
| 117 | |
| 118 | // Open (once) the front for ONE co-owner (did, firehoseKey) from its identified connection. Idempotent |
| 119 | // + dedup-guarded. Best-effort: a co-owner that doesn't verify/lacks a commit yet is simply skipped — |
| 120 | // the OTHER co-owners still front. Wakes did waiters + (re)picks the relay source on success. |
| 121 | async _ensureFront (did, fhKeyHex) { |
| 122 | const set = this._fronts.get(did) |
| 123 | if (set && set.has(fhKeyHex)) return set.get(fhKeyHex) |
| 124 | const openKey = did + '|' + fhKeyHex |
| 125 | const inflight = this._opening.get(openKey) |
| 126 | if (inflight) return inflight // AWAIT the in-progress open (a concurrent front()/learn started it) |
| 127 | const entry = this._byDid.get(did) && this._byDid.get(did).get(fhKeyHex) |
| 128 | if (!entry) return null |
| 129 | const p = (async () => { |
| 130 | const verifyKey = await this._verifyKey(did) |
| 131 | const f = await PeerFront.fromEntry(this.corestore, this.swarm, entry, did, { verifyKey }) |
| 132 | if (!this._fronts.has(did)) this._fronts.set(did, new Map()) |
| 133 | this._fronts.get(did).set(fhKeyHex, f) |
| 134 | if (f.seq && f.seq.core && !f._appendWired) { |
| 135 | f._appendWired = true |
| 136 | this._lastAppend.set(did, Date.now()) |
| 137 | f.seq.core.on('append', () => { this._lastAppend.set(did, Date.now()) }) |
| 138 | } |
| 139 | if (f.handle) this._setHandle(f.handle, did) |
| 140 | const ws = this._waiters.get(did); if (ws) { this._waiters.delete(did); for (const fn of ws) fn() } |
| 141 | this._pickRelaySource(did) // fold the freshest into the aggregate (single source per did) |
| 142 | return f |
| 143 | })().catch(() => null) // a co-owner that doesn't verify / has no commit yet just yields null; others still front |
| 144 | this._opening.set(openKey, p) |
| 145 | try { return await p } finally { this._opening.delete(openKey) } |
| 146 | } |
| 147 | |
| 148 | // Keep every cached front live-following its peer. Cores are REMOTE replicas kept live by the standing |
| 149 | // download (PeerFront.fromEntry); this timer is a pull-only staleness ceiling — a bounded update({wait}) |
| 150 | // nudge when a front has seen no 'append' for a window. It NEVER destroys connections. It ALSO re-picks |
| 151 | // the relay source (failover): if the folded co-owner stalled while another advanced, switch to the fresher. |
| 152 | async _refreshFronts () { |
| 153 | for (const [did, set] of this._fronts) { |
| 154 | for (const [fhKeyHex, f] of set) { |
| 155 | if (!f || !f.seq || !f.seq.core) continue |
| 156 | if (Date.now() - (this._lastAppend.get(did) || 0) < this._refreshMs) { this._staleStrikes.set(did + '|' + fhKeyHex, 0); continue } |
| 157 | try { await withTimeout(f.seq.core.update({ wait: true }), this._refreshTimeoutMs); this._staleStrikes.set(did + '|' + fhKeyHex, 0) } |
| 158 | catch { this._staleStrikes.set(did + '|' + fhKeyHex, (this._staleStrikes.get(did + '|' + fhKeyHex) || 0) + 1) } |
| 159 | } |
| 160 | this._pickRelaySource(did) // failover: ensure the relay is folding the freshest co-owner |
| 120 | 161 | } |
| 121 | 162 | } |
| 122 | 163 | |
| 123 | | // Symmetric join: server:true is load-bearing for firewalled↔firewalled holepunch (a one-sided |
| 124 | | // member never punches outbound). Idempotent — hyperswarm returns the existing discovery session. |
| 125 | 164 | _joinDid (did) { |
| 126 | 165 | try { this.swarm.join(discoveryTopic(did), { server: true, client: true }); this.swarm.flush().catch(() => {}) } catch {} |
| 127 | 166 | } |
| 128 | 167 | |
| 129 | | // Open exactly ONE ProtomuxRPC per connection (REUSED for every later call to that peer — |
| 130 | | // apply-write, sessions, get-service-auth, get-blob — so there is never a second channel on the |
| 131 | | // stream) and learn the DID it serves from the peer's PUSHED announce. No get-coords round-trip. |
| 168 | // Open exactly ONE ProtomuxRPC per connection and learn the did+cores it serves from the peer's PUSHED |
| 169 | // announce. A co-owner is tracked by (did, firehoseKey); a NEW announce for the same (did, firehoseKey) |
| 170 | // just refreshes conn/rpc, a NEW firehoseKey under the same did is a DISTINCT co-owner (tracked, not evicting). |
| 132 | 171 | _identify (conn) { |
| 133 | 172 | if (this._identified.has(conn)) return |
| 134 | 173 | this._identified.add(conn) |
| 135 | | // Learn the DID→cores mapping from the pushed announce, cache it, and resolve waiters. Idempotent |
| 136 | | // per conn. On RECONNECT (front already cached) refresh its write rpc + conn (its repo/firehose |
| 137 | | // cores re-replicate via corestore, so reads live-follow regardless). |
| 138 | 174 | const learn = (coords, rpcForConn) => { |
| 139 | 175 | if (!coords || !coords.did || !coords.repoKey || !coords.firehoseKey) return |
| 140 | | if (coords.handle) this._setHandle(coords.handle, coords.did) // learn+persist on EVERY announce (not just first front) — survives restart + offline (blind-relayed) peers |
| 141 | | // DURABILITY (no restart on a peer core rebuild): a peer can re-announce NEW repoKey/firehoseKey under |
| 142 | | // the SAME DID — e.g. after a storage repair that rebuilt its membrane cores. A cached PeerFront is |
| 143 | | // bound to the DEAD old keys and would serve a FROZEN repo forever (previously only a process restart |
| 144 | | // fixed it). Detect the key change here and EVICT the stale front (+ drop its aggregate relay source) |
| 145 | | // so front() re-opens the NEW cores on the next request. Keys compared as hex (announce carries hex). |
| 146 | | const stale = this._fronts.get(coords.did) |
| 147 | | if (stale && (b4a.toString(stale.repoKey, 'hex') !== coords.repoKey || b4a.toString(stale.firehoseKey, 'hex') !== coords.firehoseKey)) { |
| 148 | | this._fronts.delete(coords.did) |
| 149 | | if (this.relay) { try { this.relay.removeSource(b4a.toString(stale.firehoseKey, 'hex')) } catch {} } |
| 150 | | try { stale.close() } catch {} |
| 151 | | } |
| 152 | | const cur = this._byDid.get(coords.did) |
| 153 | | // Skip only when NOTHING changed (same connection AND same cores). If the cores changed we must fall |
| 154 | | // through to refresh _byDid.coords, else front() would rebuild against the old announced keys. |
| 155 | | const unchanged = cur && cur.conn === conn && cur.coords && cur.coords.repoKey === coords.repoKey && cur.coords.firehoseKey === coords.firehoseKey |
| 156 | | if (unchanged) return |
| 157 | | conn._frontDid = coords.did |
| 158 | | this._byDid.set(coords.did, { conn, rpc: rpcForConn, coords }) |
| 159 | | const f = this._fronts.get(coords.did) |
| 176 | if (coords.handle) this._setHandle(coords.handle, coords.did) // learn+persist on EVERY announce |
| 177 | const did = coords.did |
| 178 | const fhKeyHex = coords.firehoseKey |
| 179 | if (!this._byDid.has(did)) this._byDid.set(did, new Map()) |
| 180 | const entries = this._byDid.get(did) |
| 181 | entries.set(fhKeyHex, { conn, rpc: rpcForConn, coords }) |
| 182 | conn._frontDid = did |
| 183 | conn._frontFhKey = fhKeyHex |
| 184 | // Refresh an already-open front's write rpc + conn (its cores re-replicate via corestore regardless). |
| 185 | const f = this._fronts.get(did) && this._fronts.get(did).get(fhKeyHex) |
| 160 | 186 | if (f) { f.rpc = rpcForConn; f._conn = conn } |
| 161 | | const ws = this._waiters.get(coords.did) |
| 162 | | if (ws) { this._waiters.delete(coords.did); for (const fn of ws) fn() } |
| 187 | // Wake anyone waiting for this did (a co-owner is now available), and start opening this front. |
| 188 | const ws = this._waiters.get(did); if (ws) { this._waiters.delete(did); for (const fn of ws) fn() } |
| 189 | this._ensureFront(did, fhKeyHex).catch(() => {}) |
| 163 | 190 | } |
| 164 | 191 | try { |
| 165 | 192 | const rpc = new ProtomuxRPC(conn, { protocol: RPC_PROTOCOL, valueEncoding: c.json }) |
| 166 | | rpc.on('error', () => {}) // a channel error must never become an uncaught exception |
| 167 | | // ANNOUNCE-PUSH discovery (atproto/did-announce/v1): the peer PUSHES its core keys on connect — |
| 168 | | // no round-trip, no serverKey directionality. This channel populates _byDid and is what front() |
| 169 | | // resolves against; announce-only fronting is proven for the real multi-DID deployment |
| 170 | | // (test/announce-discovery.test.js). The rpc above is kept for the write/session/blob relay. |
| 193 | rpc.on('error', () => {}) |
| 171 | 194 | const achan = Protomux.from(conn).createChannel({ protocol: ANNOUNCE_PROTOCOL }) |
| 172 | 195 | if (achan) { |
| 173 | 196 | achan.addMessage({ encoding: c.json, onmessage: (a) => { try { learn(a, rpc) } catch {} } }) |
| 174 | 197 | achan.open() |
| 175 | 198 | } |
| 176 | | } catch {} // not a membrane peer / no RPC surface — rpc dies with the connection |
| 199 | } catch {} |
| 177 | 200 | } |
| 178 | 201 | |
| 179 | | // A peer disconnected: drop only its identified-connection entry, so _identify re-runs when the |
| 180 | | // peer reconnects. Do NOT close the cached PeerFront — its repo/firehose are corestore cores that |
| 181 | | // keep replicating over the next connection (the persistent replicate handler), so a fronted |
| 182 | | // peer's reads LIVE-FOLLOW across a reconnect. The front's write RPC is refreshed on reconnect |
| 183 | | // (see _identify). Closing the front here broke live-follow-after-reconnect (peer-router test). |
| 202 | // A peer disconnected: drop only its (did, firehoseKey) identified-connection entry, so _identify re-runs |
| 203 | // when it reconnects. Do NOT close the cached PeerFront — its repo/firehose are corestore cores that keep |
| 204 | // replicating over any remaining/next connection (reads live-follow across a reconnect). Its write RPC is |
| 205 | // refreshed on reconnect (see _identify). Other co-owners' fronts are untouched. |
| 184 | 206 | _onConnClose (conn) { |
| 185 | 207 | const did = conn._frontDid |
| 186 | | if (!did) return |
| 187 | | const entry = this._byDid.get(did) |
| 188 | | if (entry && entry.conn === conn) this._byDid.delete(did) |
| 208 | const fhKeyHex = conn._frontFhKey |
| 209 | if (!did || !fhKeyHex) return |
| 210 | const entries = this._byDid.get(did) |
| 211 | if (!entries) return |
| 212 | const entry = entries.get(fhKeyHex) |
| 213 | if (entry && entry.conn === conn) entries.delete(fhKeyHex) |
| 189 | 214 | } |
| 190 | 215 | |
| 191 | | // Resolve the identified connection for a DID — immediately if already connected, else wait for it |
| 192 | | // to connect + get identified (up to timeout). This is what makes front() a lookup, not a probe. |
| 216 | // Resolve that AT LEAST ONE co-owner of a did is identified — immediately if any is connected, else wait. |
| 193 | 217 | _waitForDid (did, timeout) { |
| 194 | 218 | const e = this._byDid.get(did) |
| 195 | | if (e) return Promise.resolve(e) |
| 219 | if (e && e.size > 0) return Promise.resolve(e) |
| 196 | 220 | return new Promise((resolve, reject) => { |
| 197 | 221 | const fn = () => { |
| 198 | 222 | clearTimeout(timer) |
| 199 | 223 | const e2 = this._byDid.get(did) |
| 200 | | e2 ? resolve(e2) : reject(new Error(`PeerRouter: ${did} identified but entry missing`)) |
| 224 | e2 && e2.size > 0 ? resolve(e2) : reject(new Error(`PeerRouter: ${did} identified but entry missing`)) |
| 201 | 225 | } |
| 202 | 226 | const timer = setTimeout(() => { |
| 203 | 227 | const ws = this._waiters.get(did); if (ws) ws.delete(fn) |
| … | … | @@ -208,26 +232,27 @@ export class PeerRouter { |
| 208 | 232 | }) |
| 209 | 233 | } |
| 210 | 234 | |
| 211 | | // Introspection — a snapshot of what the membrane is actually doing (see it, don't guess). |
| 212 | | // All public keys + counts, no secrets. Exposed over the UDS control plane + a debug HTTP route. |
| 213 | 235 | diagnose () { |
| 214 | 236 | const conns = (this.swarm && this.swarm.connections) ? [...this.swarm.connections] : [] |
| 215 | 237 | const fronts = [] |
| 216 | | for (const [did, f] of this._fronts) { |
| 217 | | fronts.push({ |
| 218 | | did, |
| 219 | | handle: f.handle || null, |
| 220 | | repoKey: f.repoKey ? b4a.toString(f.repoKey, 'hex') : null, |
| 221 | | firehoseKey: f.firehoseKey ? b4a.toString(f.firehoseKey, 'hex') : null, |
| 222 | | firehoseLen: (f.seq && f.seq.core) ? f.seq.core.length : null, |
| 223 | | rpcConnected: !!(f.rpc && !f.rpc.closed), |
| 224 | | staleStrikes: this._staleStrikes.get(did) || 0 // watchdog: consecutive failed refreshes |
| 225 | | }) |
| 238 | for (const [did, set] of this._fronts) { |
| 239 | for (const [fhKeyHex, f] of set) { |
| 240 | fronts.push({ |
| 241 | did, |
| 242 | handle: f.handle || null, |
| 243 | repoKey: f.repoKey ? b4a.toString(f.repoKey, 'hex') : null, |
| 244 | firehoseKey: fhKeyHex, |
| 245 | firehoseLen: (f.seq && f.seq.core) ? f.seq.core.length : null, |
| 246 | rpcConnected: !!(f.rpc && !f.rpc.closed), |
| 247 | relaySource: this._relaySourceKey.get(did) === fhKeyHex, |
| 248 | staleStrikes: this._staleStrikes.get(did + '|' + fhKeyHex) || 0 |
| 249 | }) |
| 250 | } |
| 226 | 251 | } |
| 227 | 252 | return { |
| 228 | 253 | fronted: [...this._fronted], |
| 229 | | cachedFronts: this._fronts.size, |
| 230 | | fronts, // per-front state (firehoseLen/rpcConnected/staleStrikes) — was BUILT but not returned |
| 254 | cachedFronts: fronts.length, |
| 255 | fronts, |
| 231 | 256 | swarm: { |
| 232 | 257 | connections: conns.length, |
| 233 | 258 | peers: conns.map((c) => (c.remotePublicKey ? b4a.toString(c.remotePublicKey, 'hex') : null)) |
| … | … | @@ -237,11 +262,8 @@ export class PeerRouter { |
| 237 | 262 | } |
| 238 | 263 | |
| 239 | 264 | isFronted (did) { return !!did && this._fronted.has(did) } |
| 240 | | // handle -> did for fronted peers (so the bridge resolves e.g. p2p.7rnx.net -> wj62). |
| 241 | 265 | resolveHandle (handle) { return this._handleToDid.get(handle) || null } |
| 242 | 266 | |
| 243 | | // Learn (and PERSIST) a handle->did. Idempotent; only persists when the mapping actually changes, so a |
| 244 | | // steady stream of re-announces is cheap. Persistence is what survives a bridge restart for OFFLINE peers. |
| 245 | 267 | _setHandle (handle, did) { |
| 246 | 268 | if (!handle || !did || this._handleToDid.get(handle) === did) return |
| 247 | 269 | this._handleToDid.set(handle, did) |
| … | … | @@ -250,42 +272,39 @@ export class PeerRouter { |
| 250 | 272 | add (did) { this._fronted.add(did); this._joinDid(did); return this } |
| 251 | 273 | list () { return [...this._fronted] } |
| 252 | 274 | |
| 253 | | // Get (discovering + verifying on first use, then cached) the PeerFront for a fronted DID. |
| 275 | // Get the FRESHEST verified PeerFront for a fronted did. Opens fronts for every identified co-owner |
| 276 | // (idempotent), then returns the one with the longest firehose. This is what every read (getRepo, |
| 277 | // getLatestCommit, listRecords, …) resolves against — so a lagging co-owner never freezes the served head. |
| 254 | 278 | async front (did, { timeout } = {}) { |
| 255 | 279 | if (!this.isFronted(did)) throw new Error(`PeerRouter: ${did} is not a fronted DID`) |
| 256 | | const cached = this._fronts.get(did) |
| 257 | | if (cached) return cached |
| 258 | | const verifyKey = await this._resolveVerifyKey(did) |
| 259 | | this._joinDid(did) // ensure the topic is joined so the peer connects + gets identified |
| 260 | | const entry = await this._waitForDid(did, timeout || 20000) |
| 261 | | const f = await PeerFront.fromEntry(this.corestore, this.swarm, entry, did, { verifyKey }) |
| 262 | | this._fronts.set(did, f) |
| 263 | | // Event-driven liveness: a firehose 'append' (a block arriving via passive replication) PROVES the |
| 264 | | // front is live — the watchdog uses this to SKIP the update({wait:true}) poll for demonstrably-live |
| 265 | | // fronts (events set the floor; the timer only catches ABSENCE = a possible stall). Attach once. |
| 266 | | if (f.seq && f.seq.core && !f._appendWired) { |
| 267 | | f._appendWired = true |
| 268 | | this._lastAppend.set(did, Date.now()) |
| 269 | | f.seq.core.on('append', () => this._lastAppend.set(did, Date.now())) |
| 280 | this._joinDid(did) |
| 281 | if (!this._byDid.has(did) || this._byDid.get(did).size === 0) { |
| 282 | await this._waitForDid(did, timeout || 20000) |
| 270 | 283 | } |
| 271 | | if (f.handle) this._setHandle(f.handle, did) // learn+persist handle->did for resolveHandle |
| 284 | const entries = this._byDid.get(did) || new Map() |
| 285 | await Promise.all([...entries.keys()].map((fhKeyHex) => this._ensureFront(did, fhKeyHex).catch(() => null))) |
| 286 | const best = this._freshestFront(did) |
| 287 | if (!best) throw new Error(`PeerRouter: no frontable co-owner for ${did}`) |
| 288 | return best |
| 289 | } |
| 272 | 290 | |
| 273 | | // fold the peer's signed firehose into the aggregate so bsky's relay crawls it in the one |
| 274 | | // monotonic subscribeRepos stream (FleetSequencer re-stamps a global seq). |
| 275 | | if (this.relay) { |
| 276 | | // Source id = the firehose CORE key, NOT the did. FleetRelay dedups frames by (sourceId, index); |
| 277 | | // a peer's firehose core can CHANGE under the same did (custodial→self-custody, or single-tenant→ |
| 278 | | // swarm migration), and the new core's indices 0..n would collide with the old core's already- |
| 279 | | // aggregated `${did}:0..n` keys → the new core's frames get deduped away and never fold (only the |
| 280 | | // #sync re-baseline gets through). Keying by core identity lets each distinct firehose fold cleanly. |
| 281 | | await this.relay.addSource(b4a.toString(f.seq.core.key, 'hex'), f.seq.core) |
| 282 | | // Emit a #sync (Sync 1.1) so a relay re-baselines to this peer's current commit. This is |
| 283 | | // load-bearing for a peer whose commit chain is discontinuous vs the relay's last-known |
| 284 | | // state (e.g. a custodial→self-custody migration): without it the relay keeps expecting |
| 285 | | // deltas from the old chain and never ingests the new one. Idempotent per rev (deduped). |
| 286 | | try { await this._emitSync(did, f) } catch { /* best-effort; retriable via emit-sync RPC */ } |
| 287 | | } |
| 288 | | return f |
| 291 | // Fold the FRESHEST co-owner's firehose into the aggregate (single source per did — two co-owners are the |
| 292 | // SAME frames under different sourceIds, so folding both duplicates every #commit to bsky's relay). On a |
| 293 | // source change (a fresher co-owner won, or the folded one stalled) swap removeSource→addSource + emit a |
| 294 | // #sync so the relay re-baselines. Idempotent when the freshest is unchanged. |
| 295 | _pickRelaySource (did) { |
| 296 | if (!this.relay) return |
| 297 | const best = this._freshestFront(did) |
| 298 | if (!best || !best.seq || !best.seq.core) return |
| 299 | const key = b4a.toString(best.seq.core.key, 'hex') |
| 300 | if (this._relaySourceKey.get(did) === key) return |
| 301 | const old = this._relaySourceKey.get(did) |
| 302 | this._relaySourceKey.set(did, key) |
| 303 | ;(async () => { |
| 304 | if (old) { try { this.relay.removeSource(old) } catch {} } |
| 305 | try { await this.relay.addSource(key, best.seq.core) } catch {} |
| 306 | try { await this._emitSync(did, best) } catch {} |
| 307 | })().catch(() => {}) |
| 289 | 308 | } |
| 290 | 309 | |
| 291 | 310 | // Build + inject a #sync frame for a fronted peer's current signed commit (re-baseline signal). |
| … | … | @@ -299,7 +318,7 @@ export class PeerRouter { |
| 299 | 318 | return { did, rev: signed.rev, emitted } |
| 300 | 319 | } |
| 301 | 320 | |
| 302 | | // Public trigger (control plane): force a #sync re-baseline for a fronted DID. |
| 321 | // Public trigger (control plane): force a #sync re-baseline for a fronted DID (from its freshest front). |
| 303 | 322 | async emitSync (did) { |
| 304 | 323 | const f = await this.front(did) |
| 305 | 324 | return this._emitSync(did, f) |
| … | … | @@ -308,9 +327,9 @@ export class PeerRouter { |
| 308 | 327 | async close () { |
| 309 | 328 | if (this._refreshTimer) { try { clearInterval(this._refreshTimer) } catch {} this._refreshTimer = null } |
| 310 | 329 | if (this._onconnection) { try { this.swarm.off('connection', this._onconnection) } catch {} } |
| 311 | | for (const f of this._fronts.values()) { try { await f.close() } catch {} } |
| 330 | for (const set of this._fronts.values()) { for (const f of set.values()) { try { await f.close() } catch {} } } |
| 312 | 331 | this._fronts.clear() |
| 313 | | for (const { rpc } of this._byDid.values()) { try { await rpc.end() } catch {} } |
| 332 | for (const entries of this._byDid.values()) { for (const { rpc } of entries.values()) { try { await rpc.end() } catch {} } } |
| 314 | 333 | this._byDid.clear() |
| 315 | 334 | } |
| 316 | 335 | } |