| … | … | @@ -1,6 +1,6 @@ |
| 1 | 1 | { |
| 2 | 2 | "name": "atproto-bridge", |
| 3 | "version": "0.9.51", | |
| 3 | "version": "0.9.52", | |
| 4 | 4 | "description": "Stateless keyless atproto MEMBRANE \u2014 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", |
| … | … | @@ -1,41 +1,24 @@ |
| 1 | // Runnable atproto MEMBRANE — a stateless bridge (Model B). Holds NO key, NO account, NO | |
| 2 | // custodial storage. It fronts self-custody hyper-proto-peers (cfg.frontedDids) discovered | |
| 3 | // over the DHT, verifies each against its DID key, and serves them keylessly to bsky. All | |
| 4 | // identity/content/signing lives in the peers (hyper-proto-peer). Two surfaces: | |
| 5 | // 1. the bsky-facing HTTP+WS membrane (server.js / server-bare.js) on cfg.port | |
| 1 | // Runnable atproto MEMBRANE — the standalone app around lib/membrane.js. Two surfaces: | |
| 2 | // 1. the bsky-facing HTTP+WS membrane (lib/membrane.js) on cfg.port | |
| 6 | 3 | // 2. a UDS control plane (control.js) at cfg.uds — the pgh readiness/management surface, |
| 7 | 4 | // which also registers with pgh-sidecar so `pgh app status atproto-bridge` resolves us. |
| 5 | // | |
| 6 | // The membrane itself is a module since 0.9.52 so the favela gateway (hyper-social #80) can run it | |
| 7 | // beside ap-bridge in one process. This file owns what only a standalone process may own: the | |
| 8 | // control socket and its sidecar name, the OTA exit, and the signal handlers. | |
| 8 | 9 | |
| 9 | 10 | import '../lib/bare-globals.js' // FIRST — polyfills TextEncoder/TextDecoder under Bare |
| 10 | 11 | import process from '#node:process' |
| 11 | 12 | import fs from '#node:fs' |
| 12 | 13 | import path from '#node:path' |
| 13 | 14 | import { fileURLToPath } from '#node:url' |
| 14 | import Hyperswarm from 'hyperswarm' | |
| 15 | import b4a from 'b4a' | |
| 16 | import { openCorestore } from 'atproto-store/lib/corestore.js' | |
| 17 | 15 | import { loadConfig } from '../lib/config.js' |
| 18 | import { loadOrCreateSwarmKeyPair } from '../lib/swarm-keypair.js' | |
| 16 | import { startMembrane } from '../lib/membrane.js' | |
| 19 | 17 | import { startControl } from '../lib/control.js' |
| 20 | 18 | import { createShutdown } from '../lib/shutdown.js' |
| 21 | import { FleetRelay } from '../lib/fleet-relay.js' | |
| 22 | import { DidPeerRouter } from 'atproto-did-peer' | |
| 23 | import { resolvePlcVerifyKey } from '../lib/verify-key.js' | |
| 24 | import { discoveryTopic } from 'atproto-identity/lib/discovery.js' | |
| 25 | 19 | |
| 26 | 20 | const cfg = loadConfig() |
| 27 | 21 | |
| 28 | // The HTTP membrane is transport-specific: bare-http1/bare-ws under Bare, node http+ws | |
| 29 | // otherwise. (bsky-interop edge only; P2P sync is Hypercore, not this.) | |
| 30 | const isBare = typeof Bare !== 'undefined' | |
| 31 | const { createPdsServer } = await import(isBare ? '../lib/server-bare.js' : '../lib/server.js') | |
| 32 | ||
| 33 | // Bare has no global `fetch`; PLC resolution (verify a fronted peer's DID key before serving | |
| 34 | // it — verify-key.js resolvePlcVerifyKey) needs one. Install the WHATWG-standard bare-fetch | |
| 35 | // global so it works on the prod (bare) runtime. Without this, front() throws "fetch is not | |
| 36 | // defined" and NO peer is fronted — silently dropping wj62. Node already has global fetch. | |
| 37 | if (isBare) await import('bare-fetch/global') | |
| 38 | ||
| 39 | 22 | // App version (for sidecar registration) — best-effort from the deploy-root package.json. |
| 40 | 23 | let version = null |
| 41 | 24 | try { |
| … | … | @@ -43,107 +26,10 @@ try { |
| 43 | 26 | version = JSON.parse(fs.readFileSync(path.join(here, '..', '..', 'package.json'), 'utf8')).version |
| 44 | 27 | } catch {} |
| 45 | 28 | |
| 46 | // The front-store holds the aggregate firehose (relay) + replicated peer cores. It is content, | |
| 47 | // NOT accounts — no ACCOUNT key material ever lives here. (The swarm keypair below is the one | |
| 48 | // piece of key material the membrane does keep: a transport/DHT identity that never signs a repo | |
| 49 | // commit and holds no account. Amended from "no key material ever" — see swarm-keypair.js.) | |
| 50 | const frontStore = openCorestore(path.join(cfg.storage, 'front-store')) | |
| 51 | // PERSISTENT swarm identity. A random per-boot keypair (plain `new Hyperswarm()`) made the bridge's | |
| 52 | // Noise public key unstable, so hyper-proto-peer's PEER_WRITERS allowlist — which gates apply-write, | |
| 53 | // delegate and get-service-auth on the CLIENT's Noise key — could never be populated: any entry | |
| 54 | // authorised exactly one boot. Stable key ⇒ the bridge is allowlistable. | |
| 55 | const swarmKeyPair = loadOrCreateSwarmKeyPair(cfg.storage) | |
| 56 | const swarm = new Hyperswarm({ keyPair: swarmKeyPair }) | |
| 57 | // Logged at BOOT, deliberately: this is the value an operator must paste into each DID's `writers` | |
| 58 | // array. Without it, populating PEER_WRITERS means fishing a key out of a peer's connection logs and | |
| 59 | // hoping it was ours. Public key only — the secret stays in the file. | |
| 60 | console.log(JSON.stringify({ msg: 'swarm identity', publicKey: b4a.toString(swarmKeyPair.publicKey, 'hex') })) | |
| 61 | // Durable emitted-seq floor — kept in a plain file BESIDE the front-store (survives a corestore | |
| 62 | // reset), so subscribeRepos seq stays monotonic across redeploys and can be lifted above a relay's | |
| 63 | // stored cursor for recovery (see FleetRelay seqBase / the getHostStatus recovery). | |
| 64 | const seqBaseFile = path.join(cfg.storage, 'relay-seq-base') | |
| 65 | let seqBase = 0 | |
| 66 | try { seqBase = Number(fs.readFileSync(seqBaseFile, 'utf8').trim()) || 0 } catch {} | |
| 67 | const persistSeqBase = (n) => { try { fs.mkdirSync(cfg.storage, { recursive: true }) } catch {}; fs.writeFileSync(seqBaseFile, String(n)) } | |
| 68 | // The relay aggregates each fronted peer's signed firehose into one monotonic subscribeRepos | |
| 69 | // stream, so bsky's relay crawls the whole set in one shot (the egress membrane). | |
| 70 | const relay = await new FleetRelay(frontStore.get({ name: 'peer-relay-firehose', valueEncoding: 'json' }), { seqBase, onSeqBase: persistSeqBase }).ready() | |
| 71 | // PERSISTED handle->did map (survives restart — an offline blind-relayed peer, e.g. ashatron, can't | |
| 72 | // re-teach its handle at boot, so an in-memory-only map left it unresolvable after a restart). | |
| 73 | const handleMapFile = path.join(cfg.storage, 'handle-map.json') | |
| 74 | let handles = {} | |
| 75 | try { handles = JSON.parse(fs.readFileSync(handleMapFile, 'utf8')) || {} } catch {} | |
| 76 | const persistHandles = (obj) => { try { fs.mkdirSync(cfg.storage, { recursive: true }) } catch {}; fs.writeFileSync(handleMapFile, JSON.stringify(obj)) } | |
| 77 | // The canonical consumer primitive (atproto-did-peer DidPeerRouter — unified across bridge/favela in | |
| 78 | // EXTRACTION-PLAN Phase 1). Owns-swarm mode: it auto-wires onconnection + joins each fronted topic in its | |
| 79 | // ctor (via `join`) + runs the pull-only refresh/failover timer. `rpc: true` opens one protomux-rpc per | |
| 80 | // connection for the peer-signs-over-RPC write/session/blob path. `resolveVerifyKey` fronts a DID with no | |
| 81 | // pre-seeded key via PLC. `relay` folds the freshest co-owner's firehose into the aggregate subscribeRepos. | |
| 82 | const peers = new DidPeerRouter(frontStore, { | |
| 83 | swarm, | |
| 84 | join: (did) => { try { swarm.join(discoveryTopic(did), { server: true, client: true }); swarm.flush().catch(() => {}) } catch {} }, | |
| 85 | fronted: cfg.frontedDids, | |
| 86 | relay, | |
| 87 | resolveVerifyKey: resolvePlcVerifyKey, | |
| 88 | rpc: true, | |
| 89 | handles, | |
| 90 | onHandles: persistHandles | |
| 91 | }) | |
| 92 | ||
| 93 | // ⛔ THE SWARM DOOR onto fronting. `allowFront()` only UN-DENIES; a DID enters the fronted set | |
| 94 | // through `add()`, reachable from config at boot or the owner-gated control socket — so being | |
| 95 | // carried was an operator action and a peer had no way to ask. This is that way. | |
| 96 | // | |
| 97 | // ⚠ PERSISTED, exactly like the handle map above it: `frontedDids` is CONFIG, and what peers asked | |
| 98 | // for is not. A restart that silently unfronted everyone who asked would make the door a lie. | |
| 99 | const requestedFile = path.join(cfg.storage, 'fronted-requested.json') | |
| 100 | const readRequested = () => { try { return JSON.parse(fs.readFileSync(requestedFile, 'utf8')) } catch { return {} } } | |
| 101 | const requested = readRequested() | |
| 102 | // ⛔ The handle goes back in too, or a restart forgets the name it granted and | |
| 103 | // `<handle>/.well-known/atproto-did` stops answering for someone already published on-chain. | |
| 104 | for (const [id, r] of Object.entries(requested)) { | |
| 105 | try { peers.add(id, r && r.handle ? { handle: r.handle } : {}) } catch {} | |
| 106 | } | |
| 107 | if (Object.keys(requested).length) console.log(JSON.stringify({ msg: 'refronted on boot', count: Object.keys(requested).length })) | |
| 108 | ||
| 109 | try { | |
| 110 | const { serveFronting, Admission } = await import('../lib/fronting.js') | |
| 111 | const door = serveFronting({ | |
| 112 | swarm, | |
| 113 | peers, | |
| 114 | admission: new Admission(), | |
| 115 | policy: process.env.PDS_FRONT_POLICY || 'open', | |
| 116 | // What the membrane OFFERS: the domain it grants names under, and the endpoint a DID must point | |
| 117 | // at. A peer needs both before it commits, because accepting is a permanent append to a global log. | |
| 118 | userDomain: cfg.userDomain, | |
| 119 | // ⛔ DERIVED FROM serverDid, not invented. `did:web:<host>` IS this membrane's public identity — | |
| 120 | // config.js already takes crawlHostname's default from exactly that host part, so using any other | |
| 121 | // source here would be a second answer to "where is this membrane" that could drift from the first. | |
| 122 | pdsUrl: 'https://' + (/^did:web:/.test(cfg.serverDid) ? cfg.serverDid.slice('did:web:'.length) : 'pds.7rnx.net'), | |
| 123 | onFronted: ({ id, root, handle }) => { | |
| 124 | const all = readRequested() | |
| 125 | all[id] = { root, handle: handle || null, at: Date.now() } | |
| 126 | fs.mkdirSync(cfg.storage, { recursive: true }) | |
| 127 | fs.writeFileSync(requestedFile, JSON.stringify(all, null, 2)) | |
| 128 | }, | |
| 129 | log: (m) => console.log(JSON.stringify(m)) | |
| 130 | }) | |
| 131 | await door.ready() | |
| 132 | // ⛔ PRINTED: it is what a peer needs in order to find this bridge at all, and a key written | |
| 133 | // nowhere is the favela-android device-key mistake one layer over. | |
| 134 | console.log(JSON.stringify({ msg: 'fronting open over the swarm', bridgeKey: door.key })) | |
| 135 | } catch (err) { | |
| 136 | // Non-fatal: the membrane serving what it already fronts is worth more than one that refuses to | |
| 137 | // boot because a second door would not open. | |
| 138 | console.log(JSON.stringify({ msg: 'swarm fronting unavailable', error: (err && err.message) || String(err) })) | |
| 139 | } | |
| 140 | ||
| 141 | // 1. bsky HTTP+WS membrane — keyless per-DID routing to the fronted peers. | |
| 142 | const server = createPdsServer({ peers, serverDid: cfg.serverDid, availableUserDomains: [cfg.userDomain], favelaWebUpstream: cfg.favelaWebUpstream }) | |
| 143 | await new Promise((r) => server.listen(cfg.port, cfg.bind, r)) | |
| 29 | const membrane = await startMembrane(cfg) | |
| 144 | 30 | |
| 145 | // 2. UDS control plane + pgh-sidecar registration. | |
| 146 | const control = await startControl({ uds: cfg.uds, name: process.env.PGH_APP_NAME || 'atproto-bridge', version, peers }) | |
| 31 | // UDS control plane + pgh-sidecar registration. | |
| 32 | const control = await startControl({ uds: cfg.uds, name: process.env.PGH_APP_NAME || 'atproto-bridge', version, peers: membrane.peers }) | |
| 147 | 33 | |
| 148 | 34 | // OTA — the app half of the push contract. The sidecar does the FULL install (tree fetch + dep closure + |
| 149 | 35 | // materialize + symlink swap) and THEN calls this; our only job is to exit cleanly so the supervisor |
| … | … | @@ -166,110 +52,16 @@ if (control && control.svc && !control.svc.hasHandler('release-ready')) { |
| 166 | 52 | console.log(JSON.stringify({ msg: 'ota: release-ready handler armed (push-based; sidecar installs, we respawn)' })) |
| 167 | 53 | } |
| 168 | 54 | |
| 169 | // 3. Warm the fronts in the BACKGROUND — never block the HTTP/UDS listen or the readiness probe. | |
| 170 | // A peer offline at boot (or one that hasn't signed its first commit yet) must not delay the bridge | |
| 171 | // coming up: awaiting fronts here blew past the 30s UDS readiness window → supervisor restart loop. | |
| 172 | // The router already joined every fronted topic in its constructor, so peers connect + get | |
| 173 | // identified in the background; front() folds each into the relay as it warms, and retries on demand. | |
| 174 | for (const did of cfg.frontedDids) { | |
| 175 | peers.front(did) | |
| 176 | .then(() => console.log(JSON.stringify({ msg: 'fronting', did }))) | |
| 177 | .catch((e) => console.error(JSON.stringify({ msg: 'front failed (will retry on demand)', did, error: e.message }))) | |
| 178 | } | |
| 179 | ||
| 180 | // 3b. Seq guard — the half of the seq-floor mechanism that was missing: something that NOTICES. | |
| 181 | // | |
| 182 | // FleetRelay.setSeqBase + the durable `relay-seq-base` file (above) have always been able to lift our | |
| 183 | // emitted seq past a relay's stored cursor. Nothing ever called them. Recovery required a human to know | |
| 184 | // to read com.atproto.sync.getHostStatus and hand-call the `set-seq-base` UDS RPC — so on 2026-07-31 the | |
| 185 | // bridge sat parked at cursor 9834 against our 79 frames for ~20 hours, with every local surface healthy. | |
| 186 | // A mechanism nothing invokes is not a mechanism; this is the trigger. | |
| 187 | // | |
| 188 | // Runs once at boot and every 15 minutes. Read-only unless the arithmetic says the relay can never reach | |
| 189 | // us. OUTWARD-FACING, so it inherits the same opt-in gate as the reactor's requestCrawl below. | |
| 190 | if (cfg.crawlRelay && cfg.crawlHostname) { | |
| 191 | const { runSeqGuard } = await import('../lib/seq-guard.js') | |
| 192 | const crawl = async () => { | |
| 193 | const res = await fetch(`${cfg.crawlRelay.replace(/\/$/, '')}/xrpc/com.atproto.sync.requestCrawl`, { | |
| 194 | method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ hostname: cfg.crawlHostname }) | |
| 195 | }) | |
| 196 | if (!res.ok) throw new Error(`requestCrawl ${res.status}`) | |
| 197 | } | |
| 198 | const pass = () => runSeqGuard({ | |
| 199 | relay: peers.relay, | |
| 200 | relayBase: cfg.crawlRelay, | |
| 201 | hostname: cfg.crawlHostname, | |
| 202 | requestCrawl: crawl, | |
| 203 | log: (o) => console.log(JSON.stringify(o)) | |
| 204 | }).catch((e) => console.error(JSON.stringify({ msg: 'seq-guard pass failed', error: e && e.message }))) | |
| 205 | pass() // boot pass, backgrounded — must never delay listen or the readiness probe | |
| 206 | const seqGuardTimer = setInterval(pass, 15 * 60 * 1000) | |
| 207 | if (seqGuardTimer.unref) seqGuardTimer.unref() | |
| 208 | } else { | |
| 209 | console.log(JSON.stringify({ msg: 'seq-guard OFF (no crawlRelay/crawlHostname) — a relay that parks us will stay parked' })) | |
| 210 | } | |
| 211 | ||
| 212 | // 4. Feed-registry reactor (community-ux.md §111) — OFF unless cfg.feedRegistries is set. Watch each keyless | |
| 213 | // FeedRegistry (the durable P2P allow-set) and, per feed entry, front its home DID + requestCrawl once, with | |
| 214 | // no per-community config and no restart. Gated + guarded: favela-community is dynamically imported only when | |
| 215 | // the feature is on, and any failure is contained to this feature (never blocks the membrane). requestCrawl | |
| 216 | // is OUTWARD-FACING (asks bsky's relay to crawl us) and stays inert unless cfg.crawlRelay is configured. | |
| 217 | if (cfg.feedRegistries.length) { | |
| 218 | ;(async () => { | |
| 219 | try { | |
| 220 | const { FeedRegistries } = await import('favela-community') | |
| 221 | const { makeMembraneReactor } = await import('../lib/membrane-reactor.js') | |
| 222 | // Live federation call: POST com.atproto.sync.requestCrawl { hostname } to the bsky relay. Off (null) | |
| 223 | // → the reactor still fronts, but fires no live crawl. Built here (not in the pure reactor) so the | |
| 224 | // outward-facing side-effect is confined to the wiring + explicitly configured. | |
| 225 | const requestCrawl = (cfg.crawlRelay && cfg.crawlHostname) | |
| 226 | ? async () => { | |
| 227 | const res = await fetch(`${cfg.crawlRelay.replace(/\/$/, '')}/xrpc/com.atproto.sync.requestCrawl`, { | |
| 228 | method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ hostname: cfg.crawlHostname }) | |
| 229 | }) | |
| 230 | if (!res.ok) throw new Error(`requestCrawl ${res.status}`) | |
| 231 | } | |
| 232 | : null | |
| 233 | if (!requestCrawl) console.log(JSON.stringify({ msg: 'feed-registry reactor: requestCrawl OFF (no crawlRelay/crawlHostname) — fronting only' })) | |
| 234 | const reconcile = makeMembraneReactor({ | |
| 235 | addFront: (did) => peers.add(did), | |
| 236 | front: (did) => peers.front(did), | |
| 237 | requestCrawl, | |
| 238 | log: (m) => console.log(m) | |
| 239 | }) | |
| 240 | const feedRegistries = new FeedRegistries(frontStore, swarm) | |
| 241 | for (const regKey of cfg.feedRegistries) { | |
| 242 | await feedRegistries.watch(regKey, (feeds) => reconcile(feeds)) | |
| 243 | console.log(JSON.stringify({ msg: 'watching feed registry', registry: regKey.slice(0, 12) + '…' })) | |
| 244 | } | |
| 245 | } catch (e) { | |
| 246 | console.error(JSON.stringify({ msg: 'feed-registry reactor setup failed (membrane unaffected)', error: e.message })) | |
| 247 | } | |
| 248 | })() | |
| 249 | } | |
| 250 | ||
| 251 | console.log(JSON.stringify({ | |
| 252 | msg: 'membrane listening', | |
| 253 | http: `${cfg.bind}:${cfg.port}`, | |
| 254 | uds: cfg.uds, | |
| 255 | swarmKey: b4a.toString(swarmKeyPair.publicKey, 'hex'), | |
| 256 | serverDid: cfg.serverDid, | |
| 257 | frontedDids: cfg.frontedDids, | |
| 258 | feedRegistries: cfg.feedRegistries | |
| 259 | })) | |
| 55 | console.log(JSON.stringify({ msg: 'control listening', uds: cfg.uds })) | |
| 260 | 56 | |
| 261 | 57 | // Bounded and timed (lib/shutdown.js): the caller's exit code wins, every step logs its duration, and |
| 262 | 58 | // a deadline exits anyway, naming the step it was stuck on. 0.9.50's OTA took ~50 s and exited 0. |
| 59 | // Control first, then the membrane's own steps in their original order (front-store last). | |
| 263 | 60 | const shutdown = createShutdown({ |
| 264 | 61 | exit: (code) => process.exit(code), |
| 265 | 62 | steps: [ |
| 266 | 63 | ['control', () => control.close()], |
| 267 | ['peers', () => peers.close()], | |
| 268 | ['relay', () => relay.close()], | |
| 269 | ['swarm', () => swarm.destroy()], | |
| 270 | ['server', () => new Promise((resolve) => server.close(resolve))], | |
| 271 | // Release the front-store Corestore fd-lock last so a redeploy can reopen. | |
| 272 | ['frontStore', () => frontStore.close()] | |
| 64 | ...membrane.steps | |
| 273 | 65 | ] |
| 274 | 66 | }) |
| 275 | 67 | process.on('SIGINT', () => shutdown(0)) |
| … | … | @@ -4,81 +4,94 @@ |
| 4 | 4 | // |
| 5 | 5 | // The membrane is keyless — there are no accounts to administer here. Handlers report what |
| 6 | 6 | // the membrane fronts (the self-custody peers) and resolve their handles. |
| 7 | // | |
| 8 | // ⛔ The verbs are a plain map (membraneHandlers), separate from the socket that serves them. The | |
| 9 | // favela gateway (hyper-social #80) serves them on ITS control socket, beside ap-bridge's verbs, and | |
| 10 | // must not bind atproto-bridge's socket or register its name: a release of atproto-bridge pushed to | |
| 11 | // a path the gateway served would restart the wrong process. | |
| 7 | 12 | |
| 8 | 13 | import { Service } from 'hyper-svc' |
| 9 | 14 | import { maintainSidecarRegistration } from './sidecar.js' |
| 10 | 15 | |
| 11 | export async function startControl ({ uds, name = 'atproto-bridge', version = null, peers = null, relay = null } = {}) { | |
| 12 | const svc = new Service({ name, uds }) // UDS only — omit `http` so we don't bind a port | |
| 16 | export function membraneHandlers ({ peers = null, relay = null } = {}) { | |
| 17 | return { | |
| 18 | status: async () => ({ | |
| 19 | ok: true, | |
| 20 | membrane: !relay, // the blind-relay mode passes `relay` instead of `peers` | |
| 21 | relay: relay ? relay.status() : undefined, | |
| 22 | fronted: peers ? peers.list() : [] | |
| 23 | }), | |
| 24 | ||
| 25 | 'list-repos': async () => ({ repos: peers ? peers.list().map((did) => ({ did, active: true })) : [] }), | |
| 26 | 'list-fronted': async () => ({ fronted: peers ? peers.list() : [] }), | |
| 13 | 27 | |
| 14 | svc.handle('status', async () => ({ | |
| 15 | ok: true, | |
| 16 | membrane: !relay, // the blind-relay mode passes `relay` instead of `peers` | |
| 17 | relay: relay ? relay.status() : undefined, | |
| 18 | fronted: peers ? peers.list() : [] | |
| 19 | })) | |
| 28 | // Introspection: a live snapshot of the membrane's swarm + per-front state (see it, don't | |
| 29 | // guess). `diagnose` is read-only + secret-free. `probe-front` forces an on-demand front for | |
| 30 | // one DID and returns the outcome/error — the fast way to see WHY a peer isn't being served. | |
| 31 | diagnose: async () => (peers ? peers.diagnose() : { membrane: true, peers: null }), | |
| 32 | // Recovery: lift the relay's emitted-seq floor above a downstream relay's stored cursor (read via | |
| 33 | // com.atproto.sync.getHostStatus), so it stops treating us as a future-cursor. Monotonic. Then | |
| 34 | // emit-sync + the relay's requestCrawl re-baselines us. See the indigo audit / the post-mortem. | |
| 35 | 'set-seq-base': async (params = {}) => { | |
| 36 | if (!peers || !peers.relay) return { ok: false, error: 'no relay' } | |
| 37 | const n = Number(params.seq) | |
| 38 | if (!Number.isFinite(n)) return { ok: false, error: 'numeric seq required' } | |
| 39 | const seqBase = await peers.relay.setSeqBase(n) | |
| 40 | return { ok: true, seqBase, length: peers.relay.length } | |
| 41 | }, | |
| 42 | 'emit-sync': async (params = {}) => { | |
| 43 | if (!peers) return { ok: false, error: 'no peers router' } | |
| 44 | if (!params.did) return { ok: false, error: 'did required' } | |
| 45 | try { return { ok: true, ...(await peers.emitSync(params.did)) } } catch (e) { return { ok: false, did: params.did, error: e.message } } | |
| 46 | }, | |
| 47 | 'probe-front': async (params = {}) => { | |
| 48 | if (!peers) return { ok: false, error: 'no peers router' } | |
| 49 | const did = params.did | |
| 50 | if (!did) return { ok: false, error: 'did required' } | |
| 51 | try { | |
| 52 | const f = await peers.front(did) | |
| 53 | const lc = await f.getLatestCommit() | |
| 54 | return { ok: true, did, handle: f.handle || null, firehoseLen: (f.seq && f.seq.core) ? f.seq.core.length : null, latest: lc } | |
| 55 | } catch (e) { | |
| 56 | return { ok: false, did, error: e.message } | |
| 57 | } | |
| 58 | }, | |
| 59 | // Eviction: stop serving a wrong head WITHOUT bouncing the process. A cached front outlives its peer's | |
| 60 | // connection by design (its cores keep replicating over the next one), so before these handlers the only | |
| 61 | // way to drop a front announcing a FORKED line — one whose rev sorts higher, so freshest-by-rev keeps | |
| 62 | // picking it — was a restart. `forget: true` also DENIES that firehose key, which is required whenever | |
| 63 | // the reason to evict is what the front is SERVING: peers re-announce on every append, so a plain drop | |
| 64 | // would be undone milliseconds later. `allow-front` is the undo. | |
| 65 | 'drop-front': async (params = {}) => { | |
| 66 | if (!peers) return { ok: false, error: 'no peers router' } | |
| 67 | if (!params.did) return { ok: false, error: 'did required' } | |
| 68 | try { | |
| 69 | return { ok: true, ...(await peers.dropFront(params.did, { firehoseKey: params.firehoseKey || null, forget: !!params.forget })) } | |
| 70 | } catch (e) { return { ok: false, did: params.did, error: e.message } } | |
| 71 | }, | |
| 72 | 'allow-front': async (params = {}) => { | |
| 73 | if (!peers) return { ok: false, error: 'no peers router' } | |
| 74 | if (!params.did) return { ok: false, error: 'did required' } | |
| 75 | try { return { ok: true, ...peers.allowFront(params.did, params.firehoseKey || null) } } catch (e) { return { ok: false, did: params.did, error: e.message } } | |
| 76 | }, | |
| 77 | // Drop EVERY cached front + resolved key for a DID so the next request re-fronts from scratch. Nothing is | |
| 78 | // denied and the DID stays fronted — this is the "re-derive, don't restart" lever. | |
| 79 | 'forget-did': async (params = {}) => { | |
| 80 | if (!peers) return { ok: false, error: 'no peers router' } | |
| 81 | if (!params.did) return { ok: false, error: 'did required' } | |
| 82 | try { return { ok: true, ...(await peers.forgetDid(params.did)) } } catch (e) { return { ok: false, did: params.did, error: e.message } } | |
| 83 | }, | |
| 20 | 84 | |
| 21 | svc.handle('list-repos', async () => ({ repos: peers ? peers.list().map((did) => ({ did, active: true })) : [] })) | |
| 22 | svc.handle('list-fronted', async () => ({ fronted: peers ? peers.list() : [] })) | |
| 85 | 'resolve-handle': async (params = {}) => ({ did: peers ? peers.resolveHandle(params.handle) : null }) | |
| 86 | } | |
| 87 | } | |
| 23 | 88 | |
| 24 | // Introspection: a live snapshot of the membrane's swarm + per-front state (see it, don't | |
| 25 | // guess). `diagnose` is read-only + secret-free. `probe-front` forces an on-demand front for | |
| 26 | // one DID and returns the outcome/error — the fast way to see WHY a peer isn't being served. | |
| 27 | svc.handle('diagnose', async () => (peers ? peers.diagnose() : { membrane: true, peers: null })) | |
| 28 | // Recovery: lift the relay's emitted-seq floor above a downstream relay's stored cursor (read via | |
| 29 | // com.atproto.sync.getHostStatus), so it stops treating us as a future-cursor. Monotonic. Then | |
| 30 | // emit-sync + the relay's requestCrawl re-baselines us. See the indigo audit / the post-mortem. | |
| 31 | svc.handle('set-seq-base', async (ctx, params = {}) => { | |
| 32 | if (!peers || !peers.relay) return { ok: false, error: 'no relay' } | |
| 33 | const n = Number(params.seq) | |
| 34 | if (!Number.isFinite(n)) return { ok: false, error: 'numeric seq required' } | |
| 35 | const seqBase = await peers.relay.setSeqBase(n) | |
| 36 | return { ok: true, seqBase, length: peers.relay.length } | |
| 37 | }) | |
| 38 | svc.handle('emit-sync', async (ctx, params = {}) => { | |
| 39 | if (!peers) return { ok: false, error: 'no peers router' } | |
| 40 | if (!params.did) return { ok: false, error: 'did required' } | |
| 41 | try { return { ok: true, ...(await peers.emitSync(params.did)) } } catch (e) { return { ok: false, did: params.did, error: e.message } } | |
| 42 | }) | |
| 43 | svc.handle('probe-front', async (ctx, params = {}) => { | |
| 44 | if (!peers) return { ok: false, error: 'no peers router' } | |
| 45 | const did = params.did | |
| 46 | if (!did) return { ok: false, error: 'did required' } | |
| 47 | try { | |
| 48 | const f = await peers.front(did) | |
| 49 | const lc = await f.getLatestCommit() | |
| 50 | return { ok: true, did, handle: f.handle || null, firehoseLen: (f.seq && f.seq.core) ? f.seq.core.length : null, latest: lc } | |
| 51 | } catch (e) { | |
| 52 | return { ok: false, did, error: e.message } | |
| 53 | } | |
| 54 | }) | |
| 55 | // Eviction: stop serving a wrong head WITHOUT bouncing the process. A cached front outlives its peer's | |
| 56 | // connection by design (its cores keep replicating over the next one), so before these handlers the only | |
| 57 | // way to drop a front announcing a FORKED line — one whose rev sorts higher, so freshest-by-rev keeps | |
| 58 | // picking it — was a restart. `forget: true` also DENIES that firehose key, which is required whenever | |
| 59 | // the reason to evict is what the front is SERVING: peers re-announce on every append, so a plain drop | |
| 60 | // would be undone milliseconds later. `allow-front` is the undo. | |
| 61 | svc.handle('drop-front', async (ctx, params = {}) => { | |
| 62 | if (!peers) return { ok: false, error: 'no peers router' } | |
| 63 | if (!params.did) return { ok: false, error: 'did required' } | |
| 64 | try { | |
| 65 | return { ok: true, ...(await peers.dropFront(params.did, { firehoseKey: params.firehoseKey || null, forget: !!params.forget })) } | |
| 66 | } catch (e) { return { ok: false, did: params.did, error: e.message } } | |
| 67 | }) | |
| 68 | svc.handle('allow-front', async (ctx, params = {}) => { | |
| 69 | if (!peers) return { ok: false, error: 'no peers router' } | |
| 70 | if (!params.did) return { ok: false, error: 'did required' } | |
| 71 | try { return { ok: true, ...peers.allowFront(params.did, params.firehoseKey || null) } } catch (e) { return { ok: false, did: params.did, error: e.message } } | |
| 72 | }) | |
| 73 | // Drop EVERY cached front + resolved key for a DID so the next request re-fronts from scratch. Nothing is | |
| 74 | // denied and the DID stays fronted — this is the "re-derive, don't restart" lever. | |
| 75 | svc.handle('forget-did', async (ctx, params = {}) => { | |
| 76 | if (!peers) return { ok: false, error: 'no peers router' } | |
| 77 | if (!params.did) return { ok: false, error: 'did required' } | |
| 78 | try { return { ok: true, ...(await peers.forgetDid(params.did)) } } catch (e) { return { ok: false, did: params.did, error: e.message } } | |
| 79 | }) | |
| 89 | export async function startControl ({ uds, name = 'atproto-bridge', version = null, peers = null, relay = null } = {}) { | |
| 90 | const svc = new Service({ name, uds }) // UDS only — omit `http` so we don't bind a port | |
| 80 | 91 | |
| 81 | svc.handle('resolve-handle', async (ctx, params = {}) => ({ did: peers ? peers.resolveHandle(params.handle) : null })) | |
| 92 | for (const [verb, fn] of Object.entries(membraneHandlers({ peers, relay }))) { | |
| 93 | svc.handle(verb, async (ctx, params) => fn(params)) | |
| 94 | } | |
| 82 | 95 | |
| 83 | 96 | await svc.listen() |
| 84 | 97 |
| … | … | @@ -0,0 +1,273 @@ |
| 1 | // The atproto MEMBRANE as a module — a stateless bridge (Model B). Holds NO key, NO account, NO | |
| 2 | // custodial storage. It fronts self-custody hyper-proto-peers (cfg.frontedDids) discovered over the | |
| 3 | // DHT, verifies each against its DID key, and serves them keylessly to bsky. All | |
| 4 | // identity/content/signing lives in the peers (hyper-proto-peer). | |
| 5 | // | |
| 6 | // ⛔ WHY A MODULE. Until 0.9.52 this was pds-serve.js's top level: it bound the port, installed signal | |
| 7 | // handlers and called process.exit, so the membrane could only be stopped by ending its process. The | |
| 8 | // favela gateway (hyper-social #80) runs it beside ap-bridge in ONE process, so it has to be startable | |
| 9 | // and stoppable as a unit. startMembrane() does exactly what pds-serve.js did and returns: | |
| 10 | // - `steps`: the ordered stop sequence, as [name, fn] pairs, so the caller's bounded shutdown | |
| 11 | // (lib/shutdown.js) times and reports each one; | |
| 12 | // - `handlers`: the control verbs, for whichever socket serves them (lib/control.js); | |
| 13 | // - `port`, `swarmKey`, `peers`, `relay`, `status()`. | |
| 14 | // Nothing here exits the process or installs a signal handler. pds-serve.js is the standalone wrapper. | |
| 15 | ||
| 16 | import './bare-globals.js' // FIRST — polyfills TextEncoder/TextDecoder under Bare | |
| 17 | import process from '#node:process' | |
| 18 | import fs from '#node:fs' | |
| 19 | import path from '#node:path' | |
| 20 | import Hyperswarm from 'hyperswarm' | |
| 21 | import b4a from 'b4a' | |
| 22 | import { openCorestore } from 'atproto-store/lib/corestore.js' | |
| 23 | import { loadOrCreateSwarmKeyPair } from './swarm-keypair.js' | |
| 24 | import { membraneHandlers } from './control.js' | |
| 25 | import { FleetRelay } from './fleet-relay.js' | |
| 26 | import { DidPeerRouter } from 'atproto-did-peer' | |
| 27 | import { resolvePlcVerifyKey } from './verify-key.js' | |
| 28 | import { discoveryTopic } from 'atproto-identity/lib/discovery.js' | |
| 29 | ||
| 30 | // The HTTP membrane is transport-specific: bare-http1/bare-ws under Bare, node http+ws | |
| 31 | // otherwise. (bsky-interop edge only; P2P sync is Hypercore, not this.) | |
| 32 | const isBare = typeof Bare !== 'undefined' | |
| 33 | const { createPdsServer } = await import(isBare ? './server-bare.js' : './server.js') | |
| 34 | ||
| 35 | // Bare has no global `fetch`; PLC resolution (verify a fronted peer's DID key before serving | |
| 36 | // it — verify-key.js resolvePlcVerifyKey) needs one. Install the WHATWG-standard bare-fetch | |
| 37 | // global so it works on the prod (bare) runtime. Without this, front() throws "fetch is not | |
| 38 | // defined" and NO peer is fronted — silently dropping wj62. Node already has global fetch. | |
| 39 | if (isBare) await import('bare-fetch/global') | |
| 40 | ||
| 41 | const defaultLog = (o) => console.log(JSON.stringify(o)) | |
| 42 | ||
| 43 | /** | |
| 44 | * @param {object} cfg loadConfig()'s shape: serverDid, userDomain, frontedDids, feedRegistries, | |
| 45 | * favelaWebUpstream, crawlRelay, crawlHostname, port, bind, storage | |
| 46 | * @param {object} [opts] | |
| 47 | * @param {Function} [opts.log] one object per line | |
| 48 | * @param {Array} [opts.bootstrap] DHT bootstrap; tests pass [] to stay off the public DHT | |
| 49 | */ | |
| 50 | export async function startMembrane (cfg, { log = defaultLog, bootstrap = null } = {}) { | |
| 51 | // The front-store holds the aggregate firehose (relay) + replicated peer cores. It is content, | |
| 52 | // NOT accounts — no ACCOUNT key material ever lives here. (The swarm keypair below is the one | |
| 53 | // piece of key material the membrane does keep: a transport/DHT identity that never signs a repo | |
| 54 | // commit and holds no account. Amended from "no key material ever" — see swarm-keypair.js.) | |
| 55 | const frontStore = openCorestore(path.join(cfg.storage, 'front-store')) | |
| 56 | // PERSISTENT swarm identity. A random per-boot keypair (plain `new Hyperswarm()`) made the bridge's | |
| 57 | // Noise public key unstable, so hyper-proto-peer's PEER_WRITERS allowlist — which gates apply-write, | |
| 58 | // delegate and get-service-auth on the CLIENT's Noise key — could never be populated: any entry | |
| 59 | // authorised exactly one boot. Stable key ⇒ the bridge is allowlistable. | |
| 60 | const swarmKeyPair = loadOrCreateSwarmKeyPair(cfg.storage) | |
| 61 | const swarm = new Hyperswarm(bootstrap ? { keyPair: swarmKeyPair, bootstrap } : { keyPair: swarmKeyPair }) | |
| 62 | const swarmKey = b4a.toString(swarmKeyPair.publicKey, 'hex') | |
| 63 | // Logged at BOOT, deliberately: this is the value an operator must paste into each DID's `writers` | |
| 64 | // array. Without it, populating PEER_WRITERS means fishing a key out of a peer's connection logs and | |
| 65 | // hoping it was ours. Public key only — the secret stays in the file. | |
| 66 | log({ msg: 'swarm identity', publicKey: swarmKey }) | |
| 67 | // Durable emitted-seq floor — kept in a plain file BESIDE the front-store (survives a corestore | |
| 68 | // reset), so subscribeRepos seq stays monotonic across redeploys and can be lifted above a relay's | |
| 69 | // stored cursor for recovery (see FleetRelay seqBase / the getHostStatus recovery). | |
| 70 | const seqBaseFile = path.join(cfg.storage, 'relay-seq-base') | |
| 71 | let seqBase = 0 | |
| 72 | try { seqBase = Number(fs.readFileSync(seqBaseFile, 'utf8').trim()) || 0 } catch {} | |
| 73 | const persistSeqBase = (n) => { try { fs.mkdirSync(cfg.storage, { recursive: true }) } catch {}; fs.writeFileSync(seqBaseFile, String(n)) } | |
| 74 | // The relay aggregates each fronted peer's signed firehose into one monotonic subscribeRepos | |
| 75 | // stream, so bsky's relay crawls the whole set in one shot (the egress membrane). | |
| 76 | const relay = await new FleetRelay(frontStore.get({ name: 'peer-relay-firehose', valueEncoding: 'json' }), { seqBase, onSeqBase: persistSeqBase }).ready() | |
| 77 | // PERSISTED handle->did map (survives restart — an offline blind-relayed peer, e.g. ashatron, can't | |
| 78 | // re-teach its handle at boot, so an in-memory-only map left it unresolvable after a restart). | |
| 79 | const handleMapFile = path.join(cfg.storage, 'handle-map.json') | |
| 80 | let handles = {} | |
| 81 | try { handles = JSON.parse(fs.readFileSync(handleMapFile, 'utf8')) || {} } catch {} | |
| 82 | const persistHandles = (obj) => { try { fs.mkdirSync(cfg.storage, { recursive: true }) } catch {}; fs.writeFileSync(handleMapFile, JSON.stringify(obj)) } | |
| 83 | // The canonical consumer primitive (atproto-did-peer DidPeerRouter — unified across bridge/favela in | |
| 84 | // EXTRACTION-PLAN Phase 1). Owns-swarm mode: it auto-wires onconnection + joins each fronted topic in its | |
| 85 | // ctor (via `join`) + runs the pull-only refresh/failover timer. `rpc: true` opens one protomux-rpc per | |
| 86 | // connection for the peer-signs-over-RPC write/session/blob path. `resolveVerifyKey` fronts a DID with no | |
| 87 | // pre-seeded key via PLC. `relay` folds the freshest co-owner's firehose into the aggregate subscribeRepos. | |
| 88 | const peers = new DidPeerRouter(frontStore, { | |
| 89 | swarm, | |
| 90 | join: (did) => { try { swarm.join(discoveryTopic(did), { server: true, client: true }); swarm.flush().catch(() => {}) } catch {} }, | |
| 91 | fronted: cfg.frontedDids, | |
| 92 | relay, | |
| 93 | resolveVerifyKey: resolvePlcVerifyKey, | |
| 94 | rpc: true, | |
| 95 | handles, | |
| 96 | onHandles: persistHandles | |
| 97 | }) | |
| 98 | ||
| 99 | // ⛔ THE SWARM DOOR onto fronting. `allowFront()` only UN-DENIES; a DID enters the fronted set | |
| 100 | // through `add()`, reachable from config at boot or the owner-gated control socket — so being | |
| 101 | // carried was an operator action and a peer had no way to ask. This is that way. | |
| 102 | // | |
| 103 | // ⚠ PERSISTED, exactly like the handle map above it: `frontedDids` is CONFIG, and what peers asked | |
| 104 | // for is not. A restart that silently unfronted everyone who asked would make the door a lie. | |
| 105 | const requestedFile = path.join(cfg.storage, 'fronted-requested.json') | |
| 106 | const readRequested = () => { try { return JSON.parse(fs.readFileSync(requestedFile, 'utf8')) } catch { return {} } } | |
| 107 | const requested = readRequested() | |
| 108 | // ⛔ The handle goes back in too, or a restart forgets the name it granted and | |
| 109 | // `<handle>/.well-known/atproto-did` stops answering for someone already published on-chain. | |
| 110 | for (const [id, r] of Object.entries(requested)) { | |
| 111 | try { peers.add(id, r && r.handle ? { handle: r.handle } : {}) } catch {} | |
| 112 | } | |
| 113 | if (Object.keys(requested).length) log({ msg: 'refronted on boot', count: Object.keys(requested).length }) | |
| 114 | ||
| 115 | let bridgeKey = null | |
| 116 | try { | |
| 117 | const { serveFronting, Admission } = await import('./fronting.js') | |
| 118 | const door = serveFronting({ | |
| 119 | swarm, | |
| 120 | peers, | |
| 121 | admission: new Admission(), | |
| 122 | policy: process.env.PDS_FRONT_POLICY || 'open', | |
| 123 | // What the membrane OFFERS: the domain it grants names under, and the endpoint a DID must point | |
| 124 | // at. A peer needs both before it commits, because accepting is a permanent append to a global log. | |
| 125 | userDomain: cfg.userDomain, | |
| 126 | // ⛔ DERIVED FROM serverDid, not invented. `did:web:<host>` IS this membrane's public identity — | |
| 127 | // config.js already takes crawlHostname's default from exactly that host part, so using any other | |
| 128 | // source here would be a second answer to "where is this membrane" that could drift from the first. | |
| 129 | pdsUrl: 'https://' + (/^did:web:/.test(cfg.serverDid) ? cfg.serverDid.slice('did:web:'.length) : 'pds.7rnx.net'), | |
| 130 | onFronted: ({ id, root, handle }) => { | |
| 131 | const all = readRequested() | |
| 132 | all[id] = { root, handle: handle || null, at: Date.now() } | |
| 133 | fs.mkdirSync(cfg.storage, { recursive: true }) | |
| 134 | fs.writeFileSync(requestedFile, JSON.stringify(all, null, 2)) | |
| 135 | }, | |
| 136 | log | |
| 137 | }) | |
| 138 | await door.ready() | |
| 139 | bridgeKey = door.key | |
| 140 | // ⛔ PRINTED: it is what a peer needs in order to find this bridge at all, and a key written | |
| 141 | // nowhere is the favela-android device-key mistake one layer over. | |
| 142 | log({ msg: 'fronting open over the swarm', bridgeKey: door.key }) | |
| 143 | } catch (err) { | |
| 144 | // Non-fatal: the membrane serving what it already fronts is worth more than one that refuses to | |
| 145 | // boot because a second door would not open. | |
| 146 | log({ msg: 'swarm fronting unavailable', error: (err && err.message) || String(err) }) | |
| 147 | } | |
| 148 | ||
| 149 | // 1. bsky HTTP+WS membrane — keyless per-DID routing to the fronted peers. | |
| 150 | const server = createPdsServer({ peers, serverDid: cfg.serverDid, availableUserDomains: [cfg.userDomain], favelaWebUpstream: cfg.favelaWebUpstream }) | |
| 151 | await new Promise((resolve, reject) => { | |
| 152 | server.once('error', reject) | |
| 153 | server.listen(cfg.port, cfg.bind, resolve) | |
| 154 | }) | |
| 155 | const port = server.address().port | |
| 156 | ||
| 157 | // 2. Warm the fronts in the BACKGROUND — never block the HTTP/UDS listen or the readiness probe. | |
| 158 | // A peer offline at boot (or one that hasn't signed its first commit yet) must not delay the bridge | |
| 159 | // coming up: awaiting fronts here blew past the 30s UDS readiness window → supervisor restart loop. | |
| 160 | // The router already joined every fronted topic in its constructor, so peers connect + get | |
| 161 | // identified in the background; front() folds each into the relay as it warms, and retries on demand. | |
| 162 | for (const did of cfg.frontedDids) { | |
| 163 | peers.front(did) | |
| 164 | .then(() => log({ msg: 'fronting', did })) | |
| 165 | .catch((e) => log({ msg: 'front failed (will retry on demand)', did, error: e.message })) | |
| 166 | } | |
| 167 | ||
| 168 | // 3. Seq guard — the half of the seq-floor mechanism that was missing: something that NOTICES. | |
| 169 | // | |
| 170 | // FleetRelay.setSeqBase + the durable `relay-seq-base` file (above) have always been able to lift our | |
| 171 | // emitted seq past a relay's stored cursor. Nothing ever called them. Recovery required a human to know | |
| 172 | // to read com.atproto.sync.getHostStatus and hand-call the `set-seq-base` UDS RPC — so on 2026-07-31 the | |
| 173 | // bridge sat parked at cursor 9834 against our 79 frames for ~20 hours, with every local surface healthy. | |
| 174 | // A mechanism nothing invokes is not a mechanism; this is the trigger. | |
| 175 | // | |
| 176 | // Runs once at boot and every 15 minutes. Read-only unless the arithmetic says the relay can never reach | |
| 177 | // us. OUTWARD-FACING, so it inherits the same opt-in gate as the reactor's requestCrawl below. | |
| 178 | // ⛔ The timer is CLEARED by a stop step: `unref` alone let it outlive a stopped membrane, and a | |
| 179 | // restart in the same process would then run two guards. | |
| 180 | let seqGuardTimer = null | |
| 181 | if (cfg.crawlRelay && cfg.crawlHostname) { | |
| 182 | const { runSeqGuard } = await import('./seq-guard.js') | |
| 183 | const crawl = async () => { | |
| 184 | const res = await fetch(`${cfg.crawlRelay.replace(/\/$/, '')}/xrpc/com.atproto.sync.requestCrawl`, { | |
| 185 | method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ hostname: cfg.crawlHostname }) | |
| 186 | }) | |
| 187 | if (!res.ok) throw new Error(`requestCrawl ${res.status}`) | |
| 188 | } | |
| 189 | const pass = () => runSeqGuard({ | |
| 190 | relay: peers.relay, | |
| 191 | relayBase: cfg.crawlRelay, | |
| 192 | hostname: cfg.crawlHostname, | |
| 193 | requestCrawl: crawl, | |
| 194 | log | |
| 195 | }).catch((e) => log({ msg: 'seq-guard pass failed', error: e && e.message })) | |
| 196 | pass() // boot pass, backgrounded — must never delay listen or the readiness probe | |
| 197 | seqGuardTimer = setInterval(pass, 15 * 60 * 1000) | |
| 198 | if (seqGuardTimer.unref) seqGuardTimer.unref() | |
| 199 | } else { | |
| 200 | log({ msg: 'seq-guard OFF (no crawlRelay/crawlHostname) — a relay that parks us will stay parked' }) | |
| 201 | } | |
| 202 | ||
| 203 | // 4. Feed-registry reactor (community-ux.md §111) — OFF unless cfg.feedRegistries is set. Watch each keyless | |
| 204 | // FeedRegistry (the durable P2P allow-set) and, per feed entry, front its home DID + requestCrawl once, with | |
| 205 | // no per-community config and no restart. Gated + guarded: favela-community is dynamically imported only when | |
| 206 | // the feature is on, and any failure is contained to this feature (never blocks the membrane). requestCrawl | |
| 207 | // is OUTWARD-FACING (asks bsky's relay to crawl us) and stays inert unless cfg.crawlRelay is configured. | |
| 208 | if (cfg.feedRegistries && cfg.feedRegistries.length) { | |
| 209 | ;(async () => { | |
| 210 | try { | |
| 211 | const { FeedRegistries } = await import('favela-community') | |
| 212 | const { makeMembraneReactor } = await import('./membrane-reactor.js') | |
| 213 | // Live federation call: POST com.atproto.sync.requestCrawl { hostname } to the bsky relay. Off (null) | |
| 214 | // → the reactor still fronts, but fires no live crawl. Built here (not in the pure reactor) so the | |
| 215 | // outward-facing side-effect is confined to the wiring + explicitly configured. | |
| 216 | const requestCrawl = (cfg.crawlRelay && cfg.crawlHostname) | |
| 217 | ? async () => { | |
| 218 | const res = await fetch(`${cfg.crawlRelay.replace(/\/$/, '')}/xrpc/com.atproto.sync.requestCrawl`, { | |
| 219 | method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ hostname: cfg.crawlHostname }) | |
| 220 | }) | |
| 221 | if (!res.ok) throw new Error(`requestCrawl ${res.status}`) | |
| 222 | } | |
| 223 | : null | |
| 224 | if (!requestCrawl) log({ msg: 'feed-registry reactor: requestCrawl OFF (no crawlRelay/crawlHostname) — fronting only' }) | |
| 225 | const reconcile = makeMembraneReactor({ | |
| 226 | addFront: (did) => peers.add(did), | |
| 227 | front: (did) => peers.front(did), | |
| 228 | requestCrawl, | |
| 229 | log: (m) => console.log(m) | |
| 230 | }) | |
| 231 | const feedRegistries = new FeedRegistries(frontStore, swarm) | |
| 232 | for (const regKey of cfg.feedRegistries) { | |
| 233 | await feedRegistries.watch(regKey, (feeds) => reconcile(feeds)) | |
| 234 | log({ msg: 'watching feed registry', registry: regKey.slice(0, 12) + '…' }) | |
| 235 | } | |
| 236 | } catch (e) { | |
| 237 | log({ msg: 'feed-registry reactor setup failed (membrane unaffected)', error: e.message }) | |
| 238 | } | |
| 239 | })() | |
| 240 | } | |
| 241 | ||
| 242 | log({ | |
| 243 | msg: 'membrane listening', | |
| 244 | http: `${cfg.bind}:${port}`, | |
| 245 | swarmKey, | |
| 246 | serverDid: cfg.serverDid, | |
| 247 | frontedDids: cfg.frontedDids, | |
| 248 | feedRegistries: cfg.feedRegistries | |
| 249 | }) | |
| 250 | ||
| 251 | return { | |
| 252 | port, | |
| 253 | swarmKey, | |
| 254 | bridgeKey, | |
| 255 | peers, | |
| 256 | relay, | |
| 257 | handlers: membraneHandlers({ peers }), | |
| 258 | status: () => ({ ok: true, membrane: true, port, fronted: peers.list() }), | |
| 259 | // The stop sequence, unchanged in order from pds-serve.js (control is the caller's, so it is not | |
| 260 | // here), plus the seq-guard timer. | |
| 261 | steps: [ | |
| 262 | ['seq-guard', () => { if (seqGuardTimer) clearInterval(seqGuardTimer) }], | |
| 263 | ['peers', () => peers.close()], | |
| 264 | ['relay', () => relay.close()], | |
| 265 | ['swarm', () => swarm.destroy()], | |
| 266 | ['server', () => new Promise((resolve) => server.close(resolve))], | |
| 267 | // Release the front-store Corestore fd-lock last so a redeploy can reopen. | |
| 268 | ['frontStore', () => frontStore.close()] | |
| 269 | ] | |
| 270 | } | |
| 271 | } | |
| 272 | ||
| 273 | export default startMembrane |
| … | … | @@ -0,0 +1,85 @@ |
| 1 | // The membrane as a module: started and stopped inside one process, with no exit. | |
| 2 | // | |
| 3 | // ⛔ WHY. The favela gateway (hyper-social #80) runs the membrane beside ap-bridge in ONE process. | |
| 4 | // pds-serve.js did everything at module top level — bound the port, installed SIGINT/SIGTERM handlers | |
| 5 | // and called process.exit — so the only way to stop it was to end the process. A role that cannot be | |
| 6 | // stopped cannot share a process: an OTA of the gateway, or a second role failing, would have no way | |
| 7 | // to release this one's port and store locks. | |
| 8 | // | |
| 9 | // So: startMembrane(cfg) returns the running membrane plus its stop steps, and running those steps | |
| 10 | // leaves the process alive with the port and the front-store free for the next start. | |
| 11 | ||
| 12 | import '../../lib/bare-globals.js' // FIRST: cborg touches TextEncoder at module load | |
| 13 | import test from 'brittle' | |
| 14 | import fs from 'bare-fs' | |
| 15 | import os from 'bare-os' | |
| 16 | import path from 'bare-path' | |
| 17 | import http from 'bare-http1' | |
| 18 | import { startMembrane } from '../../lib/membrane.js' | |
| 19 | ||
| 20 | const tmp = () => fs.mkdtempSync(path.join(os.tmpdir(), 'membrane-')) | |
| 21 | ||
| 22 | // bootstrap [] keeps the swarm off the public DHT: start/stop is what this tests, not discovery. | |
| 23 | function cfgFor (storage, port = 0) { | |
| 24 | return { | |
| 25 | serverDid: 'did:web:pds.test', | |
| 26 | userDomain: '.test', | |
| 27 | frontedDids: [], | |
| 28 | feedRegistries: [], | |
| 29 | favelaWebUpstream: null, | |
| 30 | crawlRelay: null, | |
| 31 | crawlHostname: null, | |
| 32 | port, | |
| 33 | bind: '127.0.0.1', | |
| 34 | storage | |
| 35 | } | |
| 36 | } | |
| 37 | ||
| 38 | function get (port, p) { | |
| 39 | return new Promise((resolve, reject) => { | |
| 40 | const req = http.request({ host: '127.0.0.1', port, path: p, method: 'GET' }, (res) => { | |
| 41 | let body = '' | |
| 42 | res.on('data', (c) => { body += c }) | |
| 43 | res.on('end', () => resolve({ status: res.statusCode, body })) | |
| 44 | }) | |
| 45 | req.on('error', reject) | |
| 46 | req.end() | |
| 47 | }) | |
| 48 | } | |
| 49 | ||
| 50 | async function stop (m) { | |
| 51 | for (const [, fn] of m.steps) await fn() | |
| 52 | } | |
| 53 | ||
| 54 | test('★ start → serve → stop → start again on the same port and storage, in one process', async (t) => { | |
| 55 | const storage = tmp() | |
| 56 | const one = await startMembrane(cfgFor(storage), { bootstrap: [], log: () => {} }) | |
| 57 | t.ok(one.port > 0, 'bound a port') | |
| 58 | t.is((await get(one.port, '/xrpc/com.atproto.server.describeServer')).status, 200, 'serving') | |
| 59 | const key = one.swarmKey | |
| 60 | ||
| 61 | await stop(one) | |
| 62 | // Same port and same storage: the port is released and the front-store's lock is let go. | |
| 63 | const two = await startMembrane(cfgFor(storage, one.port), { bootstrap: [], log: () => {} }) | |
| 64 | t.teardown(() => stop(two)) | |
| 65 | t.is(two.port, one.port, 'the same port binds again') | |
| 66 | t.is((await get(two.port, '/xrpc/com.atproto.server.describeServer')).status, 200, 'and serves') | |
| 67 | t.is(two.swarmKey, key, 'the persistent swarm identity is the same one') | |
| 68 | }) | |
| 69 | ||
| 70 | test('the stop steps are named, so a shutdown can time and report each', async (t) => { | |
| 71 | const m = await startMembrane(cfgFor(tmp()), { bootstrap: [], log: () => {} }) | |
| 72 | t.teardown(() => stop(m)) | |
| 73 | const names = m.steps.map(([n]) => n) | |
| 74 | t.alike(names.slice(-2), ['server', 'frontStore'], 'the store fd-lock goes last, after the server') | |
| 75 | t.ok(names.includes('swarm') && names.includes('peers') && names.includes('relay')) | |
| 76 | }) | |
| 77 | ||
| 78 | test('the control verbs are exposed as plain handlers, for whichever socket serves them', async (t) => { | |
| 79 | const m = await startMembrane(cfgFor(tmp()), { bootstrap: [], log: () => {} }) | |
| 80 | t.teardown(() => stop(m)) | |
| 81 | for (const verb of ['status', 'list-repos', 'list-fronted', 'diagnose', 'set-seq-base', 'emit-sync', 'probe-front', 'drop-front', 'allow-front', 'forget-did', 'resolve-handle']) { | |
| 82 | t.is(typeof m.handlers[verb], 'function', verb) | |
| 83 | } | |
| 84 | t.alike((await m.handlers['list-fronted']({})).fronted, [], 'answers') | |
| 85 | }) |