7.6 KiB
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
#[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)
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 withOutput = Connection(not(PeerId, Muxer)), so it cannot be used withwith_other_transport(). The correct API iswith_relay_client()which handles the transport upgrade internally and provides therelay::client::Behaviourto thewith_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
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 aSwarmCommand::TriggerSyncto the event loop, which callsSyncProtocol::send_sync_request()for each connected peer.broadcast_event()sendsSwarmCommand::BroadcastGossip, which callsGossipBehaviour::publish()on thechronogrid/state-changes/1topic.subscribe_events()returns abroadcast::Receiver<NetEvent>— multiple consumers can subscribe independently (broadcast semantics, 256-event buffer).shutdown()sendsSwarmCommand::Shutdown, causing the event loop to exit.SwarmHandleisClone— multiple components can hold a handle.
Event loop design
The event loop runs in a tokio::spawned task and uses tokio::select! to
multiplex between swarm events and control commands:
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
ConnectionEstablished→ sendSyncRequest(local_clock)to new peer.SyncRequestreceived → computemissing_from(remote_clock, local_clock)→ encode deltas → sendSyncResponse(encrypted_deltas, local_clock).SyncResponsereceived → decrypt deltas → update local clock → emitNetEvent::DeltasReceived→ emitNetEvent::SyncComplete.- If remote is behind → send follow-up
SyncRequest.
Note: Step 2 currently sends stub DeltaPackets 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) receivesNetEvent::DeltasReceivedviaSwarmHandle::subscribe_events()and merges the delta packets into the database.chronogrid_daemon(Task 2.7) callsSwarmManager::start()with identity and swarm key loaded from the encrypted vault.- The
ChronoGridBehaviourcomposite is the singleNetworkBehaviourtype used by theSwarm— no further composition is needed. SyncProtocol::send_sync_request()is called onConnectionEstablished— this means every new connection immediately triggers a vector clock exchange.- No breaking changes to existing modules.