| … | … | @@ -96,23 +96,31 @@ export class PeerRouter { |
| 96 | 96 | return this._verifyKeys.get(did) |
| 97 | 97 | } |
| 98 | 98 | |
| 99 | | // The freshest OPEN front for a did = the co-owner membrane with the longest replicated firehose. |
| 100 | | // PREFER a front whose (did, firehoseKey) is CURRENTLY announced (live in _byDid): a peer that REBUILT |
| 101 | | // its cores (or a stale membrane a daemon abandoned) leaves a frozen front behind whose OLD key is no |
| 102 | | // longer announced — and it can tie the new core on raw length (both 1 commit), so a pure length pick |
| 103 | | // would keep serving the dead one. Among live co-owners take the longest firehose; fall back to any open |
| 104 | | // front only if none are currently announced (all co-owners momentarily disconnected → serve last-known). |
| 105 | | _freshestFront (did) { |
| 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) { |
| 106 | 109 | const set = this._fronts.get(did) |
| 107 | 110 | if (!set || !set.size) return null |
| 108 | 111 | const live = this._byDid.get(did) || new Map() |
| 109 | | let best = null; let bestLive = null |
| 112 | const scored = [] |
| 110 | 113 | for (const [fhKeyHex, f] of set) { |
| 111 | 114 | if (!f || !f.seq || !f.seq.core) continue |
| 112 | | if (!best || f.seq.core.length > best.seq.core.length) best = f |
| 113 | | if (live.has(fhKeyHex) && (!bestLive || f.seq.core.length > bestLive.seq.core.length)) bestLive = f |
| 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) }) |
| 114 | 118 | } |
| 115 | | return bestLive || best |
| 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 |
| 116 | 124 | } |
| 117 | 125 | |
| 118 | 126 | // Open (once) the front for ONE co-owner (did, firehoseKey) from its identified connection. Idempotent |
| … | … | @@ -138,7 +146,7 @@ export class PeerRouter { |
| 138 | 146 | } |
| 139 | 147 | if (f.handle) this._setHandle(f.handle, did) |
| 140 | 148 | const ws = this._waiters.get(did); if (ws) { this._waiters.delete(did); for (const fn of ws) fn() } |
| 141 | | this._pickRelaySource(did) // fold the freshest into the aggregate (single source per did) |
| 149 | this._pickRelaySource(did).catch(() => {}) // fold the freshest into the aggregate (single source per did) |
| 142 | 150 | return f |
| 143 | 151 | })().catch(() => null) // a co-owner that doesn't verify / has no commit yet just yields null; others still front |
| 144 | 152 | this._opening.set(openKey, p) |
| … | … | @@ -157,7 +165,7 @@ export class PeerRouter { |
| 157 | 165 | try { await withTimeout(f.seq.core.update({ wait: true }), this._refreshTimeoutMs); this._staleStrikes.set(did + '|' + fhKeyHex, 0) } |
| 158 | 166 | catch { this._staleStrikes.set(did + '|' + fhKeyHex, (this._staleStrikes.get(did + '|' + fhKeyHex) || 0) + 1) } |
| 159 | 167 | } |
| 160 | | this._pickRelaySource(did) // failover: ensure the relay is folding the freshest co-owner |
| 168 | this._pickRelaySource(did).catch(() => {}) // failover: ensure the relay is folding the freshest co-owner |
| 161 | 169 | } |
| 162 | 170 | } |
| 163 | 171 | |
| … | … | @@ -283,7 +291,7 @@ export class PeerRouter { |
| 283 | 291 | } |
| 284 | 292 | const entries = this._byDid.get(did) || new Map() |
| 285 | 293 | await Promise.all([...entries.keys()].map((fhKeyHex) => this._ensureFront(did, fhKeyHex).catch(() => null))) |
| 286 | | const best = this._freshestFront(did) |
| 294 | const best = await this._freshestFront(did) |
| 287 | 295 | if (!best) throw new Error(`PeerRouter: no frontable co-owner for ${did}`) |
| 288 | 296 | return best |
| 289 | 297 | } |
| … | … | @@ -292,9 +300,9 @@ export class PeerRouter { |
| 292 | 300 | // SAME frames under different sourceIds, so folding both duplicates every #commit to bsky's relay). On a |
| 293 | 301 | // source change (a fresher co-owner won, or the folded one stalled) swap removeSource→addSource + emit a |
| 294 | 302 | // #sync so the relay re-baselines. Idempotent when the freshest is unchanged. |
| 295 | | _pickRelaySource (did) { |
| 303 | async _pickRelaySource (did) { |
| 296 | 304 | if (!this.relay) return |
| 297 | | const best = this._freshestFront(did) |
| 305 | const best = await this._freshestFront(did) |
| 298 | 306 | if (!best || !best.seq || !best.seq.core) return |
| 299 | 307 | const key = b4a.toString(best.seq.core.key, 'hex') |
| 300 | 308 | if (this._relaySourceKey.get(did) === key) return |