168 lines
7.6 KiB
Markdown
168 lines
7.6 KiB
Markdown
# ADR-0011: chronogrid_net — P2P Swarm Manager (Task 2.5)
|
||
|
||
**Status:** Accepted
|
||
**Date:** 2026-08-22
|
||
**Task:** 2.5 — Implement chronogrid_net: P2P Swarm Manager
|
||
**Author:** Autonomous Agent
|
||
|
||
---
|
||
|
||
## Context
|
||
|
||
Task 2.5 required implementing the top-level `SwarmManager` in `crates/chronogrid_net/`,
|
||
composing all protocol behaviours from Tasks 2.2–2.4 into a single running libp2p `Swarm`
|
||
with a clean public API for the application layer.
|
||
|
||
---
|
||
|
||
## Decision
|
||
|
||
### New module: `crates/chronogrid_net/src/swarm_manager.rs`
|
||
|
||
| Type | Role |
|
||
|-------------------------|-------------------------------------------------------------------------------|
|
||
| `ChronoGridBehaviour` | `#[derive(NetworkBehaviour)]` composite: mDNS + Kademlia + SyncProtocol + Gossipsub + Identify + Relay |
|
||
| `SwarmCommand` | Internal enum for `SwarmHandle → SwarmManager` control channel |
|
||
| `SwarmHandle` | Cloneable public API handle: `trigger_sync`, `broadcast_event`, `subscribe_events`, `shutdown` |
|
||
| `SwarmManager` | Owns the `Swarm<ChronoGridBehaviour>`, runs the event loop in a Tokio task |
|
||
|
||
### ChronoGridBehaviour composite
|
||
|
||
```rust
|
||
#[derive(NetworkBehaviour)]
|
||
pub struct ChronoGridBehaviour {
|
||
pub mdns: MdnsBehaviour,
|
||
pub kad: KadBehaviour<MemoryStore>,
|
||
pub sync: SyncProtocol,
|
||
pub gossip: GossipBehaviour,
|
||
pub identify: IdentifyBehaviour,
|
||
pub relay: relay::client::Behaviour,
|
||
}
|
||
```
|
||
|
||
The `#[derive(NetworkBehaviour)]` macro generates `ChronoGridBehaviourEvent` with one
|
||
variant per field. The `SyncProtocol` field generates `SyncProtocolEvent` with an
|
||
`Inner` variant wrapping the inner `request_response::Event<SyncRequest, SyncResponse>`,
|
||
accessed as `SyncProtocolEvent::Inner(inner_event)` in the event handler.
|
||
|
||
### SwarmBuilder chain (libp2p 0.54)
|
||
|
||
```rust
|
||
SwarmBuilder::with_existing_identity(keypair)
|
||
.with_tokio()
|
||
.with_tcp(tcp_config, noise::Config::new, yamux::Config::default)
|
||
.with_quic()
|
||
.with_relay_client(noise::Config::new, yamux::Config::default)
|
||
.with_behaviour(|_keys, relay_behaviour| Ok(ChronoGridBehaviour { ... }))
|
||
.with_swarm_config(|cfg| cfg.with_idle_connection_timeout(30s))
|
||
.build()
|
||
```
|
||
|
||
Key decisions:
|
||
- **Relay client**: In libp2p 0.54, `relay::client::new()` returns a raw transport
|
||
with `Output = Connection` (not `(PeerId, Muxer)`), so it cannot be used with
|
||
`with_other_transport()`. The correct API is `with_relay_client()` which handles
|
||
the transport upgrade internally and provides the `relay::client::Behaviour` to
|
||
the `with_behaviour()` closure.
|
||
- **TCP + QUIC**: Both transports are enabled; QUIC is preferred for low-latency
|
||
connections, TCP+Noise+Yamux is the fallback.
|
||
- **Idle timeout**: 30 seconds — balances connection churn vs. resource usage.
|
||
|
||
### SwarmHandle public API
|
||
|
||
```rust
|
||
impl SwarmHandle {
|
||
pub async fn trigger_sync(&self, clock: VectorClock) -> NetResult<()>;
|
||
pub async fn broadcast_event(&self, event: GossipEvent) -> NetResult<()>;
|
||
pub fn subscribe_events(&self) -> broadcast::Receiver<NetEvent>;
|
||
pub async fn shutdown(&self) -> NetResult<()>;
|
||
pub fn local_peer_id(&self) -> &PeerId;
|
||
}
|
||
```
|
||
|
||
- `trigger_sync()` sends a `SwarmCommand::TriggerSync` to the event loop, which
|
||
calls `SyncProtocol::send_sync_request()` for each connected peer.
|
||
- `broadcast_event()` sends `SwarmCommand::BroadcastGossip`, which calls
|
||
`GossipBehaviour::publish()` on the `chronogrid/state-changes/1` topic.
|
||
- `subscribe_events()` returns a `broadcast::Receiver<NetEvent>` — multiple
|
||
consumers can subscribe independently (broadcast semantics, 256-event buffer).
|
||
- `shutdown()` sends `SwarmCommand::Shutdown`, causing the event loop to exit.
|
||
- `SwarmHandle` is `Clone` — multiple components can hold a handle.
|
||
|
||
### Event loop design
|
||
|
||
The event loop runs in a `tokio::spawn`ed task and uses `tokio::select!` to
|
||
multiplex between swarm events and control commands:
|
||
|
||
```rust
|
||
loop {
|
||
tokio::select! {
|
||
event = self.swarm.next() => { ... } // requires futures::StreamExt
|
||
cmd = self.cmd_rx.recv() => { ... }
|
||
}
|
||
}
|
||
```
|
||
|
||
`Swarm<T>` implements `futures::Stream`, so `StreamExt::next()` is used (not
|
||
`tokio::io::AsyncReadExt`).
|
||
|
||
### SwarmKey authentication via Identify
|
||
|
||
After every `IdentifyEvent::Received`, the manager validates the remote peer's
|
||
`agent_version` string against the expected swarm key fingerprint
|
||
(`chronogrid/<version>/swarm:<fingerprint>`). Peers with a mismatched fingerprint
|
||
are immediately disconnected via `swarm.disconnect_peer_id()`.
|
||
|
||
### CRDT sync flow
|
||
|
||
1. `ConnectionEstablished` → send `SyncRequest(local_clock)` to new peer.
|
||
2. `SyncRequest` received → compute `missing_from(remote_clock, local_clock)` →
|
||
encode deltas → send `SyncResponse(encrypted_deltas, local_clock)`.
|
||
3. `SyncResponse` received → decrypt deltas → update local clock → emit
|
||
`NetEvent::DeltasReceived` → emit `NetEvent::SyncComplete`.
|
||
4. If remote is behind → send follow-up `SyncRequest`.
|
||
|
||
Note: Step 2 currently sends stub `DeltaPacket`s with empty `changeset_bytes`.
|
||
Full Automerge changeset loading from the database is implemented in Task 2.6.
|
||
|
||
### Channels
|
||
|
||
| Channel | Capacity | Direction |
|
||
|----------------------------------|----------|------------------------------|
|
||
| `mpsc::channel<SwarmCommand>` | 64 | `SwarmHandle → SwarmManager` |
|
||
| `broadcast::channel<NetEvent>` | 256 | `SwarmManager → subscribers` |
|
||
|
||
---
|
||
|
||
## Test Coverage
|
||
|
||
**10 new unit tests** in `swarm_manager::tests`:
|
||
|
||
| Test | Coverage |
|
||
|---------------------------------------------------|-------------------------------------------------------|
|
||
| `test_swarm_handle_trigger_sync_sends_command` | `trigger_sync()` sends `TriggerSync` command |
|
||
| `test_swarm_handle_broadcast_event_sends_command` | `broadcast_event()` sends `BroadcastGossip` command |
|
||
| `test_swarm_handle_shutdown_sends_command` | `shutdown()` sends `Shutdown` command |
|
||
| `test_swarm_handle_subscribe_events_receives_broadcast` | `subscribe_events()` receives broadcast events |
|
||
| `test_swarm_handle_multiple_subscribers` | Multiple receivers get the same event |
|
||
| `test_swarm_handle_local_peer_id` | `local_peer_id()` returns non-empty bytes |
|
||
| `test_swarm_handle_trigger_sync_after_shutdown_fails` | Error when event loop is gone |
|
||
| `test_swarm_handle_clone_shares_channel` | Cloned handle shares the same command channel |
|
||
| `test_swarm_manager_start_and_shutdown` | Full swarm start + graceful shutdown |
|
||
| `test_swarm_manager_start_multiple_listen_addrs` | TCP + QUIC listen addresses |
|
||
| `test_two_swarms_connect_and_sync` | Two in-process swarms start, trigger sync, shutdown |
|
||
|
||
---
|
||
|
||
## Consequences
|
||
|
||
- `chronogrid_core::sync::SyncEngine` (Task 2.6) receives `NetEvent::DeltasReceived`
|
||
via `SwarmHandle::subscribe_events()` and merges the delta packets into the database.
|
||
- `chronogrid_daemon` (Task 2.7) calls `SwarmManager::start()` with identity and
|
||
swarm key loaded from the encrypted vault.
|
||
- The `ChronoGridBehaviour` composite is the single `NetworkBehaviour` type used
|
||
by the `Swarm` — no further composition is needed.
|
||
- `SyncProtocol::send_sync_request()` is called on `ConnectionEstablished` — this
|
||
means every new connection immediately triggers a vector clock exchange.
|
||
- No breaking changes to existing modules.
|