Something went wrong. Try again.
A collaborative coding-agent orchestrator for atproto radl.app
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807// Catch-up and gossip, as a protocol rather than as an object graph (design §18, plan phase 3).//// `MemoryPrivateBus` answers a catch-up by iterating the other endpoint's fields. That is exactly// what a test wants and it is not a protocol: there is no batch, no resume point, no bound on what// one request can cost the peer serving it, and no serialisation, so nothing about it had to be// decided before iroh landed. All of it does have to be decided, and none of it is an iroh question// — which is why it is here, in `core`, isomorphic and dependency-free.//// The file has two halves and one seam://// - `SyncResponder` is the **serving** half: given what a peer says it holds and what it says it// is missing, which of this replica's envelopes to hand over, bounded, in a deterministic order,// with a cursor when the batch ran out of room.// - `WirePrivateBus` is the **asking** half, and it is a `PrivateBus`, so `PrivateSpaceIngestor`,// `EnvelopeWriter`, dispatch and the UI cannot tell it from `MemoryPrivateBus`.// - `PeerLink` is the seam. A transport moves frames between two authenticated endpoints and hands// each inbound one, plus the identity it authenticated, to `accept()`. It decides nothing else;// finished, and the same protocol runs in the daemon and in a browser tab because neither of// them holds a copy of it.//// Which links exist at all is `connections.ts`'s question — whom the directory says to dial, in what// order, and what to do with one that stops answering — and it is deliberately not this file's, nor a// transport's. This half asks and serves; that half decides who is on the other end.//// The property to keep while reading: **serving a peer is not trusting one.** A batch is a bag of// signed envelopes, and every one of them goes through `admit()` at the far end (`admission.ts`)// whether it arrived here by gossip, by catch-up, from an archive, or from this replica's own// writer. Nothing in this file can put a record into a store that a stranger could not have put// there, which is why the interesting bounds here are about cost and not about trust.
import type { WantEntry } from './admission.js'import { wantKey } from './admission.js'import { base64urlDecode, base64urlEncode, concat, utf8 } from './bytes.js'import { DEFAULT_BLOB_LIMITS, verifyBlobBytes, type BlobLimits, type BlobNeed, type PrivateBlobStore } from './blobs.js'import type { EnvelopeStore, VersionSummary } from './envelope-store.js'import { envelopeKey, type PrivateEnvelope } from './envelope.js'import type { CatchUpRequest, PrivateBus } from './bus.js'import type { PeerAuthorization } from './peers.js'import { compareCodePoints } from '../order.js'import { WIRE_VERSION, decodeFrame, encodeFrame, errorFrame, spaceTopic, type PrivateFrame,} from './wire.js'
export interface CatchUpLimits { /** Envelopes in one response. A peer asks again with the cursor for the rest. */ maxEnvelopes: number /** Serialised bytes in one response, counted on the envelopes rather than the frame. */ maxBytes: number}
/** * Deliberately modest. * * A catch-up response is work one peer asks another to do, so the ceiling is "what a replica will * spend on one request from anybody" rather than "how fast can a big space sync". A cold replica of * a large space pages; a warm one carries a handful of envelopes and never sees these numbers. */export const DEFAULT_CATCH_UP_LIMITS: CatchUpLimits = { maxEnvelopes: 256, maxBytes: 1024 * 1024,}
/** One batch, and where to resume when the responder stopped at a limit. */export interface CatchUpPage { envelopes: PrivateEnvelope[] /** Present iff there is more. Absent means the responder reached the end of its corpus. */ cursor?: string}
export interface ServeRequest extends CatchUpRequest { cursor?: string}
const summaryNames = (summaries: readonly VersionSummary[]): Set<string> => { const names = new Set<string>() for (const summary of summaries) { for (const cid of summary.cids ?? []) { names.add(`${summary.did} ${summary.collection} ${summary.rkey} ${cid}`) } } return names}
const wantNames = (wants: readonly WantEntry[]): Set<string> => { const names = new Set<string>() for (const want of wants) names.add(wantKey(want)) return names}
/** The name a summary and a want both address an envelope by — everything but the signing device. */const versionName = (envelope: PrivateEnvelope): string => `${envelope.did} ${envelope.collection} ${envelope.rkey} ${envelope.recordCid}`
/** * The serving half of catch-up: this replica's corpus, minus what the asker says it already has, * plus what the asker says it lost. * * **Ordering is `envelopeKey` ascending**, which is the order `EnvelopeStore.all()` already returns, * and the cursor is simply the last key served. That works because envelope identity is stable and a * corpus only grows — there is no delete on the private path — so "everything after this key" names * the same suffix on every page. An envelope that lands *behind* the cursor mid-round is missed by * that round and picked up by the next one, which re-asks with fresh summaries; paying a round of * latency is the right price for never having to hold a snapshot per asker. */export class SyncResponder { readonly #envelopes: EnvelopeStore readonly #limits: CatchUpLimits
constructor( readonly spaceUri: string, envelopes: EnvelopeStore, limits: CatchUpLimits = DEFAULT_CATCH_UP_LIMITS, ) { this.#envelopes = envelopes // Clamped rather than trusted: a zero would make every page empty and every cursor the one the // asker already sent, which is a sync loop that runs forever and transfers nothing. this.#limits = { maxEnvelopes: Math.max(1, Math.floor(limits.maxEnvelopes)), maxBytes: Math.max(1, Math.floor(limits.maxBytes)), } }
serve(request: ServeRequest): CatchUpPage { const held = summaryNames(request.summaries ?? []) const wanted = wantNames(request.wants ?? []) const cursor = request.cursor const out: PrivateEnvelope[] = [] let bytes = 0
// Sorted here rather than trusted from the store. `EnvelopeStore` has three implementations // (memory, SQLite, the browser's IndexedDB) and they order by the same *fields*, not by the same // *string* — SQLite's `ORDER BY did, collection, rkey, …` and this key's separator-joined form // agree for every identifier the protocol can carry, and "agree for every identifier" is not a // property worth resting the cursor on. If the iteration order were not the cursor's order, a // page would skip envelopes whose key sorts below the last one served. const ordered = this.#envelopes .all() .sort((left, right) => { const a = envelopeKey(left) const b = envelopeKey(right) return a < b ? -1 : a > b ? 1 : 0 })
for (const envelope of ordered) { if (envelope.space !== this.spaceUri) continue const key = envelopeKey(envelope) if (cursor !== undefined && key <= cursor) continue const name = versionName(envelope) // A want is served even when the summary claims the record is held: a replica that evicted an // envelope under load still summarises the projection it kept, so only the explicit want can // tell the difference between "I have this" and "I know I lost this" (design §18). if (held.has(name) && !wanted.has(name)) continue
const size = utf8(JSON.stringify(envelope)).length // The count is checked before the bytes, and the bytes only once something is already in the // batch, so a single envelope larger than `maxBytes` is still handed over rather than stalling // the round forever at the same cursor. Admission is where an oversized record is refused; a // responder that silently skipped one would leave a peer asking for it forever. const full = out.length >= this.#limits.maxEnvelopes || (out.length > 0 && bytes + size > this.#limits.maxBytes) if (full) { // `out` is non-empty here — both limits are clamped to at least one — so the cursor always // advances past something, which is what makes paging terminate. return { envelopes: out, cursor: envelopeKey(out[out.length - 1] as PrivateEnvelope) } } out.push(envelope) bytes += size } return { envelopes: out } }}
/** * The serving half of a blob fetch: one chunk, from an offset the asker chose. * * Stateless on purpose. A blob is immutable — its name is its hash — so "the bytes from here" is the * same answer whenever it is asked, and a responder holding a per-asker read position would be * holding state a stranger can allocate. The asker owns the loop; this owns only the bound. */export function serveBlobChunk( blobs: Pick<PrivateBlobStore, 'get'>, cid: string, offset: number, limits: BlobLimits = DEFAULT_BLOB_LIMITS,): { bytes: Uint8Array; total: number } | undefined { const held = blobs.get(cid) if (!held) return undefined // An offset past the end is answered with an empty final chunk rather than an error: it is what a // peer that already has the whole blob would ask for, and there is nothing wrong with the request. const start = Math.min(Math.max(0, Math.floor(offset)), held.length) const size = Math.max(1, Math.floor(limits.chunkBytes)) return { bytes: held.subarray(start, Math.min(held.length, start + size)), total: held.length }}
/** * One authenticated connection to one peer, in frames. * * A transport supplies these. `id` is the remote endpoint identity it authenticated (`peers.ts` * decides whether to keep it); `request` is a round trip; `send` is fire-and-forget. Nothing here * knows about iroh, sockets or relays, which is the entire point: the same `WirePrivateBus` runs * over a QUIC stream, a WASM endpoint against a public relay, and the loopback link below. */export interface PeerLink { readonly id: string request(frame: PrivateFrame): Promise<PrivateFrame> send(frame: PrivateFrame): Promise<void> close?(): void}
/** * Where a bus gets its peers. A transport rebuilds this list as connections come and go. * * `PeerConnections` (`connections.ts`) is the implementation a replica runs on; `fixedPeers()` below * is the one a test runs on. */export interface PeerSource { links(spaceUri: string): Promise<readonly PeerLink[]> /** * A link handed out by `links()` threw while this bus was using it. * * Optional, because a fixed list has nothing to do about it. A source that manages connections * closes the link and dials a fresh one later, which is the difference between a peer that * reconnects after a laptop lid closes and one this replica keeps writing into a dead socket. * * Called **only** when the link itself threw. A peer that answers a catch-up with the wrong frame, * repeats a cursor, or refuses with an `error` frame has answered: that is counted against the peer * in `PeerStatus` and says nothing about the connection. */ failed?(endpointId: string, error: unknown): void}
/** What a replica can say about one peer without asking it anything. For `radial private status`. */export interface PeerStatus { id: string /** Envelopes handed to this peer's `send` without an error. */ delivered: number /** Sends and requests that threw, or that answered with an `error` frame. */ failed: number /** Envelopes this peer contributed on the last catch-up. */ received: number /** Blobs fetched from this peer, whole and verified. Counted apart from `received`, which is * about envelopes and would otherwise stop meaning one thing. */ blobs: number lastError?: string}
export interface WirePrivateBusOptions { spaceUri: string envelopes: EnvelopeStore peers: PeerSource /** * Authorize the transport-authenticated remote endpoint against the current public directory. * * Required here, rather than left to a transport, so no implementation can serve the private * corpus merely by forwarding a frame. The callback is evaluated for every inbound request so a * revocation or removal takes effect without rebuilding the bus. */ authorizePeer(endpointId: string): PeerAuthorization /** * This endpoint's own transport identity, echoed in a `hello`. * * Optional because the protocol never reads it: a peer authorizes the endpoint identity its * transport authenticated, not the one the other end claims in a frame (`peers.ts`). It is here so * a `hello` is legible in a packet capture and in a diagnostic, and for nothing else. */ endpointId?: string /** * The bytes this endpoint can serve, and where a fetched blob is checked against. * * Optional because a replica that holds no blobs is a normal replica, not a broken one — it * answers `unknown-blob` and asks its own peers like anybody else. */ blobs?: Pick<PrivateBlobStore, 'get'> limits?: CatchUpLimits blobLimits?: BlobLimits /** * How many pages to pull from one peer in one `catchUp`. * * A bound rather than "until the cursor is absent", because the cursor is the *peer's* claim about * whether there is more, and a peer that always says yes would otherwise own this replica's sync * loop. Stopping early costs a round. A normal peer starts the next one from fresh summaries; a * peer whose bootstrap scan is deliberately withholding inventory resumes from its last cursor * instead, even if genesis landed meanwhile, because empty summaries cannot otherwise move past * the same bounded prefix. */ maxRounds?: number /** * Maximum time one peer may hold a frame exchange open. * * Dialling has its own bound in `PeerConnections`; this is the matching bound after a connection * has been established. Without it, a peer that disappears while retaining a QUIC connection can * leave `request()` waiting for the transport's minutes-long failure detection, which in turn * stalls the replica sync cycle containing the exchange. */ frameTimeoutMs?: number}
const DEFAULT_MAX_ROUNDS = 32/** A dead established link gets no longer than two ordinary daemon poll intervals to answer. */export const DEFAULT_FRAME_TIMEOUT_MS = 10_000/** A bootstrap cursor is tiny, but endpoint ids are still a peer-controlled namespace. */const MAX_BOOTSTRAP_CURSORS = 64
class FrameTimeout extends Error {}
/** * A `PrivateBus` over frames. * * Substitutable for `MemoryPrivateBus` everywhere, on purpose: `PrivateSpaceIngestor` and * `EnvelopeWriter` were built against the interface and not against the memory implementation, so * the whole of private mode above this line runs over a real protocol without a line changing. * * A publish that fails is recorded and not thrown — design §18: "a publish that fails is a delivery * failure, not a write failure", and the peer that missed the gossip re-requests it at catch-up. A * bus that threw here would turn one unreachable peer into a failed `goal create`. */export class WirePrivateBus implements PrivateBus { readonly spaceUri: string readonly #responder: SyncResponder readonly #peers: PeerSource readonly #authorizePeer: (endpointId: string) => PeerAuthorization readonly #maxRounds: number readonly #frameTimeoutMs: number readonly #endpointId: string readonly #blobs: Pick<PrivateBlobStore, 'get'> | undefined readonly #blobLimits: BlobLimits readonly #subscribers = new Map<string, Set<(envelope: PrivateEnvelope) => void>>() readonly #status = new Map<string, PeerStatus>() /** Per-peer progress while an inventory-free bootstrap scan is unfinished (ADR §30.3). */ readonly #bootstrapCursors = new Map<string, string>() #topic: string | undefined #closed = false
constructor(options: WirePrivateBusOptions) { this.spaceUri = options.spaceUri this.#responder = new SyncResponder( options.spaceUri, options.envelopes, options.limits ?? DEFAULT_CATCH_UP_LIMITS, ) this.#peers = options.peers this.#authorizePeer = options.authorizePeer this.#maxRounds = options.maxRounds ?? DEFAULT_MAX_ROUNDS this.#frameTimeoutMs = Math.max(1, options.frameTimeoutMs ?? DEFAULT_FRAME_TIMEOUT_MS) this.#endpointId = options.endpointId ?? '' this.#blobs = options.blobs this.#blobLimits = options.blobLimits ?? DEFAULT_BLOB_LIMITS }
/** The derived topic, once `join` has computed it. Exposed for a transport's subscription. */ get topic(): string | undefined { return this.#topic }
/** * Routes whose inventory-free scan has not reached the end yet. * * Connection management may keep these endpoints dialable after genesis folds, but the bus keeps * treating them as bootstrap-only: no gossip, no blobs, and no local inventory until their cursor * clears. Sorted so the answer is diagnostic as well as actionable. */ bootstrapPeers(): readonly string[] { return [...this.#bootstrapCursors.keys()].sort(compareCodePoints) }
#assertSpace(spaceUri: string): void { if (spaceUri !== this.spaceUri) { // One bus, one space — the same shape a replica directory has (design §18). A caller passing // another space's URI has confused two replicas, and guessing which one it meant is worse. throw new Error(`this bus serves ${this.spaceUri}, not ${spaceUri}`) } }
async join(spaceUri: string): Promise<void> { this.#assertSpace(spaceUri) this.#topic ??= await spaceTopic(spaceUri) }
async #topicFor(spaceUri: string): Promise<string> { this.#topic ??= await spaceTopic(spaceUri) return this.#topic }
#peerStatus(id: string): PeerStatus { const existing = this.#status.get(id) if (existing) return existing const fresh: PeerStatus = { id, delivered: 0, failed: 0, received: 0, blobs: 0 } this.#status.set(id, fresh) return fresh }
#fail(status: PeerStatus, error: unknown): void { status.failed += 1 status.lastError = error instanceof Error ? error.message : String(error) }
/** * A failure the *link* raised, rather than an answer this replica disagreed with. * * The distinction is the whole of what the peer source is told: a throw means the connection is * gone, and holding it would mean writing into it until the process restarts. Everything else — a * refusal, a wrong frame, a repeated cursor — is a peer behaving badly over a connection that * works, and reconnecting would not improve it. */ #failLink(status: PeerStatus, error: unknown): void { this.#fail(status, error) this.#peers.failed?.(status.id, error) }
/** * Bound an exchange on a link that was reachable when the round began. * * The losing operation remains observed by `Promise.race`, and `#failLink` closes the managed * connection as soon as this timeout is caught by the caller. That makes the transport operation * reject too, instead of leaving an unhandled promise or a live stream behind the completed sync. */ async #withinFrame<T>(link: PeerLink, kind: PrivateFrame['kind'], operation: Promise<T>): Promise<T> { let timer: ReturnType<typeof setTimeout> | undefined const expiry = new Promise<never>((_, reject) => { timer = setTimeout( () => reject(new FrameTimeout(`peer ${link.id} did not answer a ${kind} frame within ${this.#frameTimeoutMs}ms`)), this.#frameTimeoutMs, ) }) try { return await Promise.race([operation, expiry]) } finally { if (timer) clearTimeout(timer) } }
async publish(spaceUri: string, envelope: PrivateEnvelope): Promise<void> { this.#assertSpace(spaceUri) const topic = await this.#topicFor(spaceUri) const frame: PrivateFrame = { kind: 'gossip', version: WIRE_VERSION, topic, envelope } await Promise.all((await this.#links(spaceUri)).map(async (link) => { // A retained bootstrap route is not authorization. Until its inventory-free scan ends it gets // only the catch-up requests that advance that scan, never newly written private content. if (this.#bootstrapCursors.has(link.id)) return const status = this.#peerStatus(link.id) try { await this.#withinFrame(link, frame.kind, link.send(frame)) status.delivered += 1 } catch (error) { this.#failLink(status, error) } })) }
subscribe(spaceUri: string, onEnvelope: (envelope: PrivateEnvelope) => void): () => void { this.#assertSpace(spaceUri) const subscribers = this.#subscribers.get(spaceUri) ?? new Set() subscribers.add(onEnvelope) this.#subscribers.set(spaceUri, subscribers) return () => { subscribers.delete(onEnvelope) } }
async catchUp(spaceUri: string, request: CatchUpRequest): Promise<PrivateEnvelope[]> { this.#assertSpace(spaceUri) const topic = await this.#topicFor(spaceUri) // Before signed genesis lands, the ingestor sends neither summaries nor wants: disclosing either // would describe a partial private corpus to the raw bootstrap directory (ADR §30.3). A cursor // is progress through the PEER's corpus and says nothing about ours, so retain it across bounded // calls. If genesis lands before that scan ends, only that peer continues inventory-free; other, // proven members receive the caller's fresh summaries as normal. // Keyed rather than appended: several peers legitimately hold the same envelope, and returning // one copy per peer would make the cost of joining a space grow with the number of members. const collected = new Map<string, PrivateEnvelope>() for (const link of await this.#links(spaceUri)) { const status = this.#peerStatus(link.id) status.received = 0 const retained = this.#bootstrapCursors.get(link.id) const inventoryWithheld = retained !== undefined || (request.summaries.length === 0 && request.wants.length === 0) const summaries = inventoryWithheld ? [] : request.summaries const wants = inventoryWithheld ? [] : request.wants let cursor = retained for (let round = 0; round < this.#maxRounds; round += 1) { let answer: PrivateFrame try { const frame: PrivateFrame = { kind: 'catch-up', version: WIRE_VERSION, topic, summaries, wants, ...(cursor !== undefined ? { cursor } : {}), } answer = await this.#withinFrame(link, frame.kind, link.request(frame)) } catch (error) { this.#failLink(status, error) break } if (answer.kind === 'error') { this.#fail(status, `${answer.status}: ${answer.reason}`) break } if (answer.kind !== 'envelopes' || answer.topic !== topic) { this.#fail(status, 'peer answered a catch-up with something else') this.#bootstrapCursors.delete(link.id) break } for (const envelope of answer.envelopes) { collected.set(envelopeKey(envelope), envelope) status.received += 1 } if (answer.cursor === undefined) { this.#bootstrapCursors.delete(link.id) break } // A peer that repeats a cursor is not making progress; stopping is cheaper than trusting it. if (answer.cursor === cursor) { this.#fail(status, 'peer repeated a catch-up cursor') this.#bootstrapCursors.delete(link.id) break } cursor = answer.cursor if (inventoryWithheld) this.#rememberBootstrapCursor(link.id, cursor) } } // Sorted so two replicas offered the same set process it in the same order — the admission // pipeline is order-insensitive by construction, and a test that depends on engine iteration // order would be the thing hiding a case where it is not. return [...collected.entries()] .sort(([left], [right]) => (left < right ? -1 : left > right ? 1 : 0)) .map(([, envelope]) => envelope) }
/** Keep bootstrap progress bounded even when authorized endpoints churn before genesis arrives. */ #rememberBootstrapCursor(endpointId: string, cursor: string): void { this.#bootstrapCursors.delete(endpointId) this.#bootstrapCursors.set(endpointId, cursor) while (this.#bootstrapCursors.size > MAX_BOOTSTRAP_CURSORS) { this.#bootstrapCursors.delete(this.#bootstrapCursors.keys().next().value as string) } }
/** * The bytes a record named, from the first peer that has them. * * Every bound in this loop comes from something already signed or already known, never from the * peer: the total length is checked against the `size` the *record* declared, the offset must be * the one that was asked for, and the assembled bytes are hashed against `need.cid` before they * are returned. A peer can therefore waste this replica's time and nothing else — the worst it can * do is fail, and the next peer is asked. */ async fetchBlob(spaceUri: string, need: BlobNeed): Promise<Uint8Array | undefined> { this.#assertSpace(spaceUri) const topic = await this.#topicFor(spaceUri) for (const link of await this.#links(spaceUri)) { // A blob CID identifies private content. A route retained only to finish an inventory-free // scan must not learn one; once its cursor clears, the member roster decides whether it stays. if (this.#bootstrapCursors.has(link.id)) continue const status = this.#peerStatus(link.id) const bytes = await this.#fetchFrom(link, status, topic, need) if (bytes) return bytes } return undefined }
async #fetchFrom( link: PeerLink, status: PeerStatus, topic: string, need: BlobNeed, ): Promise<Uint8Array | undefined> { const parts: Uint8Array[] = [] let offset = 0 // The record says how long the blob is, so the number of chunks it can take is known before the // first one arrives — a peer answering with one byte at a time buys itself no more rounds than a // peer answering honestly, plus one for the terminating empty chunk. const rounds = Math.ceil(Math.max(1, need.size) / Math.max(1, this.#blobLimits.chunkBytes)) + 1 for (let round = 0; round < rounds; round += 1) { let answer: PrivateFrame try { const frame: PrivateFrame = { kind: 'blob-request', version: WIRE_VERSION, topic, cid: need.cid, ...(offset > 0 ? { offset } : {}), } answer = await this.#withinFrame(link, frame.kind, link.request(frame)) } catch (error) { this.#failLink(status, error) return undefined } if (answer.kind === 'error') { // `unknown-blob` is not a peer failure: it is the ordinary answer from a replica that has the // record and not yet the bytes, which is most of them for most of a blob's first minute. if (answer.status !== 'unknown-blob') this.#fail(status, `${answer.status}: ${answer.reason}`) return undefined } if (answer.kind !== 'blob' || answer.topic !== topic || answer.cid !== need.cid) { this.#fail(status, 'peer answered a blob request with something else') return undefined } if (answer.offset !== offset) { this.#fail(status, 'peer answered a blob request at the wrong offset') return undefined } if (answer.total !== need.size) { this.#fail(status, 'peer offered a blob of a length the record does not name') return undefined } let chunk: Uint8Array try { chunk = base64urlDecode(answer.bytes) } catch (error) { this.#fail(status, error) return undefined } if (chunk.length === 0) { this.#fail(status, 'peer stopped a blob short') return undefined } parts.push(chunk) offset += chunk.length if (offset > need.size) { this.#fail(status, 'peer sent more of a blob than the record names') return undefined } if (offset === need.size) { const assembled = concat(...parts) try { await verifyBlobBytes(assembled, need.cid, this.#blobLimits) } catch (error) { this.#fail(status, error) return undefined } status.blobs += 1 return assembled } } this.#fail(status, 'peer did not finish a blob within its own length') return undefined }
/** * One inbound frame and the transport-authenticated peer identity, plus the answer if any. * * This is the whole server side. Authorization is mandatory here rather than optional transport * glue: no transport following the seam can expose the corpus on topic knowledge alone. */ async accept(peerId: string, frame: PrivateFrame): Promise<PrivateFrame | undefined> { if (this.#closed) return errorFrame('unavailable', 'this endpoint is closed') const topic = await this.#topicFor(this.spaceUri) // `envelopes` and `blob` are ANSWERS; both are refused below rather than served, so there is // nothing to authorize. Everything a peer can ask for goes through the check. if (frame.kind !== 'error' && frame.kind !== 'envelopes' && frame.kind !== 'blob') { const authorization = this.#authorizePeer(peerId) if (!authorization.ok) return errorFrame('unavailable', authorization.reason) } switch (frame.kind) { case 'hello': if (frame.topic !== topic) { return errorFrame('unknown-topic', 'this endpoint does not serve that topic') } return { kind: 'hello', version: WIRE_VERSION, topic, ...(this.#endpointId ? { endpointId: this.#endpointId } : {}), } case 'catch-up': { if (frame.topic !== topic) { return errorFrame('unknown-topic', 'this endpoint does not serve that topic') } const page = this.#responder.serve({ summaries: frame.summaries, wants: frame.wants, ...(frame.cursor !== undefined ? { cursor: frame.cursor } : {}), }) return { kind: 'envelopes', version: WIRE_VERSION, topic, envelopes: page.envelopes, ...(page.cursor !== undefined ? { cursor: page.cursor } : {}), } } case 'gossip': { if (frame.topic !== topic) { return errorFrame('unknown-topic', 'this endpoint does not serve that topic') } for (const subscriber of this.#subscribers.get(this.spaceUri) ?? []) { subscriber(frame.envelope) } return undefined } case 'blob-request': { if (frame.topic !== topic) { return errorFrame('unknown-topic', 'this endpoint does not serve that topic') } const chunk = this.#blobs ? serveBlobChunk(this.#blobs, frame.cid, frame.offset ?? 0, this.#blobLimits) : undefined if (!chunk) return errorFrame('unknown-blob', 'this endpoint does not hold those bytes') return { kind: 'blob', version: WIRE_VERSION, topic, cid: frame.cid, offset: frame.offset ?? 0, total: chunk.total, bytes: base64urlEncode(chunk.bytes), } } case 'envelopes': // A response with no request. Refused rather than treated as a batched gossip, so there is // exactly one unsolicited path into a store and exactly one set of bounds on it. return errorFrame('malformed', 'an envelopes frame is an answer, not an offer') case 'blob': // Same rule for bytes: blobs are pulled against a name a signed record gave, never pushed. return errorFrame('malformed', 'a blob frame is an answer, not an offer') case 'error': return undefined } }
/** Per-peer delivery and catch-up counters, for diagnostics — never for trust. */ status(): PeerStatus[] { return [...this.#status.values()] .map((status) => ({ ...status })) .sort((left, right) => (left.id < right.id ? -1 : left.id > right.id ? 1 : 0)) }
async #links(spaceUri: string): Promise<readonly PeerLink[]> { if (this.#closed) return [] return this.#peers.links(spaceUri) }
async close(): Promise<void> { this.#closed = true this.#subscribers.clear() this.#bootstrapCursors.clear() }}
/** The far end of a loopback link: anything that answers frames the way a bus does. */export interface FrameEndpoint { accept(peerId: string, frame: PrivateFrame): Promise<PrivateFrame | undefined>}
export interface LoopbackOptions { /** The caller identity the loopback transport authenticated and presents to the remote. */ localId: string /** Deliver each gossip frame this many times. Duplication has to be harmless, so tests cause it. */ duplicates?: number /** When true at call time, the link behaves as unreachable. How a test partitions two replicas. */ offline?: () => boolean}
/** * A `PeerLink` straight into another endpoint's `accept`, through the real codec. * * It exists in `src` rather than in a test file for the same reason `MemoryPrivateBus` does: it is * the reference wiring a transport copies, and it is what lets the protocol be exercised end to end * with CI off the network. Every frame is encoded and decoded on both legs, so a change that made a * frame unserialisable fails here rather than on a relay. */export function loopbackLink( id: string, remote: FrameEndpoint, options: LoopbackOptions,): PeerLink { const roundTrip = (frame: PrivateFrame): PrivateFrame => decodeFrame(encodeFrame(frame)) const offline = (): boolean => options.offline?.() ?? false return { id, async request(frame) { if (offline()) throw new Error(`peer ${id} is unreachable`) const answer = await remote.accept(options.localId, roundTrip(frame)) if (!answer) throw new Error(`peer ${id} did not answer a ${frame.kind} frame`) return roundTrip(answer) }, async send(frame) { if (offline()) throw new Error(`peer ${id} is unreachable`) const copies = Math.max(1, options.duplicates ?? 1) for (let copy = 0; copy < copies; copy += 1) { await remote.accept(options.localId, roundTrip(frame)) } }, }}
/** * A fixed peer set: the links are handed in once and never re-read. * * The test stand-in, and only that. A replica runs on `PeerConnections` (`connections.ts`), which * re-reads the directory every round so new devices and changed addresses are observed * next time — a distinction a fixed list cannot express. */export const fixedPeers = (links: readonly PeerLink[]): PeerSource => ({ async links() { return links },})