| … | … | @@ -75,9 +75,19 @@ function revTime (rev) { |
| 75 | 75 | // serveMembrane — project the stores + PUSH the announce + replicate the corestore over `swarm`. |
| 76 | 76 | // Mirrors peer.serve: joins the DID discovery topic + the serverKey topic; on connect, replicates |
| 77 | 77 | // the corestore and pushes an announce. Call `sync()` after new signs to project the latest state. |
| 78 | | export async function serveMembrane (device, { corestore, swarm, did, handle, sessions = null, blobs = null } = {}) { |
| 78 | // `projectRepoCore` (default true) — when FALSE the membrane-repo Hyperbee is never opened and |
| 79 | // projectRepo never runs, so record bytes exist ONLY as the content-addressed files. Consumers then |
| 80 | // read the repo via the `get-repo-car` RPC above (derived on demand) or, when co-located, straight off |
| 81 | // the shared overlay files. This is SPEC §6.3's "the virtual device holds nothing". |
| 82 | // |
| 83 | // It is an OPTION rather than a deletion because the announce contract still carries `repoKey`, and |
| 84 | // atproto-did-peer's onAnnounce/holds/_announceTo all gate on it — a peer that stops projecting is |
| 85 | // simply not fronted by an unpatched consumer. Flipping this to false is therefore a COORDINATED |
| 86 | // change with the consumer; shipping the flag first lets the derive-on-read surface land with zero |
| 87 | // behavioural change and be verified in place before anything is removed. |
| 88 | export async function serveMembrane (device, { corestore, swarm, did, handle, sessions = null, blobs = null, projectRepoCore = true } = {}) { |
| 79 | 89 | if (!corestore || !swarm || !did) throw new Error('serveMembrane: corestore, swarm, did required') |
| 80 | | const repo = await new RepoStore(corestore, { name: 'membrane-repo:' + did }).ready() |
| 90 | const repo = projectRepoCore ? await new RepoStore(corestore, { name: 'membrane-repo:' + did }).ready() : null |
| 81 | 91 | const seq = await new FirehoseSequencer(corestore.get({ name: 'membrane-firehose:' + did })).ready() |
| 82 | 92 | // Project ONLY when the device's __commit__ log has advanced past what the membrane firehose already |
| 83 | 93 | // holds. WITHOUT this gate, the caller's sign loop (every signIntervalMs, ~2s) called sync() |
| … | … | @@ -89,13 +99,13 @@ export async function serveMembrane (device, { corestore, swarm, did, handle, se |
| 89 | 99 | const sync = async () => { |
| 90 | 100 | const log = await device.firehoseLog() |
| 91 | 101 | if (log.length <= seq.length) return |
| 92 | | await projectRepo(device, repo) |
| 102 | if (repo) await projectRepo(device, repo) // skipped entirely when the repo core is off |
| 93 | 103 | await projectFirehose(device, seq, log) |
| 94 | 104 | } |
| 95 | 105 | await sync() |
| 96 | 106 | |
| 97 | 107 | const serverKey = b4a.toString(swarm.keyPair.publicKey, 'hex') |
| 98 | | const repoKey = b4a.toString(repo.core.key, 'hex') |
| 108 | const repoKey = repo ? b4a.toString(repo.core.key, 'hex') : null |
| 99 | 109 | const firehoseKey = b4a.toString(seq.core.key, 'hex') |
| 100 | 110 | const onconn = (conn) => { |
| 101 | 111 | corestore.replicate(conn) |
| … | … | @@ -110,13 +120,30 @@ export async function serveMembrane (device, { corestore, swarm, did, handle, se |
| 110 | 120 | if (achan) { |
| 111 | 121 | const amsg = achan.addMessage({ encoding: c.json, onmessage: () => {} }) |
| 112 | 122 | achan.open() |
| 113 | | const pushAnnounce = () => { try { amsg.send({ did, handle: handle || null, repoKey, firehoseKey, repoLength: repo.core.length, firehoseLength: seq.core.length }) } catch {} } |
| 123 | // repoKey/repoLength are null when the repo core is off — a consumer that requires them simply |
| 124 | // does not front this peer (it keeps using a co-owner that still projects), which is the |
| 125 | // graceful degradation that makes the flag safe to roll out one node at a time. |
| 126 | const pushAnnounce = () => { try { amsg.send({ did, handle: handle || null, repoKey, firehoseKey, repoLength: repo ? repo.core.length : null, firehoseLength: seq.core.length }) } catch {} } |
| 114 | 127 | pushAnnounce() |
| 115 | 128 | const onapp = () => pushAnnounce() |
| 116 | 129 | seq.core.on('append', onapp) |
| 117 | 130 | conn.on('close', () => { try { seq.core.off('append', onapp) } catch {} }) |
| 118 | 131 | } |
| 119 | 132 | } catch {} |
| 133 | // get-repo-car (com.atproto.sync.getRepo, DERIVED) — the stateless replacement for the |
| 134 | // membrane-repo projection core. device.exportCar() rebuilds the MST + CAR from the record FILES on |
| 135 | // demand and asserts the rebuilt data root equals the signed head's, so it serves nothing that does |
| 136 | // not match its own signature. Nothing is stored to answer this. |
| 137 | // |
| 138 | // WHY: projectRepo copies EVERY record into an append-only Hyperbee (membrane-repo:<did>), which a |
| 139 | // consumer then replicates again — so record bytes live three times (files, this core, the |
| 140 | // consumer's replica) for data the files already hold. SPEC-atproto-overlay-storage §6.3: "the |
| 141 | // virtual device stores nothing of its own — it derives." This is that surface. PUBLIC, like |
| 142 | // get-blob: a repo CAR is public atproto data and is self-verifying against the DID key. |
| 143 | rpc.respond('get-repo-car', async () => { |
| 144 | const r = await device.exportCar() |
| 145 | return { car: b4a.toString(r.carBytes, 'hex'), commitCid: r.commitCid, dataCid: r.dataCid, rev: r.rev, recordCount: r.recordCount } |
| 146 | }) |
| 120 | 147 | // get-blob (com.atproto.sync.getBlob) — PUBLIC: atproto blobs are public bytes addressed by CID. |
| 121 | 148 | // Blobs live in the PEER's BlobStore; the keyless bridge relays this RPC (peer-front.getBlob) and |
| 122 | 149 | // streams the bytes back. null → the bridge answers 404. bytes are hex in the c.json RPC value. |