| … | … | @@ -417,7 +417,8 @@ 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, staleFrontMs = 300000, handles = {}, onHandles = null | |
| 420 | refreshMs = 10000, refreshTimeoutMs = 5000, staleFrontMs = 300000, continuityScan = 500, | |
| 421 | handles = {}, onHandles = null | |
| 421 | 422 | } = {}) { |
| 422 | 423 | this.corestore = corestore |
| 423 | 424 | this.swarm = swarm |
| … | … | @@ -430,6 +431,9 @@ export class DidPeerRouter { |
| 430 | 431 | // Age-out ceiling: a front whose peer has not announced for this long is dropped — but ONLY while a |
| 431 | 432 | // live, verified sibling remains to serve the DID (see _ageOutFronts). 0 disables age-out. |
| 432 | 433 | this._staleFrontMs = staleFrontMs |
| 434 | // How many frames back the continuity walk will look for the head we already serve. 0 disables the | |
| 435 | // continuity gate entirely (freshest-by-rev only — the pre-0.2.8 behaviour). | |
| 436 | this._continuityScan = continuityScan | |
| 433 | 437 | |
| 434 | 438 | this._peers = new Map() // did -> primary DidPeer (backward-compat add()/get()) |
| 435 | 439 | this._fronts = new Map() // did -> Map<fhKeyHex, DidPeer> (co-owned multi-front) |
| … | … | @@ -442,6 +446,9 @@ export class DidPeerRouter { |
| 442 | 446 | this._lastAppend = new Map() // did -> ts of last firehose 'append' (event-driven liveness) |
| 443 | 447 | this._lastLive = new Map() // "did|fhKeyHex" -> ts of that co-owner's last announce (age-out input) |
| 444 | 448 | this._denied = new Map() // did -> Set<fhKeyHex> evicted by an operator; a re-announce must NOT resurrect |
| 449 | this._servedHead = new Map() // did -> { rev, cid, fhKeyHex } the head reads currently resolve to | |
| 450 | this._continuity = new Map() // "candCid|servedCid" -> bool (does the candidate chain from what we serve?) | |
| 451 | this._forkRejected = new Map() // "did|fhKeyHex" -> count of times this front was rejected as off-line | |
| 445 | 452 | this._relaySourceKey = new Map() // did -> firehoseKeyHex currently folded into the relay |
| 446 | 453 | this._handleToDid = new Map(Object.entries(handles || {})) |
| 447 | 454 | this._onHandles = onHandles |
| … | … | @@ -645,9 +652,90 @@ export class DidPeerRouter { |
| 645 | 652 | return chosen ? chosen.f : null |
| 646 | 653 | } |
| 647 | 654 | |
| 655 | // ---- continuity --------------------------------------------------------------------------------- | |
| 656 | // freshest-by-rev alone trusts a rev with NO continuity check, so a FORKED line whose rev sorts higher | |
| 657 | // simply wins. That is how loser-club's public getLatestCommit served a rival 1-record head | |
| 658 | // (3mrnruepxss22) over the authoritative 139-record one (3mrfgnyiurk22) until the bridge was bounced: a | |
| 659 | // rev REGRESSION was guarded against, a higher rev on a DIFFERENT LINE was not. | |
| 660 | // | |
| 661 | // Real co-owners are not a different line. `atproto-store/lib/tid.js:nextRev` makes the chain a pure | |
| 662 | // function of the previous rev, so two co-owners continuing from the same head over the same finalized | |
| 663 | // record set produce BYTE-IDENTICAL commits — a lagging co-owner's firehose is a strict PREFIX of a | |
| 664 | // fresher one's. So "chains from the head we serve" is decidable: walk the candidate's own frames back | |
| 665 | // and look for the exact commit we are serving. | |
| 666 | // | |
| 667 | // Enforced only while the INCUMBENT IS STILL LIVE. A peer that legitimately rebuilds its cores (storage | |
| 668 | // repair, rebuild-did) goes away and comes back announcing new keys on a new line, and refusing that | |
| 669 | // would strand the DID on dead cores — the case test/peer-router.test.js 'REBUILDS its cores' covers. | |
| 670 | // Live incumbent + a rival line => fork, keep serving the incumbent. No incumbent (gone, or evicted by | |
| 671 | // an operator) => the new line is the only truth on offer, take it. | |
| 672 | async _chainsFromServed (front, served) { | |
| 673 | if (!front || !front.head) return false | |
| 674 | if (front.head.rev === served.rev) return front.head.cid === served.cid // same rev MUST be same commit | |
| 675 | if (front.head.rev < served.rev) return false // regression (pre-existing policy) | |
| 676 | const ck = front.head.cid + '|' + served.cid | |
| 677 | if (this._continuity.has(ck)) return this._continuity.get(ck) | |
| 678 | const verdict = await this._scanForServed(front, served) // 'same' | 'fork' | 'unknown' | |
| 679 | if (verdict !== 'unknown') this._continuity.set(ck, verdict === 'same') // never cache an inconclusive read | |
| 680 | return verdict === 'same' | |
| 681 | } | |
| 682 | ||
| 683 | // Walk the candidate's firehose backwards looking for the served commit. Bounded by continuityScan. | |
| 684 | // Reads are wait:false — a block we do not hold yet is INCONCLUSIVE, not evidence of a fork (the verdict | |
| 685 | // is left uncached so the next read re-decides once replication catches up). | |
| 686 | async _scanForServed (front, served) { | |
| 687 | const core = front.seq && front.seq.core | |
| 688 | if (!core || !core.length) return 'unknown' | |
| 689 | const limit = Math.min(core.length, this._continuityScan) | |
| 690 | for (let i = 1; i <= limit; i++) { | |
| 691 | let body = null | |
| 692 | try { | |
| 693 | const raw = await core.get(core.length - i, { wait: false }) | |
| 694 | if (!raw) return 'unknown' // not replicated locally — cannot conclude | |
| 695 | const d = decodeCommitFrame(raw) | |
| 696 | body = d && d.body | |
| 697 | } catch { return 'unknown' } | |
| 698 | if (!body || !body.rev) continue // #sync/#identity/#account frames carry no commit | |
| 699 | if (body.rev === served.rev) return (body.commit && body.commit.toString() === served.cid) ? 'same' : 'fork' | |
| 700 | if (body.rev < served.rev) return 'fork' // walked PAST it without meeting it -> different line | |
| 701 | } | |
| 702 | return limit < core.length ? 'unknown' : 'fork' // exhausted the window vs exhausted the core | |
| 703 | } | |
| 704 | ||
| 705 | // The freshest front that also CHAINS from what we already serve. Returns the incumbent when every | |
| 706 | // fresher candidate is on a rival line. Records what it picks as the new incumbent. | |
| 707 | async _freshestChained (did) { | |
| 708 | const set = this._fronts.get(did) | |
| 709 | if (!set || !set.size) return null | |
| 710 | if (!this._continuityScan) return this._freshestFront(did) // gate disabled | |
| 711 | const live = this._byDid.get(did) || new Map() | |
| 712 | const scored = [] | |
| 713 | for (const [fhKeyHex, f] of set) { | |
| 714 | if (!f || !f.head || !f.head.rev) continue | |
| 715 | scored.push({ fhKeyHex, f, rev: f.head.rev, live: live.has(fhKeyHex) }) | |
| 716 | } | |
| 717 | if (!scored.length) return null | |
| 718 | const pool = scored.some((x) => x.live) ? scored.filter((x) => x.live) : scored | |
| 719 | pool.sort((a, b) => (a.rev < b.rev ? 1 : a.rev > b.rev ? -1 : 0)) // highest rev first | |
| 720 | const served = this._servedHead.get(did) | |
| 721 | const incumbentLive = !!(served && live.has(served.fhKeyHex) && set.has(served.fhKeyHex)) | |
| 722 | for (const cand of pool) { | |
| 723 | if (served && incumbentLive && cand.fhKeyHex !== served.fhKeyHex) { | |
| 724 | if (!(await this._chainsFromServed(cand.f, served))) { | |
| 725 | const k = did + '|' + cand.fhKeyHex | |
| 726 | this._forkRejected.set(k, (this._forkRejected.get(k) || 0) + 1) | |
| 727 | continue | |
| 728 | } | |
| 729 | } | |
| 730 | this._servedHead.set(did, { rev: cand.rev, cid: cand.f.head.cid, fhKeyHex: cand.fhKeyHex }) | |
| 731 | return cand.f | |
| 732 | } | |
| 733 | return (served && set.get(served.fhKeyHex)) || null // every rival was a fork: hold the line | |
| 734 | } | |
| 735 | ||
| 648 | 736 | // Get the FRESHEST verified front for a fronted DID: ensure fronts for every identified co-owner, then |
| 649 | // return the highest-rev one. This is what every read (getRepo/listRecords/…) resolves against — a | |
| 650 | // lagging co-owner never freezes the served head. | |
| 737 | // return the highest-rev one that chains from what we serve. This is what every read | |
| 738 | // (getRepo/listRecords/…) resolves against — a lagging co-owner never freezes the served head. | |
| 651 | 739 | async front (did, { timeout } = {}) { |
| 652 | 740 | if (!this._isCared(did)) throw new Error(`DidPeerRouter: ${did} is not a fronted DID`) |
| 653 | 741 | this._joinDid(did) |
| … | … | @@ -657,7 +745,7 @@ export class DidPeerRouter { |
| 657 | 745 | // during a blip stalled for the full timeout and then threw, even though a verified head was cached. |
| 658 | 746 | // (It is also what makes age-out's "never drop the last front" worth anything.) Cores keep |
| 659 | 747 | // replicating over the next connection, so the cached head is the freshest thing that exists. |
| 660 | const cached = this._freshestFront(did) | |
| 748 | const cached = await this._freshestChained(did) | |
| 661 | 749 | if (cached) return cached |
| 662 | 750 | await this._waitForDid(did, timeout || 20000) |
| 663 | 751 | } |
| … | … | @@ -667,12 +755,12 @@ export class DidPeerRouter { |
| 667 | 755 | // commit REPLICATES + verifies (event-driven, on 'append'). For a single front() call (no retry — a |
| 668 | 756 | // real getRepo request), wait for AT LEAST ONE co-owner to reach a verified head, then serve the |
| 669 | 757 | // freshest among those ready. A slow co-owner never delays serving from a ready one (freshest-of-ready). |
| 670 | let best = this._freshestFront(did) | |
| 758 | let best = await this._freshestChained(did) | |
| 671 | 759 | if (!best) { |
| 672 | 760 | const end = Date.now() + (timeout || 20000) |
| 673 | 761 | while (!best && Date.now() < end) { |
| 674 | 762 | await new Promise((r) => setTimeout(r, 100)) |
| 675 | best = this._freshestFront(did) | |
| 763 | best = await this._freshestChained(did) | |
| 676 | 764 | } |
| 677 | 765 | } |
| 678 | 766 | if (!best) throw new Error(`DidPeerRouter: no frontable co-owner for ${did}`) |
| … | … | @@ -745,6 +833,12 @@ export class DidPeerRouter { |
| 745 | 833 | const survivor = set && set.size ? [...set.values()][0] : null |
| 746 | 834 | if (survivor) this._peers.set(did, survivor); else this._peers.delete(did) |
| 747 | 835 | } |
| 836 | // If the evicted front was the INCUMBENT, clear it: the operator's drop is an explicit decision that | |
| 837 | // this line is wrong, so a surviving rival must be allowed to take over instead of being held off by | |
| 838 | // the continuity gate as a "fork" of the head we just threw away. | |
| 839 | const served = this._servedHead.get(did) | |
| 840 | if (served && served.fhKeyHex === fhKeyHex) this._servedHead.delete(did) | |
| 841 | this._forkRejected.delete(did + '|' + fhKeyHex) | |
| 748 | 842 | if (f) { dropped.push(fhKeyHex); try { await f.close() } catch {} } |
| 749 | 843 | } |
| 750 | 844 | if (set && set.size === 0) this._fronts.delete(did) |
| … | … | @@ -774,6 +868,7 @@ export class DidPeerRouter { |
| 774 | 868 | this._verifyKeys.delete(did) |
| 775 | 869 | this._lastAppend.delete(did) |
| 776 | 870 | this._relaySourceKey.delete(did) |
| 871 | this._servedHead.delete(did) // no incumbent => the continuity gate starts clean on the next front() | |
| 777 | 872 | return { did, dropped: r.dropped } |
| 778 | 873 | } |
| 779 | 874 | |
| … | … | @@ -803,7 +898,9 @@ export class DidPeerRouter { |
| 803 | 898 | // change (fresher co-owner won, or the folded one stalled) swap removeSource->addSource + emit #sync. |
| 804 | 899 | async _pickRelaySource (did) { |
| 805 | 900 | if (!this.relay) return |
| 806 | const best = this._freshestFront(did) | |
| 901 | // The relay must fold the SAME line the read surface serves — folding a fork's frames would publish it | |
| 902 | // to every downstream subscriber, which is worse than serving it on one read. | |
| 903 | const best = await this._freshestChained(did) | |
| 807 | 904 | if (!best || !best.seq || !best.seq.core) return |
| 808 | 905 | const key = b4a.toString(best.seq.core.key, 'hex') |
| 809 | 906 | if (this._relaySourceKey.get(did) === key) return |
| … | … | @@ -864,7 +961,11 @@ export class DidPeerRouter { |
| 864 | 961 | live, |
| 865 | 962 | // How long this co-owner has been silent, and whether age-out would take it — so an operator can |
| 866 | 963 | // SEE an evictable front instead of inferring one. |
| 867 | silentMs: live ? 0 : now - (this._lastLive.get(did + '|' + fhKeyHex) || now) | |
| 964 | silentMs: live ? 0 : now - (this._lastLive.get(did + '|' + fhKeyHex) || now), | |
| 965 | // Is this the head reads resolve to, and has it been refused as a rival line? A front sitting at | |
| 966 | // a HIGHER rev than the served one with forkRejected > 0 is the loser-club signature. | |
| 967 | served: (this._servedHead.get(did) || {}).fhKeyHex === fhKeyHex, | |
| 968 | forkRejected: this._forkRejected.get(did + '|' + fhKeyHex) || 0 | |
| 868 | 969 | }) |
| 869 | 970 | } |
| 870 | 971 | } |
| … | … | @@ -1,6 +1,6 @@ |
| 1 | 1 | { |
| 2 | 2 | "name": "atproto-did-peer", |
| 3 | "version": "0.2.7", | |
| 3 | "version": "0.2.8", | |
| 4 | 4 | "description": "Keyless, direction-free P2P reader/seeder for an atproto DID's repo+firehose Hypercores: announce-push discovery + event-driven head reconciliation + a keyless read surface. The shared primitive beneath atproto-bridge, atproto-relay and favela (extracted from atproto-bridge did-peer.js).", |
| 5 | 5 | "type": "module", |
| 6 | 6 | "main": "index.js", |
| … | … | @@ -8,7 +8,7 @@ |
| 8 | 8 | ".": "./index.js" |
| 9 | 9 | }, |
| 10 | 10 | "scripts": { |
| 11 | "test": "node test/commitops.test.mjs && node test/router.test.mjs && node test/co-owned.test.mjs && node test/coowner-set.test.mjs && node test/evict.test.mjs && node test/no-repo.test.mjs", | |
| 11 | "test": "node test/commitops.test.mjs && node test/router.test.mjs && node test/co-owned.test.mjs && node test/coowner-set.test.mjs && node test/evict.test.mjs && node test/fork.test.mjs && node test/no-repo.test.mjs", | |
| 12 | 12 | "test:live": "node test/read-live.mjs" |
| 13 | 13 | }, |
| 14 | 14 | "dependencies": { |
| … | … | @@ -112,8 +112,18 @@ assert(await served(A.router) === 1, 'still serving the surviving co-owner after |
| 112 | 112 | |
| 113 | 113 | const allow = A.router.allowFront(DID, ahead.fhKey) |
| 114 | 114 | assert(allow.allowed.includes(ahead.fhKey), 'allowFront clears the denial') |
| 115 | await ahead.append() // next announce may re-front now | |
| 116 | assert(await waitServed(A.router, 4) === 4, 'after allowFront the evicted co-owner is fronted again (4 records)') | |
| 115 | await ahead.append() // its next announce may re-front it now | |
| 116 | const refronted = await (async () => { | |
| 117 | const end = Date.now() + 15000 | |
| 118 | while (Date.now() < end) { if (frontKeys(A.router).includes(ahead.fhKey)) return true; await sleep(250) } | |
| 119 | return false | |
| 120 | })() | |
| 121 | assert(refronted, 'after allowFront the evicted co-owner is FRONTED again on its next announce') | |
| 122 | // NOTE: re-fronted is not the same as re-SERVED. These two holders are independent LINES (the cheapest way | |
| 123 | // to tell them apart by record count), so the 0.2.8 continuity gate keeps reads on the incumbent while the | |
| 124 | // incumbent is live — deliberately. Line switching is proven in test/fork.test.mjs against a fixture that | |
| 125 | // models same-line co-owners properly. | |
| 126 | assert(await served(A.router) === 1, 'reads stay on the incumbent line (continuity gate) after re-fronting') | |
| 117 | 127 | await A.close() |
| 118 | 128 | |
| 119 | 129 | // ---------------- B. age-out ---------------- |
| … | … | @@ -0,0 +1,157 @@ |
| 1 | // CONTINUITY: a higher rev on a DIFFERENT LINE must not win the read surface (hyper-social#16 P0). | |
| 2 | // | |
| 3 | // loser-club: admitting a co-owner flipped the creator role, the new node authored a RIVAL GENESIS | |
| 4 | // (1 record), and because its rev sorted HIGHER than the authoritative 139-record head, the bridge's | |
| 5 | // freshest-by-rev logic served the fork from the public getLatestCommit until the process was bounced. | |
| 6 | // A rev REGRESSION was already guarded against; a higher rev on a different line was not. | |
| 7 | // | |
| 8 | // The fixture models co-owners the way the system actually produces them. atproto-store/lib/tid.js | |
| 9 | // `nextRev` makes the chain a pure function of the previous rev, so co-owners continuing from the same | |
| 10 | // head over the same record set emit BYTE-IDENTICAL commits — a lagging co-owner's firehose is a strict | |
| 11 | // PREFIX of a fresher one's. So: ONE canonical line, each co-owner replaying the same records into its own | |
| 12 | // repo and the same canonical FRAMES into its own firehose, truncated at its own lag. The fork is a | |
| 13 | // genuinely independent line whose rev sorts higher (a later clock base). | |
| 14 | // Run: node test/fork.test.mjs | |
| 15 | import { mkdtempSync, rmSync } from 'node:fs' | |
| 16 | import { join } from 'node:path' | |
| 17 | import { tmpdir } from 'node:os' | |
| 18 | import createTestnet from '@hyperswarm/testnet' | |
| 19 | import Hyperswarm from 'hyperswarm' | |
| 20 | import { RepoStore } from 'atproto-store/lib/repo-store.js' | |
| 21 | import { RepoManager } from 'atproto-store/lib/repo-manager.js' | |
| 22 | import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js' | |
| 23 | import { openCorestore } from 'atproto-store/lib/corestore.js' | |
| 24 | import { createAtprotoIdentityFromKey } from 'atproto-identity/lib/atproto-identity.js' | |
| 25 | import { discoveryTopic } from 'atproto-identity/lib/discovery.js' | |
| 26 | import { DidPeer, DidPeerRouter } from '../index.js' | |
| 27 | ||
| 28 | const assert = (c, m) => { if (!c) { console.error('FAIL:', m); cleanupSync(); process.exit(1) } console.log(' ok -', m) } | |
| 29 | const iso = '2026-07-27T00:00:00.000Z' | |
| 30 | const dirs = [] | |
| 31 | const toClose = [] | |
| 32 | function newDir (p) { const d = mkdtempSync(join(tmpdir(), p)); dirs.push(d); return d } | |
| 33 | function cleanupSync () { for (const d of dirs) { try { rmSync(d, { recursive: true, force: true }) } catch {} } } | |
| 34 | const sleep = (ms) => new Promise((r) => setTimeout(r, ms)) | |
| 35 | const post = (rk, text) => ({ action: 'create', collection: 'app.bsky.feed.post', rkey: rk, record: { $type: 'app.bsky.feed.post', text, createdAt: iso, langs: ['en'] } }) | |
| 36 | ||
| 37 | const id = createAtprotoIdentityFromKey(new Uint8Array(32).fill(3), { handle: 'fork.test', pdsUrl: 'https://pds.local' }) | |
| 38 | const DID = id.did | |
| 39 | ||
| 40 | const testnet = await createTestnet(3) | |
| 41 | const bootstrap = testnet.bootstrap | |
| 42 | ||
| 43 | // ---- the canonical line: ONE writer, N commits. Its FRAMES are what every co-owner replicates. ---- | |
| 44 | const OPS = [post('3aaaaaaaaaaa2', 'one'), post('3bbbbbbbbbbb2', 'two'), post('3ccccccccccc2', 'three'), post('3ddddddddddd2', 'four')] | |
| 45 | const canonStore = openCorestore(newDir('canon-')) | |
| 46 | const canonRepo = await new RepoStore(canonStore, { name: `atproto-repo:${DID}` }).ready() | |
| 47 | let canonMicros = 1_700_000_000_000_000n | |
| 48 | const canonMgr = await new RepoManager(canonRepo, { did: DID, privateKey: id.privateKey, clock: () => (canonMicros += 1_000_000n) }).init() | |
| 49 | const canonSeq = await new FirehoseSequencer(canonStore.get({ name: 'canon-firehose' })).ready() | |
| 50 | const canonRevs = [] | |
| 51 | for (const op of OPS) { const w = await canonMgr.applyWrite(op); await canonSeq.sequenceCommit(w, { time: iso }); canonRevs.push(w.rev) } | |
| 52 | const canonFrames = [] | |
| 53 | for (let i = 0; i < canonSeq.core.length; i++) canonFrames.push(await canonSeq.core.get(i)) | |
| 54 | toClose.push(async () => { try { await canonSeq.close() } catch {}; try { await canonRepo.close() } catch {}; try { await canonStore.close() } catch {} }) | |
| 55 | console.log('canonical line revs:', canonRevs.join(' ')) | |
| 56 | ||
| 57 | // A co-owner AT LAG k: its own repo holding the first k records, its own firehose holding the first k | |
| 58 | // CANONICAL frames (identical bytes => identical commits => a real prefix of the same line). | |
| 59 | async function coOwner (k, tag) { | |
| 60 | const store = openCorestore(newDir('co-' + tag + '-')) | |
| 61 | const repo = await new RepoStore(store, { name: `atproto-repo:${DID}` }).ready() | |
| 62 | let micros = 1_700_000_000_000_000n | |
| 63 | const mgr = await new RepoManager(repo, { did: DID, privateKey: id.privateKey, clock: () => (micros += 1_000_000n) }).init() | |
| 64 | for (let i = 0; i < k; i++) await mgr.applyWrite(OPS[i]) // same records => same MST => same data root | |
| 65 | const seq = await new FirehoseSequencer(store.get({ name: 'firehose' })).ready() | |
| 66 | for (let i = 0; i < k; i++) await seq.append(canonFrames[i]) // the canonical commits, verbatim | |
| 67 | return finishHolder(store, repo, seq, tag) | |
| 68 | } | |
| 69 | ||
| 70 | // A FORK: an independent line (different record, later clock base => its rev sorts HIGHER). | |
| 71 | async function forkOwner (tag) { | |
| 72 | const store = openCorestore(newDir('co-' + tag + '-')) | |
| 73 | const repo = await new RepoStore(store, { name: `atproto-repo:${DID}` }).ready() | |
| 74 | let micros = 1_900_000_000_000_000n // far ahead => higher TID => wins freshest-by-rev | |
| 75 | const mgr = await new RepoManager(repo, { did: DID, privateKey: id.privateKey, clock: () => (micros += 1_000_000n) }).init() | |
| 76 | const seq = await new FirehoseSequencer(store.get({ name: 'firehose' })).ready() | |
| 77 | const w = await mgr.applyWrite(post('3zzzzzzzzzzz2', 'rival genesis')) | |
| 78 | await seq.sequenceCommit(w, { time: iso }) | |
| 79 | console.log('fork rev:', w.rev, '(canonical head:', canonRevs[canonRevs.length - 1] + ')') | |
| 80 | return finishHolder(store, repo, seq, tag) | |
| 81 | } | |
| 82 | ||
| 83 | async function finishHolder (store, repo, seq, tag) { | |
| 84 | const peer = new DidPeer(store, { did: DID, verifyKey: id.publicKey }) | |
| 85 | await peer.provide({ repoKey: repo.core.key, firehoseKey: seq.core.key }) | |
| 86 | const swarm = new Hyperswarm({ bootstrap }) | |
| 87 | swarm.on('connection', (c) => peer.attach(c)) | |
| 88 | // server AND client, as a real peer joins: a holder that appears LATE must be able to find the reader | |
| 89 | // (which is already announcing) rather than waiting for the reader's next lookup. | |
| 90 | await swarm.join(discoveryTopic(DID), { server: true, client: true }).flushed() | |
| 91 | toClose.push(async () => { try { await swarm.destroy() } catch {}; try { await peer.close() } catch {}; try { await store.close() } catch {} }) | |
| 92 | return { peer, swarm, fhKey: seq.core.key.toString('hex'), tag, rev: peer.head.rev } | |
| 93 | } | |
| 94 | ||
| 95 | const behind = await coOwner(2, 'behind') // same line, lagging | |
| 96 | const ahead = await coOwner(4, 'ahead') // same line, current | |
| 97 | assert(ahead.rev === canonRevs[3] && behind.rev === canonRevs[1], 'co-owners sit on the SAME line at different lag') | |
| 98 | ||
| 99 | // Reader in the bridge's posture (owns-swarm), continuity gate on (default). | |
| 100 | const store = openCorestore(newDir('reader-')) | |
| 101 | const swarm = new Hyperswarm({ bootstrap }) | |
| 102 | const router = new DidPeerRouter(store, { | |
| 103 | swarm, | |
| 104 | // exactly as atproto-bridge/pds/bin/pds-serve.js joins: server AND client | |
| 105 | join: (did) => { try { swarm.join(discoveryTopic(did), { server: true, client: true }); swarm.flush().catch(() => {}) } catch {} }, | |
| 106 | fronted: [DID], | |
| 107 | resolveVerifyKey: async () => id.publicKey | |
| 108 | }) | |
| 109 | toClose.push(async () => { try { await swarm.destroy() } catch {}; try { await router.close() } catch {}; try { await store.close() } catch {} }) | |
| 110 | ||
| 111 | const servedRev = async () => { const f = await router.front(DID).catch(() => null); const lc = f && f.getLatestCommit(); return lc ? lc.rev : null } | |
| 112 | const servedRecs = async () => { const f = await router.front(DID).catch(() => null); return f ? (await f.listRecords('app.bsky.feed.post').catch(() => [])).length : -1 } | |
| 113 | async function waitRev (want, ms = 20000) { const end = Date.now() + ms; let last = null; while (Date.now() < end) { last = await servedRev(); if (last === want) return last; await sleep(250) } return last } | |
| 114 | ||
| 115 | // 1. Same line: the fresher co-owner wins, continuity permits it (its frames CONTAIN the lagging head). | |
| 116 | assert(await waitRev(canonRevs[3]) === canonRevs[3], 'on ONE line the freshest co-owner is served (continuity permits the upgrade)') | |
| 117 | assert(await servedRecs() === 4, 'serving all 4 records of the canonical head') | |
| 118 | ||
| 119 | // 2. The fork appears, live, with a HIGHER rev. It must NOT take the read surface. | |
| 120 | const fork = await forkOwner('fork') | |
| 121 | // Poll until the rival is actually FRONTED + verified, so the assertions below test the gate rather than | |
| 122 | // a rival that simply hadn't arrived yet. | |
| 123 | const forkArrived = await (async () => { | |
| 124 | const end = Date.now() + 30000 | |
| 125 | while (Date.now() < end) { | |
| 126 | await servedRev() // drive front() so the gate runs | |
| 127 | const row = router.diagnose().fronts.find((f) => f.firehoseKey === fork.fhKey) | |
| 128 | if (row && row.rev) return true | |
| 129 | await sleep(250) | |
| 130 | } | |
| 131 | return false | |
| 132 | })() | |
| 133 | assert(forkArrived, 'the rival co-owner was fronted + verified (the gate is being exercised)') | |
| 134 | const afterFork = await servedRev() | |
| 135 | assert(afterFork === canonRevs[3], `a higher-rev rival LINE does not win the read surface (still ${afterFork})`) | |
| 136 | assert(await servedRecs() === 4, 'the 4-record authoritative head is still served, not the 1-record rival') | |
| 137 | ||
| 138 | const diag = router.diagnose() | |
| 139 | console.log(' fronts:', diag.fronts.map((f) => `${f.firehoseKey.slice(0, 6)} rev=${f.rev} live=${f.live} served=${f.served} forkRejected=${f.forkRejected}`).join(' | ')) | |
| 140 | console.log(' looking for fork fh', fork.fhKey.slice(0, 6)) | |
| 141 | const forkRow = diag.fronts.find((f) => f.firehoseKey === fork.fhKey) | |
| 142 | assert(!!forkRow, 'the rival is still FRONTED (it is tracked, not ignored)') | |
| 143 | assert(forkRow.rev > canonRevs[3], 'the rival really does carry the higher rev') | |
| 144 | assert(forkRow.forkRejected > 0, 'diagnose NAMES the rejection (forkRejected > 0) — an operator can see the disagreement') | |
| 145 | assert(diag.fronts.find((f) => f.served).rev === canonRevs[3], 'diagnose reports the authoritative front as the served one') | |
| 146 | ||
| 147 | // 3. The gate is not a permanent lock-out: with the whole canonical line gone (the rebuild case — a peer | |
| 148 | // that rebuilt its cores comes back on a new line), the surviving line is served. | |
| 149 | await ahead.swarm.destroy() | |
| 150 | await behind.swarm.destroy() | |
| 151 | const afterLineGone = await waitRev(fork.rev) | |
| 152 | assert(afterLineGone === fork.rev, `once the incumbent line is gone the new line IS served (${afterLineGone})`) | |
| 153 | ||
| 154 | console.log('\nFORK OK: a higher rev on a rival line loses to the served line while that line is live') | |
| 155 | for (const close of toClose) await close() | |
| 156 | await testnet.destroy() | |
| 157 | cleanupSync() |