network resilience

crates/data/src/bin/hypha-data.rs

@@ -36,7 +36,7 @@ use tokio::{
 };
 use tokio_retry::{
     Retry,
-    strategy::{ExponentialBackoff, jitter},
+    strategy::{FixedInterval, jitter},
 };
 use tracing::level_filters::LevelFilter;
 use tracing_subscriber::{
@@ -156,7 +156,9 @@ async fn run(config: ConfigWithMetadata<Config>) -> Result<()> {
 
     // Dial each gateway and, on success, set up a relay circuit listen via it.
     let gateway_peer_ids = Retry::spawn(
-        ExponentialBackoff::from_millis(100).map(jitter).take(3),
+        FixedInterval::from_millis(config.network().rtt_ms().max(100))
+            .map(jitter)
+            .take(6),
         || {
             let network = network.clone();
 

crates/data/src/network.rs

@@ -1,4 +1,4 @@
-use std::{collections::HashMap, sync::Arc};
+use std::{collections::HashMap, sync::Arc, time::Duration};
 
 use futures_util::StreamExt;
 use hypha_config::NetworkConfig;
@@ -84,8 +84,10 @@ impl Network {
         exclude_cidrs: Vec<IpNet>,
         network_config: &NetworkConfig,
     ) -> Result<(Self, NetworkDriver), SwarmError> {
-        let (action_sender, action_receiver) = mpsc::channel(5);
+        let (action_sender, action_receiver) = mpsc::channel(64);
         let meter = metrics::global::meter();
+        let request_timeout =
+            (Duration::from_millis(network_config.rtt_ms()) * 10).max(Duration::from_secs(10));
 
         let swarm = SwarmBuilder::with_existing_identity(cert_chain, private_key, ca_certs, crls)
             .with_tokio()
@@ -150,20 +152,22 @@ impl Network {
                                 StreamProtocol::new(data_record::IDENTIFIER),
                                 request_response::ProtocolSupport::Inbound,
                             )],
-                            request_response::Config::default(),
+                            request_response::Config::default()
+                                .with_request_timeout(request_timeout),
                         ),
                     health_request_response: request_response::Behaviour::<health::Codec>::new(
                         [(
                             StreamProtocol::new(health::IDENTIFIER),
                             request_response::ProtocolSupport::Full,
                         )],
-                        request_response::Config::default(),
+                        request_response::Config::default().with_request_timeout(request_timeout),
                     ),
                 }
             })
             .map_err(|_| {
                 SwarmError::BehaviourCreation("Failed to create swarm behavior.".to_string())
             })?
+            .with_swarm_config(|c| c.with_idle_connection_timeout(Duration::from_secs(30)))
             .build();
 
         Ok((

crates/gateway/src/network.rs

@@ -3,7 +3,7 @@
 //! This module wires together the various networking primitives to run the
 //! gateway's event loop.
 
-use std::{collections::HashMap, sync::Arc};
+use std::{collections::HashMap, sync::Arc, time::Duration};
 
 use futures_util::stream::StreamExt;
 use hypha_config::NetworkConfig;
@@ -163,8 +163,7 @@ impl Network {
                 .map_err(|_| {
                     SwarmError::BehaviourCreation("Failed to create swarm behavior.".to_string())
                 })?
-                // TODO: Tune swarm configuration
-                .with_swarm_config(|config| config)
+                .with_swarm_config(|c| c.with_idle_connection_timeout(Duration::from_secs(30)))
                 .build();
 
         swarm

crates/messages/src/lib.rs

@@ -163,13 +163,11 @@ pub mod action {
         },
         SendModel {
             target: Reference,
-            timeout: SystemTime,
         },
         ExecuteBatch,
         SendUpdate {
             target: Reference,
             weight: f32,
-            timeout: SystemTime,
         },
         ApplyUpdate {
             source: Reference,

crates/network/src/request_response.rs

@@ -672,32 +672,34 @@ where
                         let handler = handlers.iter().find(|h| (h.matcher)(&request));
 
                         if let Some(handler) = handler {
-                            match handler
-                                .sender
-                                .send(Ok(InboundRequest {
-                                    request_id,
-                                    channel,
-                                    peer_id: peer,
-                                    request,
-                                }))
-                                .await
-                            {
-                                Ok(_) => {
-                                    tracing::trace!(
-                                        peer = %peer,
-                                        request_id = ?request_id,
-                                        handler_id = %handler.id,
-                                        "Successfully sent request to handler channel"
-                                    );
+                            let sender = handler.sender.clone();
+                            let handler_id = handler.id;
+                            let inbound_request = InboundRequest {
+                                request_id,
+                                channel,
+                                peer_id: peer,
+                                request,
+                            };
+
+                            tokio::spawn(async move {
+                                match sender.send(Ok(inbound_request)).await {
+                                    Ok(_) => {
+                                        tracing::trace!(
+                                            peer = %peer,
+                                            request_id = ?request_id,
+                                            handler_id = %handler_id,
+                                            "Successfully sent request to handler channel"
+                                        );
+                                    }
+                                    Err(_) => {
+                                        tracing::warn!(
+                                            peer = %peer,
+                                            handler_id = %handler_id,
+                                            "Handler channel closed, request dropped"
+                                        );
+                                    }
                                 }
-                                Err(_) => {
-                                    tracing::warn!(
-                                        peer = %peer,
-                                        handler_id = %handler.id,
-                                        "Handler channel closed, request dropped"
-                                    );
-                                }
-                            }
+                            });
                         } else {
                             tracing::warn!(
                                 peer = %peer,
@@ -861,6 +863,9 @@ where
 ///
 /// This enables `network.on::<P>(...)` and `network.request::<P>(...)` where `P: Protocol`.
 pub trait RequestResponseInterfaceExt: Clone + Sized + Send + Sync + 'static {
+    /// Default channel capacity for handler streams. Sized to tolerate bursty arrivals under RTT.
+    const DEFAULT_HANDLER_BUFFER: usize = 512;
+
     /// Create a handler builder for the protocol `P` using the given pattern.
     fn on<TCodec, Pat>(&self, pattern: Pat) -> HandlerBuilder<'_, TCodec, Self>
     where
@@ -872,7 +877,7 @@ pub trait RequestResponseInterfaceExt: Clone + Sized + Send + Sync + 'static {
         HandlerBuilder {
             interface: self,
             matcher: pattern.into_matcher(),
-            buffer_size: 32,
+            buffer_size: Self::DEFAULT_HANDLER_BUFFER,
         }
     }
 

crates/scheduler/src/bin/hypha-scheduler.rs

@@ -37,7 +37,7 @@ use miette::{IntoDiagnostic, Result};
 use serde_json::Value;
 use tokio_retry::{
     Retry,
-    strategy::{ExponentialBackoff, jitter},
+    strategy::{FixedInterval, jitter},
 };
 use tokio_stream::wrappers::ReceiverStream;
 use tokio_util::sync::CancellationToken;
@@ -144,7 +144,9 @@ async fn run(config: ConfigWithMetadata<Config>) -> Result<()> {
 
     // Dial each gateway and, on success, set up a relay circuit listen via it.
     let gateway_peer_ids = Retry::spawn(
-        ExponentialBackoff::from_millis(100).map(jitter).take(3),
+        FixedInterval::from_millis(config.network().rtt_ms().max(100))
+            .map(jitter)
+            .take(6),
         || {
             let network = network.clone();
 

crates/scheduler/src/network.rs

@@ -3,7 +3,7 @@
 //! The scheduler orchestrates workers via libp2p. This module brings together
 //! the networking primitives and drives the underlying swarm.
 
-use std::{collections::HashMap, sync::Arc};
+use std::{collections::HashMap, sync::Arc, time::Duration};
 
 use futures_util::stream::StreamExt;
 use hypha_config::NetworkConfig;
@@ -110,8 +110,10 @@ impl Network {
         exclude_cidrs: Vec<IpNet>,
         network_config: &NetworkConfig,
     ) -> Result<(Self, NetworkDriver), SwarmError> {
-        let (action_sender, action_receiver) = mpsc::channel(5);
+        let (action_sender, action_receiver) = mpsc::channel(64);
         let meter = metrics::global::meter();
+        let request_timeout =
+            (Duration::from_millis(network_config.rtt_ms()) * 10).max(Duration::from_secs(10));
 
         // Build libp2p Swarm using the derived identity and mTLS config
         let swarm = SwarmBuilder::with_existing_identity(cert_chain, private_key, ca_certs, crls)
@@ -182,35 +184,37 @@ impl Network {
                             StreamProtocol::new(api::IDENTIFIER),
                             request_response::ProtocolSupport::Full,
                         )],
-                        request_response::Config::default(),
+                        request_response::Config::default().with_request_timeout(request_timeout),
                     ),
                     health_request_response: request_response::Behaviour::<health::Codec>::new(
                         [(
                             StreamProtocol::new(health::IDENTIFIER),
                             request_response::ProtocolSupport::Outbound,
                         )],
-                        request_response::Config::default(),
+                        request_response::Config::default().with_request_timeout(request_timeout),
                     ),
                     action_request_response: request_response::Behaviour::<action::Codec>::new(
                         [(
                             StreamProtocol::new(action::IDENTIFIER),
                             request_response::ProtocolSupport::Full,
                         )],
-                        request_response::Config::default(),
+                        request_response::Config::default().with_request_timeout(request_timeout),
                     ),
                     data_record_request_response:
                         request_response::Behaviour::<data_record::Codec>::new(
                             [(
                                 StreamProtocol::new(data_record::IDENTIFIER),
                                 request_response::ProtocolSupport::Outbound,
                             )],
-                            request_response::Config::default(),
+                            request_response::Config::default()
+                                .with_request_timeout(request_timeout),
                         ),
                 }
             })
             .map_err(|_| {
                 SwarmError::BehaviourCreation("Failed to create swarm behavior.".to_string())
             })?
+            .with_swarm_config(|c| c.with_idle_connection_timeout(Duration::from_secs(30)))
             .build();
 
         Ok((