// REPRO (deep-debugging): does a fronted peer's new commit reach the bridge AFTER the PEER PROCESS
// RESTARTS — i.e. the peer's corestore is fully CLOSED and REOPENED from disk (a new core instance,
// same key), not just a swarm swap with the core kept open? This is the ashatron prod scenario:
// `pgh app restart hyper-proto-peer` → the membrane firehose core is re-opened from disk, the bridge
// reconnects (diagnose showed a LIVE connection), yet the cached front's seq.core stayed frozen at
// the pre-restart length. The existing peer-router test only swaps the swarm with the peerStore kept
// OPEN, so it never exercises the reopen path.
import test from 'brittle'
import { mkdtempSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import b4a from 'b4a'
import c from 'compact-encoding'
import crypto from 'hypercore-crypto'
import Hyperswarm from 'hyperswarm'
import createTestnet from '@hyperswarm/testnet'
import ProtomuxRPC from 'protomux-rpc'
import { pushAnnounce } from './serve-announce.js'
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'
const waitFor = async (fn, { timeout = 25000, interval = 150 } = {}) => {
const end = Date.now() + timeout
while (Date.now() < end) { if (await fn()) return true; await new Promise((r) => setTimeout(r, interval)) }
return false
}
test('REPRO: front must live-follow a new commit AFTER the peer PROCESS restarts (corestore close→reopen)', async (t) => {
const testnet = await createTestnet(3)
const DID = 'did:plc:reopenme'
const HANDLE = 'reopen.7rnx.net'
const PRIV = new Uint8Array(32).fill(9)
const T = '2026-07-11T00:00:00.000Z'
const peerDir = mkdtempSync(join(tmpdir(), 'reopen-peer-'))
const peerSwarmKeyPair = crypto.keyPair(new Uint8Array(32).fill(13))
// Boot (or re-boot) the peer FROM DISK: fresh corestore instance + fresh cores from peerDir,
// restoring the saved head — exactly what the daemon does on `pgh app restart`.
let micros = 1_700_000_000_000_000n
async function bootPeer () {
const store = openCorestore(peerDir)
const repo = await new RepoStore(store, { name: `atproto-repo:${DID}` }).ready()
const mgr = await new RepoManager(repo, { did: DID, privateKey: PRIV, clock: () => (micros += 1_000_000n) }).init()
const seq = await new FirehoseSequencer(store.get({ name: 'firehose' })).ready()
const swarm = new Hyperswarm({ bootstrap: testnet.bootstrap, keyPair: peerSwarmKeyPair })
swarm.on('connection', (conn) => {
store.replicate(conn)
pushAnnounce(conn, { did: DID, handle: HANDLE, repo, seq })
})
await swarm.join(discoveryTopic(DID), { server: true, client: true }).flushed()
const post = async (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
}
const shutdown = async () => { await swarm.destroy(); await seq.close(); await repo.close(); await store.close() }
return { store, repo, mgr, seq, swarm, post, shutdown, firehoseKey: b4a.toString(seq.core.key, 'hex') }
}
let peer = await bootPeer()
const rev1 = await peer.post('a1', 'first')
const routerStore = openCorestore(mkdtempSync(join(tmpdir(), 'reopen-router-')))
const routerSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap })
const router = mkDidRouter(routerSwarm, routerStore, { fronted: [DID], resolveVerifyKey: async () => keyFromPrivate(PRIV).publicKey })
t.teardown(async () => {
await router.close(); await routerSwarm.destroy()
await peer.shutdown().catch(() => {})
await routerStore.close()
await testnet.destroy()
rmSync(peerDir, { recursive: true, force: true })
})
const front = await router.front(DID)
t.is((await front.getLatestCommit()).rev, rev1, 'fronted at the first commit')
// sanity: live-follow over the ORIGINAL connection still works
const rev2 = await peer.post('a2', 'second')
t.ok(await waitFor(async () => (await front.getLatestCommit())?.rev === rev2), 'live-follows over the original connection')
// ---- THE PROD SCENARIO: full peer PROCESS restart (corestore close → reopen) ----
const keyBefore = peer.firehoseKey
await peer.shutdown()
peer = await bootPeer()
t.is(peer.firehoseKey, keyBefore, 'firehose key is stable across restart (same corestore seed + name)')
const rev3 = await peer.post('a3', 'after process restart')
const followed = await waitFor(async () => (await front.getLatestCommit())?.rev === rev3)
t.ok(followed, 'front live-follows a new commit AFTER the peer PROCESS restarts (corestore reopen)')
})
// Deterministic unit test of the watchdog's PULL-ONLY backstop (post-audit). A flaky-link stall
// (counted-but-not-syncing) can't be reproduced over local testnet (a clean reconnect self-heals — see
// the test above). The REAL recovery is PeerFront's standing download (flips downloading:true → corestore
// re-attaches direction-free on reconnect); the watchdog only NUDGES a stalled front with a bounded pull
// and MUST NEVER destroy the connection (force-destroying fought hyperswarm's own keepalive+reconnect and
// masked the missing download). This drives `_refreshFronts` directly and asserts the nudge-not-destroy.
test('watchdog nudges a stalled front with a pull but NEVER destroys the connection', async (t) => {
const fakeSwarm = { on () {}, off () {}, join () { return { flushed: async () => {} } }, flush: async () => {}, connections: new Set() }
// refreshMs:0 disables the auto-timer so we drive _refreshFronts manually.
const router = mkDidRouter(fakeSwarm, {}, { fronted: [], refreshMs: 0, staleStrikeLimit: 3 })
let stalledDestroyed = 0
let stalledPulls = 0
const stalled = { seq: { core: { update: async () => { stalledPulls++; throw new Error('stalled link') } } }, _conn: { destroy: () => { stalledDestroyed++ } } }
let healthyDestroyed = 0
const healthy = { seq: { core: { update: async () => {} } }, _conn: { destroy: () => { healthyDestroyed++ } } }
router._fronts.set('did:plc:stalled', new Map([['ks', stalled]]))
router._fronts.set('did:plc:healthy', new Map([['kh', healthy]]))
await router._refreshFronts() // pull 1
await router._refreshFronts() // pull 2
await router._refreshFronts() // pull 3 — the OLD strike-limit "destroy" point
await router._refreshFronts() // pull 4 — well past it
t.is(stalledPulls, 4, 'a stalled front is NUDGED with an update({wait}) pull each window')
t.is(stalledDestroyed, 0, 'stalled connection is NEVER destroyed — recovery is the standing download + hyperswarm reconnect')
t.is(healthyDestroyed, 0, 'a healthy front (update resolves) is never destroyed either')
await router.close()
})