diff --git a/docs/specs/p2p/README.md b/docs/specs/p2p/README.md new file mode 100644 index 00000000000..ff54f9df23e --- /dev/null +++ b/docs/specs/p2p/README.md @@ -0,0 +1,132 @@ +# Zakura P2P Stack — Specification + +## Overview + +Zakura is Zebra's **second-generation peer-to-peer transport (P2P v2)**. It runs over +**iroh QUIC** alongside the legacy Zcash TCP gossip network, and carries the native +header-sync, block-sync, discovery, and legacy-gossip-compatibility services. This +directory specifies the parts of the stack that are common to every service — how peers +**connect**, how they **find each other**, and how they learn **which services** each +other speak. + +Three properties frame the whole design: + +- **Bounded** — every queue, allocation, loop, and wait over attacker-influenced data has + an explicit cap or deadline. A peer cannot make this node allocate, spin, or block + without limit. +- **Authenticated and scoped** — a peer is its iroh node id (an Ed25519 public key). Every + connection is scoped to a network and a chain (genesis hash), so mainnet, testnet, and + private dev cohorts never cross-connect. +- **Direct-only** — iroh relays and side-channel discovery are disabled + (`direct_endpoint_builder`); Zakura dials direct QUIC paths so that every peer has a real + source IP to key per-IP admission on. + +The three specs: + +- **[connections.md](connections.md)** — the transport: connection lifecycle, the per-peer + registry and duplicate resolution (the **generation guard**), framing, streams, + timeouts, backpressure, rate limiting, and identity. +- **[peer_discovery.md](peer_discovery.md)** — how a node learns peer addresses (bootstrap, + gossip, self-records), the signed-record discovery protocol, candidate selection, + dialing, redial/backoff, and target connection counts. +- **[service_discovery.md](service_discovery.md)** — how connected peers negotiate + capabilities, how a stream is tagged to a service, and how each service (block-sync, + header-sync, discovery, legacy gossip) acquires and admits peers. + +## Architecture + +```text + ┌───────────────────────────────────────────┐ + │ iroh QUIC endpoint │ + │ ALPN p2p-v2/1 · relays & discovery off │ + └───────────────────────────────────────────┘ + dial (outbound) ─────────────►│ │◄───────── accept (inbound) + ▼ ▼ + control handshake (hello / ack): version · path · role · + network_id · chain_id · node id · nonces · capabilities · limits + │ + ▼ + ┌─────────────────────────────┐ register(conn_id, peer_id, ip, …) + │ ZakuraSupervisor registry │──► per-peer entry keyed by node id, + │ dedup · per-IP cap · gen │ guarded by monotonic conn generation + └─────────────────────────────┘ + │ fan out ordered / request streams by capability + ┌────────────────────┼────────────────────┬────────────────────┐ + ▼ ▼ ▼ ▼ + legacy gossip discovery header-sync block-sync + (kinds 2,3) (kind 4) (kind 5) (kind 6) + │ │ + │ └── learns peer addresses, ranks dial candidates ──┐ + │ ▼ + └────────────────────────────────────────────────► outbound dialer / redial +``` + +## Shared vocabulary + +Three **independent namespaces** identify "a service"; the spec keeps them distinct: + +1. **Stream kind** (`u16`) — tags one QUIC stream to the service that owns it. Carried in + every stream's `StreamPrelude`. Routes bytes. +2. **Capability bit** (`u64` bitmask) — negotiated once in the control handshake; gates + _which stream kinds may be opened_ on a connection. Authorises. +3. **Service id** (`ZakuraServiceId`, bounded ASCII, e.g. `zakura.block_sync.v1`) — + advertised in signed discovery records; used only to decide _who to dial_. Never routes + bytes. + +A single **ALPN** (`P2P_V2_ALPN = b"p2p-v2/1"`) gates the whole connection — there is no +per-service ALPN in production. + +### Stream kinds + +| Kind | Constant | Mode | Service | Version | Capability bit | +| --- | --- | --- | --- | --- | --- | +| 0 | — (`control`) | — | control handshake (trace label only) | — | — | +| 1 | — (`request`) | — | request/response (trace label only) | — | — | +| 2 | `ZAKURA_STREAM_GOSSIP` | Ordered | legacy gossip | 1 | `ZAKURA_CAP_LEGACY_GOSSIP` | +| 3 | `ZAKURA_STREAM_LEGACY_REQUESTS` | RequestResponse | legacy inventory req/resp | 1 | `ZAKURA_CAP_LEGACY_GOSSIP` | +| 4 | `ZAKURA_STREAM_DISCOVERY` | Ordered | native discovery | 1 | `ZAKURA_CAP_DISCOVERY` | +| 5 | `ZAKURA_STREAM_HEADER_SYNC` | Ordered | native header sync | 5 | `ZAKURA_CAP_HEADER_SYNC` | +| 6 | `ZAKURA_STREAM_BLOCK_SYNC` | Ordered | native block sync | 2 | `ZAKURA_CAP_BLOCK_SYNC` | + +Kinds 0 and 1 are trace/metric labels only, not registered streams. Kinds 2 and 3 share +one capability bit. Comments that say "stream-6" / "kind-6" always mean block-sync. + +### Capability bits + +| Constant | Value | Gates stream kind(s) | +| --- | --- | --- | +| `ZAKURA_CAP_LEGACY_GOSSIP` | `1 << 0` | 2, 3 | +| `ZAKURA_CAP_HEADER_SYNC` | `1 << 1` | 5 | +| `ZAKURA_CAP_DISCOVERY` | `1 << 2` | 4 | +| `ZAKURA_CAP_BLOCK_SYNC` | `1 << 3` | 6 | + +A node advertises exactly the OR of the capability bits of the services it actually runs +(`registry.supported_capabilities()`); it cannot claim a capability whose service is not +registered. + +### Network and chain scoping + +Every handshake and every signed discovery record carries `network_id` (`ZakuraNetworkId`: +Mainnet=1, Testnet=2, Regtest=3, Configured=4) and `chain_id` (the network's genesis hash). +A mismatch is rejected (`WrongNetwork` / `WrongChain`). A private dev cohort +(`ZakuraConfig::dev_network` tag) advertises `Configured` and a `chain_id` derived from the +genesis and the tag, so cohorts and the public network reject each other **without changing +consensus** (same genesis, magic, and activation heights). See +[`private-zakura-network.md`](../../../book/src/dev/private-zakura-network.md). + +### Wire magics + +`P2P_V2_ALPN = b"p2p-v2/1"` · `PRELUDE_MAGIC = b"ZAKURA1\0"` (legacy-upgrade prelude) · +`CONTROL_HELLO_MAGIC = b"ZAKCTRL\0"` · `CONTROL_ACK_MAGIC = b"ZAKACK\0\0"` · +`STREAM_PRELUDE_MAGIC = b"ZKST"` · signed-record domain +`ZAKURA_NODE_RECORD_SIG_DOMAIN = b"zakura-node-record-v1"`. + +All wire codecs are little-endian, hard-capped, and reject trailing bytes. + +## Status + +Pinned iroh version: `IROH_VERSION = "0.92.0"`. This spec reflects the transport as of the +generation-guard connection-registration work (Zakura WS-1) and the native per-IP +connection default (Zakura WS-4, default 3). Behavior gated behind future work streams +(e.g. WS-5 block-sync parking, see `docs/plans/zakura-ws5.md`) is out of scope and called +out where relevant. diff --git a/docs/specs/p2p/connections.md b/docs/specs/p2p/connections.md new file mode 100644 index 00000000000..ef9ddf73786 --- /dev/null +++ b/docs/specs/p2p/connections.md @@ -0,0 +1,352 @@ +# Zakura P2P Connections / Transport — Specification + +## Overview + +The transport carries every Zakura service over **iroh QUIC**. A connection is opened by a +**dial** (outbound) or an **accept** (inbound), authenticated by the peer's iroh node id, +negotiated by a **control handshake**, and then **registered** in a single per-peer +registry before any service sees it. One `ZakuraProtocolHandler` owns the endpoint; one +`ZakuraSupervisor` owns the registry; one `ServiceRegistry` fans a registered peer out to +the services it negotiated. + +The transport's job is to turn an unauthenticated, attacker-reachable QUIC socket into a +small set of authenticated, deduplicated, rate-limited, bounded peer sessions. It balances +three goals: **liveness** (a genuinely dead or wedged peer is reaped promptly and its slot +reused), **stability** (a healthy long-lived peer is never churned by a redial race or a +late teardown), and **bounded cost** (connections, streams, handshakes, frames, and queues +are all capped, so no peer can exhaust local resources). + +The subtle invariant that ties liveness and stability together is the **generation guard** +(Zakura WS-1): a per-peer registry entry is keyed by node id _and_ a monotonic connection +generation, so the teardown of a superseded connection can never delete its live +successor. + +## Glossary + +Plain term (code identifier): + +- **Endpoint** (`ZakuraEndpoint` / `ZakuraProtocolHandler`) — the single iroh QUIC endpoint + and its ALPN protocol handler. +- **Peer identity** (`ZakuraPeerId`) — the authenticated iroh node id (Ed25519 public key), + bounded to `MAX_IROH_NODE_ID_BYTES` (128). +- **Connection generation** (`ZakuraConnId`, a `u64`) — a monotonically increasing id + stamped on every accepted/dialed connection from one `next_conn_id` counter. +- **Registry / supervisor** (`ZakuraSupervisor`, `active_by_peer`) — the one map from peer + identity to its current live connection entry (`ZakuraPeerConnectionEntry`). +- **Transcript hash** (`transcript_hash`, `TRANSCRIPT_HASH_BYTES` = 32) — a deterministic + per-connection value (derived from the initiator's node id) that both ends of a + simultaneous-open race compute identically; the duplicate-resolution seed. +- **Control handshake** (`ZakuraControlHello` / `ZakuraControlAck`) — the version / + identity / capability / limits exchange on the first bidirectional stream. +- **Negotiated limits** (`ZakuraConnectionLimits`, from `ZakuraLocalLimits::clamp`) — the + per-connection frame/message/stream/idle ceilings, each the min of local and peer. +- **Frame** (`Frame { message_type, flags, payload }`) — the length-prefixed unit on a + stream; header `FRAME_HEADER_BYTES` = 8. +- **Stream kind / mode** (`StreamPrelude.stream_kind`, `StreamMode::{Ordered, RequestResponse}`) + — which service a stream belongs to, and whether it is long-lived or per-request. +- **Freshness reaper** (`freshness_reaper`, `freshness_tx`) — the app-level idle reaper, + bumped on any inbound frame. +- **Close cause** (`CloseCause`) — first-writer-wins attribution of why a connection closed. + +## Connection lifecycle + +**MUST:** + +- Every connection MUST take a distinct generation `conn_id = next_conn_id.fetch_add(1)` + before anything else, on both the accept and dial paths. +- Every connection MUST hold a **global admission** permit (`admission`, a + `Semaphore(max_connections)`) for its whole life. Accept without a permit closes with + `ZAKURA_CLOSE_RESOURCE`; dial without one fails `ResourceLimit("admission")`. +- Every connection MUST hold a **pending-handshake** permit + (`Semaphore(max_pending_handshakes)`) for the duration of the control handshake only, so + the number of concurrent un-negotiated handshakes is bounded. +- The peer identity MUST be the QUIC-authenticated `connection.remote_node_id()`, read + before the handshake, and the handshake MUST bind the claimed id to it (see + [Security](#security-identity-and-authentication)). +- A connection becomes **active** only when `register_and_serve` calls + `supervisor.register(conn_id, peer_id, remote_ip, transcript_hash, …)` and it returns + `Registered`. Registration MUST precede service fan-out; a rejected registration MUST + close the connection without ever handing it to a service. +- Teardown MUST cancel the connection's `disconnect_token`, drain stream workers (bounded + by `STREAM_WORKER_DRAIN_TIMEOUT`, 1 s) then abort them, and call — in order — + `registry.remove_peer(peer_id, conn_id, …)` and `supervisor.deregister(peer_id, conn_id)`, + **both carrying this connection's generation** (see [generation guard](#registry-deduplication-and-the-generation-guard)). + +**Handshake sequence.** + +- The control handshake runs on the **first bidirectional QUIC stream**. The initiator + `open_bi` and sends `ZakuraControlHello`; the responder `accept_bi` (under + `control_timeout`), validates, and replies `ZakuraControlAck`. +- `ZakuraControlHello` carries: magic, `control_version` (`CONTROL_VERSION` = 1), selected + protocol, `handshake_path` (`Native` for a direct dial, `Upgraded` for a legacy TCP→QUIC + hand-off), `role`, `network_id`, `chain_id`, the sender's `iroh_node_id`, a random 32-byte + `peer_nonce`, the upgrade nonces + transcript (all-zero on `Native`), `capabilities`, and + `initial_limits` (`ZakuraLimits`). +- `ZakuraControlAck` echoes the nonces and returns `accepted_capabilities`, + `accepted_channels`, and `accepted_limits`. The initiator MUST validate the nonce + round-trip and the returned limits before serving. +- The negotiated per-connection limits are `ZakuraLocalLimits::clamp(&peer_limits)` — each + ceiling is the min of the local hard limit and the peer's advertised value. + +**SHOULD:** + +- The endpoint SHOULD bind loopback-only when `listen_addr` is unset, so the experimental + `p2p-v2/1` ALPN surface is not exposed on all interfaces for a dial-only node. +- On the inbound path the remote IP SHOULD be recovered from the connection's confirmed + direct path (iroh's router consumes the address before `accept`); a relay-only path + yields no attributable IP and falls back to the global cap alone. + +## Registry, deduplication, and the generation guard + +The registry (`ZakuraSupervisorState`) holds one entry per peer +(`active_by_peer: HashMap`), a per-IP count +(`active_by_ip: HashMap`), and a tiny authenticated-peer transcript map. An +entry is `{ conn_id, outbound_handle, disconnect_token, registered_at, remote_ip }`. + +**MUST — single live connection per peer.** + +- At most one connection per peer identity MUST be `active_by_peer` at a time. A second + connection to the same peer MUST be resolved deterministically, never left as a silent + duplicate. +- Resolution uses the **transcript-hash tiebreak**: in `register_authenticated`, if an + incumbent hash exists and `incumbent_hash <= new_hash` the incumbent wins (`Duplicate`), + otherwise the newcomer replaces it (`Upgraded`). Ties keep the incumbent. Because the + transcript hash is derived from the **initiator's node id** + (`native_connection_transcript_hash`), both ends of a simultaneous open compute the same + winner and converge. +- The IP accounting invariant `sum(active_by_ip) == count(active_by_peer with Some(ip))` + MUST hold (checked by `debug_assert_accounting`). Duplicate eviction MUST cancel exactly + the **incumbent's** `disconnect_token` (never the newcomer's) and adjust `active_by_ip` + symmetrically. + +**MUST — the generation guard (Zakura WS-1).** + +A registry entry is addressed by `(peer_id, conn_id)`, not `peer_id` alone. When a +duplicate replaces an incumbent (the `Upgraded` path) or a peer restarts and redials, the +_old_ connection's serve loop is still unwinding and will call +`deregister(peer_id, old_conn_id)` and `remove_peer(peer_id, old_conn_id)`. Without the +guard, that late teardown would delete the **new** connection's entry and tear down its +healthy service sessions. + +- `deregister` and `remove_peer` MUST be **no-ops unless the caller's `conn_id` equals the + currently-registered generation**: `if entry.conn_id != conn_id { return }`. +- Every layer MUST thread the generation: `Peer.conn_id` carries it into each service + session; `Service::remove_peer(&self, peer, conn_id)` and + `ServiceRegistry::remove_peer(peer, conn_id, negotiated)` compare it (e.g. block-sync + removes only if `record.conn_id == conn_id`; discovery keys its admitted-peer entry by + `conn_id`). +- A superseded connection's teardown MUST NOT decrement the peer's IP count, cancel the + successor's token, or deregister the successor's transcript. + +**SHOULD — reclaim stale slots fast without flapping.** + +- On the `Upgraded` path the replaced incumbent's token SHOULD be cancelled immediately so + its slot frees in milliseconds. +- On the `Duplicate` path (incumbent wins) the newcomer is closed neutrally + (`b"duplicate"`) and relies on its redial; but if the incumbent has been registered for + at least `ZAKURA_DUPLICATE_EVICT_MIN_AGE` (300 s) and its token is not already cancelled, + the incumbent SHOULD be cancelled so a restarted peer reclaims the slot in milliseconds + rather than waiting the ~150 s QUIC idle timeout. A younger incumbent is kept, so a redial + race does not flap a fresh healthy connection. + +## Framing and streams + +**MUST:** + +- A `Frame` is an 8-byte header (`message_type: u16`, `flags: u16`, `payload_len: u32`, all + LE) followed by `payload_len` bytes. Encode MUST reject a payload larger than + `max_frame_bytes - FRAME_HEADER_BYTES`; decode MUST reject an oversize length **before + allocating** the payload and MUST reject trailing bytes. +- The streaming reader (`read_frame`) MUST enforce `frame_len <= max_frame_bytes` before + allocation. The receiver's cap is the _authoritative_ one: + `inbound_frame_cap_for_stream_kind` clamps to `max_message_bytes + header` so a peer that + negotiated a large `max_frame_bytes` still cannot force a large allocation. +- Every stream opens with a `StreamPrelude` (magic `ZKST`, `stream_kind`, `stream_version`, + optional `request_id`, `max_frame_bytes`). The prelude read MUST be bounded by + `prelude_timeout` (3 s). An `Ordered` stream MUST NOT carry a `request_id`; a + `RequestResponse` stream MUST carry one — a violation closes the connection with + `ZAKURA_CLOSE_BAD_PRELUDE`. +- A stream whose `(stream_kind, stream_version)` is not a registered pair MUST be reset with + `ZAKURA_CLOSE_UNKNOWN_STREAM` (this is how a superseded protocol version — e.g. + header-sync v1 after the v5 break — is refused). +- All Zakura streams MUST be **bidirectional**: the QUIC transport config sets + `max_concurrent_uni_streams = 0` and disables datagrams. + +**Stream ↔ service mapping.** + +- Each `Stream { kind, version, frame_cap, capability, mode }` MUST map to exactly one + service and exactly one **single-bit** capability. The registry build MUST reject a + non-single-bit capability (`InvalidCapability`) and two services claiming one kind + (`DuplicateKind`). +- **Ordered** streams are one long-lived stream per protocol per peer; **RequestResponse** + streams are one short-lived stream per request. The `read_frame` reader is **not + cancellation-safe** once the first header byte is consumed, so a persistent ordered + stream MUST run its reader in a dedicated task rather than as a `select!` branch, so an + outbound-write branch can never drop a mid-read future and desync the stream. + +**SHOULD:** the dialer (initiator) SHOULD proactively open all the ordered streams it +demands; the responder SHOULD additionally open only block-sync (the sole symmetric +service). See [service_discovery.md](service_discovery.md) for the symmetric-collision +tiebreak. + +## Timeouts, keepalive, and backpressure + +**MUST — no unbounded wait, no idle survivor.** + +- The endpoint MUST fail to build unless `keep_alive_interval < quic_idle_timeout` and the + advertised initial idle timeout `< quic_idle_timeout` (`validate_idle_invariant`), so + keepalive (10 s) always fires well within the idle timeout (150 s). +- The negotiated idle timeout MUST be clamped to `quic_idle_timeout - 1ms` and floored at + 1 ms, and to at most `LOCAL_MAX_IDLE_TIMEOUT_MILLIS` (10 min) from a peer. +- An in-progress frame read MUST be bounded by `read_timeout = idle_timeout`. The + **first-byte** wait is `None` for persistent ordered streams (legitimately quiet between + frames — inter-frame silence MUST NOT cancel the connection) and `Some(idle_timeout)` for + request streams. +- The **freshness reaper** MUST close a connection (`b"idle"`, cause `idle_timeout`) after + `idle_timeout` with no inbound frame on any stream; every stream worker bumps + `freshness_tx` on any inbound frame. Genuinely dead connections are additionally caught by + the QUIC idle timeout. + +**MUST — bounded queues, natural backpressure.** + +- Every inbound and outbound path MUST be a bounded `mpsc` channel. The per-connection + inbound depth is split across its streams + (`per_stream_inbound_queue_depth = (depth / count).max(1)`). +- When a service's inbound queue is full the stream worker MUST **await** the send rather + than drop — this stalls QUIC flow control (the 32 MiB receive windows) and applies + backpressure to the peer. +- Outbound work MUST be non-blocking from the service's view: `try_send_frame` returns + `Full`/`Closed` and the caller sheds load; `outbound_peer_handles()` MUST only return + handles with free capacity, so a peer that stopped reading is skipped for new work rather + than blocking others. +- When a peer stops reading, a `Stopped` write MUST close only that stream; any other write + error resets the stream and cancels the connection. + +## Rate limiting and DoS resistance + +**MUST — every admission surface is bounded:** + +- **Global connections** — `Semaphore(max_connections)`, default 256, on both accept and + dial. +- **Per-IP connections** — `max_connections_per_ip`, default 3 (Zakura WS-4; + `DEFAULT_ZAKURA_MAX_CONNS_PER_IP`, coerced back to 3 if configured 0), enforced in + `register` and pre-checked by the dialer via `can_accept_remote_ip_with_in_flight`. A + same-`(peer_id, ip)` re-registration is exempt (a duplicate redial, not a new host). + Relay-only inbound (no attributable IP) falls back to the global cap only. +- **Pending handshakes** — `Semaphore(max_pending_handshakes)`, default 32. +- **Stream-open rate** — a per-connection `TokenBucket(stream_open_rate_per_second)`, + default 32/s. The token MUST be charged **before** parsing the prelude, so + protocol-invalid stream churn (bad prelude, unknown kind, unnegotiated capability) still + spends budget. Over-rate resets that stream with `ZAKURA_CLOSE_RATE_LIMIT` (connection + kept). +- **Message rate** — one shared `TokenBucket(message_rate_per_second)` per stream-kind per + connection (default 2048/s), so N same-kind streams draw one budget. An oversize message + MUST disconnect (`ZAKURA_CLOSE_OVERSIZE`); a throttled **ordered** frame MUST disconnect + (`ZAKURA_CLOSE_RATE_LIMIT`, because dropping a solicited ordered frame is a permanent + gap); a throttled **request** frame is intentionally not rejected. + +**MUST — bounded blast radius and allocation:** + +- Each peer's work MUST run under a panic-containing supervised task + (`spawn_supervised_pipe` / `spawn_supervised_peer_task`) so a panicking or hostile peer is + contained to itself; the build MUST refuse `panic = "abort"` where containment cannot + work. +- Frame, message, control (16 KiB), and retained request/response + (`LEGACY_RESPONSE_MAX_AGGREGATE_BYTES`) sizes MUST all be capped, enforced before + allocation. + +## Security, identity, and authentication + +**MUST:** + +- The peer identity MUST be the QUIC/TLS-authenticated iroh node id + (`connection.remote_node_id()`). `ZakuraControlHello::validate` MUST reject a hello whose + claimed `iroh_node_id` does not equal the authenticated id (`IdentityMismatch`). +- The handshake MUST enforce magic, `control_version`, selected protocol, + `handshake_path`, `role`, `network_id`, and `chain_id`; on the `Native` path all + upgrade-nonce and transcript fields MUST be zero. +- The endpoint MUST disable iroh relays and side-channel discovery + (`direct_endpoint_builder`), so every peer has a direct source IP for per-IP admission. +- The node's own secret key MUST be stable across restarts (configured or persisted) so its + node id is consistent. +- Every wire decoder MUST use hard caps and reject trailing bytes + (`MAX_PRELUDE_PAYLOAD_BYTES` 4 KiB, `MAX_CONTROL_PAYLOAD_BYTES` 16 KiB, + `MAX_IROH_NODE_ID_BYTES` 128, `MAX_IROH_DIRECT_ADDRESSES` 8, etc.). + +**SHOULD:** the legacy TCP→QUIC upgrade path SHOULD bind a Blake2b transcript over both +legacy version messages plus the upgrade init/accept, so the hand-off cannot be spliced; +the `Native` path skips the transcript (there is no prior TCP leg to bind). + +## Edge cases and bounds + +**A superseded connection tears down its successor.** The classic use-after-replace: peer +restarts, its new connection registers, then the old serve loop unwinds and deregisters. +The **generation guard** makes the old teardown a no-op because +`old_conn_id != current_conn_id`. Test coverage: +`upgraded_winner_registers_second_loser_cleanup_is_generation_guarded` proves the winner +stays registered with correct IP accounting after the loser's late deregister. + +**A simultaneous open races both directions.** Both ends dial each other. The transcript +hash is derived from the initiator's node id, so both compute the same winner; the loser is +closed neutrally and does not churn. Block-sync additionally opens symmetrically and +resolves its stream collision by the node-id tiebreak (see service_discovery.md), keeping +the connection. + +**A restarted peer's slot is pinned for ~150 s.** A peer that restarts reconnects from a +fresh ephemeral path; the incumbent entry would otherwise survive until the QUIC idle +timeout. `ZAKURA_DUPLICATE_EVICT_MIN_AGE` (300 s) lets the registry cancel a _stale_ +incumbent immediately while keeping a _young_ one, trading flap-resistance against reclaim +latency. + +**A peer stops reading and holds our outbound full.** QUIC flow control (32 MiB windows) +plus the bounded app queue apply backpressure; `outbound_peer_handles()` skips the full +peer for new work; the freshness reaper and QUIC idle timeout eventually reap it. No other +peer is blocked. + +**A peer opens invalid streams in a loop.** The stream-open token is charged before parsing, +so churn is rate-limited regardless of validity; unknown/unnegotiated/oversize streams are +reset with the appropriate close code, and a mode/request-id violation cancels the whole +connection. + +**A peer advertises huge limits.** Negotiated limits are the min of local and peer, floored +where needed; the receiver's frame cap is clamped to `max_message_bytes + header` +independent of the negotiated `max_frame_bytes`, so a large advertised frame size cannot +buy a large allocation. + +**Numeric safety (MUST).** All rate refills, byte budgets, and limit clamps MUST saturate or +be checked; token buckets use nanosecond-precision saturating refill; every `.max(1)` +floor prevents a zero limit from wedging a stream. + +**Observability (SHOULD).** The transport SHOULD emit connection counters +(`zakura.p2p.conn.{accepted,active,closed.*,rejected.*,duplicate.*}`), queue depth +(`zakura.p2p.queue.depth`), and a first-writer-wins `CloseCause` on every close, so a trace +can attribute why any connection ended. + +## Defaults + +| Knob | Default | Meaning | +| --- | --- | --- | +| `IROH_VERSION` | `0.92.0` | pinned iroh (QUIC) version | +| `P2P_V2_ALPN` | `p2p-v2/1` | single whole-connection ALPN | +| `DEFAULT_ZAKURA_LISTEN_ADDR` | `0.0.0.0:8234` | native endpoint bind (loopback-only if `None`) | +| `max_connections` | 256 | global connection cap (inbound + outbound) | +| `max_connections_per_ip` | 3 | per-source-IP cap (WS-4; 0 ⇒ 3) | +| `max_pending_handshakes` | 32 | concurrent control handshakes | +| `stream_open_rate_per_second` | 32 | per-connection stream-open token rate | +| `message_rate_per_second` | 2048 | per-stream-kind message token rate | +| `DEFAULT_ZAKURA_QUIC_IDLE_TIMEOUT` | 150 s | QUIC idle + app idle-reaper bound | +| `DEFAULT_ZAKURA_KEEP_ALIVE_INTERVAL` | 10 s | QUIC keepalive | +| `DEFAULT_ZAKURA_PRELUDE_TIMEOUT` | 3 s | stream prelude read deadline | +| `DEFAULT_ZAKURA_CONTROL_TIMEOUT` | 10 s | control read/write + dial connect deadline | +| `DEFAULT_ZAKURA_*_WINDOW` | 32 MiB | QUIC stream/connection send/receive windows | +| `ZAKURA_DUPLICATE_EVICT_MIN_AGE` | 300 s | min incumbent age before stale-eviction on redial | +| `STREAM_WORKER_DRAIN_TIMEOUT` | 1 s | worker drain before abort on teardown | +| `OUTBOUND_STREAM_WRITE_TIMEOUT` | 10 s | ordered-frame write deadline | +| `OUTBOUND_REQUEST_RESPONSE_TIMEOUT` | 30 s | request/response round-trip deadline | +| `FRAME_HEADER_BYTES` | 8 | frame header (type + flags + len) | +| `MAX_CONTROL_PAYLOAD_BYTES` | 16 KiB | control payload hard cap | +| `LOCAL_MAX_CONTROL_FRAME_BYTES` / `LOCAL_MAX_MESSAGE_BYTES` | 1 MiB / 4 MiB | control frame / message ceilings | +| `LOCAL_MAX_OPEN_STREAMS` / `LOCAL_MAX_INBOUND_QUEUE_DEPTH` | 1024 / 4096 | per-connection stream / queue ceilings | +| `LOCAL_MAX_IDLE_TIMEOUT_MILLIS` | 600000 | largest peer-advertisable idle timeout | +| `DEFAULT_ZAKURA_REDIAL_INITIAL_BACKOFF` / `_MAX_BACKOFF` | 1 s / 30 s | supervised dial backoff (see peer_discovery.md) | +| close codes | `NEUTRAL=0 RESOURCE=1 BAD_PRELUDE=2 RATE_LIMIT=3 OVERSIZE=4 UNKNOWN_STREAM=5` | bounded QUIC reset codes | diff --git a/docs/specs/p2p/peer_discovery.md b/docs/specs/p2p/peer_discovery.md new file mode 100644 index 00000000000..d83dfabe4a2 --- /dev/null +++ b/docs/specs/p2p/peer_discovery.md @@ -0,0 +1,272 @@ +# Zakura P2P Peer Discovery — Specification + +## Overview + +Peer discovery is how a node learns **which peers exist and how to reach them**, and then +decides **which to dial**. A peer is its iroh node id (an Ed25519 public key); reaching it +means a `NodeAddr` = node id + a bounded set of direct socket addresses. Discovery keeps an +in-memory **address book** of signed peer records, fills it from four sources (operator +bootstrap, gossip responses, connected peers' self-records, and a persisted cache), and +runs a dialer that turns eligible book entries into connections up to a target count. + +Discovery runs as its own ordered service (stream kind 4) on every connection: two peers +exchange self-records and peer samples over a long-lived stream. It balances **reach** +(keep enough diverse peers connected to sync and gossip), **safety** (never dial or gossip +an address that isn't provably a routable public host owned by a key that signed for it), +and **bounded cost** (every record, sample, dial, and retry is capped, and expensive +signature checks never run under the shared lock). + +Discovery is **advisory to dialing, authoritative to nothing else**: it decides who is +_worth_ dialing; the transport ([connections.md](connections.md)) decides whether a dial is +_admitted_, and each service ([service_discovery.md](service_discovery.md)) decides whether +a connected peer is _useful_. + +## Glossary + +Plain term (code identifier): + +- **Address book** (`ZakuraDiscoveryBook`, `ZakuraDiscoveryInner` behind + `Arc>`) — the in-memory store of learned peers. +- **Signed record** (`ZakuraNodeRecord` = `ZakuraNodeRecordBody` + Ed25519 signature) — a + peer's self-description: node id, direct addrs, advertised services, protocol range, + `network_id`, `chain_id`, monotonic `sequence`, expiry. +- **Book entry** (`ZakuraDiscoveryEntry`) — a stored record plus dial bookkeeping: + `source`, `last_seen`, `last_dial_attempt`, `last_success`, + `last_short_lived_exchange`, `failure_count`. +- **Static candidate** (`ZakuraStaticDiscoveryCandidate`) — an operator bootstrap peer + (trusted unsigned dial hint), stored separately and never gossiped. +- **Dial authority** (`entry_has_confirmed_dial_authority`) — a record is dialable only if + it is a static candidate or self-authored (the peer signed for its own address). +- **Live service summary** (`ServiceSummaryEnvelope`) — a short-TTL advisory hint about a + connected peer's per-service state, used only to rank dial/admission preference. +- **Dial slot limit** (`discovery_dial_slot_limit`) — how many new dials the candidate + dialer may start right now. +- **Redial policy** (`RedialPolicy`) — the supervised reconnect/backoff for bootstrap and + upgrade dials, distinct from the book's per-candidate dial backoff. + +## Sourcing peers + +**MUST — four sources, one book, strict separation:** + +1. **Operator bootstrap** — `DEFAULT_ZAKURA_BOOTSTRAP_PEERS` (`node_id@ip:port` seeds on + port 8234) parsed into **static candidates**. Static candidates MUST be stored apart from + signed records and MUST NOT appear in any peer sample returned to another node. +2. **Gossip** — a `Peers { records }` response, imported by `import_peer_records`. +3. **Connected-peer self-records** — a `Hello { record }` on the discovery stream, imported + by `import_connected_peer_record`. +4. **Persisted cache** — records loaded from disk are re-validated on import like any + untrusted record (cache-persistence wiring is staged; the type carries no filesystem + behavior yet). + +- There are **no DNS seeds** in the native path (DNS seeding belongs to the legacy address + book, not Zakura). +- A node's **own** record MUST NOT be stored in the book (`SelfRecord`). + +**Record import validation (MUST).** `validate_record_body_for_import` MUST reject a record +whose `network_id` or `chain_id` does not match this node (`WrongNetwork` / `WrongChain`), +whose protocol range does not overlap (`IncompatibleProtocol`), whose expiry is outside +`[now - skew, now + max_record_ttl + skew]` (`Expired` / `FarFutureExpiry`), or whose +Ed25519 signature does not verify against its own `node_id` (`InvalidSignature`). The +signature domain is `zakura-node-record-v1`. + +## The discovery protocol + +The service runs one long-lived ordered stream (kind 4) per peer. Messages +(`DiscoveryMessage`) are LE, length-prefixed, one leading `u8` tag, trailing-byte-rejecting, +and hard-capped at `MAX_DISCOVERY_MESSAGE_BYTES` (16 KiB): + +| Tag | Constant | Variant | Purpose | +| --- | --- | --- | --- | +| 1 | `MSG_DISCOVERY_HELLO` | `Hello { record }` | sender's signed self-record | +| 2 | `MSG_DISCOVERY_GET_PEERS` | `GetPeers { limit, wanted_services, exclude_node_ids }` | request a bounded peer sample | +| 3 | `MSG_DISCOVERY_PEERS` | `Peers { records }` | signed peer records (gossip response) | +| 4 | `MSG_DISCOVERY_GET_SERVICES` | `GetServices(…)` | request live service summaries | +| 5 | `MSG_DISCOVERY_SERVICES` | `Services(…)` | live per-service advisory summaries | + +**Per-peer exchange (SHOULD).** On admission a peer SHOULD send `Hello` (self-record), then +`GetPeers` (excluding self, connected peers, and recently-known records, bounded by +`MAX_DISCOVERY_EXCLUDED_NODE_IDS` = 256), then `GetServices`, and settle within +`DISCOVERY_EXCHANGE_SETTLE_TIMEOUT` (2 s). + +**Live summaries are advisory only (MUST).** A `Services` summary +(`HeaderSyncServiceSummary` / `BlockSyncServiceSummary` / `DiscoveryServiceSummary`, tagged +`SUMMARY_TAG_*_V1`) MUST be accepted only from the authenticated peer it describes, MUST be +TTL-clamped to `DEFAULT_LIVE_SERVICE_SUMMARY_TTL` (30 s), and MUST only influence dial / +admission _preference_ — never authorise a dial or bypass a validation. + +## Dialing and candidate selection + +Two dialer paths, both gated by transport admission capacity +(`endpoint.has_native_admission_capacity()`) and the per-IP cap: + +- **Bootstrap dialer** — one supervised task per configured bootstrap peer, using + `RedialPolicy::maintain` (redial forever). On a Zakura-only node this is the only healing + path. +- **Candidate dialer** (`run_native_discovery_dialer`) — the loop that dials discovered + peers. It wakes on endpoint shutdown, a worker finishing, a supervisor registration + change, or every `ZAKURA_DISCOVERY_DIAL_INTERVAL` (1 s). + +**MUST — bounded, deduplicated, per-IP-safe dialing:** + +- The dialer MUST NOT start a dial unless the transport admission semaphore has a free + permit and per-IP capacity remains, counting **in-flight** dials toward the per-IP cap + (`can_accept_discovery_dial_ip` + `in_flight_by_ip`) so concurrent dials cannot overshoot + `max_connections_per_ip`. +- A node id already **in-flight** MUST NOT be dialed again concurrently + (`in_flight: HashSet`). +- The number of new dials MUST be bounded by `discovery_dial_slot_limit`: + + ```text + soft_cap = max_zakura_connections − connection_headroom + available_connection_slots = soft_cap − connected_count + available_dial_slots = max_concurrent_dials − in_flight_count + limit = min(available_connection_slots, available_dial_slots) // saturating + ``` + + If `limit == 0`, dialing stops. So the dialer targets a **soft cap** below the global + connection ceiling, reserving `connection_headroom` slots (raised to + `max(4, bootstrap_count)` in production so bootstrap dials are never starved by discovered + peers), and never runs more than `max_concurrent_discovery_dials` (4) dials at once. + +**Dial success is registration, not dial completion (MUST).** A dial "succeeds" only when +the peer appears in the supervisor's registration watch, not when the dial future returns. +Outcomes (`DiscoveryDialResult`): `Registered` (stayed registered ≥ +`ZAKURA_REDIAL_HEALTHY_CONNECTION`, 60 s ⇒ `mark_dial_success`), `ShortLivedRegistered` +(registered then dropped early ⇒ `mark_short_lived_exchange`), `Failed` +(⇒ `mark_dial_failure`), `LocalResourceLimit` (our own cap ⇒ no penalty on the peer). + +## Redial and backoff + +Two distinct backoffs — do not conflate them: + +**Supervised connection-level redial (`RedialPolicy`)** — for bootstrap and legacy-upgrade +dials. + +- Exponential doubling capped at `max_backoff`: `backoff = (backoff × 2).min(max_backoff)`, + from `DEFAULT_ZAKURA_REDIAL_INITIAL_BACKOFF` (1 s) to `_MAX_BACKOFF` (30 s). +- A connection alive ≥ `ZAKURA_REDIAL_HEALTHY_CONNECTION` (60 s) counts as healthy and + resets the backoff to initial, so a long-lived peer is not penalised on its next redial. +- **Anti-churn (MUST):** before each dial, a peer already in the supervisor registration set + MUST be skipped (it may have dialed us first), so both directions do not churn duplicates. + A `maintain` policy never self-exits; teardown MUST be driven by the endpoint shutdown + token. + +**Book-side per-candidate dial backoff** — decides candidate _eligibility_. + +- `dial_backoff_secs(failure_count) = base × 2^min(failure_count−1, 10)`, capped at `max`, + from `DEFAULT_DISCOVERY_DIAL_BACKOFF_BASE` (60 s) to `_MAX` (1 h); `failure_count == 0` + ⇒ no backoff. +- A candidate is suppressed while `now < last_dial_attempt + backoff(failure_count)` + (`entry_in_dial_backoff`), and for one base interval (60 s) after a short-lived exchange + (`entry_in_short_lived_exchange_backoff`), so a peer that connects, does one exchange, and + drops is not re-dialed immediately. + +**No hard blacklist (MUST).** A dial failure only increments `failure_count`; a bad peer is +pruned by book eviction, never permanently banned. `HIGH_DIAL_FAILURE_COUNT` (3) marks a +candidate as a preferred eviction target. + +## Target connection counts + +Discovered-peer dialing targets `soft_cap = max_zakura_connections − connection_headroom`. +Production wires `max_zakura_connections` to the transport's `max_connections` (256) and +`connection_headroom` to `max(4, bootstrap_count)`. So a node keeps dialing discovered +peers until its connected count reaches the soft cap, then stops; separately, +per-direction admission into the discovery _service_ is capped by `ServicePeerLimits` +(`max_inbound_peers` / `max_outbound_peers`, default 256 each). Inbound and outbound are +counted and capped independently. + +## Candidate lifecycle and scoring + +**Lifecycle.** unknown → **stored** (`import_*`, monotonic `sequence` gate: `< stored` +ignored, `==` metadata-refresh with first-party source preferred over gossip, `>` replaced) +→ **dialable** (passes every eligibility gate below) → **dialing** (`mark_dial_attempt`) +→ **healthy** (`mark_dial_success`, resets `failure_count`) or **short-lived** +(60 s cooldown) or **failed** (`failure_count++`, exponential backoff) → **evicted**. + +**Eligibility (MUST all hold).** A book entry is a dial candidate only if it is: not already +connected, not in-flight, not local, not expired, not in dial-backoff, not in +short-lived-exchange backoff, has **confirmed dial authority** (static or self-authored), +advertises a wanted service, and has at least one usable direct address. + +**Scoring** (`dial_candidate_sort_key`, top-k selected in O(n) via `select_nth_unstable`, +not a full sort): prefer signed records over static, then most-recent `last_success`, then +fewest `failure_count`, then most-recently-seen, then a random tie-break (deterministic +node-id tie-break for static). + +**Eviction** (`evict_to_limits`, when `discovered_len > max_records`, never evicts a static +candidate): expired records first (soonest expiry), then never-successful high-failure +records (`last_success.is_none() && failure_count ≥ 3`), then the least-recently +successful/seen. + +## Edge cases and bounds + +**A signed record proves key ownership, not address ownership.** An attacker can sign a +record pointing at a victim's address. Gossiped (untrusted) records MUST advertise only +**globally routable** addresses: `is_discovery_dialable_addr` rejects port 0, unspecified, +loopback, multicast, broadcast, link-local, RFC 1918 private, RFC 6598 CGNAT (100.64/10), +and RFC 4193 unique-local. Static/operator records are looser (loopback allowed for +regtest). Combined with "dial authority = static or self-authored," a peer can only get you +to dial _its own_ routable address. + +**Expensive verification under a global lock is a DoS.** Ed25519 verification of imported +records MUST run **outside** the shared discovery mutex; only the cheap book mutation holds +the lock. The same rationale drives reservoir sampling in `sample_peers` and top-k +selection (not full sort) in `dial_candidates`, so a large record set or a flood of records +cannot pin the lock. + +**A summary or self-record impersonates another peer.** A `Hello`, `Services`, or live +summary that does not match the authenticated peer it describes MUST be rejected +(`MismatchedConnectedPeerRecord` / `MismatchedNodeId`). + +**A malicious legacy responder leaks maintained dials.** The legacy→Zakura upgrade dials a +peer-supplied node address under `RedialPolicy::maintain`. If the peer never registers +within the wait window and did not register from another connection, the dial MUST be +cancelled and its entry dropped, so repeated failed upgrades with distinct node ids cannot +leak unbounded maintained dials. + +**A pure-discovery peer holds a connection open.** If the discovery exchange completed and +no other service owns the peer, the connection SHOULD be dropped after the exchange (and the +peer put in the 60 s short-lived cooldown), rather than held idle. + +**Self-record sequence monotonicity is best-effort.** The self-record `sequence` is seeded +from a wall-clock nanosecond value; a backward clock step can regress it until cache +persistence lands. Peers apply the monotonic `sequence` gate on import, so a regressed +sequence is ignored rather than accepted as fresh. + +**Bounded everything (MUST).** Per record: `MAX_DIRECT_ADDRS_PER_RECORD` (8), +`MAX_SERVICES_PER_RECORD` (32), `MAX_NODE_RECORD_BODY_BYTES` (16 KiB). Per response: +`MAX_DISCOVERY_RECORDS_PER_RESPONSE` (32), `MAX_SERVICE_SUMMARIES_PER_RESPONSE` (32), +`MAX_SERVICE_SUMMARY_BYTES` (1 KiB). Book: `max_records` (`DEFAULT_MAX_DISCOVERY_BOOK_RECORDS`, +10 000). Sample exclusions: `MAX_DISCOVERY_EXCLUDED_NODE_IDS` (256). + +**Numeric safety (MUST).** Heights, sequences, and expiries decoded from records MUST be +range-checked (`InvalidHeight`, `NumericOverflow`); backoff shifts are capped (shift ≤ 10) +so the doubling cannot overflow. + +**Observability (SHOULD).** The dialer SHOULD emit +`zakura.p2p.discovery.dial.{started,succeeded,failed,short_lived_registered,local_resource_limit,worker_failed}` +so a trace can distinguish a healthy dial mix from a failing or churning one. + +## Defaults + +| Knob | Default | Meaning | +| --- | --- | --- | +| `DEFAULT_ZAKURA_BOOTSTRAP_PEERS` | 9 mainnet seeds (`node_id@ip:8234`) | operator static candidates | +| `record_ttl` (`DEFAULT_DISCOVERY_RECORD_TTL`) | 24 h | self-record advertised lifetime | +| `refresh_interval` (`DEFAULT_DISCOVERY_REFRESH_INTERVAL`) | 10 min | self-record re-publish cadence | +| `peer_sample_limit` (`DEFAULT_DISCOVERY_PEER_SAMPLE_LIMIT`) | 32 | max records returned per `GetPeers` | +| `dial_backoff_base` / `_max` | 60 s / 1 h | book per-candidate dial backoff | +| `max_concurrent_discovery_dials` | 4 | concurrent discovered-peer dials | +| `connection_headroom` (`DEFAULT_DISCOVERY_CONNECTION_HEADROOM`) | 4 (prod `max(4, bootstrap)`) | slots reserved below the soft cap | +| `max_zakura_connections` | 256 (prod override of the 32 default) | soft-cap basis for dialing | +| `max_record_ttl` | 24 h | reject records expiring beyond this | +| `clock_skew_tolerance` | 5 min | expiry tolerance on import | +| `peer_limits` (`ServicePeerLimits`) | in = out = 256 | discovery-service per-direction admission | +| `ZAKURA_DISCOVERY_DIAL_INTERVAL` | 1 s | candidate-dialer wake cadence | +| `ZAKURA_REDIAL_HEALTHY_CONNECTION` | 60 s | uptime that resets redial backoff | +| `DEFAULT_ZAKURA_REDIAL_INITIAL_BACKOFF` / `_MAX_BACKOFF` | 1 s / 30 s | supervised dial backoff | +| `DISCOVERY_EXCHANGE_SETTLE_TIMEOUT` | 2 s | per-peer exchange settle wait | +| `DEFAULT_LIVE_SERVICE_SUMMARY_TTL` | 30 s | live summary freshness bound | +| `HIGH_DIAL_FAILURE_COUNT` | 3 | failure count that flags eviction | +| `MAX_DISCOVERY_MESSAGE_BYTES` | 16 KiB | discovery frame hard cap | +| `max_records` (`DEFAULT_MAX_DISCOVERY_BOOK_RECORDS`) | 10 000 | address-book size cap | diff --git a/docs/specs/p2p/service_discovery.md b/docs/specs/p2p/service_discovery.md new file mode 100644 index 00000000000..6dcd81141be --- /dev/null +++ b/docs/specs/p2p/service_discovery.md @@ -0,0 +1,207 @@ +# Zakura P2P Service Discovery — Specification + +## Overview + +Service discovery is how two connected peers learn **which services each other speaks**, and +how a service (block-sync, header-sync, discovery, legacy gossip) **acquires the peers** that +offer it. It spans two moments: the **control handshake**, where the pair negotiate a +capability mask once for the connection's life, and **stream fan-out**, where each negotiated +service is handed a tagged QUIC stream and decides whether to admit the peer. + +The design keeps **three namespaces distinct** — conflating them is the classic bug: + +1. **Stream kind** (`u16`, in every `StreamPrelude`) — _routes_ a stream to its owning + service. +2. **Capability bit** (`u64`, negotiated in the handshake) — _authorises_ which stream kinds + may be opened. +3. **Service id** (`ZakuraServiceId`, in signed discovery records) — _ranks who to dial_; + never routes or authorises anything on a live connection. + +Capability advertisement is **static per node, symmetric, and negotiated by intersection**: a +node advertises exactly the services it runs, and the accepted set is the bitwise AND of the +two sides. There is no dynamic capability probing on a live connection — what a connection can +carry is fixed at handshake time. + +## Glossary + +Plain term (code identifier): + +- **Capability mask** (`capabilities`, `accepted_capabilities`) — the `u64` bitset a node + advertises / the pair agree on. +- **Supported capabilities** (`registry.supported_capabilities()`) — the OR of every + registered service's single capability bit; a node cannot advertise more. +- **Stream kind / version** (`StreamPrelude.stream_kind`, `.stream_version`) — the routing + tag and protocol version on a stream. +- **Service registry** (`ServiceRegistry`, `Service` trait) — the transport-side router that + maps a negotiated stream to its owning service and calls `add_peer` / `remove_peer`. +- **Peer session** (`Peer`, carrying `id`, `conn_id`, `direction`, and a + `HashMap`) — the per-connection handle a service claims streams from + via `take_stream(kind)`. +- **Demand hooks** (`Service::wants_peer`, `wants_ordered_stream`) — a service's advisory + "do I have room / want this?" check before a stream is opened. +- **Service id** (`ZakuraServiceId`, e.g. `zakura.block_sync.v1`) — the discovery-namespace + advertisement of a service, used only for candidate selection. +- **Symmetric service** — block-sync, the only ordered stream both peers open (so it can + collide and needs a tiebreak). + +## Capability negotiation + +**MUST:** + +- A node's advertised `capabilities` MUST equal `registry.supported_capabilities()` — the OR + of the single capability bit of each locally registered service. A node MUST NOT advertise + a capability whose service is not registered. +- The **initiator** sends its `capabilities` in `ZakuraControlHello`. The **responder** + computes `accepted_capabilities = hello.capabilities & local_supported`, returns it in + `ZakuraControlAck`, and the initiator MUST validate it. Negotiation is a pure + **intersection** — optional capabilities present on only one side are silently dropped. +- `required_capabilities` defaults to 0. A peer MUST reject the connection with + `MissingRequiredCapability` only if a _required_ bit is absent — an unmatched _optional_ + capability MUST NOT fail the handshake. +- Each declared `Stream.capability` MUST be exactly one non-zero bit + (`is_power_of_two`, else `InvalidCapability`), and each stream kind MUST be claimed by + exactly one service (`DuplicateKind`). These are enforced at registry build. + +**Authorisation to open a stream (MUST).** A peer may open stream kind _k_ iff the owning +service's capability bit is fully within `accepted_capabilities`. On the inbound side a stream +whose `stream.capability` is not a subset of `accepted_capabilities` MUST be reset with +`ZAKURA_CLOSE_UNKNOWN_STREAM`. So "may I open a block-sync stream?" is exactly +`accepted_capabilities & ZAKURA_CAP_BLOCK_SYNC == ZAKURA_CAP_BLOCK_SYNC`. + +**Discovery-record advertisement is independent (MUST).** A signed discovery record separately +advertises `services: Vec` (default: discovery, block_sync, header_sync, +legacy_gossip, legacy_requests, service_discovery). This is read by a _remote_ node to decide +the peer is worth dialing for a given service, **before** any handshake. It MUST NOT be treated +as authorisation to open a stream — the handshake capability mask is the only such authority. +Note `service_discovery` and `legacy_requests` have service ids but no dedicated capability bit +(legacy requests ride `ZAKURA_CAP_LEGACY_GOSSIP`). + +## Stream identification and routing + +Every stream (other than the control handshake) opens with a `StreamPrelude` carrying +`stream_kind` and `stream_version`. Inbound stream admission (`admit_bi_stream`) MUST, in +order: + +1. Acquire a stream-concurrency permit and charge one stream-open token **before** parsing — + so protocol-invalid churn still costs budget. +2. Read and validate the prelude magic (`ZKST`). +3. Look up `(stream_kind, stream_version)` in the registry; an unknown kind or unsupported + version MUST reset with `ZAKURA_CLOSE_UNKNOWN_STREAM`. (This is how header-sync v1 is + refused after the v5 break.) +4. Enforce the capability gate (above). +5. Enforce mode/request-id consistency: `RequestResponse` MUST carry a `request_id`, `Ordered` + MUST NOT — a violation cancels the whole connection (`ZAKURA_CLOSE_BAD_PRELUDE`). +6. Route: an accepted request stream dispatches to the owning service's + `RequestResponseService`; an accepted ordered stream is handed to the service via + `ServiceRegistry::add_escalated_peer` with just that stream. + +The canonical kind → service → capability table lives in the [README](README.md#stream-kinds). +Per-kind receiver frame caps (`app_frame_cap_for_stream_kind` / `inbound_frame_cap_for_stream_kind`) +clamp header-sync and block-sync to their message ceilings regardless of the negotiated frame +size. + +## How a service acquires peers + +Three layers cooperate; keep them separate. + +**(a) Transport fan-out (the primary mechanism).** When a connection finishes handshake, +`serve_connection` fans the peer out to every service whose capability is in +`accepted_capabilities`: + +- It computes the demanded ordered streams + (`registry.ordered_streams_for_negotiated(accepted_capabilities)`). +- The **initiator** opens all demanded ordered streams; the **responder** additionally opens + only block-sync (the sole symmetric service). +- Each opened stream routes to its owning service via `add_escalated_peer`, which calls + `Service::add_peer(Peer)` once per service, handing only that service's streams. The `Peer` + carries `conn_id` (the generation, see [connections.md](connections.md)) into the session. +- Before opening, the demand hooks `wants_peer` / `wants_ordered_stream` give a service an + advisory chance to decline (e.g. no free slots); `add_peer` is the **authoritative** + admission. + +**(b) Per-service admission.** Each service decides independently: + +- **block-sync** — admits iff the peer is not parked and a direction slot is free + (`peer_slots_free`); `add_peer` re-checks `can_admit_peer`, takes stream 6, and spawns the + per-peer routine. A block-sync peer is _usable for serving_ only after it has sent an + `MSG_BS_STATUS` (`has_received_status`). +- **header-sync** — admits iff a direction slot is free; takes stream 5 and spawns its sink. +- **discovery** — admits with a local-room check, takes stream 4, then runs the async + self-record/peer-sample exchange. + +Per-service admission is capped by `ServicePeerLimits` (`max_inbound_peers` / +`max_outbound_peers`, default 256 each), counted independently per direction, returning +`ServiceAdmissionDecision::{Admit, RejectFull, RejectNotUseful, RejectBackoff, RejectUnsupported}`. + +**(c) Discovery-driven candidate selection (who to dial).** Separately from admission, a sync +reactor learns _candidate_ peers to dial from discovery's service-aware API +(`service_candidates` / `header_sync_candidates` / `block_sync_candidates`), which returns +connected peers advertising the service (ranked by live-summary preference) plus dialable +records filtered by that `ZakuraServiceId`. The transport then dials, and fan-out in (a) +delivers the stream. Discovery shares connected-peer state via `watch` channels. + +## The symmetric block-sync collision + +**MUST.** Block-sync is the only service both peers open (stream kind 6), so a +simultaneous double-open is expected and MUST NOT tear down the connection. It is resolved by +the node-id tiebreak `i_open_collision_winner(local, remote) = local_id < remote_id`: the +winner keeps its own stream and parks the peer's duplicate; the loser adopts the peer's +stream. Because the comparison is mirror-stable, both ends agree. + +Any **other** peer-opened ordered stream arriving at an initiator (a service the initiator did +not expect the peer to open) MUST close the connection (`unexpected_stream`); a **duplicate** +accepted kind (a second stream of a kind already wired) MUST also close the connection. A +demand re-check (`wants_ordered_stream`) that finds no demand parks that one stream locally +(cancelling only its token) while keeping the connection. + +## Edge cases and bounds + +**Capability lies.** A node cannot advertise a capability it does not run — the advertised mask +is derived from the registry, not from config, so `supported_capabilities` is exactly the set +of registered services. A peer that opens a stream for an un-negotiated capability is reset, +not trusted. + +**Version skew.** A stream is admitted only for a declared `(kind, version)` pair. A breaking +protocol change bumps the stream version (header-sync 4→5, block-sync current 2), and old +versions are refused with `ZAKURA_CLOSE_UNKNOWN_STREAM` rather than mis-parsed. + +**Two streams, one capability.** Legacy gossip (kind 2, Ordered) and legacy requests (kind 3, +RequestResponse) share `ZAKURA_CAP_LEGACY_GOSSIP`. The single-bit-per-stream invariant still +holds (each stream declares the same one bit); the capability gate authorises both, and the +registry routes each kind to its handler. + +**Stale service teardown after a reconnect.** A service's `remove_peer` carries the connection +generation `conn_id`; a superseded connection's late `remove_peer` MUST be a no-op unless it +matches the currently-admitted generation, so it cannot evict the peer's live successor session +(the generation guard, [connections.md](connections.md)). + +**Discovery advertises a service the peer no longer runs.** A record's `services` list is a +hint; if a dialed peer's negotiated capability mask lacks the service, no stream of that kind +is opened and the service simply does not acquire the peer — the record is not authoritative. + +**Block-sync admits but the peer never sends status.** A block-sync peer counts against the +service cap on admission but is not _usable for serving_ until it sends `MSG_BS_STATUS`; a peer +that connects and stays silent is handled by block-sync's own liveness/park policy (see +`congestion_control.md`), not by service discovery. + +**Numeric / bounds safety (MUST).** Every wire field in the handshake and preludes is bounded +and trailing-byte-rejecting; capability and channel masks are plain `u64` intersections; +`ZakuraServiceId` is bounded ASCII (`MAX_ZAKURA_SERVICE_ID_BYTES` = 64, ≤ 32 per record). + +**Observability (SHOULD).** Stream admission and per-service `add_peer` / `remove_peer` SHOULD +be traced with the stream-kind label and direction, so a trace shows which services each peer +was admitted to and why any admission was rejected. + +## Defaults + +| Knob | Default | Meaning | +| --- | --- | --- | +| capability bits | `LEGACY_GOSSIP=1<<0 HEADER_SYNC=1<<1 DISCOVERY=1<<2 BLOCK_SYNC=1<<3` | per-service authorisation bits | +| `required_capabilities` | 0 | no capability is mandatory by default | +| stream versions | discovery 1 · header-sync 5 · block-sync 2 · legacy 1 | current `(kind, version)` pairs | +| `ServicePeerLimits.max_inbound_peers` / `max_outbound_peers` | 256 / 256 | per-service, per-direction admission cap | +| `DEFAULT_SERVICE_INBOUND_QUEUE_DEPTH` / `_OUTBOUND_` | 128 / 128 | reserved per-service queue depths (not yet enforced) | +| `DEFAULT_SERVICE_MAX_PENDING_ESCALATIONS` | 32 | reserved lazy-escalation bound (not yet enforced) | +| advertised service ids | discovery, block_sync, header_sync, legacy_gossip, legacy_requests, service_discovery | default signed-record `services` set | +| `MAX_ZAKURA_SERVICE_ID_BYTES` / `MAX_SERVICES_PER_RECORD` | 64 / 32 | service-id string / count bounds | +| symmetric service | block-sync (kind 6) | only ordered stream both peers open |