OverviewFilesCommitsIssuesDocsReleasesPeers

favela-gateway commits / fa9c1f0

metrics verb: process, roles, membrane, fediverse, and bsky's own view
- lib/probes.js: relay getHostStatus (status + seq vs our emitted), per served DID the relay's rev vs ours, PLC #atproto_pds + handle, AppView profile and newest post. Every problem NAMED in plain words. Cached 5 min, on demand only (never a timer, never per page view — repeated outbound HTTPS is how favela-web leaked 20 GB under Bare); concurrent callers share one probe; never throws. fetch injectable. - lib/gateway.js: `metrics` (one answer for the panel and the CLI): process uptime/RSS/heap (bare-os), each role up or down with its error, the membrane's and the fediverse role's own metrics, and the probes for the membrane's host (from serverDid). `refresh: true` forces a probe. Tests first: probes 5/5 (green, named failures, unreachable never throws, cache TTL, shared in-flight); gateway 14/14. Ran once against the real relay, PLC and AppView: parses; found loser-club's relay rev (3mrnruepxss22) NEWER than the head we serve (3mrfgnyiurk22). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
fa9c1f0a8f4a3a163079f730066bd345cad2ebd9
by plankatron <fleeky@gmail.com> · 2026-10-13T22:40:02.000Z

Changed files (4) +278 -4

lib/gateway.js +40 -4
……@@ -18,6 +18,8 @@
1818
1919import { startMembrane } from 'atproto-bridge/pds/lib/membrane.js'
2020import { startApBridge } from 'ap-bridge/lib/bridge.js'
21import os from 'bare-os'
22import { createProbes } from './probes.js'
2123
2224const ROLES = {
2325 membrane: async (cfg, opts) => startMembrane(cfg, opts),
……@@ -35,8 +37,9 @@ const defaultLog = (o) => console.log(JSON.stringify(o))
3537 * @param {Array} [opts.bootstrap] DHT bootstrap for every role's swarm (tests pass [])
3638 * @param {string} [opts.version]
3739 */
38export async function startGateway ({ roles, log = defaultLog, bootstrap = null, version = null, ...configs }) {
40export async function startGateway ({ roles, log = defaultLog, bootstrap = null, version = null, fetch = null, ...configs }) {
3941 for (const name of roles) if (!ROLES[name]) throw new Error('unknown role: ' + name)
42 const startedAt = Date.now()
4043
4144 const running = {}
4245 const down = {}
……@@ -75,19 +78,52 @@ export async function startGateway ({ roles, log = defaultLog, bootstrap = null,
7578 return { ok: true, version, roles: out }
7679 }
7780
81 // bsky's own view of the membrane's host. Only when the membrane runs: it is the role bsky reaches.
82 const membraneCfg = configs.membrane && !(configs.membrane instanceof Error) ? configs.membrane : null
83 const hostname = membraneCfg
84 ? (membraneCfg.crawlHostname || (/^did:web:/.test(membraneCfg.serverDid || '') ? membraneCfg.serverDid.slice('did:web:'.length) : null))
85 : null
86 const probes = running.membrane && hostname ? createProbes({ hostname, fetch }) : null
87
88 /**
89 * One answer for the panel and the CLI. Each role reports from its own local state; bsky's view comes
90 * from the probes (cached; `refresh: true` forces a new probe). A role that is down is named with its
91 * error and has no section — never a stale one.
92 */
93 const metrics = async (params = {}) => {
94 const mem = os.memoryUsage()
95 const out = {
96 at: Date.now(),
97 version,
98 process: { uptimeS: Math.round((Date.now() - startedAt) / 1000), rss: mem.rss, heapUsed: mem.heapUsed, heapTotal: mem.heapTotal },
99 roles: {},
100 membrane: null,
101 fediverse: null,
102 bsky: null
103 }
104 for (const name of roles) out.roles[name] = running[name] ? { up: true } : { up: false, error: down[name] }
105 if (running.membrane) out.membrane = running.membrane.metrics()
106 if (running.fediverse) out.fediverse = await running.fediverse.metrics()
107 if (probes) {
108 const dids = out.membrane.dids.map((d) => ({ did: d.did, rev: d.rev }))
109 out.bsky = await probes.get({ dids, emitted: out.membrane.relay.emitted }, { force: !!params.refresh })
110 }
111 return out
112 }
113
78114 // ONE verb map. The roles' verbs do not overlap except `status`, which the gateway answers for all
79115 // of them; a collision would be a bug, so it is refused here rather than silently shadowed.
80 const handlers = { status }
116 const handlers = { status, metrics }
81117 for (const name of roles) {
82118 if (!running[name]) continue
83119 for (const [verb, fn] of Object.entries(running[name].handlers)) {
84 if (verb === 'status') continue
120 if (verb === 'status' || verb === 'metrics') continue
85121 if (handlers[verb]) throw new Error(`verb ${verb} is claimed by two roles`)
86122 handlers[verb] = fn
87123 }
88124 }
89125
90 return { roles: running, down, handlers, status, steps }
126 return { roles: running, down, handlers, status, metrics, steps }
91127}
92128
93129export default startGateway
lib/probes.js +114 -0
……@@ -0,0 +1,114 @@
1// bsky probes — "are we resolving to bsky?", asked of bsky itself (hyper-social #80).
2//
3// Nothing local can see these: on 2026-10-13 bsky's relay had us "offline" while every local surface
4// was healthy, and in 2026-07 a parked relay cursor hid a 20-hour federation outage. So the gateway asks:
5//
6// relay getHostStatus(host) status + seq, against our emitted seq
7// per DID relay getLatestCommit(did) does the relay hold the head we serve?
8// plc.directory/<did> does #atproto_pds point at us? (and the handle)
9// AppView getProfile / getAuthorFeed does bsky resolve the account, and its newest post
10//
11// ⛔ CACHED, AND ONLY ON DEMAND. Probes run when metrics are asked for and the cache is older than the
12// TTL — never on a timer and never once per page view. favela-web's 20 GB leak was repeated outbound
13// HTTPS under Bare (bare-tls never settled `_open` on destroy); a probe loop would re-create it. The
14// gateway's RSS is on the same panel, so a leak would be visible.
15//
16// ⛔ NEVER THROWS. A probe that fails is reported as unreachable with its reason; metrics always answer.
17
18const RELAY = 'https://bsky.network'
19const PLC = 'https://plc.directory'
20const APPVIEW = 'https://public.api.bsky.app'
21const TIMEOUT_MS = 8000
22
23export function createProbes ({ hostname, fetch: fetchImpl = null, ttlMs = 5 * 60 * 1000, now = () => Date.now(), timeoutMs = TIMEOUT_MS } = {}) {
24 if (!hostname) throw new Error('createProbes: hostname required')
25 const doFetch = fetchImpl || ((...a) => globalThis.fetch(...a))
26 let cache = null
27 let inFlight = null
28
29 async function getJson (url) {
30 let timer = null
31 try {
32 const res = await Promise.race([
33 doFetch(url),
34 new Promise((resolve, reject) => { timer = setTimeout(() => reject(new Error('timeout after ' + timeoutMs + ' ms')), timeoutMs) })
35 ])
36 if (!res || !res.ok) return { error: 'HTTP ' + (res && res.status) }
37 return { body: await res.json() }
38 } catch (err) {
39 return { error: (err && err.message) || String(err) }
40 } finally {
41 clearTimeout(timer)
42 }
43 }
44
45 async function probeDid ({ did, rev }, problems) {
46 const out = { did, ourRev: rev || null, relayRev: null, inSync: null, plcPds: null, handle: null, appview: { resolves: null, postsCount: null, newestPost: null, newestIndexedAt: null } }
47
48 const lc = await getJson(`${RELAY}/xrpc/com.atproto.sync.getLatestCommit?did=${encodeURIComponent(did)}`)
49 if (lc.error) problems.push(`${did}: relay getLatestCommit unreachable (${lc.error})`)
50 else {
51 out.relayRev = lc.body.rev || null
52 if (out.relayRev && out.ourRev) {
53 out.inSync = out.relayRev === out.ourRev
54 // Revs are TIDs, so string order is time order.
55 if (out.relayRev < out.ourRev) problems.push(`${did}: the relay is behind — it holds ${out.relayRev}, we serve ${out.ourRev}`)
56 else if (out.relayRev > out.ourRev) problems.push(`${did}: the relay holds a NEWER rev (${out.relayRev}) than we serve (${out.ourRev})`)
57 }
58 }
59
60 // The documented form is the bare DID in the path; DIDs are URL-safe.
61 const plc = await getJson(`${PLC}/${did}`)
62 if (plc.error) problems.push(`${did}: PLC unreachable (${plc.error})`)
63 else {
64 const svc = (plc.body.service || []).find((s) => s && (s.id === '#atproto_pds' || s.id === did + '#atproto_pds'))
65 out.plcPds = svc ? svc.serviceEndpoint : null
66 const aka = (plc.body.alsoKnownAs || []).find((a) => typeof a === 'string' && a.startsWith('at://'))
67 out.handle = aka ? aka.slice('at://'.length) : null
68 if (out.plcPds !== 'https://' + hostname) problems.push(`${did}: PLC #atproto_pds points at ${out.plcPds || 'nothing'}, not https://${hostname}`)
69 }
70
71 const prof = await getJson(`${APPVIEW}/xrpc/app.bsky.actor.getProfile?actor=${encodeURIComponent(did)}`)
72 if (prof.error) {
73 out.appview.resolves = false
74 problems.push(`${did}: the bsky AppView does not resolve the profile (${prof.error})`)
75 } else {
76 out.appview.resolves = true
77 out.appview.postsCount = prof.body.postsCount == null ? null : prof.body.postsCount
78 const feed = await getJson(`${APPVIEW}/xrpc/app.bsky.feed.getAuthorFeed?actor=${encodeURIComponent(did)}&limit=1`)
79 const top = !feed.error && feed.body.feed && feed.body.feed[0] && feed.body.feed[0].post
80 if (top) { out.appview.newestPost = top.uri; out.appview.newestIndexedAt = top.indexedAt || null }
81 }
82 return out
83 }
84
85 async function probe ({ dids = [], emitted = null } = {}) {
86 const problems = []
87 const relay = { status: null, seq: null, error: null }
88 const hs = await getJson(`${RELAY}/xrpc/com.atproto.sync.getHostStatus?hostname=${encodeURIComponent(hostname)}`)
89 if (hs.error) {
90 relay.error = hs.error
91 problems.push(`relay getHostStatus unreachable (${hs.error})`)
92 } else {
93 relay.status = hs.body.status || null
94 relay.seq = hs.body.seq == null ? null : hs.body.seq
95 if (relay.status && relay.status !== 'active') problems.push(`the bsky relay has ${hostname} ${relay.status} (seq ${relay.seq})`)
96 if (emitted != null && relay.seq != null && emitted < relay.seq) problems.push(`our emitted seq ${emitted} is behind the relay's cursor ${relay.seq} — it can never advance`)
97 }
98 const rows = []
99 for (const d of dids) rows.push(await probeDid(d, problems))
100 return { at: now(), hostname, relay, dids: rows, problems }
101 }
102
103 // The cached read every caller uses. Concurrent callers share the probe in flight.
104 async function get (args, { force = false } = {}) {
105 if (!force && cache && now() - cache.at < ttlMs) return cache
106 if (inFlight) return inFlight
107 inFlight = probe(args).then((r) => { cache = r; return r }).finally(() => { inFlight = null })
108 return inFlight
109 }
110
111 return { probe, get, cached: () => cache }
112}
113
114export default createProbes
test/gateway.test.js +26 -0
……@@ -110,3 +110,29 @@ test('the stop steps free everything: the same ports start again in one process'
110110 t.is(two.roles.fediverse.bridgeKey, one.roles.fediverse.bridgeKey, 'the published bridgeKey survives a restart')
111111 t.is(two.roles.membrane.swarmKey, one.roles.membrane.swarmKey, 'and so does the membrane\'s swarm identity')
112112})
113
114// ── metrics: one verb, every role, plus bsky's own view (hyper-social #80) ─────────────────────────────
115const offlineFetch = async (url) => ({ ok: true, status: 200, json: async () => (url.includes('getHostStatus') ? { status: 'active', seq: 0 } : {}) })
116
117test('★ metrics: process, roles, membrane, fediverse and bsky in one answer', async (t) => {
118 const g = await startGateway({ roles: ['membrane', 'fediverse'], membrane: membraneConfig(), fediverse: apConfig(), fetch: offlineFetch, ...quiet })
119 t.teardown(() => stop(g))
120 const m = await g.handlers.metrics({})
121 t.ok(m.process.uptimeS >= 0 && m.process.rss > 0, 'process uptime and RSS')
122 t.is(m.roles.membrane.up, true)
123 t.is(m.roles.fediverse.up, true)
124 t.alike(m.membrane.relay, { emitted: 0, seqBase: 0 }, 'the membrane section is the role\'s own metrics')
125 t.is(m.fediverse.users[0].username, 'p2p', 'and so is the fediverse section')
126 t.is(m.bsky.relay.status, 'active', 'bsky\'s view, through the probes')
127 t.is(m.bsky.hostname, 'pds.test', 'probed for the membrane\'s own host (from serverDid)')
128})
129
130test('metrics still answers with a role down, and names it', async (t) => {
131 const g = await startGateway({ roles: ['membrane', 'fediverse'], membrane: membraneConfig(), fediverse: new Error('fediverse config: AP_HOST is required'), fetch: offlineFetch, ...quiet })
132 t.teardown(() => stop(g))
133 const m = await g.handlers.metrics({})
134 t.is(m.roles.fediverse.up, false)
135 t.is(m.roles.fediverse.error, 'fediverse config: AP_HOST is required')
136 t.is(m.fediverse, null, 'no section for a role that is not running')
137 t.ok(m.membrane, 'the running role still reports')
138})
test/probes.test.js +98 -0
……@@ -0,0 +1,98 @@
1// bsky probes — "are we resolving to bsky?", asked of bsky itself (hyper-social #80).
2//
3// Every check here is something that silently broke at least once: the relay marking us offline
4// (2026-10-13, and 2026-08), the relay holding an older rev than we serve, PLC pointing somewhere else,
5// the AppView not resolving a profile. Each problem comes back NAMED, so the CLI and the panel can say
6// what is wrong instead of showing a red light.
7
8import test from 'brittle'
9import { createProbes } from '../lib/probes.js'
10
11const HOST = 'pds.7rnx.net'
12const WJ = 'did:plc:wj62ebbusln6gxo5azo6ogzo'
13
14// A fake fetch answering by URL prefix; anything unlisted is a 404.
15function fakeFetch (routes, calls = []) {
16 return async (url) => {
17 calls.push(url)
18 for (const [prefix, body] of Object.entries(routes)) {
19 if (url.startsWith(prefix)) {
20 if (body instanceof Error) throw body
21 return { ok: true, status: 200, json: async () => body }
22 }
23 }
24 return { ok: false, status: 404, json: async () => ({}) }
25 }
26}
27
28const healthy = {
29 ['https://bsky.network/xrpc/com.atproto.sync.getHostStatus']: { hostname: HOST, seq: 40267, status: 'active' },
30 ['https://bsky.network/xrpc/com.atproto.sync.getLatestCommit']: { cid: 'bafyA', rev: '3mrhxnkbhv622' },
31 ['https://plc.directory/' + WJ]: { alsoKnownAs: ['at://p2p.7rnx.net'], service: [{ id: '#atproto_pds', type: 'AtprotoPersonalDataServer', serviceEndpoint: 'https://pds.7rnx.net' }] },
32 ['https://public.api.bsky.app/xrpc/app.bsky.actor.getProfile']: { did: WJ, handle: 'p2p.7rnx.net', postsCount: 199 },
33 ['https://public.api.bsky.app/xrpc/app.bsky.feed.getAuthorFeed']: { feed: [{ post: { uri: 'at://' + WJ + '/app.bsky.feed.post/3mvks7khp3u6w', indexedAt: '2026-09-15T14:05:39Z' } }] }
34}
35
36const ours = [{ did: WJ, rev: '3mrhxnkbhv622' }]
37
38test('★ all green: relay active and in sync, PLC points here, the AppView resolves', async (t) => {
39 const p = createProbes({ hostname: HOST, fetch: fakeFetch(healthy) })
40 const r = await p.probe({ dids: ours, emitted: 40267 })
41 t.is(r.relay.status, 'active')
42 t.is(r.relay.seq, 40267)
43 const d = r.dids[0]
44 t.is(d.relayRev, '3mrhxnkbhv622')
45 t.is(d.inSync, true, 'the relay holds the head we serve')
46 t.is(d.plcPds, 'https://pds.7rnx.net')
47 t.is(d.handle, 'p2p.7rnx.net')
48 t.is(d.appview.resolves, true)
49 t.is(d.appview.newestPost, 'at://' + WJ + '/app.bsky.feed.post/3mvks7khp3u6w')
50 t.alike(r.problems, [], 'nothing to report')
51})
52
53test('⛔ each failure is named, not just counted', async (t) => {
54 const broken = {
55 ...healthy,
56 ['https://bsky.network/xrpc/com.atproto.sync.getHostStatus']: { hostname: HOST, seq: 40262, status: 'offline' },
57 ['https://bsky.network/xrpc/com.atproto.sync.getLatestCommit']: { cid: 'bafyOld', rev: '3mrhxnkbhv522' },
58 ['https://plc.directory/' + WJ]: { service: [{ id: '#atproto_pds', serviceEndpoint: 'https://bsky.social' }] }
59 }
60 delete broken['https://public.api.bsky.app/xrpc/app.bsky.actor.getProfile']
61 const p = createProbes({ hostname: HOST, fetch: fakeFetch(broken) })
62 const r = await p.probe({ dids: ours, emitted: 40267 })
63 const text = r.problems.join('\n')
64 t.ok(/relay.*offline/i.test(text), 'the relay has us offline')
65 t.ok(/behind/i.test(text) && text.includes('3mrhxnkbhv522'), 'the relay is behind our head, with its rev')
66 t.ok(/PLC/.test(text) && text.includes('bsky.social'), 'PLC points elsewhere, and where')
67 t.ok(/AppView/.test(text), 'the AppView does not resolve the profile')
68 t.is(r.dids[0].inSync, false)
69})
70
71test('a probe that throws is reported as unreachable, never thrown — metrics must always answer', async (t) => {
72 const p = createProbes({ hostname: HOST, fetch: fakeFetch({ 'https://': new Error('ECONNRESET') }) })
73 const r = await p.probe({ dids: ours, emitted: 1 })
74 t.ok(r.relay.error, 'the relay probe failed, with the reason')
75 t.ok(r.problems.some((s) => /unreachable|ECONNRESET/.test(s)), 'and that is a named problem')
76})
77
78test('⛔ cached: a second call within the TTL makes no request — page views must not reach bsky', async (t) => {
79 const calls = []
80 let now = 1e9
81 const p = createProbes({ hostname: HOST, fetch: fakeFetch(healthy, calls), ttlMs: 5 * 60e3, now: () => now })
82 await p.get({ dids: ours, emitted: 40267 })
83 const n = calls.length
84 now += 60e3
85 await p.get({ dids: ours, emitted: 40267 })
86 t.is(calls.length, n, 'no new requests one minute later')
87 now += 5 * 60e3
88 await p.get({ dids: ours, emitted: 40267 })
89 t.ok(calls.length > n, 'past the TTL it probes again')
90})
91
92test('concurrent callers share one probe in flight', async (t) => {
93 const calls = []
94 const p = createProbes({ hostname: HOST, fetch: fakeFetch(healthy, calls) })
95 await Promise.all([p.get({ dids: ours, emitted: 1 }), p.get({ dids: ours, emitted: 1 })])
96 const hostCalls = calls.filter((u) => u.includes('getHostStatus')).length
97 t.is(hostCalls, 1, 'one getHostStatus for two simultaneous asks')
98})