9643f51 · atproto-bridge

Changed files (6) +157 -54

package.json +1 -1

…	…	@@ -1,6 +1,6 @@
1	1	{
2	2	"name": "atproto-bridge",
3		"version": "0.9.23",
	3	"version": "0.9.24",
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",

pds/lib/peer-front.js +12 -0

…	…	@@ -69,6 +69,18 @@ export class PeerFront {
69	69	front.repo = await new RepoStore(corestore, { key: front.repoKey }).ready()
70	70	front.seq = await new FirehoseSequencer(corestore.get({ key: front.firehoseKey })).ready()
71	71
	72	// DIRECTION-FREE LIVE-FOLLOW — the corrected primitive from did-peer.js (see
	73	// memory/feedback_p2p-no-directionality-event-driven). Issue a STANDING full-range download on both
	74	// cores and re-pull on 'append'. This flips `core.replicator.downloading = true`, which is the ONLY
	75	// condition under which `corestore.replicate()` re-attaches a core to a FRESH connection after a flaky
	76	// link drops (corestore `_shouldReplicate`). WITHOUT it, the front only recovered if the REMOTE re-sent
	77	// its discovery key (a directionality dependency the project forbids) — so over a flaky link the served
	78	// head froze until a bridge restart. WITH it, a reconnect (hyperswarm's own 5s keepalive + topic-rejoin)
	79	// re-attaches the cores and the standing download actively pulls, event-driven — no watchdog needed to
	80	// force-destroy connections (see PeerRouter._refreshFronts, now pull-only).
	81	front._seed = (core) => { const pull = () => { try { core.download({ start: 0, end: -1 }) } catch {} }; pull(); core.on('append', pull) }
	82	front._seed(front.repo.core); front._seed(front.seq.core)
	83
72	84	const signed = await front._waitSignedCommit(timeout)
73	85	if (!signed \|\| !verifyCommit(signed, verifyKey)) {
74	86	await front.close() // closes only the cores this front opened; leaves the router's conn/rpc alone

pds/lib/peer-router.js +21 -24

…	…	@@ -46,16 +46,17 @@ export class PeerRouter {
46	46	this._handleToDid = new Map(Object.entries(handles \|\| {}))
47	47	this._onHandles = onHandles // (obj) => void — persist the handle map when it grows
48	48	this.relay = relay // optional FleetRelay: aggregates fronted peers' firehoses
49		// Freshness watchdog config (see _refreshFronts): the front's firehose core is a REMOTE replica;
50		// over a flaky P2P link a connection can go counted-but-stalled (swarm still lists it, but passive
51		// replication no longer delivers new blocks), freezing the served head until a bridge restart. The
52		// watchdog forces a round-trip pull and, on repeated failure, drops the stalled connection so
53		// hyperswarm re-establishes a FRESH one (proven to recover). Root cause: PeerRouter held one
54		// long-lived conn per peer and assumed passive replicate stays live across a degraded link.
	49	// Stall-backstop config (see _refreshFronts): the front's cores are REMOTE replicas kept live by a
	50	// STANDING download (PeerFront.fromEntry) that flips `downloading:true`, so `corestore.replicate`
	51	// re-attaches them direction-free on every reconnect and the download actively pulls. This timer is
	52	// ONLY a pull-only staleness ceiling — a bounded update({wait}) nudge when a front has seen no 'append'
	53	// for a window. It does NOT destroy connections: that earlier watchdog fought hyperswarm's own 5s
	54	// keepalive + topic-rejoin reconnect (and force-destroying an inbound conn is worse — no auto-requeue),
	55	// and it masked the real bug (the missing standing download). Set refreshMs<=0 to disable (tests).
55	56	this._refreshMs = refreshMs
56	57	this._refreshTimeoutMs = refreshTimeoutMs
57		this._staleStrikeLimit = staleStrikeLimit
58		this._staleStrikes = new Map() // did -> consecutive failed refreshes (reset on success)
	58	this._staleStrikeLimit = staleStrikeLimit // retained for back-compat; no longer drives a destroy
	59	this._staleStrikes = new Map() // did -> consecutive failed nudges (diagnose() visibility only)
59	60	this._lastAppend = new Map() // did -> ts of the firehose's last 'append' (event-driven liveness)
60	61
61	62	// The membrane is ONE edge fronting MANY DIDs (rooms) over ONE swarm — the keet model: keet keeps
…	…	@@ -103,23 +104,19 @@ export class PeerRouter {
103	104	async _refreshFronts () {
104	105	for (const [did, f] of this._fronts) {
105	106	if (!f \|\| !f.seq \|\| !f.seq.core) continue
106		// EVENT-INFORMED (Step 3): if the firehose appended within the last refresh window, the front is
107		// demonstrably live (passive replication is delivering) — SKIP the round-trip poll entirely.
108		// Only when there's been NO append (idle OR stalled) do we force update({wait:true}) to tell them
109		// apart + strike/drop a zombie connection. This replaces the blind per-tick data-poll with a stall
110		// BACKSTOP: events prove liveness, the timer only catches absence.
	107	// EVENT-INFORMED: if the firehose appended within the last window the front is demonstrably live
	108	// (PeerFront's standing per-core download is delivering) — skip the nudge entirely.
111	109	if (Date.now() - (this._lastAppend.get(did) \|\| 0) < this._refreshMs) { this._staleStrikes.set(did, 0); continue }
112		try {
113		await withTimeout(f.seq.core.update({ wait: true }), this._refreshTimeoutMs)
114		this._staleStrikes.set(did, 0)
115		} catch {
116		const n = (this._staleStrikes.get(did) \|\| 0) + 1
117		this._staleStrikes.set(did, n)
118		if (n >= this._staleStrikeLimit && f._conn) {
119		try { f._conn.destroy() } catch {}
120		this._staleStrikes.set(did, 0) // reconnect will re-identify + re-replicate
121		}
122		}
	110	// PULL-ONLY stall backstop (staleness ceiling): a bounded update({wait}) that pulls from whatever peer
	111	// is currently attached. It NEVER destroys the connection. The old strike-then-`f._conn.destroy()`
	112	// escalation was an anti-pattern: it fought hyperswarm's own 5s keepalive + topic-rejoin reconnect
	113	// (and force-destroying an INBOUND connection is strictly worse — hyperswarm does not auto-requeue
	114	// those), AND it masked the real bug — the front had no standing download, so `corestore.replicate`
	115	// never re-attached the core to a reconnected stream (`_shouldReplicate` needs `downloading:true`).
	116	// PeerFront.fromEntry's standing download now makes recovery direction-free + event-driven; this
	117	// backstop only nudges a quiet-but-alive link. Strikes are tracked for `diagnose()` visibility only.
	118	try { await withTimeout(f.seq.core.update({ wait: true }), this._refreshTimeoutMs); this._staleStrikes.set(did, 0) }
	119	catch { this._staleStrikes.set(did, (this._staleStrikes.get(did) \|\| 0) + 1) }
123	120	}
124	121	}
125	122

pds/test/live-follow-repro.test.js +17 -17

…	…	@@ -95,34 +95,34 @@ test('REPRO: front must live-follow a new commit AFTER the peer PROCESS restarts
95	95	t.ok(followed, 'front live-follows a new commit AFTER the peer PROCESS restarts (corestore reopen)')
96	96	})
97	97
98		// Deterministic unit test of the freshness watchdog's RECOVERY path — the actual fix. A flaky-link
99		// stall (counted-but-not-syncing connection) can't be reproduced over local testnet (a clean
100		// reconnect self-heals — see the test above), so this drives `_refreshFronts` directly against a
101		// front whose `update` always fails, and asserts the stalled connection is destroyed at the strike
102		// limit (forcing hyperswarm to re-establish a FRESH connection — the proven recovery).
103		test('watchdog destroys a stalled front connection at the strike limit (forces fresh reconnect)', async (t) => {
	98	// Deterministic unit test of the watchdog's PULL-ONLY backstop (post-audit). A flaky-link stall
	99	// (counted-but-not-syncing) can't be reproduced over local testnet (a clean reconnect self-heals — see
	100	// the test above). The REAL recovery is PeerFront's standing download (flips downloading:true → corestore
	101	// re-attaches direction-free on reconnect); the watchdog only NUDGES a stalled front with a bounded pull
	102	// and MUST NEVER destroy the connection (force-destroying fought hyperswarm's own keepalive+reconnect and
	103	// masked the missing download). This drives `_refreshFronts` directly and asserts the nudge-not-destroy.
	104	test('watchdog nudges a stalled front with a pull but NEVER destroys the connection', async (t) => {
104	105	const fakeSwarm = { on () {}, off () {}, join () { return { flushed: async () => {} } }, flush: async () => {}, connections: new Set() }
105	106	// refreshMs:0 disables the auto-timer so we drive _refreshFronts manually.
106	107	const router = new PeerRouter(fakeSwarm, {}, { fronted: [], refreshMs: 0, staleStrikeLimit: 3 })
107	108
108	109	let stalledDestroyed = 0
109		const stalled = { seq: { core: { update: async () => { throw new Error('stalled link') } } }, _conn: { destroy: () => { stalledDestroyed++ } } }
	110	let stalledPulls = 0
	111	const stalled = { seq: { core: { update: async () => { stalledPulls++; throw new Error('stalled link') } } }, _conn: { destroy: () => { stalledDestroyed++ } } }
110	112	let healthyDestroyed = 0
111	113	const healthy = { seq: { core: { update: async () => {} } }, _conn: { destroy: () => { healthyDestroyed++ } } }
112	114	router._fronts.set('did:plc:stalled', stalled)
113	115	router._fronts.set('did:plc:healthy', healthy)
114	116
115		await router._refreshFronts() // strike 1
116		await router._refreshFronts() // strike 2
117		t.is(stalledDestroyed, 0, 'not destroyed before the strike limit')
	117	await router._refreshFronts() // pull 1
	118	await router._refreshFronts() // pull 2
	119	await router._refreshFronts() // pull 3 — the OLD strike-limit "destroy" point
	120	await router._refreshFronts() // pull 4 — well past it
118	121
119		await router._refreshFronts() // strike 3 → destroy
120		t.is(stalledDestroyed, 1, 'stalled connection destroyed at the strike limit')
121
122		await router._refreshFronts() // strikes reset after destroy → no immediate re-destroy
123		t.is(stalledDestroyed, 1, 'strikes reset after destroy (one destroy per stall window)')
124
125		t.is(healthyDestroyed, 0, 'a healthy front (update resolves) is never destroyed')
	122	t.is(stalledPulls, 4, 'a stalled front is NUDGED with an update({wait}) pull each window')
	123	t.is(stalledDestroyed, 0, 'stalled connection is NEVER destroyed — recovery is the standing download + hyperswarm reconnect')
	124	t.ok((router._staleStrikes.get('did:plc:stalled') \|\| 0) >= 1, 'failed pulls accrue strikes for diagnose() visibility only')
	125	t.is(healthyDestroyed, 0, 'a healthy front (update resolves) is never destroyed either')
126	126
127	127	await router.close()
128	128	})

pds/test/peer-restart-durability.test.js +87 -0

…	…	@@ -0,0 +1,87 @@
	1	// REPRO: does a fronted peer's RESTART (same keys, new swarm session) self-heal on the membrane, or does
	2	// it need a membrane restart? Mirrors peer-router.test.js scaffolding. Run: node test/zz-repro-peer-restart.mjs
	3	import test from 'brittle'
	4	import { mkdtempSync, rmSync } from 'node:fs'
	5	import { tmpdir } from 'node:os'
	6	import { join } from 'node:path'
	7	import crypto from 'hypercore-crypto'
	8	import Hyperswarm from 'hyperswarm'
	9	import createTestnet from '@hyperswarm/testnet'
	10	import { openCorestore } from 'atproto-store/lib/corestore.js'
	11	import { RepoStore } from 'atproto-store/lib/repo-store.js'
	12	import { RepoManager } from 'atproto-store/lib/repo-manager.js'
	13	import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js'
	14	import { keyFromPrivate } from 'atproto-identity'
	15	import { discoveryTopic } from 'atproto-identity/lib/discovery.js'
	16	import { PeerRouter } from '../lib/peer-router.js'
	17	import { FleetRelay } from '../lib/fleet-relay.js'
	18	import { pushAnnounce } from './serve-announce.js'
	19
	20	const DID = 'did:plc:restartme'
	21	const HANDLE = 'restart.7rnx.net'
	22	const PRIV = new Uint8Array(32).fill(9)
	23	const NOW = '2026-07-18T00:00:00.000Z'
	24
	25	test('peer RESTART (same keys) — front self-heals, new commit replicates without a membrane restart', async (t) => {
	26	const testnet = await createTestnet(3)
	27	const peerDir = mkdtempSync(join(tmpdir(), 'restart-peer-'))
	28	const peerStore = openCorestore(peerDir)
	29	const repo = await new RepoStore(peerStore, { name: `atproto-repo:${DID}` }).ready()
	30	let micros = 1_700_000_000_000_000n
	31	const mgr = await new RepoManager(repo, { did: DID, privateKey: PRIV, clock: () => (micros += 1_000_000n) }).init()
	32	const seq = await new FirehoseSequencer(peerStore.get({ name: 'firehose' })).ready()
	33	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'] } })
	34	await seq.sequenceCommit(w1, { time: NOW })
	35
	36	const peerKeyPair = crypto.keyPair() // STABLE swarm identity across the restart (mimics loadOrCreateSwarmKeyPair)
	37	const servePeer = (conn) => { peerStore.replicate(conn); pushAnnounce(conn, { did: DID, handle: HANDLE, repo, seq }) }
	38	let peerSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap, keyPair: peerKeyPair })
	39	peerSwarm.on('connection', servePeer)
	40	await peerSwarm.join(discoveryTopic(DID), { server: true, client: false }).flushed()
	41
	42	const routerStore = openCorestore(mkdtempSync(join(tmpdir(), 'restart-router-')))
	43	const routerSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap })
	44	const relay = await new FleetRelay(routerStore.get({ name: 'peer-relay-fh', valueEncoding: 'json' })).ready()
	45	const router = new PeerRouter(routerSwarm, routerStore, {
	46	fronted: [DID], resolveVerifyKey: async () => keyFromPrivate(PRIV).publicKey, relay,
	47	refreshMs: 1000, refreshTimeoutMs: 800, staleStrikeLimit: 2 // fast watchdog so the test doesn't wait 30s
	48	})
	49
	50	t.teardown(async () => {
	51	await router.close(); await relay.close()
	52	try { await peerSwarm.destroy() } catch {}
	53	await routerSwarm.destroy()
	54	await seq.close(); await repo.close(); await peerStore.close(); await routerStore.close()
	55	await testnet.destroy(); rmSync(peerDir, { recursive: true, force: true })
	56	})
	57
	58	// baseline
	59	const front = await router.front(DID)
	60	// FIX: the standing download must put both cores in `downloading` state — the one condition under which
	61	// corestore.replicate re-attaches them to a FRESH connection after a flaky-link drop (direction-free).
	62	t.ok(front.seq.core.replicator && front.seq.core.replicator.downloading, 'FIX: firehose core is downloading (standing download)')
	63	t.ok(front.repo.core.replicator && front.repo.core.replicator.downloading, 'FIX: repo core is downloading (standing download)')
	64	const r1 = await front.listRecords('app.bsky.feed.post')
	65	t.is(r1.length, 1, 'baseline: front serves the pre-restart post')
	66
	67	// --- RESTART the peer: swarm dies (process restart), recreate SAME keypair + SAME storage, rejoin ---
	68	await peerSwarm.destroy()
	69	await new Promise(r => setTimeout(r, 500))
	70	peerSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap, keyPair: peerKeyPair })
	71	peerSwarm.on('connection', servePeer)
	72	await peerSwarm.join(discoveryTopic(DID), { server: true, client: false }).flushed()
	73
	74	// author a NEW commit AFTER the restart
	75	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'] } })
	76	await seq.sequenceCommit(w2, { time: NOW })
	77
	78	// ASSERT: the front replicates the new commit within ~12s, WITHOUT re-creating the router.
	79	let got = 0
	80	for (let i = 0; i < 60; i++) {
	81	const recs = await front.listRecords('app.bsky.feed.post').catch(() => [])
	82	got = recs.length
	83	if (got >= 2) break
	84	await new Promise(r => setTimeout(r, 200))
	85	}
	86	t.is(got, 2, 'DURABILITY: after a peer restart, the front replicates the new commit without a membrane restart')
	87	})

pds/test/watchdog-event-driven.test.js +19 -12

…	…	@@ -1,8 +1,8 @@
1		// Step 3: the stale-front watchdog is now EVENT-INFORMED. A firehose 'append' (block arriving via
2		// passive replication) proves the front is live → the watchdog SKIPS the update({wait:true}) poll.
3		// Only when there's been NO append (idle OR stalled) does it poll to distinguish idle-alive from a
4		// zombie connection, and drop the conn at the strike limit. Events set the liveness floor; the timer
5		// only catches ABSENCE (the audit's rule). This unit-drives _refreshFronts deterministically.
	1	// The stale-front watchdog is EVENT-INFORMED + PULL-ONLY. A firehose 'append' (a block arriving via the
	2	// standing download) proves the front is live → the watchdog SKIPS the update({wait:true}) poll. Only on
	3	// NO append does it nudge with a bounded pull. It NEVER destroys the connection — recovery is PeerFront's
	4	// standing download (re-attaches direction-free on reconnect) + hyperswarm's keepalive/topic-rejoin.
	5	// Events set the liveness floor; the timer only catches ABSENCE. This unit-drives _refreshFronts.
6	6	import test from 'brittle'
7	7	import { PeerRouter } from '../lib/peer-router.js'
8	8
…	…	@@ -32,20 +32,27 @@ test('watchdog SKIPS the poll for a LIVE front (recent append) — event-informe
32	32	t.absent(destroyed, 'live front connection is NOT dropped')
33	33	})
34	34
35		test('watchdog POLLS + drops a STALLED front (no append + poll times out) — safety net intact', async (t) => {
	35	test('watchdog NUDGES a stalled front but NEVER destroys the connection (pull-only backstop)', async (t) => {
	36	// Post-audit contract: recovery is PeerFront's STANDING download (flips downloading:true → corestore
	37	// re-attaches direction-free on reconnect) + hyperswarm's own keepalive/topic-rejoin. The watchdog is a
	38	// pure staleness ceiling: on no-append it nudges with a bounded update({wait}) pull, and NEVER force-
	39	// destroys the connection (that fought hyperswarm's connection management and masked the missing download).
36	40	const router = mkRouter()
	41	let updateCalls = 0
37	42	let destroyed = false
38	43	const stalled = {
39		seq: { core: { update: () => new Promise(() => {}) } }, // never resolves → withTimeout fires → strike
	44	seq: { core: { update: () => { updateCalls++; return new Promise(() => {}) } } }, // never resolves → withTimeout fires
40	45	_conn: { destroy: () => { destroyed = true } }
41	46	}
42	47	router._fronts.set('did:plc:stalled', stalled)
43	48	router._lastAppend.set('did:plc:stalled', Date.now() - 10 * 60 * 1000) // no append for 10 min → suspect stall
44	49
45		await router._refreshFronts() // strike 1 (poll times out)
46		t.is(router._staleStrikes.get('did:plc:stalled'), 1, 'stalled front (no append + poll timeout) accrues a strike')
47		t.absent(destroyed, 'not dropped before the strike limit')
	50	await router._refreshFronts() // nudge 1 (poll times out)
	51	t.is(updateCalls, 1, 'stalled front (no recent append) IS nudged with a bounded update({wait}) pull')
	52	t.is(router._staleStrikes.get('did:plc:stalled'), 1, 'a failed nudge accrues a strike (diagnose visibility only)')
	53	t.absent(destroyed, 'connection NOT destroyed after a failed nudge')
48	54
49		await router._refreshFronts() // strike 2 == staleStrikeLimit → drop the zombie conn
50		t.ok(destroyed, 'stalled front connection DROPPED at the strike limit — stall detection preserved')
	55	await router._refreshFronts() // nudge 2 — the OLD strike-limit "drop" point
	56	await router._refreshFronts() // nudge 3 — well past the old limit
	57	t.absent(destroyed, 'connection is NEVER force-destroyed — recovery is the standing download + hyperswarm reconnect')
51	58	})

atproto-bridge commits / 9643f51

Changed files (6) +157 -54