| … | … | @@ -201,7 +201,12 @@ function revTime (rev) { |
| 201 | 201 | // simply not fronted by an unpatched consumer. Flipping this to false is therefore a COORDINATED |
| 202 | 202 | // change with the consumer; shipping the flag first lets the derive-on-read surface land with zero |
| 203 | 203 | // behavioural change and be verified in place before anything is removed. |
| 204 | | export async function serveMembrane (device, { corestore, swarm, did, handle, sessions = null, blobs = null, projectRepoCore = true } = {}) { |
| 204 | // `labels` — a LabelRoom (or anything with projectTo(core)) for a LABELER DID. Its signed __label__ log |
| 205 | // lives in the StateBase, which only co-owners replicate; a keyless consumer (the gateway answering bsky's |
| 206 | // queryLabels/subscribeLabels) reaches the DID only through this announce. So the log is projected into a |
| 207 | // plain `membrane-labels:<did>` hypercore and named as labelKey/labelLength, the firehose pattern again. |
| 208 | // Absent = no core opened and labelKey:null in the announce (the field exists, so a consumer can branch). |
| 209 | export async function serveMembrane (device, { corestore, swarm, did, handle, sessions = null, blobs = null, projectRepoCore = true, labels = null } = {}) { |
| 205 | 210 | if (!corestore || !swarm || !did) throw new Error('serveMembrane: corestore, swarm, did required') |
| 206 | 211 | // Step markers. This function is `await`ed during a DID's boot, so anything slow here stalls the whole |
| 207 | 212 | // node — and on a phone it stalled with NO output at all, leaving "somewhere inside serveMembrane" as |
| … | … | @@ -213,6 +218,11 @@ export async function serveMembrane (device, { corestore, swarm, did, handle, se |
| 213 | 218 | step(projectRepoCore ? 'repo core ready' : 'repo core SKIPPED (virtual device)') |
| 214 | 219 | const seq = await new FirehoseSequencer(corestore.get({ name: 'membrane-firehose:' + did })).ready() |
| 215 | 220 | step('firehose sequencer ready') |
| 221 | const labelCore = labels ? corestore.get({ name: 'membrane-labels:' + did }) : null |
| 222 | if (labelCore) { await labelCore.ready(); step('label core ready') } |
| 223 | // Incremental and idempotent (LabelRoom.projectTo appends only labels the core lacks). Called by the |
| 224 | // host right after the label signer's tick; returns the count appended. |
| 225 | const syncLabels = async () => (labels && labelCore ? labels.projectTo(labelCore) : 0) |
| 216 | 226 | // Project ONLY when the device's __commit__ log has advanced past what the membrane firehose already |
| 217 | 227 | // holds. WITHOUT this gate, the caller's sign loop (every signIntervalMs, ~2s) called sync() |
| 218 | 228 | // UNCONDITIONALLY, and projectRepo re-put ALL records into the APPEND-ONLY Hyperbee repo core on EVERY |
| … | … | @@ -286,6 +296,11 @@ export async function serveMembrane (device, { corestore, swarm, did, handle, se |
| 286 | 296 | const serverKey = b4a.toString(swarm.keyPair.publicKey, 'hex') |
| 287 | 297 | const repoKey = repo ? b4a.toString(repo.core.key, 'hex') : null |
| 288 | 298 | const firehoseKey = b4a.toString(seq.core.key, 'hex') |
| 299 | const labelKey = labelCore ? b4a.toString(labelCore.key, 'hex') : null |
| 300 | // The announce payload, one place. repoKey/repoLength are null when the repo core is off — a consumer |
| 301 | // that requires them simply does not front this peer (it keeps using a co-owner that still projects), |
| 302 | // which is the graceful degradation that makes the flag safe to roll out one node at a time. |
| 303 | const announcement = () => ({ did, handle: handle || null, repoKey, firehoseKey, repoLength: repo ? repo.core.length : null, firehoseLength: seq.core.length, labelKey, labelLength: labelCore ? labelCore.length : null }) |
| 289 | 304 | const onconn = (conn) => { |
| 290 | 305 | corestore.replicate(conn) |
| 291 | 306 | const rpc = new ProtomuxRPC(conn, { protocol: RPC_PROTOCOL, valueEncoding: c.json }) |
| … | … | @@ -299,14 +314,13 @@ export async function serveMembrane (device, { corestore, swarm, did, handle, se |
| 299 | 314 | if (achan) { |
| 300 | 315 | const amsg = achan.addMessage({ encoding: c.json, onmessage: () => {} }) |
| 301 | 316 | achan.open() |
| 302 | | // repoKey/repoLength are null when the repo core is off — a consumer that requires them simply |
| 303 | | // does not front this peer (it keeps using a co-owner that still projects), which is the |
| 304 | | // graceful degradation that makes the flag safe to roll out one node at a time. |
| 305 | | const pushAnnounce = () => { try { amsg.send({ did, handle: handle || null, repoKey, firehoseKey, repoLength: repo ? repo.core.length : null, firehoseLength: seq.core.length }) } catch {} } |
| 317 | const pushAnnounce = () => { try { amsg.send(announcement()) } catch {} } |
| 306 | 318 | pushAnnounce() |
| 307 | 319 | const onapp = () => pushAnnounce() |
| 308 | 320 | seq.core.on('append', onapp) |
| 309 | | conn.on('close', () => { try { seq.core.off('append', onapp) } catch {} }) |
| 321 | // A new label is news to the consumer just like a new commit: re-announce so it extends its replica. |
| 322 | if (labelCore) labelCore.on('append', onapp) |
| 323 | conn.on('close', () => { try { seq.core.off('append', onapp); if (labelCore) labelCore.off('append', onapp) } catch {} }) |
| 310 | 324 | } |
| 311 | 325 | } catch {} |
| 312 | 326 | // get-repo-car (com.atproto.sync.getRepo, DERIVED) — the stateless replacement for the |
| … | … | @@ -409,7 +423,7 @@ export async function serveMembrane (device, { corestore, swarm, did, handle, se |
| 409 | 423 | const dDid = swarm.join(discoveryTopic(did), { server: true, client: true }) |
| 410 | 424 | step('topics joined — serveMembrane returning (the announce flush is the CALLER\'s to await)') |
| 411 | 425 | return { |
| 412 | | repo, seq, serverKey, repoKey, firehoseKey, sync, |
| 426 | repo, seq, serverKey, repoKey, firehoseKey, sync, labelCore, labelKey, syncLabels, announcement, |
| 413 | 427 | ready: () => Promise.all([dKey.flushed(), dDid.flushed()]), |
| 414 | 428 | close: () => swarm.off('connection', onconn) |
| 415 | 429 | } |