| … | … | @@ -1,49 +0,0 @@ |
| 1 | // Front a live hyper-proto-peer from tabertron over the REAL DHT (not testnet). Given a DID + its | |
| 2 | // did:key, PeerFront.discover derives the DID topic, finds the peer on the public DHT, | |
| 3 | // replicates, and verifies the served repo signs against the DID key — then serves getRepo. | |
| 4 | // | |
| 5 | // node front-peer.mjs <did> <didKey> | |
| 6 | ||
| 7 | import Hyperswarm from 'hyperswarm' | |
| 8 | import { mkdtempSync, rmSync } from 'node:fs' | |
| 9 | import { tmpdir } from 'node:os' | |
| 10 | import { join } from 'node:path' | |
| 11 | import { openCorestore } from 'atproto-store/lib/corestore.js' | |
| 12 | import { readCar, loadCommit, listRecords } from 'atproto-repo/lib/repo.js' | |
| 13 | import { verifyCommit } from 'atproto-repo/lib/commit.js' | |
| 14 | import { decodeDidKey } from 'atproto-identity/lib/did-key.js' | |
| 15 | import { PeerFront } from './lib/peer-front.js' | |
| 16 | ||
| 17 | const did = process.argv[2] | |
| 18 | const didKey = process.argv[3] | |
| 19 | if (!did || !didKey) { console.error('usage: node front-peer.mjs <did> <didKey>'); process.exit(1) } | |
| 20 | ||
| 21 | const { pubkey } = decodeDidKey(didKey) | |
| 22 | const dir = mkdtempSync(join(tmpdir(), 'front-')) | |
| 23 | const store = openCorestore(dir) | |
| 24 | const swarm = new Hyperswarm() | |
| 25 | ||
| 26 | try { | |
| 27 | console.error(`[front] discovering ${did} over the real DHT ...`) | |
| 28 | const front = await PeerFront.discover(swarm, store, did, { verifyKey: pubkey, timeout: 30000 }) | |
| 29 | const { carBytes } = await front.getRepo() | |
| 30 | const car = await readCar(carBytes) | |
| 31 | const signed = loadCommit(car).value | |
| 32 | const recs = await listRecords(car) | |
| 33 | console.log(JSON.stringify({ | |
| 34 | fronted: true, | |
| 35 | did: front.did, | |
| 36 | handle: front.handle, | |
| 37 | serverKey: front.serverKey, | |
| 38 | carBytes: carBytes.length, | |
| 39 | verifiesAgainstDidKey: verifyCommit(signed, pubkey), | |
| 40 | records: recs.map(r => r.value.text) | |
| 41 | }, null, 2)) | |
| 42 | await front.close() | |
| 43 | } catch (e) { | |
| 44 | console.error('[front] FAILED:', e.message) | |
| 45 | process.exitCode = 1 | |
| 46 | } finally { | |
| 47 | await swarm.destroy() | |
| 48 | rmSync(dir, { recursive: true, force: true }) | |
| 49 | } |
| … | … | @@ -1,161 +0,0 @@ |
| 1 | // DidPeer — the corrected, direction-free membrane primitive (audited against keet + holepunch, | |
| 2 | // 2026-07-13). A DID's data (repo + firehose Hypercores) lives in a swarm keyed by discoveryTopic(did); | |
| 3 | // every participant — signing device, availability relay, read/HTTP bridge — is the SAME primitive. | |
| 4 | // It is NOT a "front the origin by serverKey" client: it replicates the DID's cores from WHOEVER has | |
| 5 | // them and serves the freshest commit that VERIFIES against the DID key. | |
| 6 | // | |
| 7 | // Corrections baked in (see memory/feedback_p2p-no-directionality-event-driven): | |
| 8 | // - ANNOUNCE, don't DERIVE + don't get-coords-PULL: on connect, a holder PUSHES {repoKey,firehoseKey, | |
| 9 | // lengths} over a protomux channel (the same one-mux-per-connection model keet uses). A consumer | |
| 10 | // learns the core keys from the announce — no request/response round-trip, no serverKey directionality. | |
| 11 | // (protomux-wakeup is the heavier idiomatic version with a capability handshake; this is the minimal | |
| 12 | // announce over the shared protomux — swap up to protomux-wakeup later for GC/length-hints for free.) | |
| 13 | // - EVENT-DRIVEN head reconciliation: the served head re-derives on the firehose core's 'append' event | |
| 14 | // (fires on local append AND on replicated download) — never a data poll. | |
| 15 | // - KEEP liveness timers elsewhere (transport keepalive, a stall backstop) — this primitive adds none; | |
| 16 | // it reacts to core/connection events. "Events set the latency floor; timers set the staleness ceiling." | |
| 17 | // | |
| 18 | // Auth is unchanged and mandatory: the address (DID topic) is unauthenticated in ANY discovery scheme; | |
| 19 | // trust is established by verifying the served repo's signed commit against the DID signing key. | |
| 20 | ||
| 21 | import b4a from 'b4a' | |
| 22 | import c from 'compact-encoding' | |
| 23 | import Protomux from 'protomux' | |
| 24 | import { RepoStore } from 'atproto-store/lib/repo-store.js' | |
| 25 | import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js' | |
| 26 | import { decodeCommitFrame } from 'atproto-store/lib/firehose.js' | |
| 27 | import { readCar, loadCommit } from 'atproto-repo/lib/repo.js' | |
| 28 | import { verifyCommit, commitCid } from 'atproto-repo/lib/commit.js' | |
| 29 | ||
| 30 | const ANNOUNCE_PROTOCOL = 'atproto/did-announce/v1' | |
| 31 | const toKey = (k) => (b4a.isBuffer(k) ? k : b4a.from(k, 'hex')) | |
| 32 | ||
| 33 | export class DidPeer { | |
| 34 | // corestore: a Corestore (cores are global by key). did: the DID. verifyKey: its signing pubkey | |
| 35 | // (Uint8Array) — REQUIRED to accept a served head. onHead: (head) => void, called on every | |
| 36 | // reconciliation that produced a NEW verified head. onError: optional (err) => void. | |
| 37 | constructor (corestore, { did, verifyKey, onHead = null, onError = null } = {}) { | |
| 38 | if (!corestore) throw new Error('DidPeer: corestore required') | |
| 39 | if (!did) throw new Error('DidPeer: did required') | |
| 40 | if (!verifyKey) throw new Error('DidPeer: verifyKey (DID signing pubkey) required — auth is mandatory') | |
| 41 | this.corestore = corestore | |
| 42 | this.did = did | |
| 43 | this.verifyKey = verifyKey | |
| 44 | this._onHead = onHead | |
| 45 | this._onError = onError | |
| 46 | this.repo = null | |
| 47 | this.seq = null | |
| 48 | this._repoKey = null | |
| 49 | this._firehoseKey = null | |
| 50 | this._channels = new Set() | |
| 51 | this.head = null | |
| 52 | this._opening = null | |
| 53 | } | |
| 54 | ||
| 55 | // This node HOLDS the cores (a signing device, a relay that already seeded, or a bootstrap). Open | |
| 56 | // them so we can serve + announce. Idempotent. | |
| 57 | async provide ({ repoKey, firehoseKey } = {}) { return this._openCores(repoKey, firehoseKey) } | |
| 58 | ||
| 59 | async _openCores (repoKey, firehoseKey) { | |
| 60 | if (this.repo) return | |
| 61 | if (this._opening) return this._opening | |
| 62 | this._opening = (async () => { | |
| 63 | this._repoKey = toKey(repoKey) | |
| 64 | this._firehoseKey = toKey(firehoseKey) | |
| 65 | this.repo = await new RepoStore(this.corestore, { key: this._repoKey }).ready() | |
| 66 | this.seq = await new FirehoseSequencer(this.corestore.get({ key: this._firehoseKey })).ready() | |
| 67 | // FULL seed so this node can serve the cores to others (download all + follow appends). | |
| 68 | const seed = (core) => { const pull = () => { try { core.download({ start: 0, end: -1 }) } catch {} }; pull(); core.on('append', pull) } | |
| 69 | seed(this.repo.core); seed(this.seq.core) | |
| 70 | // EVENT-DRIVEN: re-derive the served head on every firehose append (local or replicated), and | |
| 71 | // re-announce the new length to connected peers. No poll loop. | |
| 72 | this.seq.core.on('append', () => { this._reconcile(); this._announceAll() }) | |
| 73 | await this._reconcile() | |
| 74 | })() | |
| 75 | try { await this._opening } finally { this._opening = null } | |
| 76 | } | |
| 77 | ||
| 78 | // Wire a swarm connection: replicate the corestore + open the announce channel. Symmetric — both | |
| 79 | // sides call attach(); whoever HOLDS the cores announces, whoever lacks them learns from the announce. | |
| 80 | attach (conn) { | |
| 81 | try { this.corestore.replicate(conn) } catch (e) { this._fail(e) } | |
| 82 | const mux = Protomux.from(conn) | |
| 83 | const chan = mux.createChannel({ | |
| 84 | protocol: ANNOUNCE_PROTOCOL, | |
| 85 | onopen: () => { if (this.repo) this._announceTo(chan) }, // ANNOUNCE-PUSH on open | |
| 86 | onclose: () => this._channels.delete(chan) | |
| 87 | }) | |
| 88 | if (!chan) return // already open on this mux | |
| 89 | chan._announce = chan.addMessage({ | |
| 90 | encoding: c.json, | |
| 91 | onmessage: (a) => this._onannounce(a).catch((e) => this._fail(e)) | |
| 92 | }) | |
| 93 | chan.open() | |
| 94 | this._channels.add(chan) | |
| 95 | if (this.repo) this._announceTo(chan) | |
| 96 | } | |
| 97 | ||
| 98 | _announceTo (chan) { | |
| 99 | if (!this.repo || !chan._announce) return | |
| 100 | try { | |
| 101 | chan._announce.send({ | |
| 102 | did: this.did, | |
| 103 | repoKey: b4a.toString(this._repoKey, 'hex'), | |
| 104 | firehoseKey: b4a.toString(this._firehoseKey, 'hex'), | |
| 105 | repoLength: this.repo.core.length, | |
| 106 | firehoseLength: this.seq.core.length | |
| 107 | }) | |
| 108 | } catch (e) { this._fail(e) } | |
| 109 | } | |
| 110 | ||
| 111 | _announceAll () { for (const chan of this._channels) this._announceTo(chan) } | |
| 112 | ||
| 113 | async _onannounce (a) { | |
| 114 | if (!a || a.did !== this.did || !a.repoKey || !a.firehoseKey) return | |
| 115 | if (!this.repo) { | |
| 116 | await this._openCores(a.repoKey, a.firehoseKey) | |
| 117 | this._announceAll() // we can help seed now — tell peers we have them too (symmetric) | |
| 118 | } | |
| 119 | // The announce is a HINT (the peer says its firehose is at length N). Replication pushes the | |
| 120 | // blocks; we react to the 'append' event to reconcile. One-shot update() nudges the fetch — this | |
| 121 | // is event-triggered (on announce), NOT a poll loop. | |
| 122 | if (this.seq) this.seq.core.update().catch(() => {}) | |
| 123 | } | |
| 124 | ||
| 125 | // The served head = the freshest signed commit that VERIFIES against the DID key. Re-derived on | |
| 126 | // firehose 'append'. Rejects an unverifiable head (can't be tricked into serving a forged repo). | |
| 127 | async _reconcile () { | |
| 128 | if (!this.seq || this.seq.core.length === 0) return | |
| 129 | try { | |
| 130 | const { body } = decodeCommitFrame(await this.seq.core.get(this.seq.core.length - 1)) | |
| 131 | const signed = loadCommit(await readCar(body.blocks)).value | |
| 132 | if (!verifyCommit(signed, this.verifyKey)) return | |
| 133 | const rev = signed.rev | |
| 134 | if (this.head && this.head.rev === rev) return // no change | |
| 135 | this.head = { rev, cid: commitCid(signed).toString(), signed } | |
| 136 | if (this._onHead) this._onHead(this.head) | |
| 137 | } catch (e) { this._fail(e) } | |
| 138 | } | |
| 139 | ||
| 140 | _fail (e) { if (this._onError) this._onError(e) } | |
| 141 | ||
| 142 | status () { | |
| 143 | return { | |
| 144 | did: this.did, | |
| 145 | hasCores: !!this.repo, | |
| 146 | repoLength: this.repo ? this.repo.core.length : 0, | |
| 147 | firehoseLength: this.seq ? this.seq.core.length : 0, | |
| 148 | head: this.head ? { rev: this.head.rev, cid: this.head.cid } : null, | |
| 149 | channels: this._channels.size | |
| 150 | } | |
| 151 | } | |
| 152 | ||
| 153 | async close () { | |
| 154 | for (const chan of this._channels) { try { chan.close() } catch {} } | |
| 155 | this._channels.clear() | |
| 156 | if (this.seq) { try { await this.seq.close() } catch {} } | |
| 157 | if (this.repo) { try { await this.repo.close() } catch {} } | |
| 158 | } | |
| 159 | } | |
| 160 | ||
| 161 | export { ANNOUNCE_PROTOCOL } |
| … | … | @@ -1,233 +0,0 @@ |
| 1 | // PeerFront — the bridge fronting a REMOTE hyper-proto-peer, KEYLESSLY (peer/bridge split, step 3). | |
| 2 | // | |
| 3 | // A hyper-proto-peer (self-custody, often NAT'd) serves its repo + its own signed firehose over | |
| 4 | // Hyperswarm (peer.serve()). The bridge replicates those cores and projects the peer to bsky | |
| 5 | // over HTTP — WITHOUT ever holding the peer's key. The signature comes from the peer's | |
| 6 | // firehose (getRepoFromCommit assembles the CAR from replicated records + the firehose's | |
| 7 | // signed commit; the bridge never re-signs). This is the thinned bridge's read path — the | |
| 8 | // keyless replacement for the custodial Pds.getRepo/subscribeRepos. | |
| 9 | // | |
| 10 | // Rendezvous is the DID discovery topic (discoveryTopic(did)); a holder PUSHES its repoKey/ | |
| 11 | // firehoseKey in an announce (atproto/did-announce/v1) on connect. The front borrows the router's | |
| 12 | // connection (fromEntry) — no serverKey, no directionality. See lib/peer-router.js + lib/did-peer.js. | |
| 13 | ||
| 14 | import b4a from 'b4a' | |
| 15 | import { RepoStore } from 'atproto-store/lib/repo-store.js' | |
| 16 | import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js' | |
| 17 | import { getRepoFromCommit, getRecord as rvGetRecord, listRecords as rvListRecords } from 'atproto-store/lib/read-view.js' | |
| 18 | import { decodeCommitFrame } from 'atproto-store/lib/firehose.js' | |
| 19 | import { readCar, loadCommit } from 'atproto-repo/lib/repo.js' | |
| 20 | import { verifyCommit, commitCid } from 'atproto-repo/lib/commit.js' | |
| 21 | import { discoveryTopic } from 'atproto-identity/lib/discovery.js' | |
| 22 | ||
| 23 | export const RPC_PROTOCOL = 'hyper-proto-peer/rpc/v1' // must match hyper-proto-peer's serve RPC protocol | |
| 24 | const toKey = (k) => (typeof k === 'string' ? b4a.from(k, 'hex') : k) | |
| 25 | ||
| 26 | // Reject a promise if it doesn't settle within ms. Exported for PeerRouter — it bounds the | |
| 27 | // stale-front watchdog's update({wait:true}) poll (and any RPC request on a degraded connection). | |
| 28 | export function withTimeout (p, ms) { | |
| 29 | return new Promise((resolve, reject) => { | |
| 30 | const t = setTimeout(() => reject(new Error('rpc timeout')), ms) | |
| 31 | p.then((v) => { clearTimeout(t); resolve(v) }, (e) => { clearTimeout(t); reject(e) }) | |
| 32 | }) | |
| 33 | } | |
| 34 | ||
| 35 | export class PeerFront { | |
| 36 | constructor (corestore, swarm, { repoKey, firehoseKey, did, handle } = {}) { | |
| 37 | if (!repoKey || !firehoseKey) throw new Error('PeerFront: repoKey, firehoseKey required') | |
| 38 | this.corestore = corestore | |
| 39 | this.swarm = swarm | |
| 40 | // No serverKey: discovery is ANNOUNCE-PUSH (atproto/did-announce/v1). A front is built via | |
| 41 | // fromEntry, borrowing the router's ONE connection, and needs only repoKey/firehoseKey — no | |
| 42 | // serverKey directionality, no connect-by-serverKey path. See lib/did-peer.js. | |
| 43 | this.repoKey = toKey(repoKey) | |
| 44 | this.firehoseKey = toKey(firehoseKey) | |
| 45 | this.did = did | |
| 46 | this.handle = handle | |
| 47 | this.repo = null | |
| 48 | this.seq = null | |
| 49 | this._conn = null | |
| 50 | this._ownsConn = false // fromEntry borrows the router's conn+rpc (the only construction path) | |
| 51 | } | |
| 52 | ||
| 53 | // Build a PeerFront from a connection the ROUTER already identified. The router keeps exactly ONE | |
| 54 | // ProtomuxRPC per connection (opened once per connection — see PeerRouter), the same way | |
| 55 | // keet keeps one rootCorestore.replicate() per connection and lets hypercore mux every room's cores | |
| 56 | // over it. We do NOT open our own RPC or probe connections here — that caused DUPLICATE_CHANNEL when | |
| 57 | // fronting a 2nd DID probed the 1st front's cached connection. The PeerFront BORROWS the router's | |
| 58 | // conn+rpc (does not own them), opens the repo + firehose cores (replicated via the router's | |
| 59 | // corestore.replicate), and AUTHENTICATES: the served repo MUST sign against `verifyKey` (the DID's | |
| 60 | // signing pubkey). Fail-closed — an unverifiable or empty repo is refused, but the router's | |
| 61 | // conn/rpc are left intact so a later retry (e.g. after the peer signs its first commit) can front. | |
| 62 | static async fromEntry (corestore, swarm, { conn, rpc, coords }, did, { verifyKey, timeout = 15000 } = {}) { | |
| 63 | if (!verifyKey) throw new Error('PeerFront.fromEntry: verifyKey (DID signing pubkey) required for connect-time auth') | |
| 64 | if (!coords || coords.did !== did) throw new Error(`PeerFront.fromEntry: coords DID mismatch (${coords && coords.did} != ${did})`) | |
| 65 | const front = new PeerFront(corestore, swarm, { repoKey: coords.repoKey, firehoseKey: coords.firehoseKey, did, handle: coords.handle }) | |
| 66 | front._conn = conn | |
| 67 | front.rpc = rpc | |
| 68 | front._ownsConn = false // the ROUTER owns conn+rpc (closes them on connection close), not the front | |
| 69 | front.repo = await new RepoStore(corestore, { key: front.repoKey }).ready() | |
| 70 | front.seq = await new FirehoseSequencer(corestore.get({ key: front.firehoseKey })).ready() | |
| 71 | ||
| 72 | // DIRECTION-FREE LIVE-FOLLOW — the corrected primitive from did-peer.js (see | |
| 73 | // memory/feedback_p2p-no-directionality-event-driven). Issue a STANDING full-range download on both | |
| 74 | // cores and re-pull on 'append'. This flips `core.replicator.downloading = true`, which is the ONLY | |
| 75 | // condition under which `corestore.replicate()` re-attaches a core to a FRESH connection after a flaky | |
| 76 | // link drops (corestore `_shouldReplicate`). WITHOUT it, the front only recovered if the REMOTE re-sent | |
| 77 | // its discovery key (a directionality dependency the project forbids) — so over a flaky link the served | |
| 78 | // head froze until a bridge restart. WITH it, a reconnect (hyperswarm's own 5s keepalive + topic-rejoin) | |
| 79 | // re-attaches the cores and the standing download actively pulls, event-driven — no watchdog needed to | |
| 80 | // force-destroy connections (see PeerRouter._refreshFronts, now pull-only). | |
| 81 | front._seed = (core) => { const pull = () => { try { core.download({ start: 0, end: -1 }) } catch {} }; pull(); core.on('append', pull) } | |
| 82 | front._seed(front.repo.core); front._seed(front.seq.core) | |
| 83 | ||
| 84 | const signed = await front._waitSignedCommit(timeout) | |
| 85 | if (!signed || !verifyCommit(signed, verifyKey)) { | |
| 86 | await front.close() // closes only the cores this front opened; leaves the router's conn/rpc alone | |
| 87 | throw new Error(`PeerFront.fromEntry: repo does not verify against the DID key — refusing to front ${did}`) | |
| 88 | } | |
| 89 | return front | |
| 90 | } | |
| 91 | ||
| 92 | // Wait for the replicated firehose to carry a signed commit (returns null on timeout). | |
| 93 | async _waitSignedCommit (timeout = 12000) { | |
| 94 | const end = Date.now() + timeout | |
| 95 | while (Date.now() < end) { | |
| 96 | const s = await this.latestSignedCommit() | |
| 97 | if (s) return s | |
| 98 | await new Promise((r) => setTimeout(r, 100)) | |
| 99 | } | |
| 100 | return null | |
| 101 | } | |
| 102 | ||
| 103 | // Relay a write to the peer (peer-signs-over-RPC): the PEER applies + signs it; the bridge | |
| 104 | // never holds the key. Returns the peer's { did, commitCid, rev, seq }. The signed result | |
| 105 | // flows back out on the peer's firehose (which this PeerFront already replicates). | |
| 106 | applyWrite (op) { | |
| 107 | if (!this.rpc) throw new Error('PeerFront: not connected') | |
| 108 | return this.rpc.request('apply-write', op) | |
| 109 | } | |
| 110 | ||
| 111 | // com.atproto.repo.applyWrites — batch relay (the bsky composer's post path). writes are | |
| 112 | // pre-normalized by the bridge to { action, collection, rkey, record }; peer returns { results }. | |
| 113 | applyWrites (op) { | |
| 114 | if (!this.rpc) throw new Error('PeerFront: not connected') | |
| 115 | return this.rpc.request('apply-writes', op) | |
| 116 | } | |
| 117 | ||
| 118 | // com.atproto.server.getServiceAuth — the peer mints the ES256K service token (keyless bridge | |
| 119 | // can't). Returns { token, exp }. Used to authenticate AppView proxying (timeline/notifications). | |
| 120 | getServiceAuth ({ aud, lxm, token } = {}) { | |
| 121 | if (!this.rpc) throw new Error('PeerFront: not connected') | |
| 122 | return this.rpc.request('get-service-auth', { aud, lxm, token }) | |
| 123 | } | |
| 124 | ||
| 125 | // Relay login/refresh to the peer — the PEER verifies the app-password and mints the tokens | |
| 126 | // (it holds the jwtSecret). The bridge never sees credentials-as-authority nor mints tokens. | |
| 127 | createSession ({ identifier, password } = {}) { | |
| 128 | if (!this.rpc) throw new Error('PeerFront: not connected') | |
| 129 | return this.rpc.request('create-session', { identifier, password }) | |
| 130 | } | |
| 131 | ||
| 132 | refreshSession (refreshJwt) { | |
| 133 | if (!this.rpc) throw new Error('PeerFront: not connected') | |
| 134 | return this.rpc.request('refresh-session', { refreshJwt }) | |
| 135 | } | |
| 136 | ||
| 137 | // Relay private preferences (birthdate etc.) — the peer verifies the token + stores/returns them. | |
| 138 | getPreferences ({ token } = {}) { | |
| 139 | if (!this.rpc) throw new Error('PeerFront: not connected') | |
| 140 | return this.rpc.request('get-preferences', { token }) | |
| 141 | } | |
| 142 | ||
| 143 | putPreferences ({ preferences, token } = {}) { | |
| 144 | if (!this.rpc) throw new Error('PeerFront: not connected') | |
| 145 | return this.rpc.request('put-preferences', { preferences, token }) | |
| 146 | } | |
| 147 | ||
| 148 | // com.atproto.sync.getBlob — PUBLIC read: the peer holds the blob bytes, the keyless bridge relays. | |
| 149 | // Returns { mimeType, size, bytes(hex) } or null (the server answers 404 on null). | |
| 150 | getBlob (cid) { | |
| 151 | if (!this.rpc) throw new Error('PeerFront: not connected') | |
| 152 | return this.rpc.request('get-blob', { cid }) | |
| 153 | } | |
| 154 | ||
| 155 | // com.atproto.repo.uploadBlob — token-gated write: the peer stores the bytes content-addressed and | |
| 156 | // returns { cid, mimeType, size }. bytes are hex (the server hex-encodes the raw POST body). | |
| 157 | uploadBlob ({ bytes, mimeType, token } = {}) { | |
| 158 | if (!this.rpc) throw new Error('PeerFront: not connected') | |
| 159 | return this.rpc.request('upload-blob', { bytes, mimeType, token }) | |
| 160 | } | |
| 161 | ||
| 162 | // The peer's latest SIGNED commit, read from the replicated firehose (keyless). | |
| 163 | async latestSignedCommit () { | |
| 164 | await this.seq.core.update() | |
| 165 | if (this.seq.core.length === 0) return null | |
| 166 | const { body } = decodeCommitFrame(await this.seq.core.get(this.seq.core.length - 1)) | |
| 167 | return loadCommit(await readCar(body.blocks)).value | |
| 168 | } | |
| 169 | ||
| 170 | // com.atproto.sync.getRepo -> keyless signed CAR (records rebuilt from the replicated repo, | |
| 171 | // signature taken from the firehose commit; getRepoFromCommit guards data-root integrity). | |
| 172 | async getRepo () { | |
| 173 | const signed = await this.latestSignedCommit() | |
| 174 | if (!signed) throw new Error(`PeerFront: no commit yet from peer ${this.did}`) | |
| 175 | return getRepoFromCommit(this.repo, signed) | |
| 176 | } | |
| 177 | ||
| 178 | // com.atproto.sync.subscribeRepos -> the peer's own signed firehose frames (cursor=0 backfills). | |
| 179 | subscribeRepos (cursor = 0, opts) { return this.seq.subscribe(cursor, opts) } | |
| 180 | ||
| 181 | // ---- keyless read surface (from the replicated repo/firehose; no signing) ---- | |
| 182 | // com.atproto.repo.getRecord | |
| 183 | getRecord (collection, rkey) { return rvGetRecord(this.repo, collection, rkey, { did: this.did }) } | |
| 184 | ||
| 185 | // com.atproto.repo.listRecords | |
| 186 | listRecords (collection) { return rvListRecords(this.repo, collection, { did: this.did }) } | |
| 187 | ||
| 188 | // com.atproto.sync.getLatestCommit -> { cid, rev } from the firehose's latest signed commit | |
| 189 | // (keyless — the signature travels in the firehose; we never re-sign). | |
| 190 | async getLatestCommit () { | |
| 191 | const signed = await this.latestSignedCommit() | |
| 192 | if (!signed) return null | |
| 193 | return { cid: commitCid(signed).toString(), rev: signed.rev } | |
| 194 | } | |
| 195 | ||
| 196 | // com.atproto.sync.getRepoStatus -> a fronted peer we replicate + verified is active. | |
| 197 | getRepoStatus () { return { did: this.did, active: true, status: 'active' } } | |
| 198 | ||
| 199 | // Distinct collection NSIDs present in the replicated repo (keys are "<collection>/<rkey>"). | |
| 200 | async collections () { | |
| 201 | const set = new Set() | |
| 202 | for (const r of await this.repo.records()) { | |
| 203 | const i = r.key.indexOf('/') | |
| 204 | if (i > 0) set.add(r.key.slice(0, i)) | |
| 205 | } | |
| 206 | return [...set] | |
| 207 | } | |
| 208 | ||
| 209 | // com.atproto.repo.describeRepo -> { handle, did, didDoc, collections, handleIsCorrect }. Keyless: | |
| 210 | // did/handle come from the peer's announce, collections from the replicated repo, didDoc is a | |
| 211 | // best-effort PLC fetch (null on failure). handleIsCorrect is true when the peer declared a handle | |
| 212 | // (the membrane serves it via the *.7rnx.net wildcard + resolveHandle). | |
| 213 | // doFetch matches the servers' wrapper shape: (url, opts) -> { status, contentType, body:Buffer }. | |
| 214 | async describeRepo ({ doFetch } = {}) { | |
| 215 | const collections = await this.collections() | |
| 216 | let didDoc = null | |
| 217 | try { | |
| 218 | if (doFetch) { | |
| 219 | const r = await doFetch(`https://plc.directory/${this.did}`, { method: 'GET' }) | |
| 220 | if (r && r.status === 200 && r.body) didDoc = JSON.parse(r.body.toString()) | |
| 221 | } | |
| 222 | } catch {} | |
| 223 | return { handle: this.handle || null, did: this.did, didDoc, collections, handleIsCorrect: !!this.handle } | |
| 224 | } | |
| 225 | ||
| 226 | async close () { | |
| 227 | // Only end the rpc if THIS front owns it (connect() path). A fromEntry() front borrows the | |
| 228 | // router's per-connection rpc — ending it here would kill the shared channel for that peer. | |
| 229 | if (this._ownsConn !== false && this.rpc) { try { await this.rpc.end() } catch {} } | |
| 230 | if (this.seq) await this.seq.close() | |
| 231 | if (this.repo) await this.repo.close() | |
| 232 | } | |
| 233 | } |
| … | … | @@ -1,343 +0,0 @@ |
| 1 | // PeerRouter — the bridge's per-DID router to fronted hyper-proto-peers. | |
| 2 | // | |
| 3 | // The membrane is a hybrid (atproto is per-DID/stateless): a DID is served EITHER by the local | |
| 4 | // custodial Pds OR by a remote self-custody hyper-proto-peer, never both. PeerRouter owns the "fronted" | |
| 5 | // set + a cache of live PeerFronts. For a fronted DID it discovers the peer over the DHT and | |
| 6 | // AUTHENTICATES it against the DID's signing key (resolved from the DID doc) before serving — so | |
| 7 | // the bridge holds no key and can't be tricked into fronting a squatter. `isFronted(did)` lets | |
| 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. | |
| 17 | ||
| 18 | import b4a from 'b4a' | |
| 19 | import c from 'compact-encoding' | |
| 20 | import Protomux from 'protomux' | |
| 21 | import ProtomuxRPC from 'protomux-rpc' | |
| 22 | import { decodeDidKey } from 'atproto-identity/lib/did-key.js' | |
| 23 | ||
| 24 | const ANNOUNCE_PROTOCOL = 'atproto/did-announce/v1' // corrected discovery — MUST match hyper-proto-peer serve-membrane | |
| 25 | import { discoveryTopic } from 'atproto-identity/lib/discovery.js' | |
| 26 | import { syncFrame } from 'atproto-store/lib/firehose.js' | |
| 27 | import { PeerFront, RPC_PROTOCOL, withTimeout } from './peer-front.js' | |
| 28 | ||
| 29 | // Resolve a DID's atproto signing pubkey from its DID document (PLC). The doc's | |
| 30 | // verificationMethod.publicKeyMultibase is the same multibase multikey as a did:key's suffix, | |
| 31 | // so we decode it via `did:key:<multibase>`. `doFetch` is injectable (bare-fetch in prod, | |
| 32 | // global fetch in node/tests). | |
| 33 | export async function resolvePlcVerifyKey (did, { plcUrl = 'https://plc.directory', doFetch = fetch } = {}) { | |
| 34 | const res = await doFetch(`${plcUrl}/${did}`) | |
| 35 | if (!res.ok) throw new Error(`resolvePlcVerifyKey: PLC ${res.status} for ${did}`) | |
| 36 | const doc = await res.json() | |
| 37 | const vm = (doc.verificationMethod || []).find((m) => m.id && m.id.endsWith('#atproto')) || (doc.verificationMethod || [])[0] | |
| 38 | if (!vm || !vm.publicKeyMultibase) throw new Error(`resolvePlcVerifyKey: no verificationMethod for ${did}`) | |
| 39 | const { pubkey } = decodeDidKey(`did:key:${vm.publicKeyMultibase}`) | |
| 40 | return pubkey | |
| 41 | } | |
| 42 | ||
| 43 | export class PeerRouter { | |
| 44 | constructor (swarm, corestore, { resolveVerifyKey = resolvePlcVerifyKey, fronted = [], relay = null, refreshMs = 10000, refreshTimeoutMs = 5000, staleStrikeLimit = 3, handles = {}, onHandles = null } = {}) { | |
| 45 | this.swarm = swarm | |
| 46 | this.corestore = corestore | |
| 47 | this._resolveVerifyKey = resolveVerifyKey // async (did) -> secp256k1 pubkey | |
| 48 | this._verifyKeys = new Map() // did -> Promise<pubkey> (resolve once, reuse for every co-owner front) | |
| 49 | this._fronted = new Set(fronted) // DIDs this bridge fronts (vs local Pds accounts) | |
| 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) | |
| 54 | // handle -> did. PERSISTED (loaded from `handles`, saved via onHandles) so a bridge restart does NOT | |
| 55 | // drop handle resolution — an OFFLINE peer (served only by the blind-relay, e.g. ashatron) cannot | |
| 56 | // re-teach its handle at boot, so an in-memory-only map left it unresolvable (com.atproto.identity. | |
| 57 | // resolveHandle → HandleNotFound). Learned on EVERY announce (see _identify→learn), not just first front. | |
| 58 | this._handleToDid = new Map(Object.entries(handles || {})) | |
| 59 | this._onHandles = onHandles // (obj) => void — persist the handle map when it grows | |
| 60 | this.relay = relay // optional FleetRelay: aggregates fronted peers' firehoses | |
| 61 | this._relaySourceKey = new Map() // did -> firehoseKeyHex currently folded into the relay (single source per did — avoids duplicate frames) | |
| 62 | this._refreshMs = refreshMs | |
| 63 | this._refreshTimeoutMs = refreshTimeoutMs | |
| 64 | this._staleStrikeLimit = staleStrikeLimit // retained for back-compat; no longer drives a destroy | |
| 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) | |
| 67 | ||
| 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() | |
| 73 | this._identified = new WeakSet() // connections we've already opened an identify rpc on | |
| 74 | this._waiters = new Map() // did -> Set<fn> woken when ANY co-owner of that did lands | |
| 75 | ||
| 76 | this._onconnection = (conn) => { | |
| 77 | this.corestore.replicate(conn) // keet pattern: mux ALL fronted DIDs' cores over this stream | |
| 78 | this._identify(conn) // + one rpc, learn did+cores from the announce, track by (did, firehoseKey) | |
| 79 | const rk = conn.remotePublicKey ? b4a.toString(conn.remotePublicKey, 'hex') : null | |
| 80 | console.log(JSON.stringify({ msg: 'swarm connection', peer: rk, connections: this.swarm.connections ? this.swarm.connections.size : null })) | |
| 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 })) }) | |
| 82 | } | |
| 83 | this.swarm.on('connection', this._onconnection) | |
| 84 | ||
| 85 | for (const did of this._fronted) this._joinDid(did) | |
| 86 | ||
| 87 | if (this._refreshMs > 0) { | |
| 88 | this._refreshTimer = setInterval(() => { this._refreshFronts().catch(() => {}) }, this._refreshMs) | |
| 89 | if (this._refreshTimer && this._refreshTimer.unref) this._refreshTimer.unref() | |
| 90 | } | |
| 91 | } | |
| 92 | ||
| 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 whose LATEST SIGNED COMMIT has the highest | |
| 100 | // rev. Freshness is the commit REV, NOT the firehose frame count: two co-owners project the SAME | |
| 101 | // deterministic __commit__ log but can FRAME it differently (many small incremental frames at an OLD | |
| 102 | // head vs a few batched frames at a NEW head), so a frame-count pick can serve a stale membrane with | |
| 103 | // more-but-older frames (the wj62 prod symptom: an 83-frame membrane frozen at rev 3mr436 out-"lengthed" | |
| 104 | // the current 6-frame membrane at rev 3mr74lg). PREFER a front whose key is CURRENTLY announced (live in | |
| 105 | // _byDid) — a rebuilt/abandoned membrane leaves a frozen front behind; fall back to any open front only | |
| 106 | // if none are currently announced (all co-owners momentarily disconnected → serve last-known). Revs are | |
| 107 | // TID strings (base32 lexicographically = chronologically sortable). Async: reads each front's head once. | |
| 108 | async _freshestFront (did) { | |
| 109 | const set = this._fronts.get(did) | |
| 110 | if (!set || !set.size) return null | |
| 111 | const live = this._byDid.get(did) || new Map() | |
| 112 | const scored = [] | |
| 113 | for (const [fhKeyHex, f] of set) { | |
| 114 | if (!f || !f.seq || !f.seq.core) continue | |
| 115 | let rev = '' | |
| 116 | try { const s = await f.latestSignedCommit(); if (s && s.rev) rev = s.rev } catch {} | |
| 117 | scored.push({ f, rev, live: live.has(fhKeyHex) }) | |
| 118 | } | |
| 119 | if (!scored.length) return null | |
| 120 | const pickMaxRev = (arr) => arr.reduce((b, x) => (!b || x.rev > b.rev ? x : b), null) | |
| 121 | const liveOnes = scored.filter((x) => x.live) | |
| 122 | const chosen = pickMaxRev(liveOnes) || pickMaxRev(scored) | |
| 123 | return chosen ? chosen.f : null | |
| 124 | } | |
| 125 | ||
| 126 | // Open (once) the front for ONE co-owner (did, firehoseKey) from its identified connection. Idempotent | |
| 127 | // + dedup-guarded. Best-effort: a co-owner that doesn't verify/lacks a commit yet is simply skipped — | |
| 128 | // the OTHER co-owners still front. Wakes did waiters + (re)picks the relay source on success. | |
| 129 | async _ensureFront (did, fhKeyHex) { | |
| 130 | const set = this._fronts.get(did) | |
| 131 | if (set && set.has(fhKeyHex)) return set.get(fhKeyHex) | |
| 132 | const openKey = did + '|' + fhKeyHex | |
| 133 | const inflight = this._opening.get(openKey) | |
| 134 | if (inflight) return inflight // AWAIT the in-progress open (a concurrent front()/learn started it) | |
| 135 | const entry = this._byDid.get(did) && this._byDid.get(did).get(fhKeyHex) | |
| 136 | if (!entry) return null | |
| 137 | const p = (async () => { | |
| 138 | const verifyKey = await this._verifyKey(did) | |
| 139 | const f = await PeerFront.fromEntry(this.corestore, this.swarm, entry, did, { verifyKey }) | |
| 140 | if (!this._fronts.has(did)) this._fronts.set(did, new Map()) | |
| 141 | this._fronts.get(did).set(fhKeyHex, f) | |
| 142 | if (f.seq && f.seq.core && !f._appendWired) { | |
| 143 | f._appendWired = true | |
| 144 | this._lastAppend.set(did, Date.now()) | |
| 145 | f.seq.core.on('append', () => { this._lastAppend.set(did, Date.now()) }) | |
| 146 | } | |
| 147 | if (f.handle) this._setHandle(f.handle, did) | |
| 148 | const ws = this._waiters.get(did); if (ws) { this._waiters.delete(did); for (const fn of ws) fn() } | |
| 149 | this._pickRelaySource(did).catch(() => {}) // fold the freshest into the aggregate (single source per did) | |
| 150 | return f | |
| 151 | })().catch(() => null) // a co-owner that doesn't verify / has no commit yet just yields null; others still front | |
| 152 | this._opening.set(openKey, p) | |
| 153 | try { return await p } finally { this._opening.delete(openKey) } | |
| 154 | } | |
| 155 | ||
| 156 | // Keep every cached front live-following its peer. Cores are REMOTE replicas kept live by the standing | |
| 157 | // download (PeerFront.fromEntry); this timer is a pull-only staleness ceiling — a bounded update({wait}) | |
| 158 | // nudge when a front has seen no 'append' for a window. It NEVER destroys connections. It ALSO re-picks | |
| 159 | // the relay source (failover): if the folded co-owner stalled while another advanced, switch to the fresher. | |
| 160 | async _refreshFronts () { | |
| 161 | for (const [did, set] of this._fronts) { | |
| 162 | for (const [fhKeyHex, f] of set) { | |
| 163 | if (!f || !f.seq || !f.seq.core) continue | |
| 164 | if (Date.now() - (this._lastAppend.get(did) || 0) < this._refreshMs) { this._staleStrikes.set(did + '|' + fhKeyHex, 0); continue } | |
| 165 | try { await withTimeout(f.seq.core.update({ wait: true }), this._refreshTimeoutMs); this._staleStrikes.set(did + '|' + fhKeyHex, 0) } | |
| 166 | catch { this._staleStrikes.set(did + '|' + fhKeyHex, (this._staleStrikes.get(did + '|' + fhKeyHex) || 0) + 1) } | |
| 167 | } | |
| 168 | this._pickRelaySource(did).catch(() => {}) // failover: ensure the relay is folding the freshest co-owner | |
| 169 | } | |
| 170 | } | |
| 171 | ||
| 172 | _joinDid (did) { | |
| 173 | try { this.swarm.join(discoveryTopic(did), { server: true, client: true }); this.swarm.flush().catch(() => {}) } catch {} | |
| 174 | } | |
| 175 | ||
| 176 | // Open exactly ONE ProtomuxRPC per connection and learn the did+cores it serves from the peer's PUSHED | |
| 177 | // announce. A co-owner is tracked by (did, firehoseKey); a NEW announce for the same (did, firehoseKey) | |
| 178 | // just refreshes conn/rpc, a NEW firehoseKey under the same did is a DISTINCT co-owner (tracked, not evicting). | |
| 179 | _identify (conn) { | |
| 180 | if (this._identified.has(conn)) return | |
| 181 | this._identified.add(conn) | |
| 182 | const learn = (coords, rpcForConn) => { | |
| 183 | if (!coords || !coords.did || !coords.repoKey || !coords.firehoseKey) return | |
| 184 | if (coords.handle) this._setHandle(coords.handle, coords.did) // learn+persist on EVERY announce | |
| 185 | const did = coords.did | |
| 186 | const fhKeyHex = coords.firehoseKey | |
| 187 | if (!this._byDid.has(did)) this._byDid.set(did, new Map()) | |
| 188 | const entries = this._byDid.get(did) | |
| 189 | entries.set(fhKeyHex, { conn, rpc: rpcForConn, coords }) | |
| 190 | conn._frontDid = did | |
| 191 | conn._frontFhKey = fhKeyHex | |
| 192 | // Refresh an already-open front's write rpc + conn (its cores re-replicate via corestore regardless). | |
| 193 | const f = this._fronts.get(did) && this._fronts.get(did).get(fhKeyHex) | |
| 194 | if (f) { f.rpc = rpcForConn; f._conn = conn } | |
| 195 | // Wake anyone waiting for this did (a co-owner is now available), and start opening this front. | |
| 196 | const ws = this._waiters.get(did); if (ws) { this._waiters.delete(did); for (const fn of ws) fn() } | |
| 197 | this._ensureFront(did, fhKeyHex).catch(() => {}) | |
| 198 | } | |
| 199 | try { | |
| 200 | const rpc = new ProtomuxRPC(conn, { protocol: RPC_PROTOCOL, valueEncoding: c.json }) | |
| 201 | rpc.on('error', () => {}) | |
| 202 | const achan = Protomux.from(conn).createChannel({ protocol: ANNOUNCE_PROTOCOL }) | |
| 203 | if (achan) { | |
| 204 | achan.addMessage({ encoding: c.json, onmessage: (a) => { try { learn(a, rpc) } catch {} } }) | |
| 205 | achan.open() | |
| 206 | } | |
| 207 | } catch {} | |
| 208 | } | |
| 209 | ||
| 210 | // A peer disconnected: drop only its (did, firehoseKey) identified-connection entry, so _identify re-runs | |
| 211 | // when it reconnects. Do NOT close the cached PeerFront — its repo/firehose are corestore cores that keep | |
| 212 | // replicating over any remaining/next connection (reads live-follow across a reconnect). Its write RPC is | |
| 213 | // refreshed on reconnect (see _identify). Other co-owners' fronts are untouched. | |
| 214 | _onConnClose (conn) { | |
| 215 | const did = conn._frontDid | |
| 216 | const fhKeyHex = conn._frontFhKey | |
| 217 | if (!did || !fhKeyHex) return | |
| 218 | const entries = this._byDid.get(did) | |
| 219 | if (!entries) return | |
| 220 | const entry = entries.get(fhKeyHex) | |
| 221 | if (entry && entry.conn === conn) entries.delete(fhKeyHex) | |
| 222 | } | |
| 223 | ||
| 224 | // Resolve that AT LEAST ONE co-owner of a did is identified — immediately if any is connected, else wait. | |
| 225 | _waitForDid (did, timeout) { | |
| 226 | const e = this._byDid.get(did) | |
| 227 | if (e && e.size > 0) return Promise.resolve(e) | |
| 228 | return new Promise((resolve, reject) => { | |
| 229 | const fn = () => { | |
| 230 | clearTimeout(timer) | |
| 231 | const e2 = this._byDid.get(did) | |
| 232 | e2 && e2.size > 0 ? resolve(e2) : reject(new Error(`PeerRouter: ${did} identified but entry missing`)) | |
| 233 | } | |
| 234 | const timer = setTimeout(() => { | |
| 235 | const ws = this._waiters.get(did); if (ws) ws.delete(fn) | |
| 236 | reject(new Error(`PeerRouter: timeout finding ${did}`)) | |
| 237 | }, timeout) | |
| 238 | if (!this._waiters.has(did)) this._waiters.set(did, new Set()) | |
| 239 | this._waiters.get(did).add(fn) | |
| 240 | }) | |
| 241 | } | |
| 242 | ||
| 243 | diagnose () { | |
| 244 | const conns = (this.swarm && this.swarm.connections) ? [...this.swarm.connections] : [] | |
| 245 | const fronts = [] | |
| 246 | for (const [did, set] of this._fronts) { | |
| 247 | for (const [fhKeyHex, f] of set) { | |
| 248 | fronts.push({ | |
| 249 | did, | |
| 250 | handle: f.handle || null, | |
| 251 | repoKey: f.repoKey ? b4a.toString(f.repoKey, 'hex') : null, | |
| 252 | firehoseKey: fhKeyHex, | |
| 253 | firehoseLen: (f.seq && f.seq.core) ? f.seq.core.length : null, | |
| 254 | rpcConnected: !!(f.rpc && !f.rpc.closed), | |
| 255 | relaySource: this._relaySourceKey.get(did) === fhKeyHex, | |
| 256 | staleStrikes: this._staleStrikes.get(did + '|' + fhKeyHex) || 0 | |
| 257 | }) | |
| 258 | } | |
| 259 | } | |
| 260 | return { | |
| 261 | fronted: [...this._fronted], | |
| 262 | cachedFronts: fronts.length, | |
| 263 | fronts, | |
| 264 | swarm: { | |
| 265 | connections: conns.length, | |
| 266 | peers: conns.map((c) => (c.remotePublicKey ? b4a.toString(c.remotePublicKey, 'hex') : null)) | |
| 267 | }, | |
| 268 | relay: this.relay ? { sources: this.relay.sourceCount ? this.relay.sourceCount() : (this.relay._sources ? this.relay._sources.size : null) } : null | |
| 269 | } | |
| 270 | } | |
| 271 | ||
| 272 | isFronted (did) { return !!did && this._fronted.has(did) } | |
| 273 | resolveHandle (handle) { return this._handleToDid.get(handle) || null } | |
| 274 | ||
| 275 | _setHandle (handle, did) { | |
| 276 | if (!handle || !did || this._handleToDid.get(handle) === did) return | |
| 277 | this._handleToDid.set(handle, did) | |
| 278 | if (this._onHandles) { try { this._onHandles(Object.fromEntries(this._handleToDid)) } catch {} } | |
| 279 | } | |
| 280 | add (did) { this._fronted.add(did); this._joinDid(did); return this } | |
| 281 | list () { return [...this._fronted] } | |
| 282 | ||
| 283 | // Get the FRESHEST verified PeerFront for a fronted did. Opens fronts for every identified co-owner | |
| 284 | // (idempotent), then returns the one with the longest firehose. This is what every read (getRepo, | |
| 285 | // getLatestCommit, listRecords, …) resolves against — so a lagging co-owner never freezes the served head. | |
| 286 | async front (did, { timeout } = {}) { | |
| 287 | if (!this.isFronted(did)) throw new Error(`PeerRouter: ${did} is not a fronted DID`) | |
| 288 | this._joinDid(did) | |
| 289 | if (!this._byDid.has(did) || this._byDid.get(did).size === 0) { | |
| 290 | await this._waitForDid(did, timeout || 20000) | |
| 291 | } | |
| 292 | const entries = this._byDid.get(did) || new Map() | |
| 293 | await Promise.all([...entries.keys()].map((fhKeyHex) => this._ensureFront(did, fhKeyHex).catch(() => null))) | |
| 294 | const best = await this._freshestFront(did) | |
| 295 | if (!best) throw new Error(`PeerRouter: no frontable co-owner for ${did}`) | |
| 296 | return best | |
| 297 | } | |
| 298 | ||
| 299 | // Fold the FRESHEST co-owner's firehose into the aggregate (single source per did — two co-owners are the | |
| 300 | // SAME frames under different sourceIds, so folding both duplicates every #commit to bsky's relay). On a | |
| 301 | // source change (a fresher co-owner won, or the folded one stalled) swap removeSource→addSource + emit a | |
| 302 | // #sync so the relay re-baselines. Idempotent when the freshest is unchanged. | |
| 303 | async _pickRelaySource (did) { | |
| 304 | if (!this.relay) return | |
| 305 | const best = await this._freshestFront(did) | |
| 306 | if (!best || !best.seq || !best.seq.core) return | |
| 307 | const key = b4a.toString(best.seq.core.key, 'hex') | |
| 308 | if (this._relaySourceKey.get(did) === key) return | |
| 309 | const old = this._relaySourceKey.get(did) | |
| 310 | this._relaySourceKey.set(did, key) | |
| 311 | ;(async () => { | |
| 312 | if (old) { try { this.relay.removeSource(old) } catch {} } | |
| 313 | try { await this.relay.addSource(key, best.seq.core) } catch {} | |
| 314 | try { await this._emitSync(did, best) } catch {} | |
| 315 | })().catch(() => {}) | |
| 316 | } | |
| 317 | ||
| 318 | // Build + inject a #sync frame for a fronted peer's current signed commit (re-baseline signal). | |
| 319 | async _emitSync (did, front) { | |
| 320 | if (!this.relay) return { did, emitted: false, reason: 'no relay' } | |
| 321 | const signed = await front.latestSignedCommit() | |
| 322 | if (!signed) return { did, emitted: false, reason: 'no commit' } | |
| 323 | const time = new Date().toISOString() | |
| 324 | const frameBytes = await syncFrame({ seq: 0, did, signedCommit: signed, time }) | |
| 325 | const emitted = await this.relay.emit(`sync:${did}:${signed.rev}`, frameBytes) | |
| 326 | return { did, rev: signed.rev, emitted } | |
| 327 | } | |
| 328 | ||
| 329 | // Public trigger (control plane): force a #sync re-baseline for a fronted DID (from its freshest front). | |
| 330 | async emitSync (did) { | |
| 331 | const f = await this.front(did) | |
| 332 | return this._emitSync(did, f) | |
| 333 | } | |
| 334 | ||
| 335 | async close () { | |
| 336 | if (this._refreshTimer) { try { clearInterval(this._refreshTimer) } catch {} this._refreshTimer = null } | |
| 337 | if (this._onconnection) { try { this.swarm.off('connection', this._onconnection) } catch {} } | |
| 338 | for (const set of this._fronts.values()) { for (const f of set.values()) { try { await f.close() } catch {} } } | |
| 339 | this._fronts.clear() | |
| 340 | for (const entries of this._byDid.values()) { for (const { rpc } of entries.values()) { try { await rpc.end() } catch {} } } | |
| 341 | this._byDid.clear() | |
| 342 | } | |
| 343 | } |
| … | … | @@ -18,7 +18,7 @@ import { readCar, loadCommit } from 'atproto-repo/lib/repo.js' |
| 18 | 18 | import { verifyCommit } from 'atproto-repo/lib/commit.js' |
| 19 | 19 | import { keyFromPrivate } from 'atproto-identity' |
| 20 | 20 | import { discoveryTopic } from 'atproto-identity/lib/discovery.js' |
| 21 | import { PeerRouter } from '../lib/peer-router.js' | |
| 21 | import { mkDidRouter } from './mk-did-router.js' | |
| 22 | 22 | |
| 23 | 23 | const DID = 'did:plc:announceonly2222222222' |
| 24 | 24 | const PRIV = new Uint8Array(32).fill(6) |
| … | … | @@ -58,7 +58,7 @@ test('PeerRouter fronts an ANNOUNCE-ONLY peer (no get-coords) — corrected push |
| 58 | 58 | const store = openCorestore(tmp('ao-bridge-')) |
| 59 | 59 | const swarm = new Hyperswarm({ bootstrap: testnet.bootstrap }) |
| 60 | 60 | const key = keyFromPrivate(PRIV).publicKey |
| 61 | const router = new PeerRouter(swarm, store, { fronted: [DID], resolveVerifyKey: async () => key }) | |
| 61 | const router = mkDidRouter(swarm, store, { fronted: [DID], resolveVerifyKey: async () => key }) | |
| 62 | 62 | t.teardown(async () => { await peer.close(); await router.close(); await swarm.destroy(); await store.close(); await testnet.destroy() }) |
| 63 | 63 | |
| 64 | 64 | // the router joined the DID topic in its constructor; it must IDENTIFY the peer purely from the |
| … | … | @@ -82,7 +82,7 @@ test('PeerRouter fronts MULTIPLE announce-only DIDs over ONE swarm (the real wj6 |
| 82 | 82 | const B = await announceOnlyPeer(testnet, DID_B, PRIV_B) |
| 83 | 83 | const store = openCorestore(tmp('aom-bridge-')) |
| 84 | 84 | const swarm = new Hyperswarm({ bootstrap: testnet.bootstrap }) |
| 85 | const router = new PeerRouter(swarm, store, { fronted: [DID_A, DID_B], resolveVerifyKey: async (d) => (d === DID_A ? keyA : keyB) }) | |
| 85 | const router = mkDidRouter(swarm, store, { fronted: [DID_A, DID_B], resolveVerifyKey: async (d) => (d === DID_A ? keyA : keyB) }) | |
| 86 | 86 | t.teardown(async () => { await A.close(); await B.close(); await router.close(); await swarm.destroy(); await store.close(); await testnet.destroy() }) |
| 87 | 87 | |
| 88 | 88 | // both DIDs must be identified purely from their pushed announces (no get-coords), over ONE swarm. |
| … | … | @@ -19,7 +19,7 @@ import { readCar, loadCommit } from 'atproto-repo/lib/repo.js' |
| 19 | 19 | import { verifyCommit } from 'atproto-repo/lib/commit.js' |
| 20 | 20 | import { keyFromPrivate } from 'atproto-identity' |
| 21 | 21 | import { discoveryTopic } from 'atproto-identity/lib/discovery.js' |
| 22 | import { PeerRouter } from '../lib/peer-router.js' | |
| 22 | import { mkDidRouter } from './mk-did-router.js' | |
| 23 | 23 | import { BlindRelay } from '../lib/blind-relay.js' |
| 24 | 24 | import { pushAnnounce } from './serve-announce.js' |
| 25 | 25 | |
| … | … | @@ -51,7 +51,7 @@ async function servePeer (testnet, did, priv) { |
| 51 | 51 | function mkRouter (testnet, fronted, keyByDid) { |
| 52 | 52 | const store = openCorestore(mkdtempSync(join(tmpdir(), 'br-bridge-'))) |
| 53 | 53 | const swarm = new Hyperswarm({ bootstrap: testnet.bootstrap }) |
| 54 | const router = new PeerRouter(swarm, store, { fronted, resolveVerifyKey: async (did) => keyByDid[did] }) | |
| 54 | const router = mkDidRouter(swarm, store, { fronted, resolveVerifyKey: async (did) => keyByDid[did] }) | |
| 55 | 55 | return { store, swarm, router, close: async () => { await router.close(); await swarm.destroy(); await store.close() } } |
| 56 | 56 | } |
| 57 | 57 |
| … | … | @@ -1,81 +0,0 @@ |
| 1 | // REPRO: co-owned DID (the wj62 prod topology — stortron notary + tabertron co-owner). BOTH co-owners | |
| 2 | // run serve-membrane, so BOTH announce the SAME did with THEIR OWN membrane core keys (each device has a | |
| 3 | // distinct named membrane firehose/repo core → distinct keys, even though the projected #commit content is | |
| 4 | // the same). The bridge's _identify->learn treats a DIFFERENT key under the same did as a stale-key change | |
| 5 | // and EVICTS + re-fronts. So when a SECOND co-owner announces (or a co-owner lags behind), the bridge can | |
| 6 | // flip its front onto the BEHIND co-owner and serve a stale head — exactly the "bridge regressed to an old | |
| 7 | // head after a daemon restart" symptom that a manual bridge restart fixed. | |
| 8 | import test from 'brittle' | |
| 9 | import { mkdtempSync, rmSync } from 'node:fs' | |
| 10 | import { tmpdir } from 'node:os' | |
| 11 | import { join } from 'node:path' | |
| 12 | import Hyperswarm from 'hyperswarm' | |
| 13 | import createTestnet from '@hyperswarm/testnet' | |
| 14 | import { openCorestore } from 'atproto-store/lib/corestore.js' | |
| 15 | import { RepoStore } from 'atproto-store/lib/repo-store.js' | |
| 16 | import { RepoManager } from 'atproto-store/lib/repo-manager.js' | |
| 17 | import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js' | |
| 18 | import { keyFromPrivate } from 'atproto-identity' | |
| 19 | import { discoveryTopic } from 'atproto-identity/lib/discovery.js' | |
| 20 | import { PeerRouter } from '../lib/peer-router.js' | |
| 21 | import { pushAnnounce } from './serve-announce.js' | |
| 22 | ||
| 23 | const DID = 'did:plc:coowned1' | |
| 24 | const HANDLE = 'coowned.7rnx.net' | |
| 25 | const PRIV = new Uint8Array(32).fill(7) | |
| 26 | const NOW = '2026-07-21T00:00:00.000Z' | |
| 27 | const post = (rk, text) => ({ action: 'create', collection: 'app.bsky.feed.post', rkey: rk, record: { $type: 'app.bsky.feed.post', text, createdAt: NOW, langs: ['en'] } }) | |
| 28 | ||
| 29 | // Build a co-owner peer: its own corestore/cores, N signed commits over the shared DID key. Commit revs | |
| 30 | // are wall-clock TIDs, so a peer built LATER has strictly higher revs — the caller creates the BEHIND peer | |
| 31 | // first and the AHEAD peer after a beat, modelling "ahead = a later head" (in prod the two co-owners share | |
| 32 | // the SAME deterministic log, so a lagging co-owner literally sits at an earlier commit with a lower rev). | |
| 33 | async function coOwner (bootstrap, nCommits) { | |
| 34 | const dir = mkdtempSync(join(tmpdir(), 'coowner-')) | |
| 35 | const store = openCorestore(dir) | |
| 36 | const repo = await new RepoStore(store, { name: `atproto-repo:${DID}` }).ready() | |
| 37 | let micros = 1_700_000_000_000_000n | |
| 38 | const mgr = await new RepoManager(repo, { did: DID, privateKey: PRIV, clock: () => (micros += 1_000_000n) }).init() | |
| 39 | const seq = await new FirehoseSequencer(store.get({ name: 'firehose' })).ready() | |
| 40 | for (let i = 0; i < nCommits; i++) { const w = await mgr.applyWrite(post('3post' + i + 'aaaaaa2', 'c' + i)); await seq.sequenceCommit(w, { time: NOW }) } | |
| 41 | const swarm = new Hyperswarm({ bootstrap }) | |
| 42 | swarm.on('connection', (conn) => { store.replicate(conn); pushAnnounce(conn, { did: DID, handle: HANDLE, repo, seq }) }) | |
| 43 | await swarm.join(discoveryTopic(DID), { server: true, client: false }).flushed() | |
| 44 | return { dir, store, repo, seq, swarm, close: async () => { try { await swarm.destroy() } catch {} ; await seq.close(); await repo.close(); await store.close(); rmSync(dir, { recursive: true, force: true }) } } | |
| 45 | } | |
| 46 | ||
| 47 | test('co-owned: bridge must not front the BEHIND co-owner (front stability under 2 membranes)', async (t) => { | |
| 48 | const testnet = await createTestnet(3) | |
| 49 | const BEHIND = await coOwner(testnet.bootstrap, 1) // co-owner lagging at an EARLIER head (built first → lower revs) | |
| 50 | await new Promise((r) => setTimeout(r, 1500)) // ensure AHEAD's wall-clock revs are strictly higher | |
| 51 | const AHEAD = await coOwner(testnet.bootstrap, 2) // notary at a LATER head (higher revs, more records) | |
| 52 | ||
| 53 | const routerStore = openCorestore(mkdtempSync(join(tmpdir(), 'coowned-router-'))) | |
| 54 | const routerSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap }) | |
| 55 | const router = new PeerRouter(routerSwarm, routerStore, { | |
| 56 | fronted: [DID], resolveVerifyKey: async () => keyFromPrivate(PRIV).publicKey, refreshMs: 0 | |
| 57 | }) | |
| 58 | t.teardown(async () => { await router.close(); await routerSwarm.destroy(); await routerStore.close(); await AHEAD.close(); await BEHIND.close(); await testnet.destroy() }) | |
| 59 | ||
| 60 | await router.front(DID) // establish the first front | |
| 61 | // Poll front() FRESH each time (as a real getRepo request does) so re-fronting/eviction is visible. | |
| 62 | const counts = [] | |
| 63 | const keys = [] | |
| 64 | for (let i = 0; i < 40; i++) { | |
| 65 | const f = await router.front(DID).catch(() => null) | |
| 66 | const recs = f ? await f.listRecords('app.bsky.feed.post').catch(() => []) : [] | |
| 67 | counts.push(recs.length) | |
| 68 | const d = router.diagnose() | |
| 69 | keys.push((d.fronts[0] && d.fronts[0].firehoseKey) ? d.fronts[0].firehoseKey.slice(0, 8) : '-') | |
| 70 | await new Promise(r => setTimeout(r, 300)) | |
| 71 | } | |
| 72 | t.comment('served record counts over time: ' + JSON.stringify(counts)) | |
| 73 | t.comment('fronted firehoseKey over time: ' + JSON.stringify([...new Set(keys)])) | |
| 74 | // With the freshest-serving fix, reads converge to the AHEAD co-owner (2) and never regress to the | |
| 75 | // BEHIND one (1) once the ahead is available. Early polls may read 1 before the ahead co-owner's front | |
| 76 | // opens — that's fine; what matters is it ends on the freshest and never drops back. | |
| 77 | t.is(counts[counts.length - 1], 2, 'bridge ends up serving the FRESHEST (ahead) co-owner') | |
| 78 | const firstTwo = counts.indexOf(2) | |
| 79 | t.ok(firstTwo >= 0, 'the ahead co-owner (2 records) was fronted') | |
| 80 | t.ok(counts.slice(firstTwo).every((x) => x >= 2), 'once fronting the freshest, it never REGRESSES to the behind co-owner') | |
| 81 | }) |
| … | … | @@ -1,12 +1,10 @@ |
| 1 | 1 | // com.atproto.repo.describeRepo — the bridge previously returned MethodNotImplemented. PeerFront now |
| 2 | 2 | // derives collections from the replicated repo and reports did/handle/handleIsCorrect keylessly. |
| 3 | 3 | import test from 'brittle' |
| 4 | import { PeerFront } from '../lib/peer-front.js' | |
| 5 | ||
| 6 | const HEX = '00'.repeat(32) | |
| 4 | import { DidPeer } from 'atproto-did-peer' | |
| 7 | 5 | |
| 8 | 6 | function makeFront (records, { did = 'did:plc:desc', handle = 'desc.7rnx.net' } = {}) { |
| 9 | const f = new PeerFront({}, {}, { serverKey: HEX, repoKey: HEX, firehoseKey: HEX, did, handle }) | |
| 7 | const f = new DidPeer({}, { did, verifyKey: new Uint8Array(32), handle }) | |
| 10 | 8 | f.repo = { records: async () => records } |
| 11 | 9 | return f |
| 12 | 10 | } |
| … | … | @@ -1,29 +1,26 @@ |
| 1 | // diagnose() must RETURN the per-front state it builds (firehoseLen/rpcConnected/staleStrikes). | |
| 1 | // diagnose() must RETURN the per-front state it builds (firehoseLen/rpcConnected/live). | |
| 2 | 2 | // It previously built the `fronts` array but omitted it from the return, so operators couldn't see |
| 3 | 3 | // per-front freshness — which is exactly what made the 2026-07-11 stale-front bug slow to diagnose. |
| 4 | 4 | import test from 'brittle' |
| 5 | import { PeerRouter } from '../lib/peer-router.js' | |
| 5 | import { mkDidRouter } from './mk-did-router.js' | |
| 6 | 6 | |
| 7 | 7 | test('diagnose() returns per-front state (fronts array)', (t) => { |
| 8 | 8 | const fakeSwarm = { on () {}, off () {}, join () { return { flushed: async () => {} } }, flush: async () => {}, connections: new Set() } |
| 9 | const router = new PeerRouter(fakeSwarm, {}, { fronted: [], refreshMs: 0 }) | |
| 9 | const router = mkDidRouter(fakeSwarm, {}, { fronted: [], refreshMs: 0 }) | |
| 10 | 10 | |
| 11 | 11 | router._fronts.set('did:plc:abc', new Map([['k1', { |
| 12 | 12 | handle: 'a.7rnx.net', |
| 13 | serverKey: 'aa', | |
| 14 | repoKey: null, | |
| 15 | firehoseKey: null, | |
| 13 | _repoKey: null, | |
| 16 | 14 | seq: { core: { length: 7 } }, |
| 17 | 15 | rpc: { closed: false } |
| 18 | 16 | }]])) |
| 19 | router._staleStrikes.set('did:plc:abc|k1', 2) | |
| 20 | 17 | |
| 21 | 18 | const d = router.diagnose() |
| 22 | 19 | t.ok(Array.isArray(d.fronts), 'diagnose returns a fronts array') |
| 23 | 20 | t.is(d.fronts.length, 1, 'one fronted peer reported') |
| 24 | 21 | const f = d.fronts[0] |
| 25 | 22 | t.is(f.did, 'did:plc:abc', 'did reported') |
| 23 | t.is(f.firehoseKey, 'k1', 'firehoseKey reported') | |
| 26 | 24 | t.is(f.firehoseLen, 7, 'firehoseLen (served head length) reported') |
| 27 | 25 | t.is(f.rpcConnected, true, 'rpcConnected reported') |
| 28 | t.is(f.staleStrikes, 2, 'watchdog staleStrikes reported') | |
| 29 | 26 | }) |
| … | … | @@ -1,101 +0,0 @@ |
| 1 | // DidPeer — the corrected direction-free membrane primitive. Proves, OFFLINE (local testnet): | |
| 2 | // (1) ANNOUNCE-PUSH discovery: a consumer learns a peer's core keys from a pushed announce — NO | |
| 3 | // get-coords request, NO serverKey directionality — and serves the DID-key-verified head. | |
| 4 | // (2) EVENT-DRIVEN head: when the holder signs a new commit, the consumer's head advances via the | |
| 5 | // firehose 'append' event (no poll). | |
| 6 | // (3) AUTH: only a commit that verifies against the DID key is accepted (verifyKey mandatory). | |
| 7 | // (4) SYMMETRY / no origin: a consumer that has replicated the cores itself becomes a holder that | |
| 8 | // announces + serves a THIRD peer while the original holder is OFFLINE. | |
| 9 | import test from 'brittle' | |
| 10 | import { mkdtempSync, rmSync } from 'node:fs' | |
| 11 | import { tmpdir } from 'node:os' | |
| 12 | import { join } from 'node:path' | |
| 13 | import Hyperswarm from 'hyperswarm' | |
| 14 | import createTestnet from '@hyperswarm/testnet' | |
| 15 | import b4a from 'b4a' | |
| 16 | import { openCorestore } from 'atproto-store/lib/corestore.js' | |
| 17 | import { RepoStore } from 'atproto-store/lib/repo-store.js' | |
| 18 | import { RepoManager } from 'atproto-store/lib/repo-manager.js' | |
| 19 | import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js' | |
| 20 | import { keyFromPrivate } from 'atproto-identity' | |
| 21 | import { discoveryTopic } from 'atproto-identity/lib/discovery.js' | |
| 22 | import { DidPeer } from '../lib/did-peer.js' | |
| 23 | ||
| 24 | const DID = 'did:plc:didpeerrrrrrrrrrrrrrrrr' | |
| 25 | const PRIV = new Uint8Array(32).fill(7) | |
| 26 | const VERIFY = keyFromPrivate(PRIV).publicKey | |
| 27 | const NOW = '2026-07-13T00:00:00.000Z' | |
| 28 | const tmp = (p) => mkdtempSync(join(tmpdir(), p)) | |
| 29 | async function waitFor (fn, ms = 20000) { const end = Date.now() + ms; while (Date.now() < end) { if (await fn()) return true; await new Promise(r => setTimeout(r, 80)) } return false } | |
| 30 | ||
| 31 | // A signing holder: repo + firehose + a RepoManager to sign commits, joined to the DID topic as a DidPeer. | |
| 32 | async function makeHolder (testnet, dir) { | |
| 33 | const store = openCorestore(tmp(dir)) | |
| 34 | const repo = await new RepoStore(store, { name: `atproto-repo:${DID}` }).ready() | |
| 35 | let micros = 1_700_000_000_000_000n | |
| 36 | const mgr = await new RepoManager(repo, { did: DID, privateKey: PRIV, clock: () => (micros += 1_000_000n) }).init() | |
| 37 | const seq = await new FirehoseSequencer(store.get({ name: 'firehose' })).ready() | |
| 38 | const sign = async (rkey, text) => { const w = await mgr.applyWrite({ action: 'create', collection: 'app.bsky.feed.post', rkey, record: { $type: 'app.bsky.feed.post', text, createdAt: NOW } }); await seq.sequenceCommit(w, { time: NOW }) } | |
| 39 | await sign('3a', 'first') | |
| 40 | const swarm = new Hyperswarm({ bootstrap: testnet.bootstrap }) | |
| 41 | const peer = new DidPeer(store, { did: DID, verifyKey: VERIFY }) | |
| 42 | await peer.provide({ repoKey: repo.core.key, firehoseKey: seq.core.key }) | |
| 43 | swarm.on('connection', (conn) => peer.attach(conn)) | |
| 44 | await swarm.join(discoveryTopic(DID), { server: true, client: true }).flushed() | |
| 45 | return { store, repo, seq, swarm, peer, sign, close: async () => { await swarm.destroy(); await peer.close(); await store.close() } } | |
| 46 | } | |
| 47 | ||
| 48 | // A consumer: empty corestore, DidPeer with NO cores + NO get-coords. It must learn everything from the announce. | |
| 49 | function makeConsumer (testnet, dir) { | |
| 50 | const store = openCorestore(tmp(dir)) | |
| 51 | const swarm = new Hyperswarm({ bootstrap: testnet.bootstrap }) | |
| 52 | let heads = [] | |
| 53 | const peer = new DidPeer(store, { did: DID, verifyKey: VERIFY, onHead: (h) => heads.push(h.rev) }) | |
| 54 | swarm.on('connection', (conn) => peer.attach(conn)) | |
| 55 | swarm.join(discoveryTopic(DID), { server: true, client: true }) | |
| 56 | return { store, swarm, peer, heads: () => heads, close: async () => { await swarm.destroy(); await peer.close(); await store.close() } } | |
| 57 | } | |
| 58 | ||
| 59 | test('DidPeer: consumer fronts a DID via ANNOUNCE-PUSH (no get-coords) + EVENT-DRIVEN head advance', async (t) => { | |
| 60 | const testnet = await createTestnet(3) | |
| 61 | const A = await makeHolder(testnet, 'dp-holder-') | |
| 62 | const B = makeConsumer(testnet, 'dp-consumer-') | |
| 63 | t.teardown(async () => { await A.close(); await B.close(); await testnet.destroy() }) | |
| 64 | ||
| 65 | // (1) B learns A's cores from the PUSHED announce (B never calls get-coords) and derives a verified head. | |
| 66 | t.ok(await waitFor(() => B.peer.head && B.peer.head.rev), 'consumer got a head via announce-push (no get-coords)') | |
| 67 | t.is(B.peer.status().hasCores, true, 'consumer opened the DID cores it learned from the announce') | |
| 68 | const rev1 = B.peer.head.rev | |
| 69 | t.ok(rev1, 'head has a rev') | |
| 70 | ||
| 71 | // (3) auth: the head verified against the DID key (DidPeer._reconcile rejects otherwise); getting a | |
| 72 | // head at all means verifyCommit(signed, VERIFY) passed. | |
| 73 | t.pass('served head verified against the DID signing key (auth enforced)') | |
| 74 | ||
| 75 | // (2) A signs a SECOND commit → B's head advances via the firehose 'append' event (not a poll). | |
| 76 | await A.sign('3b', 'second') | |
| 77 | t.ok(await waitFor(() => B.peer.head && B.peer.head.rev !== rev1), 'consumer head ADVANCED on the holder\'s new commit (event-driven)') | |
| 78 | t.ok(B.heads().length >= 2, 'onHead fired for each new verified head') | |
| 79 | }) | |
| 80 | ||
| 81 | test('DidPeer: SYMMETRY — a consumer becomes a holder + serves a third peer while the origin is OFFLINE', async (t) => { | |
| 82 | const testnet = await createTestnet(3) | |
| 83 | const A = await makeHolder(testnet, 'dp2-origin-') | |
| 84 | const B = makeConsumer(testnet, 'dp2-mid-') | |
| 85 | t.teardown(async () => { try { await A.close() } catch {} ; await B.close(); await testnet.destroy() }) | |
| 86 | ||
| 87 | // B replicates A's DID cores via announce. | |
| 88 | t.ok(await waitFor(() => B.peer.status().hasCores && B.peer.head), 'B replicated the DID cores from A') | |
| 89 | // Give B a moment to fully download so it can serve. | |
| 90 | await waitFor(() => B.peer.status().firehoseLength > 0) | |
| 91 | await new Promise(r => setTimeout(r, 1500)) | |
| 92 | ||
| 93 | // A (the original signer/"origin") GOES OFFLINE. | |
| 94 | await A.close() | |
| 95 | ||
| 96 | // A THIRD consumer C joins — the only holder left is B (a former consumer). C must front via B, proving | |
| 97 | // there is no designated origin: any peer that holds the cores serves them. | |
| 98 | const C = makeConsumer(testnet, 'dp2-late-') | |
| 99 | t.teardown(async () => { await C.close() }) | |
| 100 | t.ok(await waitFor(() => C.peer.head && C.peer.head.rev, 25000), 'C fronted the DID via B (a former consumer) with the origin OFFLINE — no directionality') | |
| 101 | }) |
| … | … | @@ -9,7 +9,7 @@ import { join } from 'node:path' |
| 9 | 9 | import Hyperswarm from 'hyperswarm' |
| 10 | 10 | import createTestnet from '@hyperswarm/testnet' |
| 11 | 11 | import { openCorestore } from 'atproto-store/lib/corestore.js' |
| 12 | import { PeerRouter } from '../lib/peer-router.js' | |
| 12 | import { mkDidRouter } from './mk-did-router.js' | |
| 13 | 13 | |
| 14 | 14 | test('handle map persists + survives a bridge restart (offline peer stays resolvable)', async (t) => { |
| 15 | 15 | const testnet = await createTestnet(3) |
| … | … | @@ -18,12 +18,12 @@ test('handle map persists + survives a bridge restart (offline peer stays resolv |
| 18 | 18 | t.teardown(async () => { try { await swarm.destroy() } catch {} try { await store.close() } catch {} await testnet.destroy() }) |
| 19 | 19 | |
| 20 | 20 | // Fix 1: constructing with a persisted map restores handle resolution — no live re-front needed. |
| 21 | const r1 = new PeerRouter(swarm, store, { refreshMs: 0, handles: { 'ashatron.7rnx.net': 'did:plc:ash' } }) | |
| 21 | const r1 = mkDidRouter(swarm, store, { refreshMs: 0, handles: { 'ashatron.7rnx.net': 'did:plc:ash' } }) | |
| 22 | 22 | t.is(r1.resolveHandle('ashatron.7rnx.net'), 'did:plc:ash', 'loads a persisted handle on construct') |
| 23 | 23 | |
| 24 | 24 | // Fix 1+2: learning a new handle persists it via onHandles (this is what _identify→learn does per announce). |
| 25 | 25 | let saved = null |
| 26 | const r2 = new PeerRouter(swarm, store, { refreshMs: 0, onHandles: (o) => { saved = o } }) | |
| 26 | const r2 = mkDidRouter(swarm, store, { refreshMs: 0, onHandles: (o) => { saved = o } }) | |
| 27 | 27 | r2._setHandle('p2p.7rnx.net', 'did:plc:wj62') |
| 28 | 28 | t.is(r2.resolveHandle('p2p.7rnx.net'), 'did:plc:wj62', 'learns a new handle') |
| 29 | 29 | t.ok(saved && saved['p2p.7rnx.net'] === 'did:plc:wj62', 'persists via onHandles when the map grows') |
| … | … | @@ -34,7 +34,7 @@ test('handle map persists + survives a bridge restart (offline peer stays resolv |
| 34 | 34 | // "Restart": a fresh router loaded with the persisted map resolves BOTH — even the offline peer that |
| 35 | 35 | // never reconnects. This is the exact scenario that was broken (ashatron after a bridge restart). |
| 36 | 36 | const persisted = { 'ashatron.7rnx.net': 'did:plc:ash', 'p2p.7rnx.net': 'did:plc:wj62' } |
| 37 | const r3 = new PeerRouter(swarm, store, { refreshMs: 0, handles: persisted }) | |
| 37 | const r3 = mkDidRouter(swarm, store, { refreshMs: 0, handles: persisted }) | |
| 38 | 38 | t.is(r3.resolveHandle('ashatron.7rnx.net'), 'did:plc:ash', 'after restart, the OFFLINE peer handle STILL resolves') |
| 39 | 39 | t.is(r3.resolveHandle('p2p.7rnx.net'), 'did:plc:wj62', 'after restart, the learned handle persists') |
| 40 | 40 | }) |
| … | … | @@ -22,7 +22,7 @@ import { RepoManager } from 'atproto-store/lib/repo-manager.js' |
| 22 | 22 | import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js' |
| 23 | 23 | import { keyFromPrivate } from 'atproto-identity' |
| 24 | 24 | import { discoveryTopic } from 'atproto-identity/lib/discovery.js' |
| 25 | import { PeerRouter } from '../lib/peer-router.js' | |
| 25 | import { mkDidRouter } from './mk-did-router.js' | |
| 26 | 26 | |
| 27 | 27 | const waitFor = async (fn, { timeout = 25000, interval = 150 } = {}) => { |
| 28 | 28 | const end = Date.now() + timeout |
| … | … | @@ -67,7 +67,7 @@ test('REPRO: front must live-follow a new commit AFTER the peer PROCESS restarts |
| 67 | 67 | |
| 68 | 68 | const routerStore = openCorestore(mkdtempSync(join(tmpdir(), 'reopen-router-'))) |
| 69 | 69 | const routerSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap }) |
| 70 | const router = new PeerRouter(routerSwarm, routerStore, { fronted: [DID], resolveVerifyKey: async () => keyFromPrivate(PRIV).publicKey }) | |
| 70 | const router = mkDidRouter(routerSwarm, routerStore, { fronted: [DID], resolveVerifyKey: async () => keyFromPrivate(PRIV).publicKey }) | |
| 71 | 71 | |
| 72 | 72 | t.teardown(async () => { |
| 73 | 73 | await router.close(); await routerSwarm.destroy() |
| … | … | @@ -104,7 +104,7 @@ test('REPRO: front must live-follow a new commit AFTER the peer PROCESS restarts |
| 104 | 104 | test('watchdog nudges a stalled front with a pull but NEVER destroys the connection', async (t) => { |
| 105 | 105 | const fakeSwarm = { on () {}, off () {}, join () { return { flushed: async () => {} } }, flush: async () => {}, connections: new Set() } |
| 106 | 106 | // refreshMs:0 disables the auto-timer so we drive _refreshFronts manually. |
| 107 | const router = new PeerRouter(fakeSwarm, {}, { fronted: [], refreshMs: 0, staleStrikeLimit: 3 }) | |
| 107 | const router = mkDidRouter(fakeSwarm, {}, { fronted: [], refreshMs: 0, staleStrikeLimit: 3 }) | |
| 108 | 108 | |
| 109 | 109 | let stalledDestroyed = 0 |
| 110 | 110 | let stalledPulls = 0 |
| … | … | @@ -121,7 +121,6 @@ test('watchdog nudges a stalled front with a pull but NEVER destroys the connect |
| 121 | 121 | |
| 122 | 122 | t.is(stalledPulls, 4, 'a stalled front is NUDGED with an update({wait}) pull each window') |
| 123 | 123 | t.is(stalledDestroyed, 0, 'stalled connection is NEVER destroyed — recovery is the standing download + hyperswarm reconnect') |
| 124 | t.ok((router._staleStrikes.get('did:plc:stalled|ks') || 0) >= 1, 'failed pulls accrue strikes for diagnose() visibility only') | |
| 125 | 124 | t.is(healthyDestroyed, 0, 'a healthy front (update resolves) is never destroyed either') |
| 126 | 125 | |
| 127 | 126 | await router.close() |
| … | … | @@ -0,0 +1,15 @@ |
| 1 | // Test helper: construct the canonical atproto-did-peer DidPeerRouter EXACTLY as pds-serve.js does | |
| 2 | // (owns-swarm + injectable join with the NAT-symmetric opts + rpc for the write path). Signature mirrors | |
| 3 | // the retired PeerRouter's `new PeerRouter(swarm, corestore, opts)` so migrated tests only swap the call. | |
| 4 | // Extra/unknown opts (e.g. the retired staleStrikeLimit) are harmlessly ignored by DidPeerRouter. | |
| 5 | import { DidPeerRouter } from 'atproto-did-peer' | |
| 6 | import { discoveryTopic } from 'atproto-identity/lib/discovery.js' | |
| 7 | ||
| 8 | export function mkDidRouter (swarm, corestore, opts = {}) { | |
| 9 | return new DidPeerRouter(corestore, { | |
| 10 | swarm, | |
| 11 | join: (did) => { try { swarm.join(discoveryTopic(did), { server: true, client: true }); if (swarm.flush) swarm.flush().catch(() => {}) } catch {} }, | |
| 12 | rpc: true, | |
| 13 | ...opts | |
| 14 | }) | |
| 15 | } |
| … | … | @@ -21,8 +21,8 @@ import { keyFromPrivate } from 'atproto-identity' |
| 21 | 21 | import { discoveryTopic } from 'atproto-identity/lib/discovery.js' |
| 22 | 22 | import b4a from 'b4a' |
| 23 | 23 | import { serveReplication, connectReplica } from '../lib/fleet-replicate.js' |
| 24 | import { PeerFront, RPC_PROTOCOL } from '../lib/peer-front.js' | |
| 25 | import { PeerRouter } from '../lib/peer-router.js' | |
| 24 | import { DidPeer, RPC_PROTOCOL } from 'atproto-did-peer' | |
| 25 | import { mkDidRouter } from './mk-did-router.js' | |
| 26 | 26 | import { pushAnnounce } from './serve-announce.js' |
| 27 | 27 | |
| 28 | 28 | const DID = 'did:plc:frontme' |
| … | … | @@ -36,16 +36,16 @@ async function waitFor (fn, timeout = 12000) { |
| 36 | 36 | return false |
| 37 | 37 | } |
| 38 | 38 | |
| 39 | // Test harness replacing the removed PeerFront.connect() (the serverKey-directional construction). | |
| 40 | // A standalone front, for the two read/write-surface tests: connect to the peer by its swarm key, | |
| 41 | // replicate, open the repo+firehose cores, and open the write RPC — the same steps the router does | |
| 42 | // per connection, but for one peer without the full PeerRouter. Uses connectReplica (a live helper | |
| 43 | // in a separate subsystem) purely as test plumbing; the shipped PeerFront has no serverKey. | |
| 44 | async function connectFront (front, serverKeyHex) { | |
| 45 | front._conn = await connectReplica(front.corestore, front.swarm, serverKeyHex) | |
| 46 | front.rpc = new ProtomuxRPC(front._conn, { protocol: RPC_PROTOCOL, valueEncoding: c.json }) | |
| 47 | front.repo = await new RepoStore(front.corestore, { key: front.repoKey }).ready() | |
| 48 | front.seq = await new FirehoseSequencer(front.corestore.get({ key: front.firehoseKey })).ready() | |
| 39 | // Test harness: a standalone DidPeer front, for the two read/write-surface tests. Connect to the peer by | |
| 40 | // its swarm key, replicate, open the repo+firehose cores via provide() (which seeds + reconciles the head), | |
| 41 | // and open the write RPC — the same steps the router does per connection, but for one peer without the full | |
| 42 | // DidPeerRouter. Uses connectReplica (a live helper in a separate subsystem) purely as test plumbing. | |
| 43 | // Does NOT wait for a head: test 2's peer has no commit yet (it signs on the relayed applyWrite). | |
| 44 | async function connectFront (corestore, swarm, serverKeyHex, { repoKey, firehoseKey, did, handle, verifyKey }) { | |
| 45 | const conn = await connectReplica(corestore, swarm, serverKeyHex) | |
| 46 | const front = new DidPeer(corestore, { did, verifyKey, handle }) | |
| 47 | front.rpc = new ProtomuxRPC(conn, { protocol: RPC_PROTOCOL, valueEncoding: c.json }) | |
| 48 | await front.provide({ repoKey, firehoseKey }) | |
| 49 | 49 | return front |
| 50 | 50 | } |
| 51 | 51 | |
| … | … | @@ -69,8 +69,7 @@ test('bridge PeerFront serves a remote peer getRepo + subscribeRepos — KEYLESS |
| 69 | 69 | // --- bridge side: PeerFront holds NO key --- |
| 70 | 70 | const bridgeStore = openCorestore(bridgeDir) |
| 71 | 71 | const bridgeSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap }) |
| 72 | const front = new PeerFront(bridgeStore, bridgeSwarm, { repoKey: repo.core.key, firehoseKey: seq.core.key, did: DID, handle: HANDLE }) | |
| 73 | await connectFront(front, served.serverKey) | |
| 72 | const front = await connectFront(bridgeStore, bridgeSwarm, served.serverKey, { repoKey: repo.core.key, firehoseKey: seq.core.key, did: DID, handle: HANDLE, verifyKey: keyFromPrivate(PRIV).publicKey }) | |
| 74 | 73 | |
| 75 | 74 | t.teardown(async () => { |
| 76 | 75 | served.close() |
| … | … | @@ -132,8 +131,7 @@ test('PeerFront relays a write over RPC — the peer signs, the bridge stays key |
| 132 | 131 | |
| 133 | 132 | const bridgeStore = openCorestore(bridgeDir) |
| 134 | 133 | const bridgeSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap }) |
| 135 | const front = new PeerFront(bridgeStore, bridgeSwarm, { repoKey: repo.core.key, firehoseKey: seq.core.key, did: DID, handle: HANDLE }) | |
| 136 | await connectFront(front, served.serverKey) | |
| 134 | const front = await connectFront(bridgeStore, bridgeSwarm, served.serverKey, { repoKey: repo.core.key, firehoseKey: seq.core.key, did: DID, handle: HANDLE, verifyKey: keyFromPrivate(PRIV).publicKey }) | |
| 137 | 135 | |
| 138 | 136 | t.teardown(async () => { |
| 139 | 137 | served.close() |
| … | … | @@ -149,7 +147,7 @@ test('PeerFront relays a write over RPC — the peer signs, the bridge stays key |
| 149 | 147 | t.ok(res.commitCid, 'peer returned a signed commit for the bridge-relayed write') |
| 150 | 148 | |
| 151 | 149 | // the signed write flows back on the peer's firehose; the bridge keylessly serves the updated repo |
| 152 | t.ok(await waitFor(async () => (await front.repo.records()).length === 1), 'bridge replicated the peer-signed record') | |
| 150 | t.ok(await waitFor(async () => front.head && (await front.repo.records()).length === 1), 'bridge replicated the peer-signed record') | |
| 153 | 151 | const { carBytes } = await front.getRepo() |
| 154 | 152 | const signed = loadCommit(await readCar(carBytes)).value |
| 155 | 153 | const pub = keyFromPrivate(PRIV).publicKey |
| … | … | @@ -181,7 +179,7 @@ async function servePeer (testnet, did, priv) { |
| 181 | 179 | function mkRouter (testnet, fronted, keyByDid) { |
| 182 | 180 | const store = openCorestore(mkdtempSync(join(tmpdir(), 'pr-bridge-'))) |
| 183 | 181 | const swarm = new Hyperswarm({ bootstrap: testnet.bootstrap }) |
| 184 | const router = new PeerRouter(swarm, store, { fronted, resolveVerifyKey: async (did) => keyByDid[did] }) | |
| 182 | const router = mkDidRouter(swarm, store, { fronted, resolveVerifyKey: async (did) => keyByDid[did] }) | |
| 185 | 183 | return { store, swarm, router, close: async () => { await router.close(); await swarm.destroy(); await store.close() } } |
| 186 | 184 | } |
| 187 | 185 | |
| … | … | @@ -200,7 +198,7 @@ test('PeerRouter fronts a peer discovered by DID + rejects a wrong key (connect- |
| 200 | 198 | t.ok(verifyCommit(loadCommit(await readCar(carBytes)).value, goodKey), 'discovered peer serves a signed CAR (keyless)') |
| 201 | 199 | |
| 202 | 200 | // a squatter/wrong-key: connect-time auth refuses to front |
| 203 | await t.exception(bad.router.front(DID), 'refuses to front a repo that does not verify against the DID key') | |
| 201 | await t.exception(bad.router.front(DID, { timeout: 3000 }), 'refuses to front a repo that does not verify against the DID key') | |
| 204 | 202 | }) |
| 205 | 203 | |
| 206 | 204 | // Regression for the discover RACE (2026-07-06): the old discover used swarm.once('connection'), |
| … | … | @@ -264,7 +262,7 @@ test('PeerRouter joins each fronted DID topic symmetrically (server:true is load |
| 264 | 262 | flush () { return Promise.resolve() } |
| 265 | 263 | } |
| 266 | 264 | // the router joins every fronted DID's topic in its constructor (like keet activateRoom) |
| 267 | const r = new PeerRouter(fakeSwarm, {}, { fronted: [DID], resolveVerifyKey: async () => new Uint8Array(33).fill(1) }) | |
| 265 | const r = mkDidRouter(fakeSwarm, {}, { fronted: [DID], resolveVerifyKey: async () => new Uint8Array(33).fill(1) }) | |
| 268 | 266 | t.ok(r, 'router constructed') |
| 269 | 267 | t.is(joins.length, 1, 'router joins the fronted DID topic once') |
| 270 | 268 | t.is(joins[0].client, true, 'joins as client (look up)') |
| … | … | @@ -13,7 +13,7 @@ import { RepoManager } from 'atproto-store/lib/repo-manager.js' |
| 13 | 13 | import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js' |
| 14 | 14 | import { keyFromPrivate } from 'atproto-identity' |
| 15 | 15 | import { discoveryTopic } from 'atproto-identity/lib/discovery.js' |
| 16 | import { PeerRouter } from '../lib/peer-router.js' | |
| 16 | import { mkDidRouter } from './mk-did-router.js' | |
| 17 | 17 | import { FleetRelay } from '../lib/fleet-relay.js' |
| 18 | 18 | import { pushAnnounce } from './serve-announce.js' |
| 19 | 19 | |
| … | … | @@ -42,7 +42,7 @@ test('peer RESTART (same keys) — front self-heals, new commit replicates witho |
| 42 | 42 | const routerStore = openCorestore(mkdtempSync(join(tmpdir(), 'restart-router-'))) |
| 43 | 43 | const routerSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap }) |
| 44 | 44 | const relay = await new FleetRelay(routerStore.get({ name: 'peer-relay-fh', valueEncoding: 'json' })).ready() |
| 45 | const router = new PeerRouter(routerSwarm, routerStore, { | |
| 45 | const router = mkDidRouter(routerSwarm, routerStore, { | |
| 46 | 46 | fronted: [DID], resolveVerifyKey: async () => keyFromPrivate(PRIV).publicKey, relay, |
| 47 | 47 | refreshMs: 1000, refreshTimeoutMs: 800, staleStrikeLimit: 2 // fast watchdog so the test doesn't wait 30s |
| 48 | 48 | }) |
| … | … | @@ -18,7 +18,7 @@ import { RepoManager } from 'atproto-store/lib/repo-manager.js' |
| 18 | 18 | import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js' |
| 19 | 19 | import { keyFromPrivate } from 'atproto-identity' |
| 20 | 20 | import { discoveryTopic } from 'atproto-identity/lib/discovery.js' |
| 21 | import { PeerRouter } from '../lib/peer-router.js' | |
| 21 | import { mkDidRouter } from './mk-did-router.js' | |
| 22 | 22 | import { FleetRelay } from '../lib/fleet-relay.js' |
| 23 | 23 | import { pushAnnounce } from './serve-announce.js' |
| 24 | 24 | |
| … | … | @@ -49,7 +49,7 @@ test('PeerRouter: isFronted gate + discover-on-first-use + caching', async (t) = |
| 49 | 49 | const routerStore = openCorestore(mkdtempSync(join(tmpdir(), 'router-'))) |
| 50 | 50 | const routerSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap }) |
| 51 | 51 | const relay = await new FleetRelay(routerStore.get({ name: 'peer-relay-fh', valueEncoding: 'json' })).ready() |
| 52 | const router = new PeerRouter(routerSwarm, routerStore, { | |
| 52 | const router = mkDidRouter(routerSwarm, routerStore, { | |
| 53 | 53 | fronted: [DID], |
| 54 | 54 | resolveVerifyKey: async () => keyFromPrivate(PRIV).publicKey, // injected; no PLC fetch in the test |
| 55 | 55 | relay |
| … | … | @@ -132,7 +132,7 @@ test('PeerRouter live-follows a fronted peer across new commits AND a peer recon |
| 132 | 132 | |
| 133 | 133 | const routerStore = openCorestore(mkdtempSync(join(tmpdir(), 'follow-router-'))) |
| 134 | 134 | const routerSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap }) |
| 135 | const router = new PeerRouter(routerSwarm, routerStore, { fronted: [DID2], resolveVerifyKey: async () => keyFromPrivate(PRIV2).publicKey }) | |
| 135 | const router = mkDidRouter(routerSwarm, routerStore, { fronted: [DID2], resolveVerifyKey: async () => keyFromPrivate(PRIV2).publicKey }) | |
| 136 | 136 | |
| 137 | 137 | t.teardown(async () => { |
| 138 | 138 | await router.close() |
| … | … | @@ -181,12 +181,12 @@ test('PeerRouter re-fronts a peer that REBUILDS its cores with new keys — no r |
| 181 | 181 | let peerSwarm = mkSwarm(); await peerSwarm.join(discoveryTopic(D), { server: true, client: false }).flushed() |
| 182 | 182 | const routerStore = openCorestore(mkdtempSync(join(tmpdir(), 'rebuild-router-'))); const routerSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap }) |
| 183 | 183 | const relay = await new FleetRelay(routerStore.get({ name: 'fh', valueEncoding: 'json' })).ready() |
| 184 | const router = new PeerRouter(routerSwarm, routerStore, { fronted: [D], resolveVerifyKey: async () => keyFromPrivate(P).publicKey, relay }) | |
| 184 | const router = mkDidRouter(routerSwarm, routerStore, { fronted: [D], resolveVerifyKey: async () => keyFromPrivate(P).publicKey, relay }) | |
| 185 | 185 | t.teardown(async () => { await router.close(); await relay.close(); await peerSwarm.destroy(); await routerSwarm.destroy(); await cores.seq.close(); await cores.repo.close(); await peerStore.close(); await routerStore.close(); await testnet.destroy(); rmSync(peerDir, { recursive: true, force: true }) }) |
| 186 | 186 | |
| 187 | 187 | const f1 = await router.front(D) |
| 188 | 188 | t.is((await f1.getLatestCommit()).rev, rev1, 'fronted the v1 cores') |
| 189 | const v1key = b4a.toString(f1.repoKey, 'hex') | |
| 189 | const v1key = b4a.toString(f1._repoKey, 'hex') | |
| 190 | 190 | |
| 191 | 191 | // REBUILD: brand-new cores (new keys) under the same DID; peer reconnects announcing them. |
| 192 | 192 | await peerSwarm.destroy() |
| … | … | @@ -195,8 +195,8 @@ test('PeerRouter re-fronts a peer that REBUILDS its cores with new keys — no r |
| 195 | 195 | |
| 196 | 196 | const refronted = await waitFor(async () => { |
| 197 | 197 | const f = await router.front(D).catch(() => null) |
| 198 | if (!f || b4a.toString(f.repoKey, 'hex') === v1key) return false | |
| 199 | const lc = await f.getLatestCommit().catch(() => null) | |
| 198 | if (!f || b4a.toString(f._repoKey, 'hex') === v1key) return false | |
| 199 | const lc = f.getLatestCommit() | |
| 200 | 200 | return !!lc && lc.rev === rev2 |
| 201 | 201 | }) |
| 202 | 202 | t.ok(refronted, 'router re-fronts with the NEW cores after a rebuild (no restart, new repoKey + rev2)') |
| … | … | @@ -4,11 +4,11 @@ |
| 4 | 4 | // standing download (re-attaches direction-free on reconnect) + hyperswarm's keepalive/topic-rejoin. |
| 5 | 5 | // Events set the liveness floor; the timer only catches ABSENCE. This unit-drives _refreshFronts. |
| 6 | 6 | import test from 'brittle' |
| 7 | import { PeerRouter } from '../lib/peer-router.js' | |
| 7 | import { mkDidRouter } from './mk-did-router.js' | |
| 8 | 8 | |
| 9 | 9 | function mkRouter () { |
| 10 | 10 | const swarm = { on () {}, off () {}, join () { return { flushed: async () => {} } }, flush () { return Promise.resolve() }, connections: [], destroy: async () => {} } |
| 11 | const router = new PeerRouter(swarm, {}, { fronted: [], resolveVerifyKey: async () => null, refreshMs: 1000, refreshTimeoutMs: 100, staleStrikeLimit: 2 }) | |
| 11 | const router = mkDidRouter(swarm, {}, { fronted: [], resolveVerifyKey: async () => null, refreshMs: 1000, refreshTimeoutMs: 100 }) | |
| 12 | 12 | clearInterval(router._refreshTimer) // drive _refreshFronts manually |
| 13 | 13 | return router |
| 14 | 14 | } |
| … | … | @@ -28,7 +28,6 @@ test('watchdog SKIPS the poll for a LIVE front (recent append) — event-informe |
| 28 | 28 | await router._refreshFronts() |
| 29 | 29 | |
| 30 | 30 | t.absent(updateCalled, 'live front (recent append) SKIPS the update({wait:true}) poll') |
| 31 | t.is(router._staleStrikes.get('did:plc:live|klive') || 0, 0, 'no strike accrued for a live front') | |
| 32 | 31 | t.absent(destroyed, 'live front connection is NOT dropped') |
| 33 | 32 | }) |
| 34 | 33 | |
| … | … | @@ -49,7 +48,6 @@ test('watchdog NUDGES a stalled front but NEVER destroys the connection (pull-on |
| 49 | 48 | |
| 50 | 49 | await router._refreshFronts() // nudge 1 (poll times out) |
| 51 | 50 | t.is(updateCalls, 1, 'stalled front (no recent append) IS nudged with a bounded update({wait}) pull') |
| 52 | t.is(router._staleStrikes.get('did:plc:stalled|kstalled'), 1, 'a failed nudge accrues a strike (diagnose visibility only)') | |
| 53 | 51 | t.absent(destroyed, 'connection NOT destroyed after a failed nudge') |
| 54 | 52 | |
| 55 | 53 | await router._refreshFronts() // nudge 2 — the OLD strike-limit "drop" point |