| … | … | @@ -0,0 +1,112 @@ |
| 1 | // Co-owner SET integrity — the invariant co-owned.test.mjs could NOT see. |
| 2 | // |
| 3 | // co-owned.test.mjs asserted `diagnose().cachedFronts === 2`, which counts MAP ENTRIES. It stayed green |
| 4 | // while both entries aliased ONE DidPeer: _ensureFront reused the primary for a second co-owner because |
| 5 | // `peer.holds` (=> !!seq) is false until the primary's first provide() COMPLETES, so the second front's |
| 6 | // provide() returned the FIRST co-owner's in-flight open. The co-owner SET collapsed to a single front, |
| 7 | // freshest-by-rev had nothing to choose between, and which co-owner got served was a coin flip — which is |
| 8 | // why co-owned.test.mjs flaked (~3 in 4 runs served the BEHIND co-owner forever). |
| 9 | // |
| 10 | // This test asserts the real invariant: N announced co-owners => N DISTINCT DidPeers, each bound to the |
| 11 | // firehose key it is filed under. Run: node test/coowner-set.test.mjs |
| 12 | import { mkdtempSync, rmSync } from 'node:fs' |
| 13 | import { join } from 'node:path' |
| 14 | import { tmpdir } from 'node:os' |
| 15 | import createTestnet from '@hyperswarm/testnet' |
| 16 | import Hyperswarm from 'hyperswarm' |
| 17 | import b4a from 'b4a' |
| 18 | import { RepoStore } from 'atproto-store/lib/repo-store.js' |
| 19 | import { RepoManager } from 'atproto-store/lib/repo-manager.js' |
| 20 | import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js' |
| 21 | import { openCorestore } from 'atproto-store/lib/corestore.js' |
| 22 | import { createAtprotoIdentityFromKey } from 'atproto-identity/lib/atproto-identity.js' |
| 23 | import { discoveryTopic } from 'atproto-identity/lib/discovery.js' |
| 24 | import { DidPeer, DidPeerRouter } from '../index.js' |
| 25 | |
| 26 | const assert = (c, m) => { if (!c) { console.error('FAIL:', m); cleanupSync(); process.exit(1) } console.log(' ok -', m) } |
| 27 | const iso = '2026-07-21T00:00:00.000Z' |
| 28 | const dirs = [] |
| 29 | const toClose = [] |
| 30 | function newDir (p) { const d = mkdtempSync(join(tmpdir(), p)); dirs.push(d); return d } |
| 31 | function cleanupSync () { for (const d of dirs) { try { rmSync(d, { recursive: true, force: true }) } catch {} } } |
| 32 | const post = (rk, text) => ({ action: 'create', collection: 'app.bsky.feed.post', rkey: rk, record: { $type: 'app.bsky.feed.post', text, createdAt: iso, langs: ['en'] } }) |
| 33 | |
| 34 | const id = createAtprotoIdentityFromKey(new Uint8Array(32).fill(7), { handle: 'coowned.test', pdsUrl: 'https://pds.local' }) |
| 35 | const DID = id.did |
| 36 | |
| 37 | const testnet = await createTestnet(3) |
| 38 | const bootstrap = testnet.bootstrap |
| 39 | |
| 40 | // A co-owner: its own cores, nCommits signed over the SHARED DID key. All co-owners join the DID topic |
| 41 | // AT ONCE, which is what puts their announces inside the primary's first in-flight provide(). |
| 42 | async function coOwner (nCommits) { |
| 43 | const store = openCorestore(newDir('coowner-')) |
| 44 | const repo = await new RepoStore(store, { name: `atproto-repo:${DID}` }).ready() |
| 45 | let micros = 1_700_000_000_000_000n |
| 46 | const mgr = await new RepoManager(repo, { did: DID, privateKey: id.privateKey, clock: () => (micros += 1_000_000n) }).init() |
| 47 | const seq = await new FirehoseSequencer(store.get({ name: 'firehose' })).ready() |
| 48 | for (let i = 0; i < nCommits; i++) { const w = await mgr.applyWrite(post('3post' + i + 'aaaaaa2', 'c' + i)); await seq.sequenceCommit(w, { time: iso }) } |
| 49 | const peer = new DidPeer(store, { did: DID, verifyKey: id.publicKey }) |
| 50 | await peer.provide({ repoKey: repo.core.key, firehoseKey: seq.core.key }) |
| 51 | const swarm = new Hyperswarm({ bootstrap }) |
| 52 | swarm.on('connection', (c) => peer.attach(c)) |
| 53 | await swarm.join(discoveryTopic(DID), { server: true, client: false }).flushed() |
| 54 | toClose.push(async () => { try { await swarm.destroy() } catch {}; try { await peer.close() } catch {}; try { await store.close() } catch {} }) |
| 55 | return { peer, fhKey: seq.core.key.toString('hex'), nCommits, rev: peer.head.rev } |
| 56 | } |
| 57 | |
| 58 | // THREE co-owners: with two, an aliasing bug still had a 50% chance of aliasing onto the freshest and |
| 59 | // looking correct. The record counts also disambiguate WHICH co-owner is being served. |
| 60 | const owners = [await coOwner(1), await coOwner(3), await coOwner(2)] |
| 61 | const expectedKeys = new Set(owners.map((o) => o.fhKey)) |
| 62 | // Derive the expected winner from the holders' ACTUAL head revs rather than assuming "most commits wins". |
| 63 | // The injected clock is shared-base, so two co-owners can land the same TID timestamp and be ordered by |
| 64 | // the trailing clock-id — a fixture artifact, not a router property. Record counts differ per co-owner, |
| 65 | // so the served count still identifies WHICH front was picked. |
| 66 | const winner = owners.reduce((b, o) => (!b || o.rev > b.rev ? o : b), null) |
| 67 | console.log('co-owners:', owners.map((o) => `${o.fhKey.slice(0, 8)}(n=${o.nCommits} rev=${o.rev})`).join(' ')) |
| 68 | console.log('highest-rev co-owner:', winner.fhKey.slice(0, 8), `(n=${winner.nCommits})`) |
| 69 | |
| 70 | const store = openCorestore(newDir('reader-')) |
| 71 | const swarm = new Hyperswarm({ bootstrap }) |
| 72 | const router = new DidPeerRouter(store) |
| 73 | swarm.on('connection', (c) => router.onconnection(c)) |
| 74 | router.add(DID, { verifyKey: id.publicKey }) |
| 75 | await swarm.join(discoveryTopic(DID), { server: false, client: true }).flushed() |
| 76 | |
| 77 | // Let every co-owner announce and open, polling front() the way a real getRepo request does. A front |
| 78 | // EXISTING is not the same as its head having replicated + verified, so poll for convergence rather than |
| 79 | // breaking on front count (an early break reads a set that is complete but not yet reconciled). |
| 80 | const counts = [] |
| 81 | for (let i = 0; i < 40; i++) { |
| 82 | const f = await router.front(DID).catch(() => null) |
| 83 | const recs = f ? await f.listRecords('app.bsky.feed.post').catch(() => []) : [] |
| 84 | counts.push(recs.length) |
| 85 | await new Promise((r) => setTimeout(r, 300)) |
| 86 | } |
| 87 | console.log(' served record counts:', JSON.stringify(counts)) |
| 88 | |
| 89 | const set = router._fronts.get(DID) || new Map() |
| 90 | assert(set.size === owners.length, `every announced co-owner has a front entry (${set.size}/${owners.length})`) |
| 91 | |
| 92 | // THE INVARIANT: distinct objects, each bound to the key it is filed under. |
| 93 | const objects = new Set() |
| 94 | let mismatched = null |
| 95 | for (const [mapKey, f] of set) { |
| 96 | objects.add(f) |
| 97 | const boundFh = f._firehoseKey ? b4a.toString(f._firehoseKey, 'hex') : null |
| 98 | if (boundFh !== mapKey) mismatched = `${mapKey.slice(0, 8)} is bound to ${boundFh ? boundFh.slice(0, 8) : 'null'}` |
| 99 | } |
| 100 | assert(!mismatched, `each front is bound to its OWN firehose key (${mismatched || 'all match'})`) |
| 101 | assert(objects.size === owners.length, `fronts are DISTINCT objects, not aliases (${objects.size}/${owners.length})`) |
| 102 | assert([...set.keys()].every((k) => expectedKeys.has(k)), 'front keys are exactly the announced co-owners') |
| 103 | |
| 104 | // With a real SET to choose from, freshest-by-rev serves the highest-rev co-owner. |
| 105 | assert(counts[counts.length - 1] === winner.nCommits, |
| 106 | `freshest-by-rev serves the highest-rev co-owner (${winner.nCommits} records), got ${counts[counts.length - 1]}`) |
| 107 | |
| 108 | console.log('\nCO-OWNER SET OK: N announced co-owners => N distinct fronts, each correctly bound; freshest served') |
| 109 | await swarm.destroy(); await router.close(); await store.close() |
| 110 | for (const close of toClose) await close() |
| 111 | await testnet.destroy() |
| 112 | cleanupSync() |