atproto-bridge / pds / test / peer-restart-durability.test.js

// REPRO: does a fronted peer's RESTART (same keys, new swarm session) self-heal on the membrane, or does
// it need a membrane restart? Mirrors peer-router.test.js scaffolding. Run: node test/zz-repro-peer-restart.mjs
import test from 'brittle'
import { mkdtempSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import crypto from 'hypercore-crypto'
import Hyperswarm from 'hyperswarm'
import createTestnet from '@hyperswarm/testnet'
import { openCorestore } from 'atproto-store/lib/corestore.js'
import { RepoStore } from 'atproto-store/lib/repo-store.js'
import { RepoManager } from 'atproto-store/lib/repo-manager.js'
import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js'
import { keyFromPrivate } from 'atproto-identity'
import { discoveryTopic } from 'atproto-identity/lib/discovery.js'
import { mkDidRouter } from './mk-did-router.js'
import { FleetRelay } from '../lib/fleet-relay.js'
import { pushAnnounce } from './serve-announce.js'

const DID = 'did:plc:restartme'
const HANDLE = 'restart.7rnx.net'
const PRIV = new Uint8Array(32).fill(9)
const NOW = '2026-07-18T00:00:00.000Z'

test('peer RESTART (same keys) — front self-heals, new commit replicates without a membrane restart', async (t) => {
  const testnet = await createTestnet(3)
  const peerDir = mkdtempSync(join(tmpdir(), 'restart-peer-'))
  const peerStore = openCorestore(peerDir)
  const repo = await new RepoStore(peerStore, { name: `atproto-repo:${DID}` }).ready()
  let micros = 1_700_000_000_000_000n
  const mgr = await new RepoManager(repo, { did: DID, privateKey: PRIV, clock: () => (micros += 1_000_000n) }).init()
  const seq = await new FirehoseSequencer(peerStore.get({ name: 'firehose' })).ready()
  const w1 = await mgr.applyWrite({ action: 'create', collection: 'app.bsky.feed.post', rkey: '3aaaaaaaaaaa2', record: { $type: 'app.bsky.feed.post', text: 'before-restart', createdAt: NOW, langs: ['en'] } })
  await seq.sequenceCommit(w1, { time: NOW })

  const peerKeyPair = crypto.keyPair() // STABLE swarm identity across the restart (mimics loadOrCreateSwarmKeyPair)
  const servePeer = (conn) => { peerStore.replicate(conn); pushAnnounce(conn, { did: DID, handle: HANDLE, repo, seq }) }
  let peerSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap, keyPair: peerKeyPair })
  peerSwarm.on('connection', servePeer)
  await peerSwarm.join(discoveryTopic(DID), { server: true, client: false }).flushed()

  const routerStore = openCorestore(mkdtempSync(join(tmpdir(), 'restart-router-')))
  const routerSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap })
  const relay = await new FleetRelay(routerStore.get({ name: 'peer-relay-fh', valueEncoding: 'json' })).ready()
  const router = mkDidRouter(routerSwarm, routerStore, {
    fronted: [DID], resolveVerifyKey: async () => keyFromPrivate(PRIV).publicKey, relay,
    refreshMs: 1000, refreshTimeoutMs: 800, staleStrikeLimit: 2 // fast watchdog so the test doesn't wait 30s
  })

  t.teardown(async () => {
    await router.close(); await relay.close()
    try { await peerSwarm.destroy() } catch {}
    await routerSwarm.destroy()
    await seq.close(); await repo.close(); await peerStore.close(); await routerStore.close()
    await testnet.destroy(); rmSync(peerDir, { recursive: true, force: true })
  })

  // baseline
  const front = await router.front(DID)
  // FIX: the standing download must put both cores in `downloading` state — the one condition under which
  // corestore.replicate re-attaches them to a FRESH connection after a flaky-link drop (direction-free).
  t.ok(front.seq.core.replicator && front.seq.core.replicator.downloading, 'FIX: firehose core is downloading (standing download)')
  t.ok(front.repo.core.replicator && front.repo.core.replicator.downloading, 'FIX: repo core is downloading (standing download)')
  const { records: r1 } = await front.listRecords('app.bsky.feed.post')
  t.is(r1.length, 1, 'baseline: front serves the pre-restart post')

  // --- RESTART the peer: swarm dies (process restart), recreate SAME keypair + SAME storage, rejoin ---
  await peerSwarm.destroy()
  await new Promise(r => setTimeout(r, 500))
  peerSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap, keyPair: peerKeyPair })
  peerSwarm.on('connection', servePeer)
  await peerSwarm.join(discoveryTopic(DID), { server: true, client: false }).flushed()

  // author a NEW commit AFTER the restart
  const w2 = await mgr.applyWrite({ action: 'create', collection: 'app.bsky.feed.post', rkey: '3bbbbbbbbbbb2', record: { $type: 'app.bsky.feed.post', text: 'after-restart', createdAt: NOW, langs: ['en'] } })
  await seq.sequenceCommit(w2, { time: NOW })

  // ASSERT: the front replicates the new commit within ~12s, WITHOUT re-creating the router.
  let got = 0
  for (let i = 0; i < 60; i++) {
    const recs = (await front.listRecords('app.bsky.feed.post').catch(() => ({ records: [] }))).records
    got = recs.length
    if (got >= 2) break
    await new Promise(r => setTimeout(r, 200))
  }
  t.is(got, 2, 'DURABILITY: after a peer restart, the front replicates the new commit without a membrane restart')
})