| … | … | @@ -1,6 +1,6 @@ |
| 1 | 1 | { |
| 2 | 2 | "name": "atproto-bridge", |
| 3 | "version": "0.9.21", | |
| 3 | "version": "0.9.22", | |
| 4 | 4 | "description": "Stateless keyless atproto MEMBRANE — fronts self-custody hyper-proto-peers to bluesky over HTTP/WS, holds no keys and never signs. Bare-runnable. Deploy root / flat closure over the identity+repo+store+pds packages.", |
| 5 | 5 | "type": "module", |
| 6 | 6 | "main": "index.js", |
| … | … | @@ -54,6 +54,8 @@ export class FleetRelay { |
| 54 | 54 | // pre-existing #commit history never enters the aggregate (only a #sync re-baseline does), so the |
| 55 | 55 | // AppView never indexes it. update({wait:true}) syncs the length so _drain ingests the backlog. |
| 56 | 56 | try { await firehoseCore.update({ wait: true }) } catch { /* offline/first-connect — 'append' catches up */ } |
| 57 | const prev = this._sources.get(sourceId) // idempotent re-add: detach any prior listener so we never double-drain | |
| 58 | if (prev) { try { prev.core.off('append', prev.onappend) } catch {} } | |
| 57 | 59 | const onappend = () => this._drain(sourceId, firehoseCore).catch(() => {}) |
| 58 | 60 | this._sources.set(sourceId, { core: firehoseCore, onappend }) |
| 59 | 61 | await this._drain(sourceId, firehoseCore) |
| … | … | @@ -61,6 +63,17 @@ export class FleetRelay { |
| 61 | 63 | return this |
| 62 | 64 | } |
| 63 | 65 | |
| 66 | // Remove a source whose firehose core was rebuilt/retired (new core key ⇒ new sourceId). Detaches the | |
| 67 | // live-follow listener so a dead core stops being drained; already-aggregated frames stay in the | |
| 68 | // monotonic seq (never rewritten). Returns true iff a source was removed. | |
| 69 | removeSource (sourceId) { | |
| 70 | const s = this._sources.get(sourceId) | |
| 71 | if (!s) return false | |
| 72 | try { s.core.off('append', s.onappend) } catch {} | |
| 73 | this._sources.delete(sourceId) | |
| 74 | return true | |
| 75 | } | |
| 76 | ||
| 64 | 77 | // Inject a synthetic frame (e.g. a #sync re-baseline event, Sync 1.1) into the aggregate. |
| 65 | 78 | // `key` dedups it (same key → ingested once), so re-emitting the same rev is idempotent. The |
| 66 | 79 | // frame's internal seq is re-stamped to the relay's global seq on subscribe(), like any frame. |
| … | … | @@ -141,8 +141,22 @@ export class PeerRouter { |
| 141 | 141 | const learn = (coords, rpcForConn) => { |
| 142 | 142 | if (!coords || !coords.did || !coords.repoKey || !coords.firehoseKey) return |
| 143 | 143 | if (coords.handle) this._setHandle(coords.handle, coords.did) // learn+persist on EVERY announce (not just first front) — survives restart + offline (blind-relayed) peers |
| 144 | // DURABILITY (no restart on a peer core rebuild): a peer can re-announce NEW repoKey/firehoseKey under | |
| 145 | // the SAME DID — e.g. after a storage repair that rebuilt its membrane cores. A cached PeerFront is | |
| 146 | // bound to the DEAD old keys and would serve a FROZEN repo forever (previously only a process restart | |
| 147 | // fixed it). Detect the key change here and EVICT the stale front (+ drop its aggregate relay source) | |
| 148 | // so front() re-opens the NEW cores on the next request. Keys compared as hex (announce carries hex). | |
| 149 | const stale = this._fronts.get(coords.did) | |
| 150 | if (stale && (b4a.toString(stale.repoKey, 'hex') !== coords.repoKey || b4a.toString(stale.firehoseKey, 'hex') !== coords.firehoseKey)) { | |
| 151 | this._fronts.delete(coords.did) | |
| 152 | if (this.relay) { try { this.relay.removeSource(b4a.toString(stale.firehoseKey, 'hex')) } catch {} } | |
| 153 | try { stale.close() } catch {} | |
| 154 | } | |
| 144 | 155 | const cur = this._byDid.get(coords.did) |
| 145 | if (cur && cur.conn === conn) return | |
| 156 | // Skip only when NOTHING changed (same connection AND same cores). If the cores changed we must fall | |
| 157 | // through to refresh _byDid.coords, else front() would rebuild against the old announced keys. | |
| 158 | const unchanged = cur && cur.conn === conn && cur.coords && cur.coords.repoKey === coords.repoKey && cur.coords.firehoseKey === coords.firehoseKey | |
| 159 | if (unchanged) return | |
| 146 | 160 | conn._frontDid = coords.did |
| 147 | 161 | this._byDid.set(coords.did, { conn, rpc: rpcForConn, coords }) |
| 148 | 162 | const f = this._fronts.get(coords.did) |
| … | … | @@ -158,3 +158,46 @@ test('PeerRouter live-follows a fronted peer across new commits AND a peer recon |
| 158 | 158 | const followedReconnect = await waitFor(async () => { const lc = await front.getLatestCommit(); return !!lc && lc.rev === rev3 }) |
| 159 | 159 | t.ok(followedReconnect, 'front live-follows a new commit AFTER a peer reconnect (persistent replicate)') |
| 160 | 160 | }) |
| 161 | ||
| 162 | // DURABILITY regression: a peer can REBUILD its membrane cores (new repoKey/firehoseKey under the same DID | |
| 163 | // — e.g. a storage repair). Before the fix, front() cached the PeerFront by DID and never re-checked its | |
| 164 | // keys, so it served the DEAD old cores forever until the bridge process restarted. Fix: learn() evicts the | |
| 165 | // stale front on a key-change announce so front() re-opens the NEW cores. No restart. | |
| 166 | test('PeerRouter re-fronts a peer that REBUILDS its cores with new keys — no restart', async (t) => { | |
| 167 | const testnet = await createTestnet(3) | |
| 168 | const D = 'did:plc:rebuildme'; const P = new Uint8Array(32).fill(9); const T = '2026-07-05T00:00:00.000Z' | |
| 169 | const peerDir = mkdtempSync(join(tmpdir(), 'rebuild-peer-')); const peerStore = openCorestore(peerDir) | |
| 170 | let micros = 1_700_000_000_000_000n | |
| 171 | const mkCores = async (v) => { | |
| 172 | const repo = await new RepoStore(peerStore, { name: `atproto-repo:${D}:${v}` }).ready() | |
| 173 | const mgr = await new RepoManager(repo, { did: D, privateKey: P, clock: () => (micros += 1_000_000n) }).init() | |
| 174 | const seq = await new FirehoseSequencer(peerStore.get({ name: `firehose:${v}` })).ready() | |
| 175 | return { repo, mgr, seq } | |
| 176 | } | |
| 177 | const post = async (mgr, seq, rkey, text) => { const w = await mgr.applyWrite({ action: 'create', collection: 'app.bsky.feed.post', rkey, record: { $type: 'app.bsky.feed.post', text, createdAt: T } }); await seq.sequenceCommit(w, { time: T }); return w.rev } | |
| 178 | let cores = await mkCores('v1'); const rev1 = await post(cores.mgr, cores.seq, 'a1', 'v1') | |
| 179 | const kp = crypto.keyPair(new Uint8Array(32).fill(13)) | |
| 180 | const mkSwarm = () => { const s = new Hyperswarm({ bootstrap: testnet.bootstrap, keyPair: kp }); s.on('connection', (conn) => { peerStore.replicate(conn); pushAnnounce(conn, { did: D, handle: 'rebuild.7rnx.net', repo: cores.repo, seq: cores.seq }) }); return s } | |
| 181 | let peerSwarm = mkSwarm(); await peerSwarm.join(discoveryTopic(D), { server: true, client: false }).flushed() | |
| 182 | const routerStore = openCorestore(mkdtempSync(join(tmpdir(), 'rebuild-router-'))); const routerSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap }) | |
| 183 | const relay = await new FleetRelay(routerStore.get({ name: 'fh', valueEncoding: 'json' })).ready() | |
| 184 | const router = new PeerRouter(routerSwarm, routerStore, { fronted: [D], resolveVerifyKey: async () => keyFromPrivate(P).publicKey, relay }) | |
| 185 | t.teardown(async () => { await router.close(); await relay.close(); await peerSwarm.destroy(); await routerSwarm.destroy(); await cores.seq.close(); await cores.repo.close(); await peerStore.close(); await routerStore.close(); await testnet.destroy(); rmSync(peerDir, { recursive: true, force: true }) }) | |
| 186 | ||
| 187 | const f1 = await router.front(D) | |
| 188 | t.is((await f1.getLatestCommit()).rev, rev1, 'fronted the v1 cores') | |
| 189 | const v1key = b4a.toString(f1.repoKey, 'hex') | |
| 190 | ||
| 191 | // REBUILD: brand-new cores (new keys) under the same DID; peer reconnects announcing them. | |
| 192 | await peerSwarm.destroy() | |
| 193 | cores = await mkCores('v2'); const rev2 = await post(cores.mgr, cores.seq, 'b1', 'v2 after rebuild') | |
| 194 | peerSwarm = mkSwarm(); await peerSwarm.join(discoveryTopic(D), { server: true, client: false }).flushed() | |
| 195 | ||
| 196 | const refronted = await waitFor(async () => { | |
| 197 | const f = await router.front(D).catch(() => null) | |
| 198 | if (!f || b4a.toString(f.repoKey, 'hex') === v1key) return false | |
| 199 | const lc = await f.getLatestCommit().catch(() => null) | |
| 200 | return !!lc && lc.rev === rev2 | |
| 201 | }) | |
| 202 | t.ok(refronted, 'router re-fronts with the NEW cores after a rebuild (no restart, new repoKey + rev2)') | |
| 203 | }) |