feat: configurable handshake session timeout

core/handshake/src/config.rs

@@ -1,5 +1,6 @@
 use std::net::SocketAddr;
 use std::path::PathBuf;
+use std::time::Duration;
 
 use serde::{Deserialize, Serialize};
 
@@ -8,10 +9,15 @@ use crate::transports;
 #[derive(Serialize, Deserialize, Clone)]
 #[serde(default)]
 pub struct HandshakeConfig {
+    /// List of transports to enable
     #[serde(rename = "transport")]
     pub transports: Vec<TransportConfig>,
+    /// Shared tranport http address
     pub http_address: SocketAddr,
+    /// Optional http configuration
     pub https: Option<HttpsConfig>,
+    /// Timeout for disconnected sessions
+    pub timeout: Duration,
 }
 
 impl Default for HandshakeConfig {
@@ -25,6 +31,7 @@ impl Default for HandshakeConfig {
             ],
             http_address: ([0, 0, 0, 0], 4220).into(),
             https: None,
+            timeout: Duration::from_secs(1),
         }
     }
 }

core/handshake/src/handshake.rs

@@ -50,7 +50,7 @@ impl<C: Collection> Handshake<C> {
         let config = config.get::<Self>();
         let provider = service_executor.get_provider();
         let pk = keystore.get_ed25519_pk();
-        let ctx = Context::new(provider, waiter);
+        let ctx = Context::new(provider, waiter, config.timeout);
         let handle = Handle::new();
 
         Self {
@@ -133,6 +133,7 @@ pub struct Context<P: ExecutorProviderInterface> {
     pub(crate) shutdown: ShutdownWaiter,
     connection_counter: Arc<AtomicU64>,
     connections: Arc<DashMap<u64, ConnectionEntry>>,
+    timeout: Duration,
 }
 
 struct ConnectionEntry {
@@ -146,12 +147,13 @@ struct ConnectionEntry {
 }
 
 impl<P: ExecutorProviderInterface> Context<P> {
-    pub fn new(provider: P, waiter: ShutdownWaiter) -> Self {
+    pub fn new(provider: P, waiter: ShutdownWaiter, timeout: Duration) -> Self {
         Self {
             provider,
             shutdown: waiter,
             connection_counter: AtomicU64::new(0).into(),
             connections: DashMap::new().into(),
+            timeout,
         }
     }
 
@@ -215,7 +217,7 @@ impl<P: ExecutorProviderInterface> Context<P> {
                     },
                 );
 
-                Proxy::new(connection_id, socket, rx, self.clone()).spawn(Some(
+                Proxy::new(connection_id, socket, rx, self.clone(), self.timeout).spawn(Some(
                     State::OnlyPrimaryConnection((sender, receiver).into()),
                 ));
             },

core/handshake/src/proxy.rs

@@ -39,6 +39,7 @@ pub struct Proxy<P: ExecutorProviderInterface> {
     /// payload through the transport. And randomly inserting in some other frame in the middle of
     /// an active length delimited message before reaching the promised length breaks many things.
     queued_primary_response: VecDeque<ResponseFrame>,
+    timeout: Duration,
 }
 
 pub type IsPrimary = bool;
@@ -65,6 +66,7 @@ impl<P: ExecutorProviderInterface> Proxy<P> {
         socket: UnixStream,
         connection_rx: Receiver<(IsPrimary, TransportPair)>,
         context: Context<P>,
+        timeout: Duration,
     ) -> Self {
         Self {
             context,
@@ -76,6 +78,7 @@ impl<P: ExecutorProviderInterface> Proxy<P> {
             discard_bytes: false,
             is_primary_the_current_sender: false,
             queued_primary_response: VecDeque::new(),
+            timeout,
         }
     }
 
@@ -123,7 +126,7 @@ impl<P: ExecutorProviderInterface> Proxy<P> {
 
     #[inline]
     async fn run_with_no_connection(&mut self) -> State {
-        match tokio::time::timeout(Duration::from_secs(10), self.connection_rx.recv()).await {
+        match tokio::time::timeout(self.timeout, self.connection_rx.recv()).await {
             Ok(Ok((is_primary, pair))) => {
                 if is_primary {
                     State::OnlyPrimaryConnection(pair)
@@ -592,7 +595,11 @@ mod tests {
 
     async fn start_mock_node<P: ExecutorProviderInterface>(id: u16) -> Result<ShutdownController> {
         let shutdown = ShutdownController::default();
-        let context = Context::new(MockServiceProvider, shutdown.waiter());
+        let context = Context::new(
+            MockServiceProvider,
+            shutdown.waiter(),
+            Duration::from_secs(1),
+        );
         let (transport, _) =
             MockTransport::bind::<P>(shutdown.waiter(), MockTransportConfig { port: id }).await?;
         transport.spawn_listener_task(context);