init

crates/federation-net/Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+name = "federation-net"
+version = "0.1.0"
+description = "Generic peer-to-peer networking engine built on Iroh"
+edition.workspace = true
+license.workspace = true
+rust-version.workspace = true
+
+[dependencies]
+iroh = { workspace = true }
+iroh-base = { workspace = true }
+iroh-tickets = { workspace = true }
+tokio = { workspace = true }
+serde = { workspace = true }
+postcard = { workspace = true }
+blake3 = { workspace = true }
+thiserror = { workspace = true }
+tracing = { workspace = true }
+data-encoding = { workspace = true }
+rand = { workspace = true }
+
+[dev-dependencies]
+anyhow = { workspace = true }
+tempfile = { workspace = true }
+tokio = { workspace = true, features = ["test-util"] }

crates/federation-net/src/config.rs

@@ -0,0 +1,226 @@
+//! Engine configuration.
+
+use std::path::PathBuf;
+use std::time::Duration;
+
+use crate::error::{NetworkError, Result};
+use crate::protocol::{NetworkId, SchemaId};
+
+/// Default maximum size of an encoded domain message (256 KiB).
+pub const DEFAULT_MAX_MESSAGE_SIZE: usize = 256 * 1024;
+/// Default timeout applied to handshakes and message round-trips.
+pub const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
+/// Default capacity of the event channel.
+pub const DEFAULT_EVENT_CHANNEL_CAPACITY: usize = 1024;
+/// Default limit of concurrently processed streams per peer.
+pub const DEFAULT_MAX_CONCURRENT_STREAMS_PER_PEER: usize = 64;
+
+/// Configuration for a [`crate::NetworkEngine`].
+///
+/// Use [`NetworkConfig::builder`] to construct a validated instance.
+#[derive(Debug, Clone)]
+pub struct NetworkConfig {
+    /// Directory where the persistent peer identity is stored.
+    pub data_dir: PathBuf,
+    /// Identifier of the network this peer participates in.
+    pub network_id: NetworkId,
+    /// Identifier of the domain message schema.
+    pub schema_id: SchemaId,
+    /// Maximum size in bytes of a single encoded domain message.
+    pub max_message_size: usize,
+    /// Timeout applied to handshakes and message round-trips.
+    pub request_timeout: Duration,
+    /// Capacity of the bounded event channel.
+    pub event_channel_capacity: usize,
+    /// Maximum number of concurrently processed streams per peer.
+    pub max_concurrent_streams_per_peer: usize,
+}
+
+impl NetworkConfig {
+    /// Returns a new [`NetworkConfigBuilder`].
+    pub fn builder() -> NetworkConfigBuilder {
+        NetworkConfigBuilder::default()
+    }
+}
+
+/// Builder for [`NetworkConfig`].
+///
+/// `data_dir`, `network_id` and `schema_id` are required; everything else has
+/// sensible defaults.
+#[derive(Debug, Default, Clone)]
+pub struct NetworkConfigBuilder {
+    data_dir: Option<PathBuf>,
+    network_id: Option<NetworkId>,
+    schema_id: Option<SchemaId>,
+    max_message_size: Option<usize>,
+    request_timeout: Option<Duration>,
+    event_channel_capacity: Option<usize>,
+    max_concurrent_streams_per_peer: Option<usize>,
+}
+
+impl NetworkConfigBuilder {
+    /// Sets the directory where the persistent peer identity is stored.
+    pub fn data_dir(mut self, dir: impl Into<PathBuf>) -> Self {
+        self.data_dir = Some(dir.into());
+        self
+    }
+
+    /// Sets the network identifier.
+    pub fn network_id(mut self, network_id: NetworkId) -> Self {
+        self.network_id = Some(network_id);
+        self
+    }
+
+    /// Sets the schema identifier.
+    pub fn schema_id(mut self, schema_id: SchemaId) -> Self {
+        self.schema_id = Some(schema_id);
+        self
+    }
+
+    /// Sets the maximum size in bytes of a single encoded domain message.
+    pub fn max_message_size(mut self, size: usize) -> Self {
+        self.max_message_size = Some(size);
+        self
+    }
+
+    /// Sets the timeout applied to handshakes and message round-trips.
+    pub fn request_timeout(mut self, timeout: Duration) -> Self {
+        self.request_timeout = Some(timeout);
+        self
+    }
+
+    /// Sets the capacity of the bounded event channel.
+    pub fn event_channel_capacity(mut self, capacity: usize) -> Self {
+        self.event_channel_capacity = Some(capacity);
+        self
+    }
+
+    /// Sets the maximum number of concurrently processed streams per peer.
+    pub fn max_concurrent_streams_per_peer(mut self, limit: usize) -> Self {
+        self.max_concurrent_streams_per_peer = Some(limit);
+        self
+    }
+
+    /// Validates the configuration and builds a [`NetworkConfig`].
+    pub fn build(self) -> Result<NetworkConfig> {
+        let data_dir = self
+            .data_dir
+            .ok_or_else(|| NetworkError::InvalidConfig("data_dir is required".into()))?;
+        let network_id = self
+            .network_id
+            .ok_or_else(|| NetworkError::InvalidConfig("network_id is required".into()))?;
+        let schema_id = self
+            .schema_id
+            .ok_or_else(|| NetworkError::InvalidConfig("schema_id is required".into()))?;
+
+        if data_dir.as_os_str().is_empty() {
+            return Err(NetworkError::InvalidConfig(
+                "data_dir must not be empty".into(),
+            ));
+        }
+
+        let max_message_size = self.max_message_size.unwrap_or(DEFAULT_MAX_MESSAGE_SIZE);
+        if max_message_size == 0 {
+            return Err(NetworkError::InvalidConfig(
+                "max_message_size must be greater than zero".into(),
+            ));
+        }
+        if max_message_size > u32::MAX as usize {
+            return Err(NetworkError::InvalidConfig(
+                "max_message_size must fit into a 32-bit length prefix".into(),
+            ));
+        }
+
+        let request_timeout = self.request_timeout.unwrap_or(DEFAULT_REQUEST_TIMEOUT);
+        if request_timeout.is_zero() {
+            return Err(NetworkError::InvalidConfig(
+                "request_timeout must be greater than zero".into(),
+            ));
+        }
+
+        let event_channel_capacity = self
+            .event_channel_capacity
+            .unwrap_or(DEFAULT_EVENT_CHANNEL_CAPACITY);
+        if event_channel_capacity == 0 {
+            return Err(NetworkError::InvalidConfig(
+                "event_channel_capacity must be greater than zero".into(),
+            ));
+        }
+
+        let max_concurrent_streams_per_peer = self
+            .max_concurrent_streams_per_peer
+            .unwrap_or(DEFAULT_MAX_CONCURRENT_STREAMS_PER_PEER);
+        if max_concurrent_streams_per_peer == 0 {
+            return Err(NetworkError::InvalidConfig(
+                "max_concurrent_streams_per_peer must be greater than zero".into(),
+            ));
+        }
+
+        Ok(NetworkConfig {
+            data_dir,
+            network_id,
+            schema_id,
+            max_message_size,
+            request_timeout,
+            event_channel_capacity,
+            max_concurrent_streams_per_peer,
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn base_builder() -> NetworkConfigBuilder {
+        NetworkConfig::builder()
+            .data_dir("./some-dir")
+            .network_id(NetworkId::from_name("test-network"))
+            .schema_id(SchemaId::from_name("test-schema"))
+    }
+
+    #[test]
+    fn builder_applies_defaults() {
+        let config = base_builder().build().expect("valid config");
+        assert_eq!(config.max_message_size, DEFAULT_MAX_MESSAGE_SIZE);
+        assert_eq!(config.request_timeout, DEFAULT_REQUEST_TIMEOUT);
+        assert_eq!(
+            config.event_channel_capacity,
+            DEFAULT_EVENT_CHANNEL_CAPACITY
+        );
+        assert_eq!(
+            config.max_concurrent_streams_per_peer,
+            DEFAULT_MAX_CONCURRENT_STREAMS_PER_PEER
+        );
+    }
+
+    #[test]
+    fn builder_rejects_missing_required_fields() {
+        assert!(NetworkConfig::builder().build().is_err());
+        assert!(
+            NetworkConfig::builder()
+                .data_dir("./dir")
+                .network_id(NetworkId::from_name("n"))
+                .build()
+                .is_err()
+        );
+    }
+
+    #[test]
+    fn builder_rejects_invalid_values() {
+        assert!(base_builder().max_message_size(0).build().is_err());
+        assert!(
+            base_builder()
+                .request_timeout(Duration::ZERO)
+                .build()
+                .is_err()
+        );
+        assert!(base_builder().event_channel_capacity(0).build().is_err());
+        assert!(
+            base_builder()
+                .max_concurrent_streams_per_peer(0)
+                .build()
+                .is_err()
+        );
+    }
+}

crates/federation-net/src/engine.rs

@@ -0,0 +1,757 @@
+//! The network engine: endpoint lifecycle, connection registry and messaging.
+
+use std::collections::HashMap;
+use std::fmt;
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::{Arc, Mutex, MutexGuard, PoisonError, Weak};
+use std::time::Duration;
+
+use iroh::endpoint::{Connection, RecvStream, SendStream, VarInt, presets};
+use iroh::protocol::{AcceptError, ProtocolHandler, Router};
+use iroh::{Endpoint, EndpointId};
+use serde::Serialize;
+use serde::de::DeserializeOwned;
+use tokio::sync::{Semaphore, mpsc};
+use tokio::task::JoinSet;
+use tokio::time::timeout;
+use tracing::{debug, info, warn};
+
+use crate::config::NetworkConfig;
+use crate::error::{NetworkError, Result};
+use crate::event::{ConnectionDirection, NetworkEvent, NetworkEventReceiver};
+use crate::identity;
+use crate::protocol::{
+    ALPN, Handshake, HandshakeAck, HandshakeErrorCode, MAX_HANDSHAKE_FRAME_SIZE,
+    MessageRejectReason, MessageRequest, MessageResponse, NetworkId, PROTOCOL_VERSION, SchemaId,
+};
+use crate::ticket::{PeerTicket, TICKET_VERSION};
+use crate::wire;
+
+/// Close code for a graceful, error-free close.
+const CLOSE_CODE_OK: VarInt = VarInt::from_u32(0);
+/// Close code used when the handshake was rejected by either side.
+const CLOSE_CODE_HANDSHAKE_REJECTED: VarInt = VarInt::from_u32(1);
+/// Close code used when the engine shuts down.
+const CLOSE_CODE_SHUTDOWN: VarInt = VarInt::from_u32(2);
+/// Close code used when a connection is replaced by a newer one.
+const CLOSE_CODE_REPLACED: VarInt = VarInt::from_u32(3);
+
+/// Extra bytes allowed on top of `max_message_size` for message framing
+/// (request id, length varints and enum tags).
+const MESSAGE_FRAME_OVERHEAD: usize = 64;
+/// Maximum size of an encoded [`MessageResponse`] frame.
+const MAX_RESPONSE_FRAME_SIZE: usize = 1024;
+/// How long the accepting side keeps a rejected connection open so the
+/// rejection ack reaches the initiator before the connection is torn down.
+const REJECT_LINGER: Duration = Duration::from_secs(3);
+/// How long [`NetworkEngine::shutdown`] waits for background tasks before
+/// aborting them.
+const SHUTDOWN_TASK_GRACE: Duration = Duration::from_secs(5);
+
+/// Bounds required of a domain message type.
+///
+/// This trait is implemented automatically for every type that satisfies the
+/// bounds; applications never implement it by hand.
+pub trait Message: Serialize + DeserializeOwned + Send + Sync + 'static {}
+
+impl<T> Message for T where T: Serialize + DeserializeOwned + Send + Sync + 'static {}
+
+/// Locks a mutex, recovering the guard if the mutex was poisoned.
+///
+/// The engine never panics while holding one of its locks, so poisoning can
+/// only originate from a panic in unrelated user code; recovering is safe.
+fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
+    mutex.lock().unwrap_or_else(PoisonError::into_inner)
+}
+
+/// State kept for every active peer connection.
+struct PeerState {
+    connection: Connection,
+    direction: ConnectionDirection,
+    semaphore: Arc<Semaphore>,
+    /// Monotonic registration counter, used so a stale connection's cleanup
+    /// never removes a newer connection to the same peer.
+    generation: u64,
+}
+
+/// State shared between the engine handles, the protocol handler and all
+/// background tasks.
+struct Shared<M> {
+    config: NetworkConfig,
+    endpoint: Endpoint,
+    router: Mutex<Option<Router>>,
+    peers: Mutex<HashMap<EndpointId, PeerState>>,
+    events: Mutex<Option<mpsc::Sender<NetworkEvent<M>>>>,
+    tasks: Mutex<JoinSet<()>>,
+    next_generation: AtomicU64,
+    shutting_down: AtomicBool,
+}
+
+impl<M: Message> Shared<M> {
+    fn is_shutting_down(&self) -> bool {
+        self.shutting_down.load(Ordering::SeqCst)
+    }
+
+    fn ensure_running(&self) -> Result<()> {
+        if self.is_shutting_down() {
+            Err(NetworkError::ShuttingDown)
+        } else {
+            Ok(())
+        }
+    }
+
+    /// Delivers an event to the application.
+    ///
+    /// The channel is bounded; if it is full this awaits until the
+    /// application consumes events, applying natural back-pressure without
+    /// unbounded buffering.
+    async fn emit(&self, event: NetworkEvent<M>) {
+        let sender = lock(&self.events).clone();
+        if let Some(sender) = sender
+            && sender.send(event).await.is_err()
+        {
+            debug!("event receiver dropped; event discarded");
+        }
+    }
+
+    /// Spawns a background task tracked until shutdown.
+    fn spawn_task<F>(&self, future: F)
+    where
+        F: Future<Output = ()> + Send + 'static,
+    {
+        let mut tasks = lock(&self.tasks);
+        // Reap already-finished tasks so the set does not grow unboundedly.
+        while tasks.try_join_next().is_some() {}
+        tasks.spawn(future);
+    }
+
+    /// Adds a connection to the registry, replacing (and closing) any previous
+    /// connection to the same peer. Returns the registration generation.
+    fn register_connection(
+        &self,
+        peer_id: EndpointId,
+        connection: Connection,
+        direction: ConnectionDirection,
+    ) -> u64 {
+        let generation = self.next_generation.fetch_add(1, Ordering::Relaxed);
+        let state = PeerState {
+            connection,
+            direction,
+            semaphore: Arc::new(Semaphore::new(self.config.max_concurrent_streams_per_peer)),
+            generation,
+        };
+        let replaced = lock(&self.peers).insert(peer_id, state);
+        if let Some(old) = replaced {
+            debug!(peer = %peer_id, "replacing existing connection");
+            old.connection
+                .close(CLOSE_CODE_REPLACED, b"replaced by a new connection");
+        }
+        generation
+    }
+
+    /// Removes the connection from the registry and reports the disconnect,
+    /// unless a newer connection to the same peer took its place.
+    async fn cleanup_connection(
+        &self,
+        peer_id: EndpointId,
+        generation: u64,
+        reason: Option<String>,
+    ) {
+        let removed = {
+            let mut peers = lock(&self.peers);
+            match peers.get(&peer_id) {
+                Some(state) if state.generation == generation => peers.remove(&peer_id),
+                _ => None,
+            }
+        };
+        if let Some(state) = removed {
+            info!(
+                peer = %peer_id,
+                direction = ?state.direction,
+                reason = reason.as_deref().unwrap_or("unknown"),
+                "peer disconnected"
+            );
+            self.emit(NetworkEvent::PeerDisconnected { peer_id, reason })
+                .await;
+        }
+    }
+
+    /// Validates a remote handshake against the local configuration.
+    fn validate_handshake(&self, handshake: &Handshake) -> Option<HandshakeErrorCode> {
+        if handshake.protocol_version != PROTOCOL_VERSION {
+            Some(HandshakeErrorCode::UnsupportedProtocolVersion)
+        } else if handshake.network_id != self.config.network_id {
+            Some(HandshakeErrorCode::NetworkMismatch)
+        } else if handshake.schema_id != self.config.schema_id {
+            Some(HandshakeErrorCode::SchemaMismatch)
+        } else {
+            None
+        }
+    }
+
+    /// Drives an established connection: accepts message streams until the
+    /// connection closes, then cleans up the registry entry.
+    async fn connection_loop(
+        self: Arc<Self>,
+        peer_id: EndpointId,
+        connection: Connection,
+        generation: u64,
+    ) {
+        let semaphore = {
+            let peers = lock(&self.peers);
+            match peers.get(&peer_id) {
+                Some(state) if state.generation == generation => state.semaphore.clone(),
+                // The connection was already replaced; nothing to drive.
+                _ => return,
+            }
+        };
+        loop {
+            let (send, recv) = match connection.accept_bi().await {
+                Ok(streams) => streams,
+                Err(_) => break,
+            };
+            let permit = match semaphore.clone().acquire_owned().await {
+                Ok(permit) => permit,
+                Err(_) => break,
+            };
+            let shared = self.clone();
+            self.spawn_task(async move {
+                shared
+                    .handle_message_stream(peer_id, send, recv, permit)
+                    .await;
+                // The permit is released when the task finishes.
+            });
+        }
+        let reason = connection.close_reason().map(|err| err.to_string());
+        self.cleanup_connection(peer_id, generation, reason).await;
+    }
+
+    /// Handles a single incoming message stream: decodes the request, hands
+    /// the message to the application and acknowledges it.
+    async fn handle_message_stream(
+        &self,
+        peer_id: EndpointId,
+        mut send: SendStream,
+        mut recv: RecvStream,
+        _permit: tokio::sync::OwnedSemaphorePermit,
+    ) {
+        let request_timeout = self.config.request_timeout;
+        let max_frame = self.config.max_message_size + MESSAGE_FRAME_OVERHEAD;
+        let request: MessageRequest =
+            match timeout(request_timeout, wire::read_frame(&mut recv, max_frame)).await {
+                Ok(Ok(request)) => request,
+                Ok(Err(err)) => {
+                    self.report_protocol_error(Some(peer_id), format!("bad message frame: {err}"))
+                        .await;
+                    return;
+                }
+                Err(_) => {
+                    self.report_protocol_error(
+                        Some(peer_id),
+                        "timed out reading message frame".to_string(),
+                    )
+                    .await;
+                    return;
+                }
+            };
+        let request_id = request.request_id;
+        let response = if self.is_shutting_down() {
+            MessageResponse::Rejected {
+                request_id,
+                reason: MessageRejectReason::ShuttingDown,
+            }
+        } else {
+            match wire::decode::<M>(&request.payload) {
+                Ok(message) => {
+                    self.emit(NetworkEvent::MessageReceived { peer_id, message })
+                        .await;
+                    MessageResponse::Accepted { request_id }
+                }
+                Err(err) => {
+                    self.report_protocol_error(
+                        Some(peer_id),
+                        format!("failed to decode domain message: {err}"),
+                    )
+                    .await;
+                    MessageResponse::Rejected {
+                        request_id,
+                        reason: MessageRejectReason::MalformedPayload,
+                    }
+                }
+            }
+        };
+        let write = timeout(
+            request_timeout,
+            wire::write_frame(&mut send, &response, MAX_RESPONSE_FRAME_SIZE),
+        )
+        .await;
+        match write {
+            Ok(Ok(())) => {
+                let _ = send.finish();
+            }
+            Ok(Err(err)) => {
+                debug!(peer = %peer_id, error = %err, "failed to send message response");
+            }
+            Err(_) => {
+                debug!(peer = %peer_id, "timed out sending message response");
+            }
+        }
+    }
+
+    async fn report_protocol_error(&self, peer_id: Option<EndpointId>, error: String) {
+        match peer_id {
+            Some(peer) => warn!(peer = %peer, error = %error, "protocol error"),
+            None => warn!(error = %error, "protocol error"),
+        }
+        self.emit(NetworkEvent::ProtocolError { peer_id, error })
+            .await;
+    }
+
+    /// Handles a freshly accepted incoming connection: performs the handshake
+    /// and, if accepted, drives the connection until it closes.
+    async fn handle_incoming(self: &Arc<Self>, connection: &Connection) -> Result<()> {
+        let peer_id = connection.remote_id();
+        if self.is_shutting_down() {
+            connection.close(CLOSE_CODE_SHUTDOWN, b"engine is shutting down");
+            return Err(NetworkError::ShuttingDown);
+        }
+        debug!(peer = %peer_id, "incoming connection");
+        let request_timeout = self.config.request_timeout;
+
+        let (mut send, mut recv) = timeout(request_timeout, connection.accept_bi())
+            .await
+            .map_err(|_| NetworkError::Timeout)
+            .and_then(|res| {
+                res.map_err(|err| {
+                    NetworkError::Transport(format!("failed to accept handshake stream: {err}"))
+                })
+            })
+            .inspect_err(|_| {
+                connection.close(CLOSE_CODE_HANDSHAKE_REJECTED, b"handshake failed");
+            })?;
+
+        let handshake: Result<Handshake> = timeout(
+            request_timeout,
+            wire::read_frame(&mut recv, MAX_HANDSHAKE_FRAME_SIZE),
+        )
+        .await
+        .map_err(|_| NetworkError::Timeout)
+        .and_then(|res| res);
+
+        let verdict = match &handshake {
+            Ok(handshake) => self.validate_handshake(handshake),
+            Err(_) => Some(HandshakeErrorCode::InvalidHandshake),
+        };
+
+        if let Some(code) = verdict {
+            warn!(peer = %peer_id, code = ?code, "handshake rejected");
+            let ack = HandshakeAck {
+                accepted: false,
+                error: Some(code),
+            };
+            let sent = timeout(
+                request_timeout,
+                wire::write_frame(&mut send, &ack, MAX_HANDSHAKE_FRAME_SIZE),
+            )
+            .await;
+            if matches!(sent, Ok(Ok(()))) {
+                let _ = send.finish();
+                // Keep the connection open briefly so QUIC delivers the ack;
+                // the initiator closes as soon as it has read the rejection.
+                let _ = timeout(REJECT_LINGER, connection.closed()).await;
+            }
+            connection.close(CLOSE_CODE_HANDSHAKE_REJECTED, b"handshake rejected");
+            return Err(handshake_code_to_error(code));
+        }
+
+        let ack = HandshakeAck {
+            accepted: true,
+            error: None,
+        };
+        timeout(
+            request_timeout,
+            wire::write_frame(&mut send, &ack, MAX_HANDSHAKE_FRAME_SIZE),
+        )
+        .await
+        .map_err(|_| NetworkError::Timeout)
+        .and_then(|res| res)
+        .inspect_err(|_| {
+            connection.close(CLOSE_CODE_HANDSHAKE_REJECTED, b"handshake failed");
+        })?;
+        let _ = send.finish();
+
+        let generation =
+            self.register_connection(peer_id, connection.clone(), ConnectionDirection::Incoming);
+        info!(peer = %peer_id, "peer connected (incoming)");
+        self.emit(NetworkEvent::PeerConnected {
+            peer_id,
+            direction: ConnectionDirection::Incoming,
+        })
+        .await;
+        // Drive the connection inside the router's accept task; it runs for
+        // as long as the connection lives.
+        self.clone()
+            .connection_loop(peer_id, connection.clone(), generation)
+            .await;
+        Ok(())
+    }
+
+    /// Runs the client side of the handshake on a fresh outgoing connection.
+    async fn run_client_handshake(&self, connection: &Connection) -> Result<()> {
+        let (mut send, mut recv) = connection.open_bi().await.map_err(|err| {
+            NetworkError::Transport(format!("failed to open handshake stream: {err}"))
+        })?;
+        let handshake = Handshake {
+            protocol_version: PROTOCOL_VERSION,
+            network_id: self.config.network_id,
+            schema_id: self.config.schema_id,
+        };
+        wire::write_frame(&mut send, &handshake, MAX_HANDSHAKE_FRAME_SIZE).await?;
+        let _ = send.finish();
+        let ack: HandshakeAck = wire::read_frame(&mut recv, MAX_HANDSHAKE_FRAME_SIZE).await?;
+        if ack.accepted {
+            Ok(())
+        } else {
+            let code = ack.error.unwrap_or(HandshakeErrorCode::InvalidHandshake);
+            Err(handshake_code_to_error(code))
+        }
+    }
+}
+
+fn handshake_code_to_error(code: HandshakeErrorCode) -> NetworkError {
+    match code {
+        HandshakeErrorCode::UnsupportedProtocolVersion => NetworkError::UnsupportedProtocolVersion,
+        HandshakeErrorCode::NetworkMismatch => NetworkError::NetworkMismatch,
+        HandshakeErrorCode::SchemaMismatch => NetworkError::SchemaMismatch,
+        HandshakeErrorCode::InvalidHandshake => NetworkError::HandshakeRejected,
+    }
+}
+
+/// Protocol handler registered with the Iroh [`Router`] for [`ALPN`].
+///
+/// Holds only a weak reference to the shared state so the router does not
+/// keep the engine alive after all engine handles are dropped.
+struct FederationProtocol<M> {
+    shared: Weak<Shared<M>>,
+}
+
+impl<M> fmt::Debug for FederationProtocol<M> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("FederationProtocol").finish_non_exhaustive()
+    }
+}
+
+impl<M: Message> ProtocolHandler for FederationProtocol<M> {
+    async fn accept(&self, connection: Connection) -> Result<(), AcceptError> {
+        let Some(shared) = self.shared.upgrade() else {
+            connection.close(CLOSE_CODE_SHUTDOWN, b"engine is gone");
+            return Ok(());
+        };
+        let peer_id = connection.remote_id();
+        if let Err(err) = shared.handle_incoming(&connection).await {
+            shared
+                .report_protocol_error(Some(peer_id), format!("incoming connection failed: {err}"))
+                .await;
+        }
+        // Errors of a single connection never bring down the router; the
+        // connection has already been closed with an appropriate code.
+        Ok(())
+    }
+}
+
+/// A generic peer-to-peer network engine on top of Iroh.
+///
+/// `M` is the application-defined domain message type; the engine treats it
+/// as an opaque, postcard-serializable payload. The engine is cheaply
+/// clonable: all clones share the same endpoint and connection registry.
+pub struct NetworkEngine<M> {
+    shared: Arc<Shared<M>>,
+}
+
+impl<M> Clone for NetworkEngine<M> {
+    fn clone(&self) -> Self {
+        Self {
+            shared: self.shared.clone(),
+        }
+    }
+}
+
+impl<M> fmt::Debug for NetworkEngine<M> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("NetworkEngine").finish_non_exhaustive()
+    }
+}
+
+impl<M: Message> NetworkEngine<M> {
+    /// Starts the engine: loads (or creates) the persistent identity, binds
+    /// the Iroh endpoint and starts accepting connections.
+    ///
+    /// Returns the engine handle together with the single receiver for
+    /// [`NetworkEvent`]s.
+    pub async fn start(config: NetworkConfig) -> Result<(Self, NetworkEventReceiver<M>)> {
+        let secret_key = identity::load_or_create(&config.data_dir).await?;
+        let endpoint = Endpoint::builder(presets::N0)
+            .secret_key(secret_key)
+            .bind()
+            .await
+            .map_err(|err| NetworkError::Transport(format!("failed to bind endpoint: {err}")))?;
+        let endpoint_id = endpoint.id();
+        let (sender, receiver) = mpsc::channel(config.event_channel_capacity);
+        let shared = Arc::new(Shared {
+            config,
+            endpoint: endpoint.clone(),
+            router: Mutex::new(None),
+            peers: Mutex::new(HashMap::new()),
+            events: Mutex::new(Some(sender)),
+            tasks: Mutex::new(JoinSet::new()),
+            next_generation: AtomicU64::new(0),
+            shutting_down: AtomicBool::new(false),
+        });
+        let handler = FederationProtocol {
+            shared: Arc::downgrade(&shared),
+        };
+        let router = Router::builder(endpoint).accept(ALPN, handler).spawn();
+        *lock(&shared.router) = Some(router);
+        info!(
+            endpoint_id = %endpoint_id,
+            network_id = %shared.config.network_id,
+            schema_id = %shared.config.schema_id,
+            "network engine started"
+        );
+        Ok((Self { shared }, NetworkEventReceiver::new(receiver)))
+    }
+
+    /// Returns the stable identifier of this peer.
+    pub fn endpoint_id(&self) -> EndpointId {
+        self.shared.endpoint.id()
+    }
+
+    /// Returns the network this engine participates in.
+    pub fn network_id(&self) -> NetworkId {
+        self.shared.config.network_id
+    }
+
+    /// Returns the message schema this engine uses.
+    pub fn schema_id(&self) -> SchemaId {
+        self.shared.config.schema_id
+    }
+
+    /// Creates a shareable [`PeerTicket`] for this peer.
+    ///
+    /// Waits (bounded by the configured request timeout) for the endpoint to
+    /// come online so the ticket contains relay information whenever
+    /// possible; on an isolated network the ticket falls back to the locally
+    /// known direct addresses.
+    pub async fn ticket(&self) -> Result<PeerTicket> {
+        self.shared.ensure_running()?;
+        let endpoint = &self.shared.endpoint;
+        let has_relay = endpoint.addr().relay_urls().next().is_some();
+        if !has_relay {
+            let _ = timeout(self.shared.config.request_timeout, endpoint.online()).await;
+        }
+        let endpoint_addr = endpoint.addr();
+        if endpoint_addr.is_empty() {
+            return Err(NetworkError::Transport(
+                "endpoint has no reachable addresses yet".to_string(),
+            ));
+        }
+        Ok(PeerTicket {
+            ticket_version: TICKET_VERSION,
+            protocol_version: PROTOCOL_VERSION,
+            network_id: self.shared.config.network_id,
+            schema_id: self.shared.config.schema_id,
+            endpoint_addr,
+        })
+    }
+
+    /// Connects to the peer described by `ticket` and performs the handshake.
+    ///
+    /// Incompatible tickets (different network, schema or protocol version)
+    /// are rejected before any connection is attempted. An existing healthy
+    /// connection to the same peer is reused.
+    pub async fn connect(&self, ticket: PeerTicket) -> Result<EndpointId> {
+        let shared = &self.shared;
+        shared.ensure_running()?;
+        if ticket.ticket_version != TICKET_VERSION {
+            return Err(NetworkError::InvalidTicket(format!(
+                "unsupported ticket version {}",
+                ticket.ticket_version
+            )));
+        }
+        if ticket.protocol_version != PROTOCOL_VERSION {
+            return Err(NetworkError::UnsupportedProtocolVersion);
+        }
+        if ticket.network_id != shared.config.network_id {
+            return Err(NetworkError::NetworkMismatch);
+        }
+        if ticket.schema_id != shared.config.schema_id {
+            return Err(NetworkError::SchemaMismatch);
+        }
+        let target = ticket.endpoint_id();
+        if target == self.endpoint_id() {
+            return Err(NetworkError::InvalidTicket(
+                "the ticket points to this peer itself".to_string(),
+            ));
+        }
+        if lock(&shared.peers).contains_key(&target) {
+            debug!(peer = %target, "reusing existing connection");
+            return Ok(target);
+        }
+
+        let request_timeout = shared.config.request_timeout;
+        let connection = timeout(
+            request_timeout,
+            shared.endpoint.connect(ticket.endpoint_addr, ALPN),
+        )
+        .await
+        .map_err(|_| NetworkError::Timeout)?
+        .map_err(|err| NetworkError::Transport(format!("failed to connect: {err}")))?;
+        // The remote identity comes from the authenticated Iroh connection,
+        // never from the ticket payload.
+        let peer_id = connection.remote_id();
+
+        let handshake = timeout(request_timeout, shared.run_client_handshake(&connection))
+            .await
+            .map_err(|_| NetworkError::Timeout)
+            .and_then(|res| res);
+        if let Err(err) = handshake {
+            connection.close(CLOSE_CODE_HANDSHAKE_REJECTED, b"handshake failed");
+            return Err(err);
+        }
+
+        let generation =
+            shared.register_connection(peer_id, connection.clone(), ConnectionDirection::Outgoing);
+        info!(peer = %peer_id, "peer connected (outgoing)");
+        shared
+            .emit(NetworkEvent::PeerConnected {
+                peer_id,
+                direction: ConnectionDirection::Outgoing,
+            })
+            .await;
+        let loop_shared = shared.clone();
+        shared.spawn_task(async move {
+            loop_shared
+                .connection_loop(peer_id, connection, generation)
+                .await;
+        });
+        Ok(peer_id)
+    }
+
+    /// Sends a domain message to a connected peer.
+    ///
+    /// Returns only after the peer acknowledged the message. Fails with
+    /// [`NetworkError::PeerNotConnected`] if there is no active connection to
+    /// the peer; no implicit dialing or peer lookup is performed.
+    pub async fn send(&self, peer: EndpointId, message: &M) -> Result<()> {
+        let shared = &self.shared;
+        shared.ensure_running()?;
+        let connection = lock(&shared.peers)
+            .get(&peer)
+            .map(|state| state.connection.clone())
+            .ok_or(NetworkError::PeerNotConnected(peer))?;
+        let payload = wire::encode(message, shared.config.max_message_size)?;
+        let request_id: [u8; 16] = rand::random();
+        let request = MessageRequest {
+            request_id,
+            payload,
+        };
+        let max_frame = shared.config.max_message_size + MESSAGE_FRAME_OVERHEAD;
+        timeout(shared.config.request_timeout, async move {
+            let (mut send, mut recv) = connection.open_bi().await.map_err(|err| {
+                NetworkError::Transport(format!("failed to open message stream: {err}"))
+            })?;
+            wire::write_frame(&mut send, &request, max_frame).await?;
+            let _ = send.finish();
+            let response: MessageResponse =
+                wire::read_frame(&mut recv, MAX_RESPONSE_FRAME_SIZE).await?;
+            match response {
+                MessageResponse::Accepted { request_id: id } if id == request_id => Ok(()),
+                MessageResponse::Rejected {
+                    request_id: id,
+                    reason,
+                } if id == request_id => Err(NetworkError::MessageRejected(reason.to_string())),
+                _ => Err(NetworkError::Transport(
+                    "peer responded to a different request".to_string(),
+                )),
+            }
+        })
+        .await
+        .map_err(|_| NetworkError::Timeout)?
+    }
+
+    /// Returns the ids of all currently connected peers.
+    pub fn connected_peers(&self) -> Vec<EndpointId> {
+        lock(&self.shared.peers).keys().copied().collect()
+    }
+
+    /// Closes the connection to `peer` and removes it from the registry.
+    pub async fn disconnect(&self, peer: EndpointId) -> Result<()> {
+        let state = lock(&self.shared.peers)
+            .remove(&peer)
+            .ok_or(NetworkError::PeerNotConnected(peer))?;
+        state
+            .connection
+            .close(CLOSE_CODE_OK, b"disconnected by local peer");
+        info!(peer = %peer, "peer disconnected (local request)");
+        self.shared
+            .emit(NetworkEvent::PeerDisconnected {
+                peer_id: peer,
+                reason: Some("disconnected by local peer".to_string()),
+            })
+            .await;
+        Ok(())
+    }
+
+    /// Shuts the engine down gracefully.
+    ///
+    /// Rejects new operations, closes all connections, stops the protocol
+    /// router, closes the endpoint, waits for background tasks and finally
+    /// closes the event channel. Calling `shutdown` on several clones of the
+    /// same engine is safe; only the first call does the work.
+    pub async fn shutdown(self) -> Result<()> {
+        let shared = &self.shared;
+        if shared.shutting_down.swap(true, Ordering::SeqCst) {
+            return Ok(());
+        }
+        info!("network engine shutting down");
+
+        // Close all active connections.
+        let states: Vec<PeerState> = lock(&shared.peers)
+            .drain()
+            .map(|(_, state)| state)
+            .collect();
+        for state in &states {
+            state
+                .connection
+                .close(CLOSE_CODE_SHUTDOWN, b"engine is shutting down");
+        }
+        drop(states);
+
+        // Shut down the router; this stops the protocol handler and closes
+        // the endpoint.
+        let router = lock(&shared.router).take();
+        if let Some(router) = router
+            && let Err(err) = router.shutdown().await
+        {
+            warn!(error = %err, "router shutdown reported an error");
+        }
+        shared.endpoint.close().await;
+
+        // Wait for the remaining background tasks; abort stragglers.
+        let mut tasks = std::mem::take(&mut *lock(&shared.tasks));
+        let drained = timeout(SHUTDOWN_TASK_GRACE, async {
+            while tasks.join_next().await.is_some() {}
+        })
+        .await;
+        if drained.is_err() {
+            warn!("background tasks did not finish in time; aborting them");
+            tasks.abort_all();
+        }
+
+        // Close the event channel: dropping the last sender makes
+        // `NetworkEventReceiver::recv` return `None` once drained.
+        *lock(&shared.events) = None;
+        info!("network engine shut down");
+        Ok(())
+    }
+}

crates/federation-net/src/error.rs

@@ -0,0 +1,67 @@
+//! Error types for the federation-net library.
+
+use iroh::EndpointId;
+
+/// Convenient result alias used across the library.
+pub type Result<T, E = NetworkError> = std::result::Result<T, E>;
+
+/// All errors that can be returned by the public API of this library.
+#[derive(Debug, thiserror::Error)]
+#[non_exhaustive]
+pub enum NetworkError {
+    /// The provided [`crate::NetworkConfig`] is invalid.
+    #[error("invalid configuration: {0}")]
+    InvalidConfig(String),
+
+    /// Loading or persisting the peer identity failed.
+    #[error("identity error: {0}")]
+    Identity(String),
+
+    /// A peer ticket could not be parsed or validated.
+    #[error("invalid ticket: {0}")]
+    InvalidTicket(String),
+
+    /// The remote peer belongs to a different network.
+    #[error("network id mismatch")]
+    NetworkMismatch,
+
+    /// The remote peer uses an incompatible message schema.
+    #[error("schema id mismatch")]
+    SchemaMismatch,
+
+    /// The remote peer speaks an unsupported protocol version.
+    #[error("unsupported protocol version")]
+    UnsupportedProtocolVersion,
+
+    /// The remote peer rejected the handshake as malformed.
+    #[error("handshake rejected as invalid by remote peer")]
+    HandshakeRejected,
+
+    /// No active connection to the given peer exists.
+    #[error("peer is not connected: {0}")]
+    PeerNotConnected(EndpointId),
+
+    /// A message exceeds the configured maximum size.
+    #[error("message exceeds maximum size")]
+    MessageTooLarge,
+
+    /// The operation did not complete within the configured timeout.
+    #[error("request timed out")]
+    Timeout,
+
+    /// Encoding or decoding a wire payload failed.
+    #[error("serialization error: {0}")]
+    Serialization(String),
+
+    /// The underlying Iroh transport reported an error.
+    #[error("transport error: {0}")]
+    Transport(String),
+
+    /// The remote peer rejected a message.
+    #[error("message rejected by peer: {0}")]
+    MessageRejected(String),
+
+    /// The engine is shutting down and no longer accepts operations.
+    #[error("engine is shutting down")]
+    ShuttingDown,
+}

crates/federation-net/src/event.rs

@@ -0,0 +1,79 @@
+//! Network events delivered to the application.
+
+use iroh::EndpointId;
+use tokio::sync::mpsc;
+
+/// Direction of an established peer connection.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ConnectionDirection {
+    /// The remote peer connected to us.
+    Incoming,
+    /// We connected to the remote peer.
+    Outgoing,
+}
+
+/// Events emitted by a [`crate::NetworkEngine`].
+///
+/// `M` is the application-defined domain message type.
+#[derive(Debug)]
+pub enum NetworkEvent<M> {
+    /// A peer completed the handshake and is now connected.
+    PeerConnected {
+        /// Identifier of the connected peer.
+        peer_id: EndpointId,
+        /// Whether the peer connected to us or we connected to it.
+        direction: ConnectionDirection,
+    },
+
+    /// A peer disconnected.
+    PeerDisconnected {
+        /// Identifier of the disconnected peer.
+        peer_id: EndpointId,
+        /// Human-readable reason, if one is known.
+        reason: Option<String>,
+    },
+
+    /// A domain message was received from a peer.
+    MessageReceived {
+        /// Identifier of the sending peer.
+        peer_id: EndpointId,
+        /// The decoded domain message.
+        message: M,
+    },
+
+    /// A protocol-level error occurred.
+    ///
+    /// Errors of a single connection never bring down the engine; they are
+    /// reported here instead.
+    ProtocolError {
+        /// The peer involved, if known.
+        peer_id: Option<EndpointId>,
+        /// Human-readable error description.
+        error: String,
+    },
+}
+
+/// Receiving side of the engine's event channel.
+///
+/// There is exactly one receiver per engine; it is returned from
+/// [`crate::NetworkEngine::start`]. The underlying channel is bounded, so the
+/// application should consume events promptly to avoid back-pressuring the
+/// engine.
+#[derive(Debug)]
+pub struct NetworkEventReceiver<M> {
+    rx: mpsc::Receiver<NetworkEvent<M>>,
+}
+
+impl<M> NetworkEventReceiver<M> {
+    pub(crate) fn new(rx: mpsc::Receiver<NetworkEvent<M>>) -> Self {
+        Self { rx }
+    }
+
+    /// Receives the next event.
+    ///
+    /// Returns `None` after the engine has shut down and all pending events
+    /// have been consumed.
+    pub async fn recv(&mut self) -> Option<NetworkEvent<M>> {
+        self.rx.recv().await
+    }
+}

crates/federation-net/src/identity.rs

@@ -0,0 +1,154 @@
+//! Persistent peer identity.
+//!
+//! Each peer owns a long-lived Iroh secret key stored at
+//! `<data_dir>/identity.key`. The key is created on first start and loaded on
+//! every subsequent start, so the peer's [`iroh::EndpointId`] stays stable
+//! across restarts.
+
+use std::path::Path;
+
+use iroh::SecretKey;
+
+use crate::error::{NetworkError, Result};
+
+/// File name of the persisted secret key inside the data directory.
+pub(crate) const IDENTITY_FILE_NAME: &str = "identity.key";
+
+/// Loads the peer identity from `data_dir`, creating a new one if none exists.
+///
+/// File I/O runs on a blocking thread so the async runtime is never blocked.
+pub(crate) async fn load_or_create(data_dir: &Path) -> Result<SecretKey> {
+    let dir = data_dir.to_path_buf();
+    tokio::task::spawn_blocking(move || load_or_create_blocking(&dir))
+        .await
+        .map_err(|err| NetworkError::Identity(format!("identity task panicked: {err}")))?
+}
+
+fn load_or_create_blocking(dir: &Path) -> Result<SecretKey> {
+    std::fs::create_dir_all(dir).map_err(|err| {
+        NetworkError::Identity(format!(
+            "failed to create data directory {}: {err}",
+            dir.display()
+        ))
+    })?;
+    let path = dir.join(IDENTITY_FILE_NAME);
+    if path.exists() {
+        load_key(&path)
+    } else {
+        let key = SecretKey::generate();
+        store_key(dir, &path, &key)?;
+        Ok(key)
+    }
+}
+
+fn load_key(path: &Path) -> Result<SecretKey> {
+    let content = std::fs::read_to_string(path).map_err(|err| {
+        NetworkError::Identity(format!(
+            "failed to read identity file {}: {err}",
+            path.display()
+        ))
+    })?;
+    let trimmed = content.trim();
+    let bytes = data_encoding::HEXLOWER_PERMISSIVE
+        .decode(trimmed.as_bytes())
+        .map_err(|_| corrupted(path))?;
+    let bytes: [u8; 32] = bytes.try_into().map_err(|_| corrupted(path))?;
+    Ok(SecretKey::from_bytes(&bytes))
+}
+
+fn corrupted(path: &Path) -> NetworkError {
+    NetworkError::Identity(format!(
+        "identity file {} is corrupted: expected 64 hex characters; \
+         remove the file to generate a new identity",
+        path.display()
+    ))
+}
+
+/// Persists the key atomically: write to a temporary file in the same
+/// directory, then rename it over the final path.
+fn store_key(dir: &Path, path: &Path, key: &SecretKey) -> Result<()> {
+    use std::io::Write;
+
+    let tmp_path = dir.join(format!("{IDENTITY_FILE_NAME}.tmp"));
+    let mut options = std::fs::OpenOptions::new();
+    options.write(true).create(true).truncate(true);
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::OpenOptionsExt;
+        options.mode(0o600);
+    }
+    let write_result = (|| {
+        let mut file = options.open(&tmp_path)?;
+        let mut encoded = data_encoding::HEXLOWER.encode(&key.to_bytes());
+        encoded.push('\n');
+        file.write_all(encoded.as_bytes())?;
+        file.sync_all()?;
+        drop(file);
+        std::fs::rename(&tmp_path, path)
+    })();
+    if let Err(err) = write_result {
+        // Best effort removal of the temporary file; the original error wins.
+        let _ = std::fs::remove_file(&tmp_path);
+        return Err(NetworkError::Identity(format!(
+            "failed to persist identity file {}: {err}",
+            path.display()
+        )));
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn key_is_created_and_reloaded() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let first = load_or_create(dir.path()).await.expect("create key");
+        let second = load_or_create(dir.path()).await.expect("load key");
+        assert_eq!(first.public(), second.public());
+        assert_eq!(first.to_bytes(), second.to_bytes());
+    }
+
+    #[tokio::test]
+    async fn existing_key_is_not_overwritten() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let first = load_or_create(dir.path()).await.expect("create key");
+        let path = dir.path().join(IDENTITY_FILE_NAME);
+        let before = std::fs::read(&path).expect("read identity file");
+        let second = load_or_create(dir.path()).await.expect("load key");
+        let after = std::fs::read(&path).expect("read identity file");
+        assert_eq!(before, after);
+        assert_eq!(first.public(), second.public());
+    }
+
+    #[tokio::test]
+    async fn corrupted_key_file_is_an_error() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let path = dir.path().join(IDENTITY_FILE_NAME);
+        std::fs::write(&path, "definitely not a hex key").expect("write garbage");
+        let result = load_or_create(dir.path()).await;
+        assert!(matches!(result, Err(NetworkError::Identity(_))));
+    }
+
+    #[tokio::test]
+    async fn truncated_key_file_is_an_error() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let path = dir.path().join(IDENTITY_FILE_NAME);
+        // Valid hex, but not 32 bytes.
+        std::fs::write(&path, "deadbeef").expect("write short key");
+        let result = load_or_create(dir.path()).await;
+        assert!(matches!(result, Err(NetworkError::Identity(_))));
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn key_file_has_restrictive_permissions() {
+        use std::os::unix::fs::PermissionsExt;
+
+        let dir = tempfile::tempdir().expect("tempdir");
+        load_or_create(dir.path()).await.expect("create key");
+        let metadata = std::fs::metadata(dir.path().join(IDENTITY_FILE_NAME)).expect("metadata");
+        assert_eq!(metadata.permissions().mode() & 0o777, 0o600);
+    }
+}

crates/federation-net/src/lib.rs

@@ -0,0 +1,73 @@
+//! # federation-net
+//!
+//! A generic peer-to-peer networking engine built on [Iroh](https://iroh.computer).
+//!
+//! The library establishes direct (NAT-traversing, relay-assisted) QUIC
+//! connections between peers and exchanges typed, application-defined
+//! messages over them. It deliberately contains no domain logic: the
+//! application chooses its own message type and the engine treats it as an
+//! opaque, serde-serializable payload.
+//!
+//! ## Core concepts
+//!
+//! * [`NetworkId`] — isolates independent P2P networks from each other.
+//! * [`SchemaId`] — isolates applications with incompatible message schemas.
+//! * [`PeerTicket`] — a shareable string invitation used to reach a peer.
+//! * [`NetworkEngine`] — the engine itself; [`NetworkEventReceiver`] delivers
+//!   [`NetworkEvent`]s to the application.
+//!
+//! ## Example
+//!
+//! ```no_run
+//! use federation_net::{NetworkConfig, NetworkEngine, NetworkId, SchemaId};
+//!
+//! #[derive(Debug, serde::Serialize, serde::Deserialize)]
+//! enum MyMessage {
+//!     Hello { from: String },
+//! }
+//!
+//! # async fn run() -> federation_net::Result<()> {
+//! let config = NetworkConfig::builder()
+//!     .data_dir("./peer-a")
+//!     .network_id(NetworkId::from_name("example-network"))
+//!     .schema_id(SchemaId::from_name("my-message-v1"))
+//!     .build()?;
+//!
+//! let (engine, mut events) = NetworkEngine::<MyMessage>::start(config).await?;
+//! let ticket = engine.ticket().await?;
+//! println!("share this ticket: {ticket}");
+//!
+//! while let Some(event) = events.recv().await {
+//!     println!("{event:?}");
+//! }
+//! engine.shutdown().await?;
+//! # Ok(())
+//! # }
+//! ```
+
+#![warn(missing_docs)]
+#![forbid(unsafe_code)]
+
+mod config;
+mod engine;
+mod error;
+mod event;
+mod identity;
+mod protocol;
+mod ticket;
+mod wire;
+
+pub use config::{
+    DEFAULT_EVENT_CHANNEL_CAPACITY, DEFAULT_MAX_CONCURRENT_STREAMS_PER_PEER,
+    DEFAULT_MAX_MESSAGE_SIZE, DEFAULT_REQUEST_TIMEOUT, NetworkConfig, NetworkConfigBuilder,
+};
+pub use engine::{Message, NetworkEngine};
+pub use error::{NetworkError, Result};
+pub use event::{ConnectionDirection, NetworkEvent, NetworkEventReceiver};
+pub use protocol::{ALPN, NetworkId, PROTOCOL_VERSION, SchemaId};
+pub use ticket::{PeerTicket, TICKET_VERSION};
+
+// Re-exported Iroh types that appear in the public API.
+pub use iroh::{EndpointAddr, EndpointId};
+// Re-exported so applications can use the generic ticket helpers.
+pub use iroh_tickets::Ticket;

crates/federation-net/src/protocol.rs

@@ -0,0 +1,215 @@
+//! Protocol constants, network/schema identifiers and internal wire structures.
+
+use std::fmt;
+
+use serde::{Deserialize, Serialize};
+
+/// ALPN identifier for the federation-net protocol.
+///
+/// [`NetworkId`] and [`SchemaId`] are deliberately not part of the ALPN;
+/// they are verified during the application-level handshake instead.
+pub const ALPN: &[u8] = b"/federation-net/1";
+
+/// Version of the application-level wire protocol.
+pub const PROTOCOL_VERSION: u16 = 1;
+
+/// Maximum size of an encoded handshake frame.
+///
+/// Handshake frames are tiny; this limit only guards against malicious peers.
+pub(crate) const MAX_HANDSHAKE_FRAME_SIZE: usize = 16 * 1024;
+
+fn hash_with_domain(domain: &str, name: &str) -> [u8; 32] {
+    let mut hasher = blake3::Hasher::new();
+    hasher.update(domain.as_bytes());
+    hasher.update(name.as_bytes());
+    *hasher.finalize().as_bytes()
+}
+
+fn fmt_id(bytes: &[u8; 32], f: &mut fmt::Formatter<'_>) -> fmt::Result {
+    for byte in bytes {
+        write!(f, "{byte:02x}")?;
+    }
+    Ok(())
+}
+
+/// Identifier of a distinct P2P network.
+///
+/// Peers with different network ids refuse to establish an application-level
+/// session even though they share the same transport protocol.
+#[derive(Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
+pub struct NetworkId([u8; 32]);
+
+impl NetworkId {
+    /// Creates a network id from raw bytes.
+    pub fn from_bytes(bytes: [u8; 32]) -> Self {
+        Self(bytes)
+    }
+
+    /// Derives a network id from a human-readable name.
+    ///
+    /// The derivation is deterministic: the same name always yields the same
+    /// id. Internally this computes `BLAKE3("federation-net:network:" + name)`.
+    pub fn from_name(name: &str) -> Self {
+        Self(hash_with_domain("federation-net:network:", name))
+    }
+
+    /// Returns the raw bytes of this id.
+    pub fn as_bytes(&self) -> &[u8; 32] {
+        &self.0
+    }
+}
+
+impl fmt::Display for NetworkId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt_id(&self.0, f)
+    }
+}
+
+impl fmt::Debug for NetworkId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "NetworkId(")?;
+        fmt_id(&self.0, f)?;
+        write!(f, ")")
+    }
+}
+
+/// Identifier of the domain message schema used on top of the network.
+///
+/// Peers on the same network but with different schema ids reject each other,
+/// because they would not be able to decode each other's messages. A backwards
+/// incompatible change to the domain message type requires a new schema id.
+#[derive(Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
+pub struct SchemaId([u8; 32]);
+
+impl SchemaId {
+    /// Creates a schema id from raw bytes.
+    pub fn from_bytes(bytes: [u8; 32]) -> Self {
+        Self(bytes)
+    }
+
+    /// Derives a schema id from a human-readable name.
+    ///
+    /// The derivation is deterministic: the same name always yields the same
+    /// id. Internally this computes `BLAKE3("federation-net:schema:" + name)`.
+    pub fn from_name(name: &str) -> Self {
+        Self(hash_with_domain("federation-net:schema:", name))
+    }
+
+    /// Returns the raw bytes of this id.
+    pub fn as_bytes(&self) -> &[u8; 32] {
+        &self.0
+    }
+}
+
+impl fmt::Display for SchemaId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt_id(&self.0, f)
+    }
+}
+
+impl fmt::Debug for SchemaId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "SchemaId(")?;
+        fmt_id(&self.0, f)?;
+        write!(f, ")")
+    }
+}
+
+/// Handshake sent by the connecting side on the first bidirectional stream.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct Handshake {
+    pub protocol_version: u16,
+    pub network_id: NetworkId,
+    pub schema_id: SchemaId,
+}
+
+/// Reply to a [`Handshake`], sent by the accepting side.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct HandshakeAck {
+    pub accepted: bool,
+    pub error: Option<HandshakeErrorCode>,
+}
+
+/// Reasons for rejecting a handshake.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+pub(crate) enum HandshakeErrorCode {
+    UnsupportedProtocolVersion,
+    NetworkMismatch,
+    SchemaMismatch,
+    InvalidHandshake,
+}
+
+/// A single domain message request, sent on its own bidirectional stream.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) struct MessageRequest {
+    pub request_id: [u8; 16],
+    pub payload: Vec<u8>,
+}
+
+/// Reasons for rejecting a [`MessageRequest`].
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+pub(crate) enum MessageRejectReason {
+    /// The payload could not be decoded into the domain message type.
+    MalformedPayload,
+    /// The receiving peer is shutting down.
+    ShuttingDown,
+    /// The receiving peer is overloaded and dropped the message.
+    Overloaded,
+}
+
+impl fmt::Display for MessageRejectReason {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::MalformedPayload => write!(f, "malformed payload"),
+            Self::ShuttingDown => write!(f, "peer is shutting down"),
+            Self::Overloaded => write!(f, "peer is overloaded"),
+        }
+    }
+}
+
+/// Reply to a [`MessageRequest`].
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) enum MessageResponse {
+    Accepted {
+        request_id: [u8; 16],
+    },
+    Rejected {
+        request_id: [u8; 16],
+        reason: MessageRejectReason,
+    },
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn network_id_from_name_is_deterministic() {
+        let a = NetworkId::from_name("example-network");
+        let b = NetworkId::from_name("example-network");
+        assert_eq!(a, b);
+    }
+
+    #[test]
+    fn different_network_names_produce_different_ids() {
+        let a = NetworkId::from_name("network-a");
+        let b = NetworkId::from_name("network-b");
+        assert_ne!(a, b);
+    }
+
+    #[test]
+    fn schema_id_from_name_is_deterministic() {
+        let a = SchemaId::from_name("demo-message-v1");
+        let b = SchemaId::from_name("demo-message-v1");
+        assert_eq!(a, b);
+        assert_ne!(a, SchemaId::from_name("demo-message-v2"));
+    }
+
+    #[test]
+    fn network_and_schema_domains_are_separated() {
+        // The same name hashed under different domain prefixes must differ.
+        let network = NetworkId::from_name("same-name");
+        let schema = SchemaId::from_name("same-name");
+        assert_ne!(network.as_bytes(), schema.as_bytes());
+    }
+}

crates/federation-net/src/ticket.rs

@@ -0,0 +1,193 @@
+//! Peer tickets: self-contained connection invitations.
+
+use std::fmt;
+use std::str::FromStr;
+
+use iroh::{EndpointAddr, EndpointId};
+use iroh_tickets::{ParseError, Ticket};
+use serde::{Deserialize, Serialize};
+
+use crate::protocol::{NetworkId, SchemaId};
+
+/// Current version of the ticket wire format.
+pub const TICKET_VERSION: u16 = 1;
+
+/// Maximum size of a decoded ticket in bytes.
+const MAX_TICKET_BYTES: usize = 8 * 1024;
+/// Maximum length of a ticket string accepted by [`PeerTicket::from_str`].
+const MAX_TICKET_STRING_LEN: usize = 16 * 1024;
+
+/// A shareable invitation to connect to a peer.
+///
+/// The ticket contains everything needed to dial the peer over Iroh, plus the
+/// network and schema identifiers so incompatibility is detected before any
+/// domain message is exchanged.
+///
+/// Tickets serialize to a string with the `fnet` prefix via [`fmt::Display`]
+/// and parse back via [`FromStr`].
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct PeerTicket {
+    /// Version of the ticket format itself.
+    pub ticket_version: u16,
+    /// Version of the application-level wire protocol the peer speaks.
+    pub protocol_version: u16,
+    /// Network the issuing peer participates in.
+    pub network_id: NetworkId,
+    /// Message schema the issuing peer uses.
+    pub schema_id: SchemaId,
+    /// Iroh address of the issuing peer.
+    pub endpoint_addr: EndpointAddr,
+}
+
+impl PeerTicket {
+    /// Returns the endpoint id of the peer this ticket points to.
+    pub fn endpoint_id(&self) -> EndpointId {
+        self.endpoint_addr.id
+    }
+}
+
+/// Versioned body of the ticket; everything after the leading version number.
+#[derive(Serialize, Deserialize)]
+struct TicketBody {
+    protocol_version: u16,
+    network_id: NetworkId,
+    schema_id: SchemaId,
+    endpoint_addr: EndpointAddr,
+}
+
+impl Ticket for PeerTicket {
+    const KIND: &'static str = "fnet";
+
+    fn encode_bytes(&self) -> Vec<u8> {
+        let body = TicketBody {
+            protocol_version: self.protocol_version,
+            network_id: self.network_id,
+            schema_id: self.schema_id,
+            endpoint_addr: self.endpoint_addr.clone(),
+        };
+        // Serializing plain owned data into a growable Vec is infallible;
+        // postcard only errors on unsupported types or fixed-size buffers.
+        postcard::to_stdvec(&(self.ticket_version, body))
+            .expect("postcard serialization of a ticket into a Vec cannot fail")
+    }
+
+    fn decode_bytes(bytes: &[u8]) -> Result<Self, ParseError> {
+        if bytes.len() > MAX_TICKET_BYTES {
+            return Err(ParseError::verification_failed(
+                "ticket exceeds the maximum allowed size",
+            ));
+        }
+        let (ticket_version, rest) = postcard::take_from_bytes::<u16>(bytes)?;
+        if ticket_version != TICKET_VERSION {
+            return Err(ParseError::verification_failed(
+                "unsupported ticket version",
+            ));
+        }
+        let body: TicketBody = postcard::from_bytes(rest)?;
+        Ok(Self {
+            ticket_version,
+            protocol_version: body.protocol_version,
+            network_id: body.network_id,
+            schema_id: body.schema_id,
+            endpoint_addr: body.endpoint_addr,
+        })
+    }
+}
+
+impl fmt::Display for PeerTicket {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str(&Ticket::encode_string(self))
+    }
+}
+
+impl FromStr for PeerTicket {
+    type Err = ParseError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        if s.len() > MAX_TICKET_STRING_LEN {
+            return Err(ParseError::verification_failed(
+                "ticket string exceeds the maximum allowed length",
+            ));
+        }
+        Ticket::decode_string(s)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::net::{Ipv4Addr, SocketAddr};
+
+    use iroh::SecretKey;
+
+    use super::*;
+
+    fn sample_ticket() -> PeerTicket {
+        let endpoint_id = SecretKey::generate().public();
+        let addr = EndpointAddr::new(endpoint_id)
+            .with_ip_addr(SocketAddr::from((Ipv4Addr::LOCALHOST, 4242)));
+        PeerTicket {
+            ticket_version: TICKET_VERSION,
+            protocol_version: 1,
+            network_id: NetworkId::from_name("test-network"),
+            schema_id: SchemaId::from_name("test-schema"),
+            endpoint_addr: addr,
+        }
+    }
+
+    #[test]
+    fn ticket_string_round_trip() {
+        let ticket = sample_ticket();
+        let encoded = ticket.to_string();
+        assert!(encoded.starts_with("fnet"));
+        let decoded: PeerTicket = encoded.parse().expect("parse ticket");
+        assert_eq!(decoded, ticket);
+        // The canonical string form must be stable across round-trips.
+        assert_eq!(decoded.to_string(), encoded);
+    }
+
+    #[test]
+    fn corrupted_ticket_is_rejected() {
+        let ticket = sample_ticket();
+        let mut encoded = ticket.to_string();
+        // Truncate the payload; the result must not parse.
+        encoded.truncate(encoded.len() - 10);
+        assert!(encoded.parse::<PeerTicket>().is_err());
+
+        // Corrupt the alphabet: '!' is not valid base32.
+        let corrupted = format!("fnet!{}", &ticket.to_string()[5..]);
+        assert!(corrupted.parse::<PeerTicket>().is_err());
+    }
+
+    #[test]
+    fn wrong_prefix_is_rejected() {
+        let ticket = sample_ticket();
+        let encoded = ticket.to_string();
+        let renamed = format!("blob{}", &encoded[4..]);
+        assert!(renamed.parse::<PeerTicket>().is_err());
+    }
+
+    #[test]
+    fn unknown_ticket_version_is_rejected() {
+        let ticket = sample_ticket();
+        let body = TicketBody {
+            protocol_version: ticket.protocol_version,
+            network_id: ticket.network_id,
+            schema_id: ticket.schema_id,
+            endpoint_addr: ticket.endpoint_addr.clone(),
+        };
+        let bytes = postcard::to_stdvec(&(99u16, body)).expect("encode");
+        let mut encoded = String::from("fnet");
+        data_encoding::BASE32_NOPAD.encode_append(&bytes, &mut encoded);
+        encoded.make_ascii_lowercase();
+        assert!(encoded.parse::<PeerTicket>().is_err());
+    }
+
+    #[test]
+    fn oversized_ticket_is_rejected() {
+        let bytes = vec![0u8; MAX_TICKET_BYTES + 1];
+        assert!(PeerTicket::decode_bytes(&bytes).is_err());
+
+        let huge = "fnet".to_string() + &"a".repeat(MAX_TICKET_STRING_LEN);
+        assert!(huge.parse::<PeerTicket>().is_err());
+    }
+}

crates/federation-net/src/wire.rs

@@ -0,0 +1,152 @@
+//! Length-prefixed framing of postcard-encoded values.
+//!
+//! Every frame on the wire is laid out as:
+//!
+//! ```text
+//! 4 bytes: payload length, unsigned big-endian
+//! N bytes: postcard payload
+//! ```
+
+use serde::Serialize;
+use serde::de::DeserializeOwned;
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
+
+use crate::error::{NetworkError, Result};
+
+/// Encodes a value with postcard, enforcing `max_size` on the encoded bytes.
+pub(crate) fn encode<T: Serialize>(value: &T, max_size: usize) -> Result<Vec<u8>> {
+    let bytes = postcard::to_stdvec(value)
+        .map_err(|err| NetworkError::Serialization(format!("postcard encoding failed: {err}")))?;
+    if bytes.len() > max_size {
+        return Err(NetworkError::MessageTooLarge);
+    }
+    Ok(bytes)
+}
+
+/// Decodes a postcard-encoded value.
+pub(crate) fn decode<T: DeserializeOwned>(bytes: &[u8]) -> Result<T> {
+    postcard::from_bytes(bytes)
+        .map_err(|err| NetworkError::Serialization(format!("postcard decoding failed: {err}")))
+}
+
+/// Writes one length-prefixed frame containing the postcard encoding of
+/// `value`.
+pub(crate) async fn write_frame<W, T>(writer: &mut W, value: &T, max_size: usize) -> Result<()>
+where
+    W: AsyncWrite + Unpin,
+    T: Serialize,
+{
+    let payload = encode(value, max_size)?;
+    // `max_size` is validated to fit into u32 by the configuration.
+    let len = payload.len() as u32;
+    writer
+        .write_all(&len.to_be_bytes())
+        .await
+        .map_err(|err| NetworkError::Transport(format!("failed to write frame header: {err}")))?;
+    writer
+        .write_all(&payload)
+        .await
+        .map_err(|err| NetworkError::Transport(format!("failed to write frame payload: {err}")))?;
+    Ok(())
+}
+
+/// Reads one length-prefixed frame and decodes it with postcard.
+///
+/// The length prefix is validated against `max_size` before any payload
+/// memory is allocated. A frame that exceeds the limit is rejected without
+/// reading its payload.
+pub(crate) async fn read_frame<R, T>(reader: &mut R, max_size: usize) -> Result<T>
+where
+    R: AsyncRead + Unpin,
+    T: DeserializeOwned,
+{
+    let mut len_bytes = [0u8; 4];
+    reader
+        .read_exact(&mut len_bytes)
+        .await
+        .map_err(|err| NetworkError::Transport(format!("failed to read frame header: {err}")))?;
+    let len = u32::from_be_bytes(len_bytes) as usize;
+    if len > max_size {
+        return Err(NetworkError::Transport(format!(
+            "incoming frame of {len} bytes exceeds limit of {max_size} bytes"
+        )));
+    }
+    let mut payload = vec![0u8; len];
+    reader
+        .read_exact(&mut payload)
+        .await
+        .map_err(|err| NetworkError::Transport(format!("failed to read frame payload: {err}")))?;
+    decode(&payload)
+}
+
+#[cfg(test)]
+mod tests {
+    use serde::Deserialize;
+
+    use super::*;
+
+    #[derive(Debug, PartialEq, Serialize, Deserialize)]
+    struct TestValue {
+        text: String,
+        number: u64,
+    }
+
+    #[tokio::test]
+    async fn frame_round_trip() {
+        let value = TestValue {
+            text: "hello".into(),
+            number: 42,
+        };
+        let mut buf = Vec::new();
+        write_frame(&mut buf, &value, 1024).await.expect("write");
+        let decoded: TestValue = read_frame(&mut buf.as_slice(), 1024).await.expect("read");
+        assert_eq!(decoded, value);
+    }
+
+    #[tokio::test]
+    async fn oversized_frame_is_rejected_before_allocation() {
+        // A header declaring a huge payload with no payload bytes present.
+        // If the length check happened after allocation, this would try to
+        // allocate 4 GiB; instead it must fail fast on the limit check.
+        let mut buf = Vec::new();
+        buf.extend_from_slice(&u32::MAX.to_be_bytes());
+        let result: Result<TestValue> = read_frame(&mut buf.as_slice(), 1024).await;
+        assert!(matches!(result, Err(NetworkError::Transport(_))));
+    }
+
+    #[tokio::test]
+    async fn oversized_value_is_rejected_on_write() {
+        let value = TestValue {
+            text: "x".repeat(2048),
+            number: 1,
+        };
+        let mut buf = Vec::new();
+        let result = write_frame(&mut buf, &value, 1024).await;
+        assert!(matches!(result, Err(NetworkError::MessageTooLarge)));
+        assert!(buf.is_empty());
+    }
+
+    #[tokio::test]
+    async fn malformed_payload_is_rejected() {
+        // Valid header, but payload bytes that do not decode as TestValue.
+        let payload = [0xffu8; 8];
+        let mut buf = Vec::new();
+        buf.extend_from_slice(&(payload.len() as u32).to_be_bytes());
+        buf.extend_from_slice(&payload);
+        let result: Result<TestValue> = read_frame(&mut buf.as_slice(), 1024).await;
+        assert!(matches!(result, Err(NetworkError::Serialization(_))));
+    }
+
+    #[tokio::test]
+    async fn truncated_stream_is_an_error() {
+        let value = TestValue {
+            text: "hello".into(),
+            number: 42,
+        };
+        let mut buf = Vec::new();
+        write_frame(&mut buf, &value, 1024).await.expect("write");
+        buf.truncate(buf.len() - 1);
+        let result: Result<TestValue> = read_frame(&mut buf.as_slice(), 1024).await;
+        assert!(matches!(result, Err(NetworkError::Transport(_))));
+    }
+}

crates/federation-net/tests/integration.rs

@@ -0,0 +1,299 @@
+//! Integration tests: two engines in one Tokio runtime.
+
+use std::path::Path;
+use std::time::Duration;
+
+use federation_net::{
+    ConnectionDirection, EndpointId, NetworkConfig, NetworkEngine, NetworkError, NetworkEvent,
+    NetworkEventReceiver, NetworkId, PeerTicket, SchemaId,
+};
+
+/// Hard cap on every test so a regression can never hang CI.
+const TEST_TIMEOUT: Duration = Duration::from_secs(120);
+/// Timeout used when waiting for a single event.
+const EVENT_TIMEOUT: Duration = Duration::from_secs(30);
+
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+enum TestMessage {
+    Text { sender: String, body: String },
+    Ping { nonce: u64 },
+}
+
+type Engine = NetworkEngine<TestMessage>;
+type Events = NetworkEventReceiver<TestMessage>;
+
+/// Serializes the network-facing tests: running many endpoints at once makes
+/// relay discovery contend and produces spurious connect timeouts.
+static NET_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
+
+fn config(dir: &Path, network: &str, schema: &str) -> NetworkConfig {
+    NetworkConfig::builder()
+        .data_dir(dir)
+        .network_id(NetworkId::from_name(network))
+        .schema_id(SchemaId::from_name(schema))
+        .request_timeout(Duration::from_secs(10))
+        .build()
+        .expect("valid test config")
+}
+
+async fn start(dir: &Path, network: &str, schema: &str) -> (Engine, Events) {
+    NetworkEngine::start(config(dir, network, schema))
+        .await
+        .expect("engine starts")
+}
+
+/// Waits for the next event, panicking on timeout or channel close.
+async fn next_event(events: &mut Events) -> NetworkEvent<TestMessage> {
+    tokio::time::timeout(EVENT_TIMEOUT, events.recv())
+        .await
+        .expect("timed out waiting for an event")
+        .expect("event channel closed unexpectedly")
+}
+
+/// Waits until a `PeerConnected` event for `peer` arrives, skipping unrelated
+/// events (e.g. protocol errors from earlier rejected attempts).
+async fn wait_connected(events: &mut Events, peer: EndpointId) -> ConnectionDirection {
+    loop {
+        if let NetworkEvent::PeerConnected { peer_id, direction } = next_event(events).await
+            && peer_id == peer
+        {
+            return direction;
+        }
+    }
+}
+
+async fn wait_disconnected(events: &mut Events, peer: EndpointId) {
+    loop {
+        if let NetworkEvent::PeerDisconnected { peer_id, .. } = next_event(events).await
+            && peer_id == peer
+        {
+            return;
+        }
+    }
+}
+
+async fn wait_message(events: &mut Events, peer: EndpointId) -> TestMessage {
+    loop {
+        if let NetworkEvent::MessageReceived { peer_id, message } = next_event(events).await
+            && peer_id == peer
+        {
+            return message;
+        }
+    }
+}
+
+#[tokio::test]
+async fn connect_and_exchange_messages() {
+    let _net = NET_LOCK.lock().await;
+    tokio::time::timeout(TEST_TIMEOUT, async {
+        let dir_a = tempfile::tempdir().expect("tempdir");
+        let dir_b = tempfile::tempdir().expect("tempdir");
+        let (engine_a, mut events_a) = start(dir_a.path(), "test-net", "test-schema-v1").await;
+        let (engine_b, mut events_b) = start(dir_b.path(), "test-net", "test-schema-v1").await;
+
+        // 1. Peer A creates a ticket.
+        let ticket = engine_a.ticket().await.expect("ticket");
+        assert_eq!(ticket.endpoint_id(), engine_a.endpoint_id());
+
+        // 2. Peer B connects using the ticket.
+        let peer_a = engine_b.connect(ticket).await.expect("connect");
+        assert_eq!(peer_a, engine_a.endpoint_id());
+
+        // 3. Both sides observe PeerConnected with the right direction.
+        let dir_on_b = wait_connected(&mut events_b, engine_a.endpoint_id()).await;
+        assert_eq!(dir_on_b, ConnectionDirection::Outgoing);
+        let dir_on_a = wait_connected(&mut events_a, engine_b.endpoint_id()).await;
+        assert_eq!(dir_on_a, ConnectionDirection::Incoming);
+
+        // 8. connected_peers contains the expected endpoint ids.
+        assert_eq!(engine_b.connected_peers(), vec![engine_a.endpoint_id()]);
+        assert_eq!(engine_a.connected_peers(), vec![engine_b.endpoint_id()]);
+
+        // 4-5. B sends a message; A receives the correctly typed object.
+        let hello = TestMessage::Text {
+            sender: "bob".into(),
+            body: "hello alice".into(),
+        };
+        engine_b
+            .send(engine_a.endpoint_id(), &hello)
+            .await
+            .expect("send b -> a");
+        let received = wait_message(&mut events_a, engine_b.endpoint_id()).await;
+        assert_eq!(received, hello);
+
+        // 6-7. A sends back over the same connection; B receives it.
+        let pong = TestMessage::Ping { nonce: 4242 };
+        engine_a
+            .send(engine_b.endpoint_id(), &pong)
+            .await
+            .expect("send a -> b");
+        let received = wait_message(&mut events_b, engine_a.endpoint_id()).await;
+        assert_eq!(received, pong);
+
+        engine_a.shutdown().await.expect("shutdown a");
+        engine_b.shutdown().await.expect("shutdown b");
+    })
+    .await
+    .expect("test timed out");
+}
+
+#[tokio::test]
+async fn network_mismatch_is_rejected() {
+    let _net = NET_LOCK.lock().await;
+    tokio::time::timeout(TEST_TIMEOUT, async {
+        let dir_a = tempfile::tempdir().expect("tempdir");
+        let dir_b = tempfile::tempdir().expect("tempdir");
+        let (engine_a, _events_a) = start(dir_a.path(), "network-one", "schema-v1").await;
+        let (engine_b, _events_b) = start(dir_b.path(), "network-two", "schema-v1").await;
+
+        let ticket = engine_a.ticket().await.expect("ticket");
+
+        // Local pre-validation: the ticket carries A's network id.
+        let err = engine_b
+            .connect(ticket.clone())
+            .await
+            .expect_err("must fail");
+        assert!(matches!(err, NetworkError::NetworkMismatch), "got {err:?}");
+
+        // Remote validation: forge a ticket claiming B's own network id, so
+        // the local check passes and the remote handshake must reject it.
+        let forged = PeerTicket {
+            network_id: engine_b.network_id(),
+            ..ticket
+        };
+        let err = engine_b.connect(forged).await.expect_err("must fail");
+        assert!(matches!(err, NetworkError::NetworkMismatch), "got {err:?}");
+
+        assert!(engine_a.connected_peers().is_empty());
+        assert!(engine_b.connected_peers().is_empty());
+
+        engine_a.shutdown().await.expect("shutdown a");
+        engine_b.shutdown().await.expect("shutdown b");
+    })
+    .await
+    .expect("test timed out");
+}
+
+#[tokio::test]
+async fn schema_mismatch_is_rejected() {
+    let _net = NET_LOCK.lock().await;
+    tokio::time::timeout(TEST_TIMEOUT, async {
+        let dir_a = tempfile::tempdir().expect("tempdir");
+        let dir_b = tempfile::tempdir().expect("tempdir");
+        let (engine_a, _events_a) = start(dir_a.path(), "same-network", "schema-one").await;
+        let (engine_b, _events_b) = start(dir_b.path(), "same-network", "schema-two").await;
+
+        let ticket = engine_a.ticket().await.expect("ticket");
+
+        // Local pre-validation.
+        let err = engine_b
+            .connect(ticket.clone())
+            .await
+            .expect_err("must fail");
+        assert!(matches!(err, NetworkError::SchemaMismatch), "got {err:?}");
+
+        // Remote validation with a forged schema id.
+        let forged = PeerTicket {
+            schema_id: engine_b.schema_id(),
+            ..ticket
+        };
+        let err = engine_b.connect(forged).await.expect_err("must fail");
+        assert!(matches!(err, NetworkError::SchemaMismatch), "got {err:?}");
+
+        engine_a.shutdown().await.expect("shutdown a");
+        engine_b.shutdown().await.expect("shutdown b");
+    })
+    .await
+    .expect("test timed out");
+}
+
+#[tokio::test]
+async fn shutdown_disconnects_peers() {
+    let _net = NET_LOCK.lock().await;
+    tokio::time::timeout(TEST_TIMEOUT, async {
+        let dir_a = tempfile::tempdir().expect("tempdir");
+        let dir_b = tempfile::tempdir().expect("tempdir");
+        let (engine_a, mut events_a) = start(dir_a.path(), "test-net", "schema-v1").await;
+        let (engine_b, mut events_b) = start(dir_b.path(), "test-net", "schema-v1").await;
+
+        let ticket = engine_a.ticket().await.expect("ticket");
+        engine_b.connect(ticket).await.expect("connect");
+        wait_connected(&mut events_a, engine_b.endpoint_id()).await;
+        wait_connected(&mut events_b, engine_a.endpoint_id()).await;
+
+        let id_b = engine_b.endpoint_id();
+        engine_b.shutdown().await.expect("shutdown b");
+
+        // 11. A notices the disconnect and drops the peer from its registry.
+        wait_disconnected(&mut events_a, id_b).await;
+        assert!(engine_a.connected_peers().is_empty());
+
+        // Sending to the gone peer now fails without hanging.
+        let err = engine_a
+            .send(id_b, &TestMessage::Ping { nonce: 1 })
+            .await
+            .expect_err("peer is gone");
+        assert!(
+            matches!(err, NetworkError::PeerNotConnected(_)),
+            "got {err:?}"
+        );
+
+        // The event channel of the stopped engine closes after draining.
+        while events_b.recv().await.is_some() {}
+
+        engine_a.shutdown().await.expect("shutdown a");
+    })
+    .await
+    .expect("test timed out");
+}
+
+#[tokio::test]
+async fn identity_persists_across_restarts() {
+    let _net = NET_LOCK.lock().await;
+    tokio::time::timeout(TEST_TIMEOUT, async {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let (engine, _events) = start(dir.path(), "test-net", "schema-v1").await;
+        let first_id = engine.endpoint_id();
+        engine.shutdown().await.expect("shutdown");
+
+        let (engine, _events) = start(dir.path(), "test-net", "schema-v1").await;
+        let second_id = engine.endpoint_id();
+        engine.shutdown().await.expect("shutdown");
+
+        assert_eq!(first_id, second_id);
+    })
+    .await
+    .expect("test timed out");
+}
+
+#[tokio::test]
+async fn disconnect_removes_peer() {
+    let _net = NET_LOCK.lock().await;
+    tokio::time::timeout(TEST_TIMEOUT, async {
+        let dir_a = tempfile::tempdir().expect("tempdir");
+        let dir_b = tempfile::tempdir().expect("tempdir");
+        let (engine_a, mut events_a) = start(dir_a.path(), "test-net", "schema-v1").await;
+        let (engine_b, mut events_b) = start(dir_b.path(), "test-net", "schema-v1").await;
+
+        let ticket = engine_a.ticket().await.expect("ticket");
+        let peer_a = engine_b.connect(ticket).await.expect("connect");
+        wait_connected(&mut events_a, engine_b.endpoint_id()).await;
+        wait_connected(&mut events_b, peer_a).await;
+
+        engine_b.disconnect(peer_a).await.expect("disconnect");
+        wait_disconnected(&mut events_b, peer_a).await;
+        assert!(engine_b.connected_peers().is_empty());
+
+        // Disconnecting twice reports PeerNotConnected.
+        let err = engine_b.disconnect(peer_a).await.expect_err("already gone");
+        assert!(
+            matches!(err, NetworkError::PeerNotConnected(_)),
+            "got {err:?}"
+        );
+
+        engine_a.shutdown().await.expect("shutdown a");
+        engine_b.shutdown().await.expect("shutdown b");
+    })
+    .await
+    .expect("test timed out");
+}