| … | … | @@ -65,7 +65,13 @@ export class DidPeer { |
| 65 | 65 | // Optional write/session/blob RPC to the peer (peer-signs-over-RPC). Set by DidPeerRouter from the |
| 66 | 66 | // per-connection ProtomuxRPC when the router runs with `{ rpc: true }`. Null for a read-only consumer |
| 67 | 67 | // (favela) — the keyless read surface below needs no RPC. |
| 68 | | this.rpc = null |
| 68 | // |
| 69 | // A SET, not a slot. A front is keyed by firehoseKey, but the key HOLDER and every keyless SEEDER |
| 70 | // announce the SAME core key, so a single slot was won by whoever announced last — and a seeder answers |
| 71 | // `UNKNOWN_METHOD: Unknown method 'get-blob'` because it serves no application RPC at all. That is the |
| 72 | // favela getBlob failure and e2jev's intermittent images. `rpc` stays a property (assignment still |
| 73 | // works, and consumers still read it) but is now backed by the candidate list. |
| 74 | this._rpcs = [] |
| 69 | 75 | this._repoKey = null |
| 70 | 76 | this._firehoseKey = null |
| 71 | 77 | this._channels = new Set() |
| … | … | @@ -73,6 +79,44 @@ export class DidPeer { |
| 73 | 79 | this._opening = null |
| 74 | 80 | } |
| 75 | 81 | |
| 82 | // ---- application RPC candidates ---- |
| 83 | // `rpc` reads as the preferred LIVE candidate and assigning to it adds one (or clears the set with null), |
| 84 | // so every existing caller — `if (!this.rpc) throw`, `peer.rpc = entry.rpc`, `f.rpc && !f.rpc.closed` — |
| 85 | // keeps working unchanged against a set. |
| 86 | get rpc () { const live = this._liveRpcs(); return live.length ? live[0] : null } |
| 87 | set rpc (v) { if (v) this.addRpc(v); else this._rpcs = [] } |
| 88 | |
| 89 | addRpc (rpc) { if (rpc && !this._rpcs.includes(rpc)) this._rpcs.push(rpc) } |
| 90 | _liveRpcs () { this._rpcs = this._rpcs.filter((r) => r && !r.closed); return this._rpcs } |
| 91 | |
| 92 | // Try each candidate until one ANSWERS. Only a "wrong peer" / "dead link" failure moves on: |
| 93 | // - UNKNOWN_METHOD the peer serves no such method (a seeder, not the holder) -> try the next |
| 94 | // - CHANNEL_CLOSED/DESTROYED the link died -> try the next |
| 95 | // - TIMEOUT_EXCEEDED ambiguous: retried for READS only, never for a write (a request that may have |
| 96 | // been applied must not be replayed against another peer) |
| 97 | // Anything else is the peer ANSWERING with a real error (REQUEST_ERROR: "blob not found", |
| 98 | // "unauthorized"); surfacing that beats masking it behind a retry. |
| 99 | async _call (method, params, { readOnly = false } = {}) { |
| 100 | const cands = this._liveRpcs() |
| 101 | if (!cands.length) throw new Error('DidPeer: no write RPC (read-only front)') |
| 102 | const nextOn = new Set(['UNKNOWN_METHOD', 'CHANNEL_CLOSED', 'CHANNEL_DESTROYED']) |
| 103 | if (readOnly) nextOn.add('TIMEOUT_EXCEEDED') |
| 104 | let last = null |
| 105 | for (let i = 0; i < cands.length; i++) { |
| 106 | const rpc = cands[i] |
| 107 | try { |
| 108 | const res = await rpc.request(method, params) |
| 109 | // Sticky: the peer that answered goes to the front so later calls do not re-pay the failed hops. |
| 110 | if (i > 0) { this._rpcs = [rpc, ...this._rpcs.filter((r) => r !== rpc)] } |
| 111 | return res |
| 112 | } catch (e) { |
| 113 | last = e |
| 114 | if (!nextOn.has(e && e.code)) throw e |
| 115 | } |
| 116 | } |
| 117 | throw last |
| 118 | } |
| 119 | |
| 76 | 120 | // This node HOLDS the cores (a signing device, a relay that already seeded, or a bootstrap). Open |
| 77 | 121 | // them so we can serve + announce. Idempotent. |
| 78 | 122 | async provide ({ repoKey, firehoseKey } = {}) { return this._openCores(repoKey, firehoseKey) } |
| … | … | @@ -241,7 +285,7 @@ export class DidPeer { |
| 241 | 285 | // the head we already verified off the firehose. A peer that serves a forged or rolled-back repo is |
| 242 | 286 | // rejected exactly as before — we never serve bytes we have not authenticated. |
| 243 | 287 | if (!this.rpc) throw new Error(`DidPeer: ${this.did} has no repo core and no RPC to derive one`) |
| 244 | | const r = await this.rpc.request('get-repo-car', {}) |
| 288 | const r = await this._call('get-repo-car', {}, { readOnly: true }) |
| 245 | 289 | if (!r || !r.car) throw new Error(`DidPeer: ${this.did} served no CAR`) |
| 246 | 290 | const carBytes = b4a.from(r.car, 'hex') |
| 247 | 291 | const signed = loadCommit(await readCar(carBytes)).value |
| … | … | @@ -328,15 +372,15 @@ export class DidPeer { |
| 328 | 372 | // ---- write / session / blob relay (peer-signs-over-RPC) — the PEER holds the key + applies + signs; |
| 329 | 373 | // this consumer never holds a key. Requires the router to run with `{ rpc: true }` (this.rpc set). The |
| 330 | 374 | // signed result flows back on the peer's firehose (which this DidPeer already replicates + reconciles). |
| 331 | | applyWrite (op) { if (!this.rpc) throw new Error('DidPeer: no write RPC (read-only front)'); return this.rpc.request('apply-write', op) } |
| 332 | | applyWrites (op) { if (!this.rpc) throw new Error('DidPeer: no write RPC (read-only front)'); return this.rpc.request('apply-writes', op) } |
| 333 | | createSession ({ identifier, password } = {}) { if (!this.rpc) throw new Error('DidPeer: no write RPC (read-only front)'); return this.rpc.request('create-session', { identifier, password }) } |
| 334 | | refreshSession (refreshJwt) { if (!this.rpc) throw new Error('DidPeer: no write RPC (read-only front)'); return this.rpc.request('refresh-session', { refreshJwt }) } |
| 335 | | getServiceAuth ({ aud, lxm, token } = {}) { if (!this.rpc) throw new Error('DidPeer: no write RPC (read-only front)'); return this.rpc.request('get-service-auth', { aud, lxm, token }) } |
| 336 | | getPreferences ({ token } = {}) { if (!this.rpc) throw new Error('DidPeer: no write RPC (read-only front)'); return this.rpc.request('get-preferences', { token }) } |
| 337 | | putPreferences ({ preferences, token } = {}) { if (!this.rpc) throw new Error('DidPeer: no write RPC (read-only front)'); return this.rpc.request('put-preferences', { preferences, token }) } |
| 338 | | getBlob (cid) { if (!this.rpc) throw new Error('DidPeer: no write RPC (read-only front)'); return this.rpc.request('get-blob', { cid }) } |
| 339 | | uploadBlob ({ bytes, mimeType, token } = {}) { if (!this.rpc) throw new Error('DidPeer: no write RPC (read-only front)'); return this.rpc.request('upload-blob', { bytes, mimeType, token }) } |
| 375 | applyWrite (op) { return this._call('apply-write', op) } |
| 376 | applyWrites (op) { return this._call('apply-writes', op) } |
| 377 | createSession ({ identifier, password } = {}) { return this._call('create-session', { identifier, password }) } |
| 378 | refreshSession (refreshJwt) { return this._call('refresh-session', { refreshJwt }) } |
| 379 | getServiceAuth ({ aud, lxm, token } = {}) { return this._call('get-service-auth', { aud, lxm, token }, { readOnly: true }) } |
| 380 | getPreferences ({ token } = {}) { return this._call('get-preferences', { token }, { readOnly: true }) } |
| 381 | putPreferences ({ preferences, token } = {}) { return this._call('put-preferences', { preferences, token }) } |
| 382 | getBlob (cid) { return this._call('get-blob', { cid }, { readOnly: true }) } |
| 383 | uploadBlob ({ bytes, mimeType, token } = {}) { return this._call('upload-blob', { bytes, mimeType, token }) } |
| 340 | 384 | |
| 341 | 385 | // Decode a #commit frame BODY (from firehose.decodeFrame) into normalized ops with the changed record |
| 342 | 386 | // INLINE — read from the frame's OWN CAR (body.blocks) keyed by op.cid, so it is race-free (does NOT |
| … | … | @@ -437,7 +481,8 @@ export class DidPeerRouter { |
| 437 | 481 | |
| 438 | 482 | this._peers = new Map() // did -> primary DidPeer (backward-compat add()/get()) |
| 439 | 483 | this._fronts = new Map() // did -> Map<fhKeyHex, DidPeer> (co-owned multi-front) |
| 440 | | this._byDid = new Map() // did -> Map<fhKeyHex, { conn, rpc, coords }> (identified connections) |
| 484 | this._byDid = new Map() // did -> Map<fhKeyHex, { coords, conns: Map<conn, rpc|null> }> |
| 485 | // MANY peers announce one (did, firehoseKey): the holder + every seeder |
| 441 | 486 | this._opts = new Map() // did -> { verifyKey, handle, onHead, onError } from add() |
| 442 | 487 | this._verifyKeys = new Map() // did -> Promise<pubkey> |
| 443 | 488 | this._fronted = new Set(fronted) |
| … | … | @@ -556,10 +601,17 @@ export class DidPeerRouter { |
| 556 | 601 | if (this._isDenied(did, fhKeyHex)) return |
| 557 | 602 | this._lastLive.set(did + '|' + fhKeyHex, Date.now()) |
| 558 | 603 | if (!this._byDid.has(did)) this._byDid.set(did, new Map()) |
| 559 | | this._byDid.get(did).set(fhKeyHex, { conn, rpc, coords: a }) |
| 604 | // MANY peers announce the SAME (did, firehoseKey): the key holder AND every keyless seeder. Keep them |
| 605 | // ALL — a single {conn, rpc} slot was overwritten by whoever announced last, so the front's application |
| 606 | // RPC could end up pointed at a seeder that answers UNKNOWN_METHOD for every method. |
| 607 | const entries = this._byDid.get(did) |
| 608 | if (!entries.has(fhKeyHex)) entries.set(fhKeyHex, { coords: a, conns: new Map() }) |
| 609 | const entry = entries.get(fhKeyHex) |
| 610 | entry.coords = a |
| 611 | entry.conns.set(conn, rpc || null) |
| 560 | 612 | const existing = this._fronts.get(did) && this._fronts.get(did).get(fhKeyHex) |
| 561 | 613 | if (existing) { // refresh write rpc + handle; nudge fetch (cores re-replicate via corestore regardless) |
| 562 | | if (this._useRpc && rpc) existing.rpc = rpc |
| 614 | if (this._useRpc && rpc) existing.addRpc(rpc) |
| 563 | 615 | if (a.handle) existing.handle = a.handle |
| 564 | 616 | // The peer went STATELESS while we still hold a repo-core replica (it used to project, now it |
| 565 | 617 | // announces repoKey:null). That replica is frozen at whatever it last replicated, so once the head |
| … | … | @@ -580,7 +632,13 @@ export class DidPeerRouter { |
| 580 | 632 | return |
| 581 | 633 | } |
| 582 | 634 | this._wake(did) |
| 583 | | this._ensureFront(did, fhKeyHex).catch(() => {}) |
| 635 | // Add THIS announcer's rpc once the front exists. An announce that arrives while the front is still |
| 636 | // OPENING takes the dedup path (it awaits the in-flight promise), and that open captured only the |
| 637 | // candidates present when it started — so without this the second announcer's rpc is silently dropped |
| 638 | // and the candidate set stays stuck at one, which is the single-slot bug wearing a different hat. |
| 639 | this._ensureFront(did, fhKeyHex) |
| 640 | .then((f) => { if (f && this._useRpc && rpc) f.addRpc(rpc) }) |
| 641 | .catch(() => {}) |
| 584 | 642 | } |
| 585 | 643 | |
| 586 | 644 | // Open (once, dedup-guarded) the front for ONE co-owner (did, firehoseKey). Reuses the primary DidPeer |
| … | … | @@ -613,8 +671,11 @@ export class DidPeerRouter { |
| 613 | 671 | } |
| 614 | 672 | if (!peer) peer = new DidPeer(this.corestore, { did, verifyKey, handle: entry.coords.handle || o.handle || null, onHead: o.onHead || null, onError: o.onError || null, noRepo: o.noRepo }) |
| 615 | 673 | if (entry.coords.handle) peer.handle = entry.coords.handle |
| 616 | | if (this._useRpc && entry.rpc) peer.rpc = entry.rpc |
| 674 | // EVERY peer announcing this (did, firehoseKey) is a candidate; _call tries them until one answers. |
| 675 | if (this._useRpc) for (const r of entry.conns.values()) if (r) peer.addRpc(r) |
| 617 | 676 | await peer.provide({ repoKey: entry.coords.repoKey, firehoseKey: entry.coords.firehoseKey }) |
| 677 | // Re-sync: more peers may have announced this same (did, firehoseKey) during the open above. |
| 678 | if (this._useRpc) for (const r of entry.conns.values()) if (r) peer.addRpc(r) |
| 618 | 679 | if (!this._fronts.has(did)) this._fronts.set(did, new Map()) |
| 619 | 680 | this._fronts.get(did).set(fhKeyHex, peer) |
| 620 | 681 | if (!this._peers.has(did)) this._peers.set(did, peer) // first front becomes the primary (get()) |
| … | … | @@ -791,7 +852,12 @@ export class DidPeerRouter { |
| 791 | 852 | // (Keeping it is right for a transient reconnect and WRONG forever: see _ageOutFronts + dropFront.) |
| 792 | 853 | _onConnClose (conn) { |
| 793 | 854 | for (const [did, entries] of this._byDid) { |
| 794 | | for (const [fhKeyHex, entry] of entries) { if (entry.conn === conn) entries.delete(fhKeyHex) } |
| 855 | // Drop only THIS connection's candidate. A (did, firehoseKey) announced by several peers stays |
| 856 | // identified while any of them is still connected — one seeder leaving must not un-identify the DID. |
| 857 | for (const [fhKeyHex, entry] of entries) { |
| 858 | entry.conns.delete(conn) |
| 859 | if (entry.conns.size === 0) entries.delete(fhKeyHex) |
| 860 | } |
| 795 | 861 | if (entries.size === 0) this._byDid.delete(did) |
| 796 | 862 | } |
| 797 | 863 | } |