diff --git a/CLAUDE.md b/CLAUDE.md index f7d58dd..8817015 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -111,7 +111,13 @@ turns it into the answer, so a transport sends a refusal rather than deciding th (`private/peers.ts`) is about *connections, never the index*: it stops a private corpus being enumerable by anyone who computed the topic, and it is neither a gate nor a membership check — everything an authorized peer -sends still passes both gates. **Which connections exist** is `private/connections.ts`, decided in +sends still passes both gates. A refusal is also where a replica **learns whom to poll** (ADR §30): +`catch-up` and `hello` carry an optional `from`, the dialer's claimed DIDs, which the bus buffers +only while refusing and which the answer never reflects — the refusal a stranger gets must not +change. `PrivateSpaceIngestor` drains that buffer, reads each claimed DID's *public* repo once, and +keeps it only if that repo binds the endpoint the transport authenticated. A claim is worth a bounded +number of public reads and never a directory entry; do not make it worth more, and do not settle one +anywhere but `authorizeEndpoint`. **Which connections exist** is `private/connections.ts`, decided in `core` for the same reason the wire is: `PeerConnections` re-reads `connectablePeers()` every round (never cached, so newly published devices and address changes are visible), orders it by `join.peerHints` without ever filtering on them, backs off deterministically, and drops @@ -125,7 +131,12 @@ endpoint per runtime, routes each inbound frame to the space whose wire topic it topic gets `unknown-topic`, the same answer a stranger's does — a topic is derived from a public URI and must buy no information), authorizes an inbound connection against that space's directory before adopting it, and leases one physical link across every space that wants it so none can disconnect -another. `attach`/`detach` are how a space comes and goes under a running endpoint. It names no +another. A connection **no attached space names is held, not closed** (ADR §30): both bindings call +`onLink` on the accept, so closing there destroys the dialer's first frame — and that frame is how a +replica whose directory has never heard of this peer learns whose repo to read. It waits out a +bounded grace (8 connections, 30 s), is refused frame by frame throughout exactly as before, and is +adopted only into a space that has no live link to that endpoint, because a held link is older than +anything adopted since. `attach`/`detach` are how a space comes and goes under a running endpoint. It names no binding: `bind` takes a transport factory, and a runtime supplies **only** its binding and somewhere to keep the endpoint secret. Those two things are all that differ: diff --git a/docs/adr-private-mode-iroh.md b/docs/adr-private-mode-iroh.md index a6305e6..994bcb1 100644 --- a/docs/adr-private-mode-iroh.md +++ b/docs/adr-private-mode-iroh.md @@ -1305,3 +1305,74 @@ the group has stopped trusting a person, and this answers the case where a perso trusting a machine. Nor is there repair — an attested cutoff, where a surviving device names the versions that keep counting so a removal need not be all-or-nothing, remains the shape any future revocation should take, and remains unbuilt. + +## 30. Decision: a replica learns whom to poll from whoever dials it, and an unnamed connection is held rather than closed + +*Recorded after "a fresh browser can only open a private space while another browser has it open" +turned out to be a discovery deadlock with a connection lifetime at the bottom of it.* + +The report was exact, and so was the workaround: with a daemon serving a space and no other browser +running, a fresh tab reached `waiting` (§24) and stayed there. Open the space in a second browser and +it filled in seconds. Neither store, fold nor gate was involved. The chain was: + +1. **A fresh replica polls the founder and itself.** Which is all a bookmark, a ticket or a space URI + can name — and a daemon's identities are its *agent profile* DIDs, so they are in none of them. + `connectablePeers()` therefore never listed the daemon, and the tab never dialled it. +2. **The daemon does find the browser.** Its `knownDids` grows from the fold's members, the human is + a member, so its next directory poll saw the browser's freshly published `deviceAddress` and + dialled it. Discovery already worked in exactly one direction. +3. **And that dial went nowhere.** The tab authorized the inbound endpoint against its own directory, + which had never heard of it, and `PrivateEndpoint.#adopt` closed the connection. No frame carried + the dialer's identity, so the refusal taught the tab nothing about whose repo to read next. Both + sides retried forever. A second browser fixed it by publishing an address into a repo the fresh + one already polled, which is precisely the shape of the reported workaround. + +The fix is one sentence in two halves: **a dialer says who it claims to be, and a replica keeps the +connection long enough to hear it.** Everything here is a *connection* statement in §2's vocabulary — +`admission.ts` and `devices.ts` are untouched, `materialize()` counts exactly what it counted, the +digest is unchanged — so, as with §29, there is no lexicon change, no permutation coverage and **no +protocol version bump**: `from` is an optional field an older `readFrame` ignores, and a peer that +ignores it makes a worse connection decision over an identical corpus. + +- **The claim is on the frame, and it is worth nothing on arrival.** `catch-up` and `hello` gain an + optional `from` — the dialer's DIDs, bounded at 8 — which `WirePrivateBus.accept` buffers **only + when it is refusing** the frame, before returning the identical refusal a stranger has always got. + The answer must not change: a claim that bought a different reply would be a way to ask an endpoint + which DIDs it already trusts. The buffer is drained by the layer that can settle it + (`PrivateSpaceIngestor`), which polls each claimed DID's **public** repo once and keeps it only if + that repo binds the endpoint the transport authenticated — `authorizeEndpoint`, the same function + every connection goes through. So a claim buys at most a bounded number of public reads, and the + most it can win is a directory entry its owner could have earned by publishing an address anyway. +- **An unnamed inbound connection is held for a bounded grace, and served nothing.** This is the half + the first implementation missed, and it is the half that makes the rest run: both iroh bindings + call `onLink` the moment they accept, so closing there destroys the round trip the claim travels in + — the frame is gone before it was read. `PrivateEndpoint` now keeps such a connection (8 at once, + 30 s each), refuses every frame over it exactly as before, and adopts it if the directory catches + up in the meantime. Nothing about what is *served* changed: the bus authorizes each frame against + the same directory, and a held link is not a lease. +- **A held connection is adopted only into a space with nothing to displace.** `PeerConnections.adopt` + replaces, deliberately — a peer that redialled has said which connection it believes it has — and + that rule is wrong for a link that has been waiting: it is older than anything adopted since, and + promoting it would close a live connection in favour of a socket the peer may already have hung up + on. Fresh connections still replace; held ones only fill a gap. +- **Hints reach the poller, and tickets carry them.** `openPrivateReplica` passed `meta.peerHints` as + `prefer`, which orders a roster and cannot extend one, so the hints ADR §27 put in the bookmark did + nothing in a browser; they are now bootstrap DIDs, as they always were in the CLI and the daemon. + A ticket may now carry `peerHints` too, minted from the fold's active addressed members, so an + invitee's first replica can dial a daemon directly instead of waiting to be dialled. The cost is + stated rather than hidden: those DIDs land in the invitee's **public** `join` bookmark, which names + the operator's agents beside a founder who was already named there. Anyone who considers that too + much can leave hints out and still converge by the paragraph above, one poll slower. + +**What this leaves open, deliberately.** A claim is a hint and never a handshake — there is no +signature on the wire proving the dialer holds that DID, because the repo read that follows settles +it better and costs the claimer nothing to be honest about. First-open latency is bounded by the +daemon's directory-poll interval, since it must see the browser's device before it dials; that is a +tuning question and not a correctness one. And the polled PDS operator learns that *some* replica was +interested in that DID, which is the one residual disclosure this buys. + +**One note for whoever changes this next.** The bug survived a review because the test doubles were +one-directional: a loopback where the accepter could close its half while the dialer went on being +served reports every accept-side lifecycle decision as working. Both doubles now share one open flag +across a connection's two ends, and the regressions start from the honest state — a replica that has +never heard of the peer about to dial it. diff --git a/docs/design.md b/docs/design.md index d63c0cc..5c05c92 100644 --- a/docs/design.md +++ b/docs/design.md @@ -458,8 +458,9 @@ A DID's public repo reveals that it uses Radial, how many devices it has, opaque - **Seven frames** — `hello`, `catch-up`, `envelopes`, `gossip`, `blob-request`, `blob`, `error` — UTF-8 JSON, optionally length-prefixed for a byte stream (`FrameReader` bounds a declared length *before* the body is buffered, which is the one place a stranger chooses how much memory this process holds). JSON, like the archive, because a record's authenticity does not rest on how it travelled: the receiver recomputes canonical DAG-CBOR and the `recordCid` and checks the signature over those bytes. Decoding is deliberately **shallow** — version, kind, topic, counts, sizes — and every record-level check stays in `admit()`, so there is one place an envelope is judged rather than two, one of which would drift into being the lenient one. - **Catch-up is bounded and resumable.** A responder serves its corpus in `envelopeKey` order, minus what the asker's summaries cover, plus what the asker's wants name explicitly, capped by count and bytes, with a `cursor` when it stopped at a cap. Envelope identity is stable and a corpus only grows, so "everything after this key" names the same suffix on every page and no snapshot is held per asker; an envelope landing behind the cursor mid-round is picked up by the next round's fresh summaries. An envelope larger than the byte cap is still handed over — a responder that skipped it would leave a peer asking for it forever. The asker bounds its side too: a fixed number of pages per peer, because the cursor is the *peer's* claim that there is more. - **A publish that fails is recorded, not thrown.** One unreachable peer must not fail a `goal create`; the peer that missed the gossip re-requests it at catch-up. Per-peer delivery, receipt and error counters are diagnostics and never trust. +- **A dialer may say who it claims to be**, in an optional `from` on `hello` and `catch-up` (bounded, DIDs only). It is additive and carries no weight on arrival: a receiver buffers it only while *refusing* the frame — the refusal itself is unchanged, because an answer that varied would say which DIDs an endpoint already trusts — and settles it by reading that DID's public repo and asking `authorizeEndpoint` whether it binds the endpoint the transport authenticated. This is what lets a replica whose directory names none of a space's live devices learn whose repo to poll, instead of waiting to be told by a device it already knows (ADR §30). -**Connection authorization** (`private/peers.ts`) is about connections, not the index. Before serving catch-up, a receiver requires the authenticated endpoint to be named by a published device's `deviceAddress` that its owner has not retired. Otherwise anyone who computed the topic could enumerate the corpus. This is neither membership nor a fold gate: every envelope still passes admission and materialization. A retired endpoint is refused as `unknown` rather than with a status of its own — the peer being refused is, in the case retirement exists for, the one holding the stolen key, and a typed refusal would tell them whether the theft had been noticed. +**Connection authorization** (`private/peers.ts`) is about connections, not the index. Before serving catch-up, a receiver requires the authenticated endpoint to be named by a published device's `deviceAddress` that its owner has not retired. Otherwise anyone who computed the topic could enumerate the corpus. This is neither membership nor a fold gate: every envelope still passes admission and materialization. A retired endpoint is refused as `unknown` rather than with a status of its own — the peer being refused is, in the case retirement exists for, the one holding the stolen key, and a typed refusal would tell them whether the theft had been noticed. An inbound connection no attached space names is nevertheless **held for a bounded grace rather than closed on the accept**, refused frame by frame throughout: closing it immediately destroys the frame carrying the claim above, which is the only thing that can end a discovery deadlock between two replicas neither of whose directories names the other. **Which connections exist** is protocol rather than transport (`private/connections.ts`). A transport dials an address and returns an authenticated link; `core` decides everything around it. The dial list is re-read every round so new devices and changed addresses are observed. Hints order but never filter it, local endpoints are skipped, duplicate addresses share one link, unreachable endpoints back off, and failed links are redialled. None of this is a gate. diff --git a/packages/core/src/private/endpoint.ts b/packages/core/src/private/endpoint.ts index 3af8776..4e19c88 100644 --- a/packages/core/src/private/endpoint.ts +++ b/packages/core/src/private/endpoint.ts @@ -32,6 +32,13 @@ // else opened — so it asks the same question of the same directory before keeping the link, once // per attached space, because a peer in one space is a stranger in the next. Neither is a gate: // what a peer sends still passes `admit()` either way. +// - **A connection nobody names yet is held briefly, and served nothing.** Adopting it and closing +// it are not the only two answers, and treating them as such is what made a fresh replica +// undiscoverable (ADR §30): a dialer's first frame is what tells this replica whose repo to read, +// and closing the connection on the accept — before a single frame has been read off it — +// destroys that frame along with the round trip that carried it. So an unnamed link waits out a +// bounded grace instead, during which it is refused frame by frame exactly as before and is +// adopted the moment the directory does name it. import { readDeviceDirectory, type DeviceDirectory } from '../devices.js' import { compareCodePoints } from '../order.js' @@ -91,9 +98,33 @@ export interface PrivateEndpointOptions { /** A persisted endpoint identity. Absent mints a new one, which the caller should then store. */ secretKey?: Uint8Array relays?: RelayPolicy + /** How long an inbound connection no attached space names is kept. Tests shorten it. */ + unknownLinkGraceMs?: number log?: (message: string) => void } +/** + * How long an inbound connection no attached space names is kept before it is closed. + * + * The number is a bound on a stranger's cost and nothing else, so it is chosen from the only thing + * that has to fit inside it: one dialer's opening round trip, plus enough of this replica's own sync + * cadence (a daemon interval, a tab's ten-second tick) that a claim captured on the first frame can + * be verified against the claimer's public repo and turn into an adoption rather than a redial. Half + * a minute covers both with room to spare; nothing is lost by it expiring first, because a verified + * peer is dialled from this side on the next round anyway (`PeerConnections.retryNow`). + */ +export const DEFAULT_UNKNOWN_LINK_GRACE_MS = 30_000 + +/** + * How many unnamed inbound connections are held at once. + * + * A private space is a handful of devices, so this is far above what convergence needs and well + * below what would make holding them interesting to somebody who computed the topic: past the bound + * a new stranger is closed on the accept, as every stranger was before ADR §30, and the peers a + * replica is actually waiting for arrive one or two at a time. + */ +const MAX_UNKNOWN_LINKS = 8 + export interface AttachSpaceOptions { spaceUri: string /** This replica's stores. The directory is re-read from `records` every round, never cached. */ @@ -172,11 +203,16 @@ export class PrivateEndpoint { readonly #transport: PeerTransport /** Keyed by space URI; the wire topic each answers to is resolved per routed frame. */ readonly #spaces = new Map DeviceDirectory }>() + /** Inbound connections no attached space names yet, keyed by the endpoint the transport + * authenticated. Held rather than adopted: nothing is ever served over one. */ + readonly #unknown = new Map }>() + readonly #graceMs: number readonly #log: (message: string) => void #closed = false - private constructor(transport: PeerTransport, log: (message: string) => void) { + private constructor(transport: PeerTransport, graceMs: number, log: (message: string) => void) { this.#transport = transport + this.#graceMs = graceMs this.#log = log } @@ -199,7 +235,11 @@ export class PrivateEndpoint { ...(options.secretKey ? { secretKey: options.secretKey } : {}), ...(options.relays ? { relays: options.relays } : {}), }) - endpoint = new PrivateEndpoint(transport, log) + endpoint = new PrivateEndpoint( + transport, + Math.max(0, options.unknownLinkGraceMs ?? DEFAULT_UNKNOWN_LINK_GRACE_MS), + log, + ) return endpoint } @@ -269,6 +309,10 @@ export class PrivateEndpoint { // down with it. `#route` still sees the rejection, which is where it would mean something. void attached.topic.catch(() => undefined) this.#spaces.set(options.spaceUri, { attached, directory }) + // A connection can arrive before the space it was opened for is attached — a tab navigating into + // a space a daemon is already dialling — and the peer holding it would otherwise wait out a grace + // this replica could have ended on the spot. + for (const endpointId of [...this.#unknown.keys()]) this.#reconsider(endpointId) return attached } @@ -297,13 +341,23 @@ export class PrivateEndpoint { // An `error` frame is somebody telling this endpoint about a request it made. There is nothing // to route and nothing to answer; the bus that asked has already counted it against the peer. if (topic === undefined) return undefined + let answer: PrivateFrame | undefined + let routed = false for (const { attached } of this.#spaces.values()) { - if ((await attached.topic) === topic) return attached.bus.accept(peerId, frame) + if ((await attached.topic) !== topic) continue + answer = await attached.bus.accept(peerId, frame) + routed = true + break } + // A frame from a peer on grace is what buys the round trip its claim travelled in, so the + // directory is asked again once the answer is settled: a claim captured a moment ago may have + // been verified since, and this is the connection the peer already believes it has. Deliberately + // after the answer — a promotion never changes what the frame that triggered it was told. + this.#reconsider(peerId) // Deliberately the same answer for "no such space here" and "that space is not attached yet": a // topic is derived from a public URI and confers nothing, so this must not become a way to ask // an endpoint which private spaces it holds. - return errorFrame('unknown-topic', 'this endpoint does not serve that topic') + return routed ? answer : errorFrame('unknown-topic', 'this endpoint does not serve that topic') } /** @@ -311,28 +365,46 @@ export class PrivateEndpoint { * * A link is a connection between two ENDPOINTS, and each space decides separately whether that * endpoint is one of its devices — so the same link may be adopted by one space and refused by - * the next. A link no space wants is closed: everything it could ask for would be refused frame by - * frame anyway, and a peer whose device this replica has simply not polled yet redials after its - * own backoff (`PeerRefusal` is `unknown` — "ask again"). + * the next. + * + * A link no space names is **kept for a bounded grace instead of closed**, and that is ADR §30's + * whole change here. The old answer — close it, the peer redials after its own backoff — reads as + * safe and is a deadlock in the one case that matters: a replica that has never met this space's + * other devices learns whose repos to read from what a dialer says on arrival (`from`, `sync.ts`), + * and a connection closed on the accept carries no frames at all. The grace changes nothing about + * what is SERVED — every frame over an unadopted link is refused by `WirePrivateBus.accept` frame + * by frame, because the bus authorizes against the same directory this does — only about how long + * the peer has to say who it claims to be. */ #adopt(link: PeerLink): void { if (this.#closed) { link.close?.() return } - // Each manager owns and may close its PeerLink. Give it a logical lease rather than the shared - // physical link, so a full, failed, detached, or closing space cannot disconnect another one. + if (this.#offer(link)) return + this.#hold(link) + } + + /** + * Hand one link to every attached space that names its endpoint. False when none took it. + * + * Each manager owns and may close its `PeerLink`, so what it gets is a logical lease rather than + * the shared physical link: a full, failed, detached, or closing space cannot disconnect another. + */ + #offer(link: PeerLink, options: { onlyWhenIdle?: boolean } = {}): boolean { const authorized: Array<{ attached: AttachedSpace; peer: PeerIdentity }> = [] for (const { attached, directory } of this.#spaces.values()) { const authorization = authorizeEndpoint(link.id, directory()) if (!authorization.ok) continue + // A connection that has been waiting is older than anything the space has taken up since, so + // it is offered only where there is nothing to displace. The opposite rule is right for a + // connection that has just arrived — `PeerConnections.adopt` replaces, because a peer that + // redialled has said which connection it believes it has — and applying it to a held one + // would close a live link in favour of a socket the peer may already have hung up on. + if (options.onlyWhenIdle && this.#connectedTo(attached, link.id)) continue authorized.push({ attached, peer: authorization.peer }) } - if (authorized.length === 0) { - this.#log(`private: no attached space names inbound endpoint ${link.id}; closing the connection`) - link.close?.() - return - } + if (authorized.length === 0) return false const shared = new SharedInboundLink(link) // Acquire every lease before handing any one of them to `adopt`, which may synchronously refuse // and close it at the manager's bound. A refusal cannot transiently take the count to zero. @@ -340,6 +412,70 @@ export class PrivateEndpoint { for (const [index, { attached, peer }] of authorized.entries()) { attached.connections.adopt(peer, leases[index]!) } + return true + } + + /** Whether a space already holds a live connection to one endpoint, whoever dialled it. */ + #connectedTo(attached: AttachedSpace, endpointId: string): boolean { + return attached.connections + .state() + .some((peer) => peer.endpointId === endpointId && peer.connected) + } + + /** Keep an unnamed connection open, unserved, until the grace expires or the directory catches up. */ + #hold(link: PeerLink): void { + // One endpoint, one held connection: a peer that redialled while its first attempt was still on + // grace has told this replica the earlier one is no longer the connection it believes it has. + const held = this.#unknown.get(link.id) + if (held) { + this.#unknown.delete(link.id) + clearTimeout(held.timer) + held.link.close?.() + } + if (this.#unknown.size >= MAX_UNKNOWN_LINKS) { + this.#log( + `private: no attached space names inbound endpoint ${link.id}, and ${MAX_UNKNOWN_LINKS} ` + + 'unnamed connections are already waiting; closing the connection', + ) + link.close?.() + return + } + this.#unknown.set(link.id, { link, timer: setTimeout(() => this.#expire(link), this.#graceMs) }) + } + + /** + * The directory may have caught up with a connection on grace; adopt it if it has. + * + * Cheap by construction: it folds nothing unless something is actually waiting, which for every + * replica whose peers are already in its directory is never. + */ + #reconsider(endpointId: string): void { + const held = this.#unknown.get(endpointId) + if (!held || this.#closed) return + if (!this.#offer(held.link, { onlyWhenIdle: true })) return + this.#unknown.delete(endpointId) + clearTimeout(held.timer) + this.#log(`private: inbound endpoint ${endpointId} is now named by an attached space; adopted`) + } + + /** The grace is spent. One last look at the directory, and then the connection goes. */ + #expire(link: PeerLink): void { + const held = this.#unknown.get(link.id) + if (held?.link !== link) return + this.#unknown.delete(link.id) + clearTimeout(held.timer) + if (!this.#closed && this.#offer(link, { onlyWhenIdle: true })) return + this.#log(`private: no attached space names inbound endpoint ${link.id}; closing the connection`) + link.close?.() + } + + /** Close every connection still waiting on the directory, with the grace they had left. */ + #dropUnknown(): void { + for (const [endpointId, held] of [...this.#unknown]) { + this.#unknown.delete(endpointId) + clearTimeout(held.timer) + held.link.close?.() + } } /** @@ -352,6 +488,7 @@ export class PrivateEndpoint { async close(): Promise { if (this.#closed) return this.#closed = true + this.#dropUnknown() for (const { attached } of this.#spaces.values()) { await attached.bus.close() attached.connections.close() diff --git a/packages/core/test/private-endpoint.test.mjs b/packages/core/test/private-endpoint.test.mjs new file mode 100644 index 0000000..1861fc8 --- /dev/null +++ b/packages/core/test/private-endpoint.test.mjs @@ -0,0 +1,234 @@ +// The endpoint's inbound half: what happens to a connection nobody has named yet (ADR §30). +// +// The lifecycle this covers is the one a bus-level test cannot: `PrivateEndpoint` is what a +// transport hands an accepted connection to, and until §30 it answered "no attached space names +// that endpoint" by closing the connection on the accept — before a single frame had been read off +// it. That is the state a fresh replica is in by definition, and the frame it destroyed is the one +// carrying the dialer's claimed DIDs, which is the whole of how the fresh replica learns whose +// public repo to read (`sync.ts`, `ingest/private.ts`). +// +// So the properties here are about a connection's LIFETIME and never about trust: an unnamed link is +// held, is served nothing while it is held, is adopted the moment the directory names its endpoint, +// and goes away on its own if it never does. + +import assert from 'node:assert/strict' +import { describe, it } from 'node:test' +import { COLLECTIONS } from '../dist/generated/records.js' +import { MemoryRecordStore } from '../dist/store.js' +import { PrivateEndpoint } from '../dist/private/endpoint.js' +import { MemoryEnvelopeStore } from '../dist/private/envelope-store.js' +import { WIRE_VERSION, spaceTopic } from '../dist/private/wire.js' + +const SPACE = 'at://did:plc:privatefounder/com.disnetdev.radial.space/space1' +const AGENT = 'did:plc:agent' +const AGENT_ENDPOINT = 'a'.repeat(64) + +/** A transport that binds, never dials, and hands the test the endpoint's own inbound hooks. */ +function testTransport() { + const bound = {} + return { + bound, + factory: async (options) => { + bound.receiver = options.endpoint + bound.onLink = options.onLink + return { + endpointId: 'e'.repeat(64), + relays: [], + secretKey: new Uint8Array(32), + async dial() { + throw new Error('this endpoint does not dial') + }, + async close() {}, + } + }, + } +} + +/** One accepted connection, as a transport presents it. */ +const inboundLink = (id) => { + const link = { + id, + closed: false, + async request() { + throw new Error('a held connection was used') + }, + async send() {}, + close() { + link.closed = true + }, + } + return link +} + +/** The public half: what `${did}`'s repo says about one device, once it has been polled. */ +function publishDevice(records, did, endpointId) { + const deviceKeyId = 'node-endpoint001' + for (const [collection, value] of [ + [ + COLLECTIONS.device, + { + $type: COLLECTIONS.device, + deviceKeyId, + publicKey: 'A'.repeat(43), + algorithm: 'ed25519', + kind: 'node', + createdAt: '2026-03-01T00:00:00Z', + }, + ], + [ + COLLECTIONS.deviceAddress, + { $type: COLLECTIONS.deviceAddress, deviceKeyId, endpointId, createdAt: '2026-03-01T00:00:01Z' }, + ], + ]) { + records.put({ + did, + collection, + rkey: deviceKeyId, + uri: `at://${did}/${collection}/${deviceKeyId}`, + cid: `cid-${collection}-${did}`, + rev: '0000000000001', + value, + }) + } +} + +const claimFrame = async (from) => ({ + kind: 'catch-up', + version: WIRE_VERSION, + topic: await spaceTopic(SPACE), + summaries: [], + wants: [], + ...(from ? { from } : {}), +}) + +async function endpointOver(records, options = {}) { + const transport = testTransport() + const endpoint = await PrivateEndpoint.bind({ transport: transport.factory, ...options }) + const attached = endpoint.attach({ + spaceUri: SPACE, + records, + envelopes: new MemoryEnvelopeStore(), + }) + return { endpoint, attached, ...transport.bound } +} + +describe('an inbound connection no attached space names', () => { + it('is held open and refused, so the claim it carries survives the round trip', async () => { + const records = new MemoryRecordStore() + const { endpoint, attached, receiver, onLink } = await endpointOver(records) + try { + const link = inboundLink(AGENT_ENDPOINT) + onLink(link) + // The connection this replica is waiting on is the one it used to close first. Nothing about + // the answer changes: the bus authorizes every frame against the same directory this does. + assert.equal(link.closed, false, 'the connection was closed before it could say anything') + const answer = await receiver.accept(AGENT_ENDPOINT, await claimFrame([AGENT])) + assert.equal(answer.kind, 'error') + assert.equal(answer.status, 'unavailable') + assert.equal(attached.connections.connected, 0, 'an unnamed endpoint was adopted') + // ...and the claim is now where the ingestor drains it from, to be verified against AGENT's + // own public repo. That verification is not this file's, and neither is trusting it. + assert.deepEqual(attached.bus.drainPeerCandidates(), [ + { endpointId: AGENT_ENDPOINT, dids: [AGENT] }, + ]) + } finally { + await endpoint.close() + } + }) + + it('is adopted on its next frame once the directory names it, without a redial', async () => { + const records = new MemoryRecordStore() + const { endpoint, attached, receiver, onLink } = await endpointOver(records) + try { + const link = inboundLink(AGENT_ENDPOINT) + onLink(link) + await receiver.accept(AGENT_ENDPOINT, await claimFrame([AGENT])) + + // The claim was verified out of band: AGENT's public repo does bind that endpoint, so the poll + // it bought put the directory records in this replica's store. + publishDevice(records, AGENT, AGENT_ENDPOINT) + const answer = await receiver.accept(AGENT_ENDPOINT, await claimFrame([AGENT])) + assert.equal(answer.kind, 'envelopes', 'a named endpoint was still refused') + + const [state] = attached.connections.state() + assert.equal(state.endpointId, AGENT_ENDPOINT) + assert.equal(state.did, AGENT) + assert.equal(state.inbound, true) + assert.equal(state.connected, true) + assert.equal(link.closed, false) + // Adopted exactly once: a second frame over an adopted link is an ordinary served frame. + await receiver.accept(AGENT_ENDPOINT, await claimFrame([AGENT])) + assert.equal(attached.connections.connected, 1) + } finally { + await endpoint.close() + } + }) + + it('is adopted when the space it was opened for attaches under it', async () => { + const records = new MemoryRecordStore() + publishDevice(records, AGENT, AGENT_ENDPOINT) + const transport = testTransport() + const endpoint = await PrivateEndpoint.bind({ transport: transport.factory }) + try { + const link = inboundLink(AGENT_ENDPOINT) + transport.bound.onLink(link) + assert.equal(link.closed, false, 'a connection that arrived before its space was closed') + const attached = endpoint.attach({ + spaceUri: SPACE, + records, + envelopes: new MemoryEnvelopeStore(), + }) + assert.equal(attached.connections.connected, 1, 'the held connection was not adopted') + } finally { + await endpoint.close() + } + }) + + it('is closed when the grace runs out with nobody naming it', async () => { + const records = new MemoryRecordStore() + const { endpoint, receiver, onLink } = await endpointOver(records, { unknownLinkGraceMs: 5 }) + try { + const link = inboundLink(AGENT_ENDPOINT) + onLink(link) + await receiver.accept(AGENT_ENDPOINT, await claimFrame([AGENT])) + assert.equal(link.closed, false) + await new Promise((resolve) => setTimeout(resolve, 40)) + assert.equal(link.closed, true, 'a connection nobody named was kept past its grace') + } finally { + await endpoint.close() + } + }) + + it('holds a bounded number of them, and closes the rest on the accept', async () => { + const records = new MemoryRecordStore() + const { endpoint, onLink } = await endpointOver(records) + try { + // Anybody who computed the topic can open one of these, so the bound is what stops a stranger + // holding connections for as long as this replica is willing to wait for a peer. + const links = Array.from({ length: 9 }, (_, index) => + inboundLink(`${index}`.repeat(64).slice(0, 64)), + ) + for (const link of links) onLink(link) + assert.deepEqual( + links.map((link) => link.closed), + [...Array(8).fill(false), true], + ) + } finally { + await endpoint.close() + } + }) + + it('keeps only the newest connection from one endpoint, and closes them all on shutdown', async () => { + const records = new MemoryRecordStore() + const { endpoint, onLink } = await endpointOver(records) + const first = inboundLink(AGENT_ENDPOINT) + const second = inboundLink(AGENT_ENDPOINT) + onLink(first) + onLink(second) + // A peer that redialled has told this replica which connection it believes it has. + assert.equal(first.closed, true) + assert.equal(second.closed, false) + await endpoint.close() + assert.equal(second.closed, true, 'a held connection outlived the endpoint holding it') + }) +}) diff --git a/packages/daemon/README.md b/packages/daemon/README.md index 3c43815..caf8200 100644 --- a/packages/daemon/README.md +++ b/packages/daemon/README.md @@ -221,7 +221,11 @@ radiald device retire # withdraw its address: nobody dials either way. - **Inbound connections are authorized before they are adopted**, per attached space, against the same directory `accept()` authorizes each frame against. Adoption is what lets a replica that can - only dial out be gossiped to. + only dial out be gossiped to. One the directory does not name is **held briefly rather than + closed** (ADR §30) — refused frame by frame throughout, but kept long enough for the dialer's + `catch-up` to arrive, because the DIDs it claims in `from` are how a replica that has never met + this peer learns whose public repo to read. `PrivateSpaceIngestor` verifies each claim against that + repo before it polls the DID again, so a claim buys public reads and never trust. - **Startup order is load-bearing**: device key → endpoint → address → replica → ticket check. Each is the next one's precondition; `private-run.ts` is that sequence and says why at each step. diff --git a/packages/daemon/test/private-transport.test.mjs b/packages/daemon/test/private-transport.test.mjs index 9f7e317..1f40587 100644 --- a/packages/daemon/test/private-transport.test.mjs +++ b/packages/daemon/test/private-transport.test.mjs @@ -126,6 +126,11 @@ function publicRepos() { * `dial` hands the caller a link to the remote AND hands the remote a link back, which is what a * connection is — one QUIC connection carries frames both ways, and it is exactly the inbound half * that `PrivateEndpoint`'s adoption exists to keep. + * + * The two ends share one open flag, because a connection is one object however many handles are held + * on it: an accepter that closes its half ends the dialer's too. That fidelity is load-bearing rather + * than decorative (ADR §30) — a double where the accepter could hang up and the dialer went on being + * served would report every accept-side lifecycle decision as working. */ function transportNetwork() { const endpoints = new Map() @@ -157,10 +162,19 @@ function transportNetwork() { async dial(peer) { const remote = endpoints.get(peer.endpointId) if (!remote) throw new Error(`no endpoint ${peer.endpointId} is listening`) - remote.onLink?.( - loopbackLink(endpointId, options.endpoint, { localId: peer.endpointId }), - ) - return loopbackLink(peer.endpointId, remote.receiver, { localId: endpointId }) + let open = true + const offline = () => !open + const close = () => { + open = false + } + remote.onLink?.({ + ...loopbackLink(endpointId, options.endpoint, { localId: peer.endpointId, offline }), + close, + }) + return { + ...loopbackLink(peer.endpointId, remote.receiver, { localId: endpointId, offline }), + close, + } }, async online() {}, async close() { @@ -599,11 +613,13 @@ describe('a daemon serving a private space over a transport', () => { } }) - it('closes an inbound connection no attached space will serve', async () => { + it('holds an inbound connection no attached space names, and serves nothing over it', async () => { const repos = publicRepos() const network = transportNetwork() const founder = await operator('founder', FOUNDER, repos, network) try { + await seedSpace(founder) + await founder.sync() let closed = false const listener = network.endpoints.get(founder.opened.endpoint.endpointId) listener.onLink({ @@ -616,12 +632,84 @@ describe('a daemon serving a private space over a transport', () => { closed = true }, }) - assert.equal(closed, true, 'an unauthorized inbound link was kept') + // Kept, because closing it on the accept is what left a fresh replica undiscoverable: the + // frame that says who is dialling has not arrived yet (ADR §30). How long it is kept, and that + // it does go, is `core`'s to state — `private-endpoint.test.mjs` holds the clock. + assert.equal(closed, false, 'an inbound connection was closed before it could say anything') + // Held is not adopted and not served. This corpus has a space, a project and a goal in it, and + // an endpoint the directory does not name gets exactly the answer it got before. + const answer = await listener.receiver.accept('nobody-published-this-endpoint', { + kind: 'catch-up', + version: WIRE_VERSION, + topic: await spaceTopic(SPACE_URI), + summaries: [], + wants: [], + }) + assert.equal(answer.kind, 'error') + assert.equal(answer.status, 'unavailable') + assert.equal(founder.opened.endpoint.attached()[0].connections.connected, 0) } finally { await founder.close() } }) + it('bootstraps a fresh replica that knows only the founder, from the daemon that dials it', async () => { + const repos = publicRepos() + const network = transportNetwork() + // The corpus as it stands the day the founder's machine is gone: the founder wrote the space, an + // agent daemon caught up over the transport, and the agent is the only thing still running. + const founder = await operator('founder', FOUNDER, repos, network) + await seedSpace(founder) + await founder.sync() + const agent = await operator('agent', PEER, repos, network) + await assert.rejects(agent.sync(), /Space record not found/) + await founder.sync() + assert.equal((await agent.sync()).goals.length, 1) + await founder.close() + + // A fresh machine under the FOUNDER's own identity: a new data directory, so a new endpoint and + // a new device key, and nothing to poll but its own repo — which names the machine that is gone + // and this one, and nowhere says that the agent exists. Hints cannot help here either: they were + // written before any agent had joined. This is the reported bug, exactly. + const fresh = await operator('founder-fresh', FOUNDER, repos, network) + try { + await assert.rejects(fresh.sync(), /Space record not found/) + assert.deepEqual( + fresh.opened.endpoint.attached()[0].connections.state().map((peer) => peer.connected), + [false], + 'the fresh replica reached somebody it should not have been able to reach', + ) + + // The agent is the side that can move: the founder is a member, so it polls the founder's repo, + // finds the device this replica published a moment ago, and dials it. The dial is refused — + // this directory has never heard of the agent — and the refusal is where the agent's claimed + // DID is captured, which it can only be if the connection outlived the accept. + await agent.sync() + + // One ordinary cycle on the fresh side is now enough: drain the claim, read the agent's public + // repo, find the address that binds the endpoint that dialled, and dial it back. + const index = await fresh.sync() + assert.deepEqual( + index.goals.map((goal) => goal.target.value.title), + ['Ship the transport'], + ) + assert.ok( + fresh.runtime.ingestor.knownDids.has(PEER), + 'a verified claim did not become a repo this replica polls', + ) + assert.deepEqual( + fresh.opened.endpoint + .attached()[0] + .connections.state() + .filter((peer) => peer.connected) + .map((peer) => peer.did), + [PEER], + ) + } finally { + await Promise.all([fresh.close(), agent.close()]) + } + }) + it('leases one inbound connection independently across authorized spaces', async () => { const network = transportNetwork() const listener = await PrivateEndpoint.bind({ transport: network.factory('shared-listener') }) diff --git a/packages/ingest/src/private.ts b/packages/ingest/src/private.ts index 88047a8..4ac2445 100644 --- a/packages/ingest/src/private.ts +++ b/packages/ingest/src/private.ts @@ -8,11 +8,11 @@ import { admit, assertSpacePin, compareCodePoints, + connectablePeers, deviceKey, materialize, missingBlobs, readDeviceDirectory, - authorizeEndpoint, spacePinMismatch, summarize, wantFor, @@ -161,6 +161,18 @@ const emptyReport = (): AdmissionReport => ({ /** How many blobs one cycle will pull. Modest: a picture is two orders larger than a record. */ const DEFAULT_MAX_BLOBS_PER_SYNC = 8 +/** + * What one cycle will spend on peers that claimed an identity while being refused (ADR §30). + * + * Every number here bounds work a *stranger* can ask for, so each is set to what convergence + * actually needs and no higher: a replica waiting for its first peer is waiting for one or two + * devices, not for a crowd, and a claim that misses this cycle is re-made on the claimer's next dial. + */ +const MAX_PEER_CANDIDATES_PER_SYNC = 4 +const MAX_CLAIMED_DIDS = 8 +const PEER_CANDIDATE_COOLDOWN_MS = 60_000 +const MAX_PEER_CANDIDATE_COOLDOWNS = 64 + export class PrivateSpaceIngestor { readonly spaceDid: string readonly knownDids = new Set() @@ -467,25 +479,60 @@ export class PrivateSpaceIngestor { } } + /** + * Whom a refused peer said it was, checked against the only thing that can settle it: its repo. + * + * A claim is not a credential and is never kept as one. What it buys is a bounded number of + * *public, unauthenticated* repo reads — the same reads `pollDirectory` makes of a member — and a + * DID survives them only if its own repo publishes a `deviceAddress` naming the endpoint the + * transport authenticated, which is the question `authorizeEndpoint` already answers for every + * connection. So the worst a stranger can do by claiming somebody else's DID is make this replica + * read a public repo, and the most it can win by claiming its own is a directory entry it would + * have earned anyway by publishing an address it controls. + * + * Bounded three ways, because the claim arrives from anybody who can reach this endpoint: + * candidates per cycle, DIDs per candidate, and a per-endpoint cooldown so a peer redialling in a + * loop cannot buy a poll per dial. The cooldown table is bounded too — its keys are a stranger's + * to invent. + */ async #pollPeerCandidates(signal?: AbortSignal): Promise { if (!this.#poller || !this.bus.drainPeerCandidates) return const now = Date.now() - for (const candidate of this.bus.drainPeerCandidates().slice(0, 4)) { + for (const candidate of this.bus.drainPeerCandidates().slice(0, MAX_PEER_CANDIDATES_PER_SYNC)) { const last = this.#candidateCooldown.get(candidate.endpointId) ?? 0 - if (now - last < 60_000) continue - this.#candidateCooldown.set(candidate.endpointId, now) - const verified: string[] = [] - for (const did of [...new Set(candidate.dids)].slice(0, 8)) { + if (now - last < PEER_CANDIDATE_COOLDOWN_MS) continue + this.#rememberCandidate(candidate.endpointId, now) + let verified = false + for (const did of [...new Set(candidate.dids)].slice(0, MAX_CLAIMED_DIDS)) { await this.#poller.pollDid(did, signal) + // The claimed DID and the authenticated endpoint have to be bound by a record of THIS + // author: a directory that names the endpoint under somebody else has answered a different + // question, and taking that as a verification would let anybody's claim be settled by + // anybody's repo. Asked through `connectablePeers` rather than `authorizeEndpoint`, which + // answers with the first device it finds for an endpoint — and a daemon running two + // identities legitimately publishes two addresses for one. const directory = readDeviceDirectory(this.records.records(), []) - if ([...directory.addresses].some(([key, address]) => key.startsWith(`${did}\u0000`) && address.endpointId === candidate.endpointId)) { - verified.push(did) - } - } - if (verified.length > 0 && authorizeEndpoint(candidate.endpointId, readDeviceDirectory(this.records.records(), [])).ok) { - for (const did of verified) this.knownDids.add(did) - this.#onPeerCandidate?.() + const bound = connectablePeers(directory).some( + (peer) => peer.did === did && peer.endpointId === candidate.endpointId, + ) + if (!bound) continue + // Membership is what keeps it here afterwards, if it belongs here at all: this decides whose + // repo is read and never what counts, so a DID that turns out to be nobody costs one poll. + this.knownDids.add(did) + verified = true } + // The peer is dialable from this side now, and the backoff rung it is on was earned by + // refusals from a directory that had not read it yet — which is what `retryNow` is for. + if (verified) this.#onPeerCandidate?.() + } + } + + /** One cooldown entry per endpoint, oldest evicted first: a stranger chooses these keys. */ + #rememberCandidate(endpointId: string, now: number): void { + this.#candidateCooldown.delete(endpointId) + this.#candidateCooldown.set(endpointId, now) + while (this.#candidateCooldown.size > MAX_PEER_CANDIDATE_COOLDOWNS) { + this.#candidateCooldown.delete(this.#candidateCooldown.keys().next().value as string) } } diff --git a/packages/ingest/test/private.test.mjs b/packages/ingest/test/private.test.mjs index 29905b4..0b1af9e 100644 --- a/packages/ingest/test/private.test.mjs +++ b/packages/ingest/test/private.test.mjs @@ -9,6 +9,7 @@ import { MemoryRecordStore, MemoryWantList, Quarantine, + WIRE_VERSION, WirePrivateBus, authorizeEndpoint, connectablePeers, @@ -22,6 +23,7 @@ import { readDeviceDirectory, recordCid, seal, + spaceTopic, } from '../../core/dist/index.js' import { MemorySyncStateStore, @@ -970,3 +972,143 @@ describe('private ingestion over the wire', () => { ) }) }) + +// A claim a refused peer made about itself, and what this replica spends checking it (ADR §30). +// +// The claim exists because a replica that has never met a space's live devices has no other way to +// find them: it polls the founder and itself, the daemon serving the space is neither, and a refusal +// that taught it nothing left both sides retrying forever. So a dialer names its DIDs on the frame, +// and this is the half that decides what that is worth — which is a bounded number of public repo +// reads, and never a directory entry, a served frame, or anything the fold can see. +describe('a claim from a peer this replica refused', () => { + const AGENT = 'did:plc:privateagent' + const IMPOSTOR = 'did:plc:privateimpostor' + const AGENT_ENDPOINT = 'endpoint-agent' + + /** A replica whose directory names nobody, and a fake set of public repos behind its poller. */ + async function refusingReplica(repos = new Map()) { + const { spaceUri, rootDevice, memberDevice, envelopes } = await privateCorpus() + const records = new MemoryRecordStore() + for (const record of [ + directoryRecord(rootDevice, '2026-02-01T00:00:00Z'), + directoryRecord(memberDevice, '2026-02-01T00:00:01Z'), + ]) { + records.put(record) + } + const wants = new MemoryWantList() + const envelopeStore = new MemoryEnvelopeStore(records) + const bus = new WirePrivateBus({ + spaceUri, + envelopes: envelopeStore, + peers: { async links() { return [] } }, + // No address in this directory names anybody: the honest state of a replica whose peers it + // has never read. Every inbound frame is refused, which is where a claim is captured. + authorizePeer: () => ({ ok: false, status: 'unknown', reason: 'no published device address' }), + }) + const polled = [] + const poller = { + collections: DEVICE_DIRECTORY_COLLECTIONS, + async pollDid(did) { + polled.push(did) + for (const record of repos.get(did) ?? []) records.put(record) + }, + } + let woken = 0 + const ingestor = new PrivateSpaceIngestor(spaceUri, bus, envelopeStore, records, { + wants, + quarantine: new Quarantine(wants), + bootstrapDids: [ROOT], + poller, + onPeerCandidate: () => { + woken += 1 + }, + }) + // Folded already, so `sync()` is exercising the candidate poll and not a cold start. + for (const envelope of envelopes) await ingestor.offer(envelope) + const topic = await spaceTopic(spaceUri) + return { + ingestor, + records, + polled, + woken: () => woken, + /** One refused catch-up carrying a claim — a dialer's first frame, exactly. */ + claim: (endpointId, from) => + bus.accept(endpointId, { + kind: 'catch-up', + version: WIRE_VERSION, + topic, + summaries: [], + wants: [], + from, + }), + } + } + + /** One agent's public repo: a device, and an address for the endpoint it dials from. */ + const agentRepo = async (did, endpointId) => { + const signer = await device(did, `${did}-node`) + return [directoryRecord(signer, '2026-02-01T00:00:03Z'), addressRecord(signer, endpointId)] + } + + it('polls the claimed repo once, and keeps the DID only if that repo binds the endpoint', async () => { + const replica = await refusingReplica(new Map([[AGENT, await agentRepo(AGENT, AGENT_ENDPOINT)]])) + const answer = await replica.claim(AGENT_ENDPOINT, [AGENT]) + // The refusal is the one a stranger gets, unchanged. A claim that bought a different answer + // would be a way of asking an endpoint which DIDs it already trusts. + assert.equal(answer.kind, 'error') + assert.equal(answer.status, 'unavailable') + + await replica.ingestor.sync() + assert.deepEqual(replica.polled.filter((did) => did === AGENT), [AGENT]) + assert.equal(replica.ingestor.knownDids.has(AGENT), true) + // ...and connection management is told, because the peer's backoff was earned by refusals from + // a directory that had not read it yet. + assert.equal(replica.woken(), 1) + assert.equal( + authorizeEndpoint(AGENT_ENDPOINT, readDeviceDirectory(replica.records.records(), [])).ok, + true, + ) + }) + + it('forgets a claim the claimed repo does not support, whoever else names the endpoint', async () => { + // The endpoint IS named — by somebody else's repo. Settling one author's claim with another + // author's records is the mistake this guards: `connectablePeers` is asked for the pair. + const impostor = await agentRepo(IMPOSTOR, AGENT_ENDPOINT) + const replica = await refusingReplica(new Map([[IMPOSTOR, impostor]])) + for (const record of impostor) replica.records.put(record) + + await replica.claim(AGENT_ENDPOINT, [AGENT]) + await replica.ingestor.sync() + assert.deepEqual(replica.polled.filter((did) => did === AGENT), [AGENT], 'one poll, no more') + assert.equal(replica.ingestor.knownDids.has(AGENT), false) + assert.equal(replica.woken(), 0) + }) + + it('will not be made to poll again by a peer that redials', async () => { + // A claim nothing supports, made over and over: the cooldown is per endpoint, so a stranger + // redialling in a loop cannot buy a public read per dial. (A claim that VERIFIES is polled every + // cycle after that, by `pollDirectory` — it is a peer by then, which is the whole point.) + const replica = await refusingReplica() + for (let dial = 0; dial < 3; dial += 1) { + await replica.claim(AGENT_ENDPOINT, [AGENT]) + await replica.ingestor.sync() + } + assert.deepEqual(replica.polled.filter((did) => did === AGENT), [AGENT]) + assert.equal(replica.ingestor.knownDids.has(AGENT), false) + }) + + it('spends a bounded number of polls per cycle, however many peers claim something', async () => { + const replica = await refusingReplica() + for (let index = 0; index < 6; index += 1) { + await replica.claim(`endpoint-stranger-${index}`, [`did:plc:stranger${index}`]) + } + await replica.ingestor.sync() + assert.equal( + replica.polled.filter((did) => did.startsWith('did:plc:stranger')).length, + 4, + 'a crowd of claims bought more than one cycle of polls', + ) + // Nothing was kept: none of those repos says anything at all. + assert.equal(replica.woken(), 0) + }) +}) diff --git a/packages/ui/src/lib/private-space.test.ts b/packages/ui/src/lib/private-space.test.ts index 61950fc..9f2b06a 100644 --- a/packages/ui/src/lib/private-space.test.ts +++ b/packages/ui/src/lib/private-space.test.ts @@ -939,6 +939,9 @@ describe('a tab with a transport endpoint', () => { // 64 lowercase hex characters, which is what the lexicon requires of an endpoint id. const FOUNDER_ENDPOINT = 'f0'.repeat(32) const TAB_ENDPOINT = 'ab'.repeat(32) + /** An agent daemon: a member whose DID no bookmark, ticket or hint of this tab's ever names. */ + const AGENT = 'did:plc:agent' + const AGENT_ENDPOINT = '0a'.repeat(32) /** A transport that reaches exactly one peer, through the real codec. */ function loopbackTransport(options: { @@ -948,13 +951,13 @@ describe('a tab with a transport endpoint', () => { /** Captures what `PrivateEndpoint.bind` hands down, so the far side can reach back. */ onBound?: (bound: { endpoint: FrameEndpoint; onLink?: (link: PeerLink) => void }) => void /** - * Told when this endpoint dials out, with its own `accept`. + * Told when this endpoint dials out, with the connection's inbound end. * * A real transport gives the ACCEPTING side an inbound connection, which `PeerConnections.adopt` * then gossips back over — the whole reason adoption exists. Without this the loopback is * one-directional and could never exercise it. */ - onDial?: (frames: FrameEndpoint) => void + onDial?: (link: PeerLink) => void }): PeerTransportFactory { return async (bound) => { options.onBound?.(bound) @@ -965,8 +968,25 @@ describe('a tab with a transport endpoint', () => { async dial(peer) { const remote = options.remote() if (!remote) throw new Error(`nothing is listening at ${peer.endpointId}`) - options.onDial?.(bound.endpoint) - return loopbackLink(peer.endpointId, remote, { localId: options.endpointId }) + // One connection with two ends: closing either end ends both, the way a real one does. A + // double where the accepter could hang up and the dialer went on being served would report + // every accept-side lifecycle decision as working, whatever it did (ADR §30). + let open = true + const offline = (): boolean => !open + const close = (): void => { + open = false + } + options.onDial?.({ + ...loopbackLink(options.endpointId, bound.endpoint, { + localId: peer.endpointId, + offline, + }), + close, + }) + return { + ...loopbackLink(peer.endpointId, remote, { localId: options.endpointId, offline }), + close, + } }, async close() {}, } @@ -1127,6 +1147,131 @@ describe('a tab with a transport endpoint', () => { } }) + /** + * The peer in the reported bug: a daemon holding the corpus under an identity NO repo this tab + * reads points at, dialling in from outside the tab's directory. + * + * It is `peerHolding` with two differences, and both are the bug: it answers to an agent DID that + * is neither the founder nor this member — so no bookmark, hint or ticket can name it — and its + * dial really does hand the tab an inbound connection, which is the connection the claim travels + * on. `onDial` is what makes the loopback bidirectional, exactly as `PeerConnections.adopt` needs. + */ + async function daemonPeer( + h: Harness, + envelope: PrivateEnvelope, + tab: () => { endpoint: FrameEndpoint; onLink?: (link: PeerLink) => void } | undefined, + ) { + const keys = await BrowserDeviceKeyStore.open(memoryDatabase()) + const key = await keys.create(AGENT, { kind: 'node' }) + const writer = publicDirectoryWriter(AGENT, repoWriter(h, AGENT), h.transport) + await publishDevice(writer, key) + await publishDeviceAddress(writer, key.deviceKeyId, { + endpointId: AGENT_ENDPOINT, + relays: ['https://relay.test/'], + }) + + const records = new MemoryRecordStore() + const envelopes = new MemoryEnvelopeStore(records) + let bound: { endpoint: FrameEndpoint; onLink?: (link: PeerLink) => void } | undefined + const endpoint = await PrivateEndpoint.bind({ + transport: loopbackTransport({ + endpointId: AGENT_ENDPOINT, + remote: () => tab()?.endpoint, + onBound: (accepted) => { + bound = accepted + }, + onDial: (link) => tab()?.onLink?.(link), + }), + }) + const attached = endpoint.attach({ + spaceUri: SPACE, + records, + envelopes, + // What a daemon passes: the identities it will claim while being refused (ADR §30). + selfDids: () => [AGENT], + }) + const ingestor = new PrivateSpaceIngestor(SPACE, attached.bus, envelopes, records, { + poller: directoryPoller(h.transport, records, new MemorySyncStateStore()), + bootstrapDids: [FOUNDER, MEMBER, AGENT], + }) + await ingestor.start() + await ingestor.pollDirectory() + expect((await ingestor.offer(envelope)).admitted).toBe(1) + return { + endpoint, + frames: () => bound?.endpoint, + poll: () => ingestor.pollDirectory(), + /** One daemon round: dial everyone the directory names, and ask them for the difference. */ + catchUp: () => attached.bus.catchUp(SPACE, { summaries: [], wants: [] }), + } + } + + it('fills from a daemon nothing it polls names, out of the dial the daemon made', async () => { + const h = await harness() + const founder = await founderDevice(h, h.keys) + const envelope = await spaceEnvelope(h.keys, founder.key) + // The reported setup: the corpus is on an agent daemon, the founder's own machine is off, and + // this browser has never held the space. Its bootstrap is the founder and itself — the ticket, + // the bookmark and the hints can carry nothing else, because none of them existed when the + // agent joined. Every repo this tab is willing to read is therefore silent about the daemon. + signInAs(h, MEMBER) + await memberDevice(h) + let tabBound: { endpoint: FrameEndpoint; onLink?: (link: PeerLink) => void } | undefined + const daemon = await daemonPeer(h, envelope, () => tabBound) + const endpoint = await browserEndpoint({ + transport: loopbackTransport({ + endpointId: TAB_ENDPOINT, + remote: () => daemon.frames(), + onBound: (bound) => { + tabBound = bound + }, + }), + keys: await BrowserEndpointKeyStore.open(memoryDatabase()), + }) + const replica = await openPrivateReplica({ + spaceUri: SPACE, + transport: h.transport, + actorDid: MEMBER, + meta: { + spaceUri: SPACE, + spaceCid: envelope.recordCid, + founder: FOUNDER, + protocol: 1, + peerHints: [FOUNDER], + createdAt: NOW, + }, + keys: h.keys, + openStores: h.openStores, + endpoint, + publicWriter: signedIn()?.writer as RepoWriter, + }) + try { + await replica.pollDirectory() + // Say where this tab is, and let the daemon read it. This is the only half of the bootstrap + // that worked before: the daemon polls its members' repos, so it CAN find this browser. + await replica.refreshAddress() + await daemon.poll() + // The tab still cannot move. Its directory names one dialable device — its own — so there is + // nobody to ask, which is precisely the state a second open browser used to resolve. + await expect(replica.sync()).rejects.toThrow(/Space record not found/) + expect(replica.connections?.state()).toEqual([]) + + // The daemon dials in and is refused: this tab's directory has never heard of that endpoint. + // What survives the refusal is the DID it claimed on the way, which it can only do because the + // connection outlived the accept (ADR §30). + await daemon.catchUp() + + // One ordinary tick: read the claimed repo, find the address that binds the endpoint that + // dialled, dial it back, and catch up. No second browser, and no configuration. + const index = await replica.sync() + expect(index.space.value.name).toBe('Skunkworks') + expect(replica.connections?.state().map((peer) => peer.did)).toContain(AGENT) + } finally { + replica.close() + await daemon.endpoint.close() + } + }) + it('refuses to serve a tab the far side’s directory does not name', async () => { const h = await harness() const founder = await founderDevice(h, h.keys)