| … | … | @@ -417,7 +417,7 @@ export class DidPeer { |
| 417 | 417 | export class DidPeerRouter { |
| 418 | 418 | constructor (corestore, { |
| 419 | 419 | swarm = null, fronted = [], relay = null, resolveVerifyKey = null, rpc = false, join = null, |
| 420 | | refreshMs = 10000, refreshTimeoutMs = 5000, handles = {}, onHandles = null |
| 420 | refreshMs = 10000, refreshTimeoutMs = 5000, staleFrontMs = 300000, handles = {}, onHandles = null |
| 421 | 421 | } = {}) { |
| 422 | 422 | this.corestore = corestore |
| 423 | 423 | this.swarm = swarm |
| … | … | @@ -427,6 +427,9 @@ export class DidPeerRouter { |
| 427 | 427 | this._join = join |
| 428 | 428 | this._refreshMs = refreshMs |
| 429 | 429 | this._refreshTimeoutMs = refreshTimeoutMs |
| 430 | // Age-out ceiling: a front whose peer has not announced for this long is dropped — but ONLY while a |
| 431 | // live, verified sibling remains to serve the DID (see _ageOutFronts). 0 disables age-out. |
| 432 | this._staleFrontMs = staleFrontMs |
| 430 | 433 | |
| 431 | 434 | this._peers = new Map() // did -> primary DidPeer (backward-compat add()/get()) |
| 432 | 435 | this._fronts = new Map() // did -> Map<fhKeyHex, DidPeer> (co-owned multi-front) |
| … | … | @@ -437,6 +440,8 @@ export class DidPeerRouter { |
| 437 | 440 | this._opening = new Map() // "did|fhKeyHex" -> in-flight open promise |
| 438 | 441 | this._waiters = new Map() // did -> Set<fn> woken when ANY co-owner of that did lands |
| 439 | 442 | this._lastAppend = new Map() // did -> ts of last firehose 'append' (event-driven liveness) |
| 443 | this._lastLive = new Map() // "did|fhKeyHex" -> ts of that co-owner's last announce (age-out input) |
| 444 | this._denied = new Map() // did -> Set<fhKeyHex> evicted by an operator; a re-announce must NOT resurrect |
| 440 | 445 | this._relaySourceKey = new Map() // did -> firehoseKeyHex currently folded into the relay |
| 441 | 446 | this._handleToDid = new Map(Object.entries(handles || {})) |
| 442 | 447 | this._onHandles = onHandles |
| … | … | @@ -538,6 +543,11 @@ export class DidPeerRouter { |
| 538 | 543 | const fhKeyHex = a.firehoseKey |
| 539 | 544 | if (a.handle) this._setHandle(a.handle, did) |
| 540 | 545 | if (!this._isCared(did)) return |
| 546 | // An operator-EVICTED co-owner must stay evicted. Without this, dropFront({forget:true}) is undone by |
| 547 | // the offender's very next announce (peers re-announce on connect AND on every append), which is the |
| 548 | // difference between an eviction an operator can rely on and one that lasts milliseconds. |
| 549 | if (this._isDenied(did, fhKeyHex)) return |
| 550 | this._lastLive.set(did + '|' + fhKeyHex, Date.now()) |
| 541 | 551 | if (!this._byDid.has(did)) this._byDid.set(did, new Map()) |
| 542 | 552 | this._byDid.get(did).set(fhKeyHex, { conn, rpc, coords: a }) |
| 543 | 553 | const existing = this._fronts.get(did) && this._fronts.get(did).get(fhKeyHex) |
| … | … | @@ -569,6 +579,7 @@ export class DidPeerRouter { |
| 569 | 579 | // Open (once, dedup-guarded) the front for ONE co-owner (did, firehoseKey). Reuses the primary DidPeer |
| 570 | 580 | // (from add()) for its FIRST co-owner; a co-owner with a different firehoseKey gets its own DidPeer. |
| 571 | 581 | async _ensureFront (did, fhKeyHex) { |
| 582 | if (this._isDenied(did, fhKeyHex)) return null // operator-evicted; re-front only after allowFront() |
| 572 | 583 | const set = this._fronts.get(did) |
| 573 | 584 | if (set && set.has(fhKeyHex)) return set.get(fhKeyHex) |
| 574 | 585 | const openKey = did + '|' + fhKeyHex |
| … | … | @@ -640,7 +651,16 @@ export class DidPeerRouter { |
| 640 | 651 | async front (did, { timeout } = {}) { |
| 641 | 652 | if (!this._isCared(did)) throw new Error(`DidPeerRouter: ${did} is not a fronted DID`) |
| 642 | 653 | this._joinDid(did) |
| 643 | | if (!this._byDid.has(did) || this._byDid.get(did).size === 0) await this._waitForDid(did, timeout || 20000) |
| 654 | if (!this._byDid.has(did) || this._byDid.get(did).size === 0) { |
| 655 | // Every co-owner is momentarily away. _freshestFront already documents a fallback to any VERIFIED |
| 656 | // front for exactly this case, but it was unreachable from here: _waitForDid blocked first, so a read |
| 657 | // during a blip stalled for the full timeout and then threw, even though a verified head was cached. |
| 658 | // (It is also what makes age-out's "never drop the last front" worth anything.) Cores keep |
| 659 | // replicating over the next connection, so the cached head is the freshest thing that exists. |
| 660 | const cached = this._freshestFront(did) |
| 661 | if (cached) return cached |
| 662 | await this._waitForDid(did, timeout || 20000) |
| 663 | } |
| 644 | 664 | const entries = this._byDid.get(did) || new Map() |
| 645 | 665 | await Promise.all([...entries.keys()].map((fh) => this._ensureFront(did, fh).catch(() => null))) |
| 646 | 666 | // A front is opened once provide() kicks off replication, but its head is only set once the firehose |
| … | … | @@ -680,6 +700,7 @@ export class DidPeerRouter { |
| 680 | 700 | |
| 681 | 701 | // A peer disconnected: drop its (did, firehoseKey) identified-connection entries so _onAnnounce re-runs |
| 682 | 702 | // on reconnect. Do NOT close the cached front — its cores keep replicating over any remaining/next conn. |
| 703 | // (Keeping it is right for a transient reconnect and WRONG forever: see _ageOutFronts + dropFront.) |
| 683 | 704 | _onConnClose (conn) { |
| 684 | 705 | for (const [did, entries] of this._byDid) { |
| 685 | 706 | for (const [fhKeyHex, entry] of entries) { if (entry.conn === conn) entries.delete(fhKeyHex) } |
| … | … | @@ -687,6 +708,96 @@ export class DidPeerRouter { |
| 687 | 708 | } |
| 688 | 709 | } |
| 689 | 710 | |
| 711 | // ---- eviction ---------------------------------------------------------------------------------- |
| 712 | // A cached front used to live for the life of the PROCESS: _onConnClose deliberately keeps it, and |
| 713 | // _ensureFront returns the existing entry on every later announce. So the only way to stop serving a |
| 714 | // wrong head (a peer that stopped announcing, or one announcing a FORKED line) was to restart the |
| 715 | // bridge — unacceptable for a production read surface. These are the surgical alternatives. |
| 716 | |
| 717 | _isDenied (did, fhKeyHex) { const s = this._denied.get(did); return !!(s && s.has(fhKeyHex)) } |
| 718 | |
| 719 | // Evict one co-owner's front (or every front of a DID with no firehoseKey). `forget: true` also |
| 720 | // DENIES that firehose key, so the peer's next announce cannot resurrect it — required whenever the |
| 721 | // reason for evicting is the content the front is serving, because the offender keeps announcing. |
| 722 | // Never denies when dropping ALL of a DID's fronts implicitly (that is forgetDid's job, and denying a |
| 723 | // whole DID would make it permanently unservable). |
| 724 | async dropFront (did, { firehoseKey = null, forget = false } = {}) { |
| 725 | const set = this._fronts.get(did) |
| 726 | const targets = firehoseKey ? [firehoseKey] : (set ? [...set.keys()] : []) |
| 727 | const dropped = [] |
| 728 | for (const fhKeyHex of targets) { |
| 729 | const f = set && set.get(fhKeyHex) |
| 730 | if (set) set.delete(fhKeyHex) |
| 731 | const entries = this._byDid.get(did) |
| 732 | if (entries) { entries.delete(fhKeyHex); if (entries.size === 0) this._byDid.delete(did) } |
| 733 | this._lastLive.delete(did + '|' + fhKeyHex) |
| 734 | if (forget && firehoseKey) { |
| 735 | if (!this._denied.has(did)) this._denied.set(did, new Set()) |
| 736 | this._denied.get(did).add(fhKeyHex) |
| 737 | } |
| 738 | // Stop folding an evicted front into the aggregate relay before it can emit another frame. |
| 739 | if (this._relaySourceKey.get(did) === fhKeyHex) { |
| 740 | this._relaySourceKey.delete(did) |
| 741 | if (this.relay) { try { this.relay.removeSource(fhKeyHex) } catch {} } |
| 742 | } |
| 743 | // Re-point the primary (get()/announce-held) at a surviving front before closing this one. |
| 744 | if (this._peers.get(did) === f) { |
| 745 | const survivor = set && set.size ? [...set.values()][0] : null |
| 746 | if (survivor) this._peers.set(did, survivor); else this._peers.delete(did) |
| 747 | } |
| 748 | if (f) { dropped.push(fhKeyHex); try { await f.close() } catch {} } |
| 749 | } |
| 750 | if (set && set.size === 0) this._fronts.delete(did) |
| 751 | // A surviving co-owner must take over the relay slot immediately, not at the next refresh tick. |
| 752 | this._pickRelaySource(did).catch(() => {}) |
| 753 | return { did, dropped, forgotten: forget && firehoseKey ? [firehoseKey] : [] } |
| 754 | } |
| 755 | |
| 756 | // Un-deny a previously forgotten front (omit firehoseKey to clear every denial for the DID) so the |
| 757 | // peer's next announce re-fronts it. An eviction with no undo is an outage waiting to happen. |
| 758 | allowFront (did, firehoseKey = null) { |
| 759 | const s = this._denied.get(did) |
| 760 | if (!s) return { did, allowed: [] } |
| 761 | const allowed = firehoseKey ? (s.delete(firehoseKey) ? [firehoseKey] : []) : [...s] |
| 762 | if (!firehoseKey) s.clear() |
| 763 | if (s.size === 0) this._denied.delete(did) |
| 764 | return { did, allowed } |
| 765 | } |
| 766 | |
| 767 | // Forget EVERYTHING cached for a DID — every co-owner front, its identified connections and its |
| 768 | // resolved verify key — so the next request re-fronts from scratch. The DID stays allow-listed (it must |
| 769 | // still be servable) and nothing is denied: this is the "re-derive, don't restart" lever. To evict a |
| 770 | // specific misbehaving co-owner and keep it out, use dropFront({ firehoseKey, forget: true }). |
| 771 | async forgetDid (did) { |
| 772 | const r = await this.dropFront(did) |
| 773 | this._byDid.delete(did) |
| 774 | this._verifyKeys.delete(did) |
| 775 | this._lastAppend.delete(did) |
| 776 | this._relaySourceKey.delete(did) |
| 777 | return { did, dropped: r.dropped } |
| 778 | } |
| 779 | |
| 780 | // Age out fronts whose peer has stopped announcing. Conservative by construction: a front is only |
| 781 | // dropped while a LIVE, VERIFIED sibling remains to serve the DID, so age-out can never be the reason a |
| 782 | // DID becomes unservable — the single-holder case (the overwhelmingly common one) is never touched. |
| 783 | async _ageOutFronts (did) { |
| 784 | if (!this._staleFrontMs) return [] |
| 785 | const set = this._fronts.get(did) |
| 786 | if (!set || set.size < 2) return [] |
| 787 | const live = this._byDid.get(did) || new Map() |
| 788 | const hasLiveVerified = [...set].some(([k, f]) => live.has(k) && f && f.head) |
| 789 | if (!hasLiveVerified) return [] |
| 790 | const now = Date.now() |
| 791 | const stale = [...set.keys()].filter((k) => !live.has(k) && now - (this._lastLive.get(did + '|' + k) || now) > this._staleFrontMs) |
| 792 | const dropped = [] |
| 793 | for (const k of stale) { |
| 794 | if (this._fronts.get(did) && this._fronts.get(did).size < 2) break // never drop the last front |
| 795 | const r = await this.dropFront(did, { firehoseKey: k }) |
| 796 | dropped.push(...r.dropped) |
| 797 | } |
| 798 | return dropped |
| 799 | } |
| 800 | |
| 690 | 801 | // Fold the FRESHEST co-owner's firehose into the aggregate relay (single source per DID — two co-owners |
| 691 | 802 | // are the SAME frames under different sourceIds, so folding both duplicates every #commit). On a source |
| 692 | 803 | // change (fresher co-owner won, or the folded one stalled) swap removeSource->addSource + emit #sync. |
| … | … | @@ -722,20 +833,25 @@ export class DidPeerRouter { |
| 722 | 833 | // cores live-follow via the standing download; this just nudges a front that saw no 'append' for a window |
| 723 | 834 | // and re-picks the relay source if the folded co-owner stalled while another advanced. |
| 724 | 835 | async _refreshFronts () { |
| 725 | | for (const [did, set] of this._fronts) { |
| 836 | for (const did of [...this._fronts.keys()]) { // snapshot: _ageOutFronts mutates _fronts |
| 837 | const set = this._fronts.get(did) |
| 838 | if (!set) continue |
| 726 | 839 | for (const [, f] of set) { |
| 727 | 840 | if (!f || !f.seq || !f.seq.core) continue |
| 728 | 841 | if (Date.now() - (this._lastAppend.get(did) || 0) < this._refreshMs) continue |
| 729 | 842 | try { await withTimeout(f.seq.core.update({ wait: true }), this._refreshTimeoutMs) } catch {} |
| 730 | 843 | } |
| 844 | try { await this._ageOutFronts(did) } catch {} |
| 731 | 845 | this._pickRelaySource(did).catch(() => {}) |
| 732 | 846 | } |
| 733 | 847 | } |
| 734 | 848 | |
| 735 | 849 | diagnose () { |
| 736 | 850 | const fronts = [] |
| 851 | const now = Date.now() |
| 737 | 852 | for (const [did, set] of this._fronts) { |
| 738 | 853 | for (const [fhKeyHex, f] of set) { |
| 854 | const live = (this._byDid.get(did) || new Map()).has(fhKeyHex) |
| 739 | 855 | fronts.push({ |
| 740 | 856 | did, |
| 741 | 857 | handle: f.handle || null, |
| … | … | @@ -745,14 +861,24 @@ export class DidPeerRouter { |
| 745 | 861 | rev: f.head ? f.head.rev : null, |
| 746 | 862 | rpcConnected: !!(f.rpc && !f.rpc.closed), |
| 747 | 863 | relaySource: this._relaySourceKey.get(did) === fhKeyHex, |
| 748 | | live: (this._byDid.get(did) || new Map()).has(fhKeyHex) |
| 864 | live, |
| 865 | // How long this co-owner has been silent, and whether age-out would take it — so an operator can |
| 866 | // SEE an evictable front instead of inferring one. |
| 867 | silentMs: live ? 0 : now - (this._lastLive.get(did + '|' + fhKeyHex) || now) |
| 749 | 868 | }) |
| 750 | 869 | } |
| 751 | 870 | } |
| 871 | const denied = {} |
| 872 | for (const [did, s] of this._denied) denied[did] = [...s] |
| 752 | 873 | return { |
| 753 | 874 | fronted: [...this._fronted], |
| 754 | 875 | cachedFronts: fronts.length, |
| 876 | // Distinct front OBJECTS. cachedFronts counts map entries, which is why an aliased set (one DidPeer |
| 877 | // filed under N keys) read as healthy right up to 0.2.6. |
| 878 | distinctFronts: new Set([].concat(...[...this._fronts.values()].map((s) => [...s.values()]))).size, |
| 755 | 879 | fronts, |
| 880 | denied, |
| 881 | staleFrontMs: this._staleFrontMs, |
| 756 | 882 | relay: this.relay ? { sources: this.relay.sourceCount ? this.relay.sourceCount() : null } : null |
| 757 | 883 | } |
| 758 | 884 | } |