diff --git a/distant-core/Cargo.toml b/distant-core/Cargo.toml index 9f5faaf32..15d4a0d5a 100644 --- a/distant-core/Cargo.toml +++ b/distant-core/Cargo.toml @@ -39,10 +39,10 @@ serde_json = "1.0.149" sha2 = "0.10.6" strum = { version = "0.28.0", features = ["derive"] } tokio = { version = "1.50.0", features = ["full"] } +socket2 = "0.6" [target.'cfg(unix)'.dependencies] libc = "0.2" -socket2 = "0.6" [lints] workspace = true diff --git a/distant-core/src/api.rs b/distant-core/src/api.rs index dfd7c7ab1..0e34edc13 100644 --- a/distant-core/src/api.rs +++ b/distant-core/src/api.rs @@ -32,9 +32,18 @@ impl ApiServerHandler where T: Api, { + /// Creates a new handler that wraps the given API implementation in an [`Arc`]. pub fn new(api: T) -> Self { Self { api: Arc::new(api) } } + + /// Creates a new handler from a pre-existing [`Arc`]-wrapped API implementation. + /// + /// This is useful when the caller needs to retain a reference to the API + /// (e.g., for health monitoring) while also passing it to the server handler. + pub fn from_arc(api: Arc) -> Self { + Self { api } + } } #[inline] @@ -1402,4 +1411,95 @@ mod tests { let batch = resp.payload.into_batch().unwrap(); assert!(batch.is_empty()); } + + // --------------------------------------------------------------- + // ApiServerHandler::from_arc + // --------------------------------------------------------------- + + #[test_log::test(tokio::test)] + async fn from_arc_creates_handler_that_dispatches_requests() { + let api = Arc::new(MockApi); + let handler = ApiServerHandler::from_arc(api); + + let (ctx, mut rx) = + make_request_ctx(Msg::Single(protocol::Request::Version {}), Header::new()); + + handler.on_request(ctx).await; + + let resp = rx.recv().await.unwrap(); + let msg = resp.payload.into_single().unwrap(); + match msg { + protocol::Response::Version(v) => { + assert_eq!(v.server_version, semver::Version::new(1, 0, 0)); + assert_eq!(v.protocol_version, semver::Version::new(0, 1, 0)); + assert_eq!(v.capabilities, vec![String::from("test")]); + } + other => panic!("Expected Version response, got {other:?}"), + } + } + + #[test_log::test(tokio::test)] + async fn from_arc_shares_api_with_caller() { + let api = Arc::new(MockApi); + let api_clone = Arc::clone(&api); + let handler = ApiServerHandler::from_arc(api); + + // The caller still holds a reference via api_clone + assert_eq!(Arc::strong_count(&api_clone), 2); + + // Handler still works + let (ctx, mut rx) = + make_request_ctx(Msg::Single(protocol::Request::SystemInfo {}), Header::new()); + + handler.on_request(ctx).await; + + let resp = rx.recv().await.unwrap(); + let msg = resp.payload.into_single().unwrap(); + match msg { + protocol::Response::SystemInfo(info) => { + assert_eq!(info.family, "unix"); + assert_eq!(info.os, "linux"); + assert_eq!(info.arch, "x86_64"); + } + other => panic!("Expected SystemInfo response, got {other:?}"), + } + } + + #[test_log::test(tokio::test)] + async fn from_arc_and_new_produce_equivalent_handlers() { + // from_arc handler + let arc_handler = ApiServerHandler::from_arc(Arc::new(MockApi)); + let (ctx1, mut rx1) = make_request_ctx( + Msg::Single(protocol::Request::FileRead { + path: RemotePath::from("/test"), + }), + Header::new(), + ); + arc_handler.on_request(ctx1).await; + let resp1 = rx1.recv().await.unwrap(); + let msg1 = resp1.payload.into_single().unwrap(); + + // new handler + let new_handler = ApiServerHandler::new(MockApi); + let (ctx2, mut rx2) = make_request_ctx( + Msg::Single(protocol::Request::FileRead { + path: RemotePath::from("/test"), + }), + Header::new(), + ); + new_handler.on_request(ctx2).await; + let resp2 = rx2.recv().await.unwrap(); + let msg2 = resp2.payload.into_single().unwrap(); + + // Both should produce identical Blob responses + match (msg1, msg2) { + (protocol::Response::Blob { data: d1 }, protocol::Response::Blob { data: d2 }) => { + assert_eq!(d1, d2); + assert_eq!(d1, [1, 2, 3]); + } + (other1, other2) => { + panic!("Expected Blob responses from both, got {other1:?} and {other2:?}") + } + } + } } diff --git a/distant-core/src/net/client/reconnect.rs b/distant-core/src/net/client/reconnect.rs index 9b2ba3c99..f1d1450cd 100644 --- a/distant-core/src/net/client/reconnect.rs +++ b/distant-core/src/net/client/reconnect.rs @@ -2,6 +2,7 @@ use std::io; use std::time::Duration; use log::*; +use serde::{Deserialize, Serialize}; use strum::Display; use tokio::sync::watch; use tokio::task::JoinHandle; @@ -47,8 +48,9 @@ impl ConnectionWatcher { } /// Represents the state of a connection. -#[derive(Copy, Clone, Debug, Display, PartialEq, Eq)] +#[derive(Copy, Clone, Debug, Display, PartialEq, Eq, Serialize, Deserialize)] #[strum(serialize_all = "snake_case")] +#[serde(rename_all = "snake_case")] pub enum ConnectionState { /// Connection is not active, but currently going through reconnection process. Reconnecting, @@ -254,8 +256,8 @@ impl ReconnectStrategy { } } - /// Returns the initial duration to sleep. - fn initial_sleep_duration(&self) -> Duration { + /// Returns the initial duration to sleep based on the strategy variant. + pub fn initial_sleep_duration(&self) -> Duration { match self { ReconnectStrategy::Fail => Duration::new(0, 0), ReconnectStrategy::ExponentialBackoff { base, .. } => *base, @@ -264,8 +266,8 @@ impl ReconnectStrategy { } } - /// Adjusts next sleep duration based on the strategy. - fn adjust_sleep(&self, prev: Option, curr: Duration) -> Duration { + /// Adjusts next sleep duration based on the strategy variant. + pub fn adjust_sleep(&self, prev: Option, curr: Duration) -> Duration { match self { ReconnectStrategy::Fail => Duration::new(0, 0), ReconnectStrategy::ExponentialBackoff { factor, .. } => { @@ -768,4 +770,303 @@ mod tests { let result = strategy.reconnect(&mut mock).await; assert!(result.is_err()); } + + // --------------------------------------------------------------- + // ConnectionState serde round-trip tests + // --------------------------------------------------------------- + + #[test] + fn connection_state_reconnecting_serde_round_trip() { + let state = ConnectionState::Reconnecting; + let json = serde_json::to_string(&state).unwrap(); + assert_eq!(json, "\"reconnecting\""); + let restored: ConnectionState = serde_json::from_str(&json).unwrap(); + assert_eq!(restored, ConnectionState::Reconnecting); + } + + #[test] + fn connection_state_connected_serde_round_trip() { + let state = ConnectionState::Connected; + let json = serde_json::to_string(&state).unwrap(); + assert_eq!(json, "\"connected\""); + let restored: ConnectionState = serde_json::from_str(&json).unwrap(); + assert_eq!(restored, ConnectionState::Connected); + } + + #[test] + fn connection_state_disconnected_serde_round_trip() { + let state = ConnectionState::Disconnected; + let json = serde_json::to_string(&state).unwrap(); + assert_eq!(json, "\"disconnected\""); + let restored: ConnectionState = serde_json::from_str(&json).unwrap(); + assert_eq!(restored, ConnectionState::Disconnected); + } + + #[test] + fn connection_state_should_reject_unknown_variant() { + let result = serde_json::from_str::("\"invalid\""); + assert!(result.is_err()); + } + + #[test] + fn connection_state_should_reject_capitalized_variant() { + // Ensures snake_case renaming is enforced + let result = serde_json::from_str::("\"Reconnecting\""); + assert!(result.is_err()); + } + + #[test] + fn connection_state_should_reject_non_string_json_types() { + // Number + let result = serde_json::from_str::("42"); + assert!(result.is_err()); + + // Null + let result = serde_json::from_str::("null"); + assert!(result.is_err()); + + // Object + let result = serde_json::from_str::(r#"{"variant":"connected"}"#); + assert!(result.is_err()); + + // Array + let result = serde_json::from_str::(r#"["connected"]"#); + assert!(result.is_err()); + } + + #[test] + fn connection_state_serde_display_and_serde_names_are_consistent() { + // Verify that Display (strum) and serde both use snake_case and agree + for state in [ + ConnectionState::Reconnecting, + ConnectionState::Connected, + ConnectionState::Disconnected, + ] { + let display_name = state.to_string(); + let serde_json_str = serde_json::to_string(&state).unwrap(); + // serde_json wraps in quotes: "reconnecting" + let serde_name = serde_json_str.trim_matches('"'); + assert_eq!( + display_name, serde_name, + "Display and serde names should match for {state:?}" + ); + } + } + + // --------------------------------------------------------------- + // ReconnectStrategy::initial_sleep_duration() tests + // --------------------------------------------------------------- + + #[test] + fn initial_sleep_duration_should_return_zero_for_fail() { + let strategy = ReconnectStrategy::Fail; + assert_eq!(strategy.initial_sleep_duration(), Duration::new(0, 0)); + } + + #[test] + fn initial_sleep_duration_should_return_base_for_exponential_backoff() { + let base = Duration::from_millis(250); + let strategy = ReconnectStrategy::ExponentialBackoff { + base, + factor: 2.0, + max_duration: None, + max_retries: None, + timeout: None, + }; + assert_eq!(strategy.initial_sleep_duration(), base); + } + + #[test] + fn initial_sleep_duration_should_return_base_for_fibonacci_backoff() { + let base = Duration::from_secs(1); + let strategy = ReconnectStrategy::FibonacciBackoff { + base, + max_duration: None, + max_retries: None, + timeout: None, + }; + assert_eq!(strategy.initial_sleep_duration(), base); + } + + #[test] + fn initial_sleep_duration_should_return_interval_for_fixed_interval() { + let interval = Duration::from_millis(500); + let strategy = ReconnectStrategy::FixedInterval { + interval, + max_retries: None, + timeout: None, + }; + assert_eq!(strategy.initial_sleep_duration(), interval); + } + + // --------------------------------------------------------------- + // ReconnectStrategy::adjust_sleep() tests + // --------------------------------------------------------------- + + #[test] + fn adjust_sleep_should_return_zero_for_fail() { + let strategy = ReconnectStrategy::Fail; + let result = strategy.adjust_sleep(None, Duration::from_millis(100)); + assert_eq!(result, Duration::new(0, 0)); + } + + #[test] + fn adjust_sleep_should_return_zero_for_fail_with_previous() { + let strategy = ReconnectStrategy::Fail; + let result = + strategy.adjust_sleep(Some(Duration::from_millis(50)), Duration::from_millis(100)); + assert_eq!(result, Duration::new(0, 0)); + } + + #[test] + fn adjust_sleep_should_multiply_by_factor_for_exponential_backoff() { + let strategy = ReconnectStrategy::ExponentialBackoff { + base: Duration::from_millis(100), + factor: 2.0, + max_duration: None, + max_retries: None, + timeout: None, + }; + // 100ms * 2.0 = 200ms + let result = strategy.adjust_sleep(None, Duration::from_millis(100)); + assert_eq!(result, Duration::from_millis(200)); + } + + #[test] + fn adjust_sleep_should_handle_fractional_factor_for_exponential_backoff() { + let strategy = ReconnectStrategy::ExponentialBackoff { + base: Duration::from_millis(100), + factor: 1.5, + max_duration: None, + max_retries: None, + timeout: None, + }; + // 200ms * 1.5 = 300ms + let result = strategy.adjust_sleep(None, Duration::from_millis(200)); + assert_eq!(result, Duration::from_millis(300)); + } + + #[test] + fn adjust_sleep_should_handle_large_values_for_exponential_backoff() { + let strategy = ReconnectStrategy::ExponentialBackoff { + base: Duration::from_millis(100), + factor: 2.0, + max_duration: None, + max_retries: None, + timeout: None, + }; + // Very large current sleep should not panic; should cap at u64::MAX ms + let result = strategy.adjust_sleep(None, Duration::from_millis(u64::MAX)); + assert_eq!(result, Duration::from_millis(u64::MAX)); + } + + #[test] + fn adjust_sleep_should_add_previous_and_current_for_fibonacci_backoff() { + let strategy = ReconnectStrategy::FibonacciBackoff { + base: Duration::from_millis(100), + max_duration: None, + max_retries: None, + timeout: None, + }; + // prev=100ms, curr=200ms => next=300ms + let result = + strategy.adjust_sleep(Some(Duration::from_millis(100)), Duration::from_millis(200)); + assert_eq!(result, Duration::from_millis(300)); + } + + #[test] + fn adjust_sleep_should_use_zero_when_previous_is_none_for_fibonacci_backoff() { + let strategy = ReconnectStrategy::FibonacciBackoff { + base: Duration::from_millis(100), + max_duration: None, + max_retries: None, + timeout: None, + }; + // prev=None (treated as 0), curr=100ms => next=100ms + let result = strategy.adjust_sleep(None, Duration::from_millis(100)); + assert_eq!(result, Duration::from_millis(100)); + } + + #[test] + fn adjust_sleep_should_handle_overflow_for_fibonacci_backoff() { + let strategy = ReconnectStrategy::FibonacciBackoff { + base: Duration::from_millis(100), + max_duration: None, + max_retries: None, + timeout: None, + }; + // Adding two max durations should saturate to Duration::MAX + let result = strategy.adjust_sleep(Some(Duration::MAX), Duration::MAX); + assert_eq!(result, Duration::MAX); + } + + #[test] + fn adjust_sleep_should_return_current_unchanged_for_fixed_interval() { + let strategy = ReconnectStrategy::FixedInterval { + interval: Duration::from_millis(500), + max_retries: None, + timeout: None, + }; + let current = Duration::from_millis(500); + let result = strategy.adjust_sleep(None, current); + assert_eq!(result, current); + } + + #[test] + fn adjust_sleep_should_ignore_previous_for_fixed_interval() { + let strategy = ReconnectStrategy::FixedInterval { + interval: Duration::from_millis(500), + max_retries: None, + timeout: None, + }; + let current = Duration::from_millis(500); + let result = strategy.adjust_sleep(Some(Duration::from_millis(100)), current); + assert_eq!(result, current); + } + + #[test] + fn adjust_sleep_should_grow_exponential_backoff_sequence_correctly() { + let strategy = ReconnectStrategy::ExponentialBackoff { + base: Duration::from_millis(100), + factor: 2.0, + max_duration: None, + max_retries: None, + timeout: None, + }; + // Simulate a full backoff sequence: 100 -> 200 -> 400 -> 800 + let step1 = strategy.adjust_sleep(None, Duration::from_millis(100)); + assert_eq!(step1, Duration::from_millis(200)); + + let step2 = strategy.adjust_sleep(Some(Duration::from_millis(100)), step1); + assert_eq!(step2, Duration::from_millis(400)); + + let step3 = strategy.adjust_sleep(Some(step1), step2); + assert_eq!(step3, Duration::from_millis(800)); + } + + #[test] + fn adjust_sleep_should_grow_fibonacci_backoff_sequence_correctly() { + let strategy = ReconnectStrategy::FibonacciBackoff { + base: Duration::from_millis(100), + max_duration: None, + max_retries: None, + timeout: None, + }; + // Fibonacci sequence starting from base=100ms: + // initial = 100, then: 0+100=100, 100+100=200, 100+200=300, 200+300=500 + let step1 = strategy.adjust_sleep(None, Duration::from_millis(100)); + assert_eq!(step1, Duration::from_millis(100)); + + let step2 = + strategy.adjust_sleep(Some(Duration::from_millis(100)), Duration::from_millis(100)); + assert_eq!(step2, Duration::from_millis(200)); + + let step3 = + strategy.adjust_sleep(Some(Duration::from_millis(100)), Duration::from_millis(200)); + assert_eq!(step3, Duration::from_millis(300)); + + let step4 = + strategy.adjust_sleep(Some(Duration::from_millis(200)), Duration::from_millis(300)); + assert_eq!(step4, Duration::from_millis(500)); + } } diff --git a/distant-core/src/net/common/listener/tcp.rs b/distant-core/src/net/common/listener/tcp.rs index 6bc5aee2d..04c419e89 100644 --- a/distant-core/src/net/common/listener/tcp.rs +++ b/distant-core/src/net/common/listener/tcp.rs @@ -4,6 +4,7 @@ use std::{fmt, io}; use tokio::net::TcpListener as TokioTcpListener; use super::Listener; +use crate::net::common::transport::configure_tcp_keepalive; use crate::net::common::{PortRange, TcpTransport}; /// Represents a [`Listener`] for incoming connections over TCP @@ -55,6 +56,9 @@ impl Listener for TcpListener { async fn accept(&mut self) -> io::Result { let (stream, peer_addr) = TokioTcpListener::accept(&self.inner).await?; + if let Err(e) = configure_tcp_keepalive(&stream) { + log::warn!("Failed to configure TCP keepalive: {e}"); + } Ok(TcpTransport { addr: peer_addr.ip(), port: peer_addr.port(), @@ -167,4 +171,37 @@ mod tests { // Verify that the task has completed by waiting on it let _ = task.await.expect("Listener task failed unexpectedly"); } + + #[test(tokio::test)] + async fn accept_should_produce_transport_with_keepalive_enabled() { + let addr = IpAddr::V6(Ipv6Addr::LOCALHOST); + let mut listener = TcpListener::bind(addr, 0u16) + .await + .expect("Failed to bind listener"); + + let server_addr = SocketAddr::from((addr, listener.port())); + + // Connect a client in a background task so accept() can complete + let connect_task: JoinHandle> = + tokio::spawn(async move { + let stream = tokio::net::TcpStream::connect(server_addr).await?; + Ok(stream) + }); + + // Accept the connection through our TcpListener (which calls configure_tcp_keepalive) + let transport = listener + .accept() + .await + .expect("Failed to accept connection"); + + // Verify keepalive is enabled on the accepted transport's inner stream + let sock_ref = socket2::SockRef::from(&transport.inner); + assert!( + sock_ref.keepalive().expect("Failed to query keepalive"), + "SO_KEEPALIVE should be enabled on accepted TcpTransport" + ); + + // Clean up + let _ = connect_task.await; + } } diff --git a/distant-core/src/net/common/transport.rs b/distant-core/src/net/common/transport.rs index 2138d68a2..84293ff58 100644 --- a/distant-core/src/net/common/transport.rs +++ b/distant-core/src/net/common/transport.rs @@ -10,6 +10,7 @@ mod inmemory; pub use inmemory::*; mod tcp; +pub(crate) use tcp::configure_tcp_keepalive; pub use tcp::*; #[cfg(test)] diff --git a/distant-core/src/net/common/transport/tcp.rs b/distant-core/src/net/common/transport/tcp.rs index c2ae66f70..db11dcccb 100644 --- a/distant-core/src/net/common/transport/tcp.rs +++ b/distant-core/src/net/common/transport/tcp.rs @@ -1,10 +1,30 @@ use std::net::IpAddr; +use std::time::Duration; use std::{fmt, io}; use tokio::net::{TcpStream, ToSocketAddrs}; use super::{Interest, Ready, Reconnectable, Transport}; +/// Idle time before the first keepalive probe is sent. +const TCP_KEEPALIVE_TIME: Duration = Duration::from_secs(15); + +/// Interval between successive keepalive probes after the initial probe. +const TCP_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(5); + +/// Configures TCP keepalive on the given stream for dead connection detection. +/// +/// Sets `SO_KEEPALIVE` with a 15-second idle time and 5-second probe interval. +/// Uses [`socket2::SockRef`] which works on both Unix (`AsRawFd`) and Windows +/// (`AsRawSocket`) via tokio's [`TcpStream`]. +pub(crate) fn configure_tcp_keepalive(stream: &TcpStream) -> io::Result<()> { + let sock_ref = socket2::SockRef::from(stream); + let keepalive = socket2::TcpKeepalive::new() + .with_time(TCP_KEEPALIVE_TIME) + .with_interval(TCP_KEEPALIVE_INTERVAL); + sock_ref.set_tcp_keepalive(&keepalive) +} + /// Represents a [`Transport`] that leverages a TCP stream pub struct TcpTransport { pub(crate) addr: IpAddr, @@ -17,6 +37,9 @@ impl TcpTransport { /// IP address and port pub async fn connect(addrs: impl ToSocketAddrs) -> io::Result { let stream = TcpStream::connect(addrs).await?; + if let Err(e) = configure_tcp_keepalive(&stream) { + log::warn!("Failed to configure TCP keepalive: {e}"); + } let addr = stream.peer_addr()?; Ok(Self { addr: addr.ip(), @@ -50,7 +73,11 @@ impl Reconnectable for TcpTransport { &'a mut self, ) -> std::pin::Pin> + Send + 'a>> { Box::pin(async move { - self.inner = TcpStream::connect((self.addr, self.port)).await?; + let stream = TcpStream::connect((self.addr, self.port)).await?; + if let Err(e) = configure_tcp_keepalive(&stream) { + log::warn!("Failed to configure TCP keepalive: {e}"); + } + self.inner = stream; Ok(()) }) } @@ -229,4 +256,113 @@ mod tests { // Verify that the task has completed by waiting on it let _ = task.await.expect("Server task failed unexpectedly"); } + + #[test(tokio::test)] + async fn configure_tcp_keepalive_should_enable_keepalive_on_stream() { + let listener = TcpListener::bind((IpAddr::V6(Ipv6Addr::LOCALHOST), 0u16)) + .await + .expect("Failed to bind listener"); + let addr = listener.local_addr().expect("Failed to get local addr"); + + let (client, _server) = tokio::join!(TcpStream::connect(addr), listener.accept()); + let client = client.expect("Failed to connect"); + + configure_tcp_keepalive(&client).expect("configure_tcp_keepalive failed"); + + // Verify SO_KEEPALIVE is actually enabled on the underlying socket + let sock_ref = socket2::SockRef::from(&client); + assert!( + sock_ref.keepalive().expect("Failed to query keepalive"), + "SO_KEEPALIVE should be enabled after configure_tcp_keepalive" + ); + } + + #[test(tokio::test)] + async fn connect_should_produce_transport_with_keepalive_enabled() { + let listener = TcpListener::bind((IpAddr::V6(Ipv6Addr::LOCALHOST), 0u16)) + .await + .expect("Failed to bind listener"); + let addr = listener.local_addr().expect("Failed to get local addr"); + + // Accept in a background task so connect() can complete + let accept_task: JoinHandle> = tokio::spawn(async move { + let (stream, _) = listener.accept().await?; + Ok(stream) + }); + + let transport = TcpTransport::connect(addr) + .await + .expect("Failed to connect"); + + // Verify keepalive is enabled on the transport's inner stream + let sock_ref = socket2::SockRef::from(&transport.inner); + assert!( + sock_ref.keepalive().expect("Failed to query keepalive"), + "SO_KEEPALIVE should be enabled on TcpTransport after connect()" + ); + + // Clean up the accept task + let _ = accept_task.await; + } + + #[test(tokio::test)] + async fn reconnect_should_produce_transport_with_keepalive_enabled() { + let (tx, rx) = oneshot::channel(); + + // Spawn a task that will wait for a connection, send data, + // and receive data that it will return in the task + let task: JoinHandle> = tokio::spawn(start_and_run_server(tx)); + + // Wait for the server to be ready + let addr = rx.await.expect("Failed to get server address"); + + // Connect to the server + let mut conn = TcpTransport::connect(&addr) + .await + .expect("Conn failed to connect"); + + // Kill the server to make the connection fail + task.abort(); + + // Verify the connection fails by trying to read from it (should get connection reset) + conn.readable() + .await + .expect("Failed to wait for conn to be readable"); + let res = conn.read_exact(&mut [0; 10]).await; + assert!( + matches!(res, Ok(0) | Err(_)), + "Unexpected read result: {res:?}" + ); + + // Restart the server on the same address + let task: JoinHandle> = tokio::spawn(run_server( + TcpListener::bind(addr) + .await + .expect("Failed to rebind server"), + )); + + // Reconnect to the server + conn.reconnect().await.expect("Conn failed to reconnect"); + + // Verify keepalive is enabled on the reconnected transport's inner stream + let sock_ref = socket2::SockRef::from(&conn.inner); + assert!( + sock_ref.keepalive().expect("Failed to query keepalive"), + "SO_KEEPALIVE should be enabled on TcpTransport after reconnect()" + ); + + // Complete the server protocol so the task finishes cleanly + let mut buf: [u8; 10] = [0; 10]; + conn.read_exact(&mut buf) + .await + .expect("Conn failed to read after reconnect"); + assert_eq!(&buf, b"hello conn"); + + conn.write_all(b"hello server") + .await + .expect("Conn failed to write after reconnect"); + + // Verify that the task has completed by waiting on it + let _ = task.await.expect("Server task failed unexpectedly"); + } } diff --git a/distant-core/src/net/manager/client.rs b/distant-core/src/net/manager/client.rs index 16826653c..8abb0e676 100644 --- a/distant-core/src/net/manager/client.rs +++ b/distant-core/src/net/manager/client.rs @@ -4,8 +4,8 @@ use crate::auth::AuthHandler; use crate::auth::msg::{Authentication, AuthenticationResponse}; use log::*; -use crate::net::client::Client; -use crate::net::common::{ConnectionId, Destination, Map, Request}; +use crate::net::client::{Client, Mailbox}; +use crate::net::common::{ConnectionId, Destination, Map, Request, Response}; use crate::net::manager::data::{ ConnectionInfo, ConnectionList, ManagerRequest, ManagerResponse, SemVer, }; @@ -294,6 +294,54 @@ impl ManagerClient { )), } } + + /// Subscribe to connection state change events. + /// + /// Returns a mailbox that will receive `ConnectionStateChanged` responses + /// whenever a managed connection changes state. + pub async fn subscribe_connection_events( + &mut self, + ) -> io::Result>> { + trace!("subscribe_connection_events()"); + let mut mailbox = self.mail(ManagerRequest::SubscribeConnectionEvents).await?; + + // Wait for the subscription confirmation + while let Some(res) = mailbox.next().await { + match res.payload { + ManagerResponse::SubscribedConnectionEvents => return Ok(mailbox), + ManagerResponse::Error { description } => { + return Err(io::Error::other(description)); + } + _ => continue, + } + } + + Err(io::Error::new( + io::ErrorKind::ConnectionAborted, + "Connection closed before subscription confirmed", + )) + } + + /// Request reconnection of a specific connection. + pub async fn reconnect(&mut self, id: ConnectionId) -> io::Result<()> { + trace!("reconnect({})", id); + let mut mailbox = self.mail(ManagerRequest::Reconnect { id }).await?; + + while let Some(res) = mailbox.next().await { + match res.payload { + ManagerResponse::ReconnectInitiated { .. } => return Ok(()), + ManagerResponse::Error { description } => { + return Err(io::Error::other(description)); + } + _ => continue, + } + } + + Err(io::Error::new( + io::ErrorKind::ConnectionAborted, + "Connection closed before reconnect confirmed", + )) + } } #[cfg(test)] diff --git a/distant-core/src/net/manager/data/request.rs b/distant-core/src/net/manager/data/request.rs index 28a1e971d..b77c6b7af 100644 --- a/distant-core/src/net/manager/data/request.rs +++ b/distant-core/src/net/manager/data/request.rs @@ -68,4 +68,133 @@ pub enum ManagerRequest { /// Retrieve list of connections being managed List, + + /// Subscribe to connection state change events. + /// After subscribing, the client receives unsolicited `ConnectionStateChanged` responses. + SubscribeConnectionEvents, + + /// Request reconnection of a specific connection. + Reconnect { + /// Id of the connection to reconnect + id: ConnectionId, + }, +} + +#[cfg(test)] +mod tests { + use test_log::test; + + use super::*; + + #[test] + fn subscribe_connection_events_serde_round_trip() { + let req = ManagerRequest::SubscribeConnectionEvents; + let json = serde_json::to_string(&req).unwrap(); + let val: serde_json::Value = serde_json::from_str(&json).unwrap(); + assert_eq!(val["type"], "subscribe_connection_events"); + + let restored: ManagerRequest = serde_json::from_str(&json).unwrap(); + assert!(matches!( + restored, + ManagerRequest::SubscribeConnectionEvents + )); + } + + #[test] + fn subscribe_connection_events_from_raw_json() { + let json = r#"{"type":"subscribe_connection_events"}"#; + let req: ManagerRequest = serde_json::from_str(json).unwrap(); + assert!(matches!(req, ManagerRequest::SubscribeConnectionEvents)); + } + + #[test] + fn reconnect_serde_round_trip() { + let req = ManagerRequest::Reconnect { id: 42 }; + let json = serde_json::to_string(&req).unwrap(); + let val: serde_json::Value = serde_json::from_str(&json).unwrap(); + assert_eq!(val["type"], "reconnect"); + assert_eq!(val["id"], 42); + + let restored: ManagerRequest = serde_json::from_str(&json).unwrap(); + match restored { + ManagerRequest::Reconnect { id } => assert_eq!(id, 42), + other => panic!("Expected Reconnect, got {other:?}"), + } + } + + #[test] + fn reconnect_from_raw_json() { + let json = r#"{"type":"reconnect","id":99}"#; + let req: ManagerRequest = serde_json::from_str(json).unwrap(); + match req { + ManagerRequest::Reconnect { id } => assert_eq!(id, 99), + other => panic!("Expected Reconnect, got {other:?}"), + } + } + + #[test] + fn reconnect_should_reject_missing_id() { + let json = r#"{"type":"reconnect"}"#; + let result = serde_json::from_str::(json); + assert!(result.is_err()); + } + + #[test] + fn subscribe_connection_events_ignores_extra_fields_due_to_serde_unit_variant_limitation() { + // NOTE: serde's `deny_unknown_fields` with `#[serde(tag = "type")]` does NOT + // reject extra fields on unit variants -- the tag is consumed and remaining + // fields are silently ignored. This test documents this known serde behavior. + let json = r#"{"type":"subscribe_connection_events","extra":true}"#; + let result = serde_json::from_str::(json); + assert!(matches!( + result.unwrap(), + ManagerRequest::SubscribeConnectionEvents + )); + } + + #[test] + fn reconnect_should_reject_extra_fields() { + let json = r#"{"type":"reconnect","id":1,"extra":"field"}"#; + let result = serde_json::from_str::(json); + assert!(result.is_err()); + } + + #[test] + fn reconnect_should_reject_wrong_id_type() { + // String instead of number + let json = r#"{"type":"reconnect","id":"not_a_number"}"#; + let result = serde_json::from_str::(json); + assert!(result.is_err()); + } + + #[test] + fn subscribe_connection_events_should_reject_wrong_case_type() { + // PascalCase instead of snake_case + let json = r#"{"type":"SubscribeConnectionEvents"}"#; + let result = serde_json::from_str::(json); + assert!(result.is_err()); + } + + #[test] + fn reconnect_should_reject_wrong_case_type() { + let json = r#"{"type":"Reconnect","id":1}"#; + let result = serde_json::from_str::(json); + assert!(result.is_err()); + } + + #[test] + fn reconnect_should_preserve_id_value_through_round_trip() { + // Test boundary values for ConnectionId (u32) + for id in [0u32, 1, u32::MAX] { + let req = ManagerRequest::Reconnect { id }; + let json = serde_json::to_string(&req).unwrap(); + let restored: ManagerRequest = serde_json::from_str(&json).unwrap(); + match restored { + ManagerRequest::Reconnect { id: restored_id } => { + assert_eq!(restored_id, id, "id {id} should survive round-trip") + } + other => panic!("Expected Reconnect, got {other:?}"), + } + } + } } diff --git a/distant-core/src/net/manager/data/response.rs b/distant-core/src/net/manager/data/response.rs index 4183b9427..e29ce430d 100644 --- a/distant-core/src/net/manager/data/response.rs +++ b/distant-core/src/net/manager/data/response.rs @@ -2,6 +2,7 @@ use crate::auth::msg::Authentication; use serde::{Deserialize, Serialize}; use super::{ConnectionInfo, ConnectionList, ManagerAuthenticationId, ManagerChannelId, SemVer}; +use crate::net::client::ConnectionState; use crate::net::common::{ConnectionId, Destination, UntypedResponse}; #[derive(Clone, Debug, Serialize, Deserialize)] @@ -60,6 +61,24 @@ pub enum ManagerResponse { /// Id of the channel id: ManagerChannelId, }, + + /// Unsolicited notification of a connection state change. + /// Sent to clients that have subscribed via `SubscribeConnectionEvents`. + ConnectionStateChanged { + /// Id of the connection whose state changed + id: ConnectionId, + /// New connection state + state: ConnectionState, + }, + + /// Confirmation that connection event subscription was established. + SubscribedConnectionEvents, + + /// Confirmation that a reconnection attempt has been initiated. + ReconnectInitiated { + /// Id of the connection + id: ConnectionId, + }, } impl From for ManagerResponse { @@ -69,3 +88,288 @@ impl From for ManagerResponse { } } } + +#[cfg(test)] +mod tests { + use test_log::test; + + use super::*; + + // --------------------------------------------------------------- + // ConnectionStateChanged serde round-trip + // --------------------------------------------------------------- + + #[test] + fn connection_state_changed_serde_round_trip() { + let resp = ManagerResponse::ConnectionStateChanged { + id: 7, + state: ConnectionState::Disconnected, + }; + let json = serde_json::to_string(&resp).unwrap(); + let val: serde_json::Value = serde_json::from_str(&json).unwrap(); + assert_eq!(val["type"], "connection_state_changed"); + assert_eq!(val["id"], 7); + assert_eq!(val["state"], "disconnected"); + + let restored: ManagerResponse = serde_json::from_str(&json).unwrap(); + match restored { + ManagerResponse::ConnectionStateChanged { id, state } => { + assert_eq!(id, 7); + assert_eq!(state, ConnectionState::Disconnected); + } + other => panic!("Expected ConnectionStateChanged, got {other:?}"), + } + } + + #[test] + fn connection_state_changed_with_reconnecting_state() { + let resp = ManagerResponse::ConnectionStateChanged { + id: 15, + state: ConnectionState::Reconnecting, + }; + let json = serde_json::to_string(&resp).unwrap(); + let val: serde_json::Value = serde_json::from_str(&json).unwrap(); + assert_eq!(val["state"], "reconnecting"); + + let restored: ManagerResponse = serde_json::from_str(&json).unwrap(); + match restored { + ManagerResponse::ConnectionStateChanged { id, state } => { + assert_eq!(id, 15); + assert_eq!(state, ConnectionState::Reconnecting); + } + other => panic!("Expected ConnectionStateChanged, got {other:?}"), + } + } + + #[test] + fn connection_state_changed_with_connected_state() { + let resp = ManagerResponse::ConnectionStateChanged { + id: 0, + state: ConnectionState::Connected, + }; + let json = serde_json::to_string(&resp).unwrap(); + let val: serde_json::Value = serde_json::from_str(&json).unwrap(); + assert_eq!(val["state"], "connected"); + + let restored: ManagerResponse = serde_json::from_str(&json).unwrap(); + match restored { + ManagerResponse::ConnectionStateChanged { id, state } => { + assert_eq!(id, 0); + assert_eq!(state, ConnectionState::Connected); + } + other => panic!("Expected ConnectionStateChanged, got {other:?}"), + } + } + + #[test] + fn connection_state_changed_from_raw_json() { + let json = r#"{"type":"connection_state_changed","id":42,"state":"disconnected"}"#; + let resp: ManagerResponse = serde_json::from_str(json).unwrap(); + match resp { + ManagerResponse::ConnectionStateChanged { id, state } => { + assert_eq!(id, 42); + assert_eq!(state, ConnectionState::Disconnected); + } + other => panic!("Expected ConnectionStateChanged, got {other:?}"), + } + } + + #[test] + fn connection_state_changed_should_reject_missing_fields() { + let json = r#"{"type":"connection_state_changed","id":1}"#; + let result = serde_json::from_str::(json); + assert!(result.is_err()); + + let json = r#"{"type":"connection_state_changed","state":"connected"}"#; + let result = serde_json::from_str::(json); + assert!(result.is_err()); + } + + #[test] + fn connection_state_changed_should_reject_invalid_state() { + let json = r#"{"type":"connection_state_changed","id":1,"state":"unknown"}"#; + let result = serde_json::from_str::(json); + assert!(result.is_err()); + } + + // --------------------------------------------------------------- + // SubscribedConnectionEvents serde round-trip + // --------------------------------------------------------------- + + #[test] + fn subscribed_connection_events_serde_round_trip() { + let resp = ManagerResponse::SubscribedConnectionEvents; + let json = serde_json::to_string(&resp).unwrap(); + let val: serde_json::Value = serde_json::from_str(&json).unwrap(); + assert_eq!(val["type"], "subscribed_connection_events"); + + let restored: ManagerResponse = serde_json::from_str(&json).unwrap(); + assert!(matches!( + restored, + ManagerResponse::SubscribedConnectionEvents + )); + } + + #[test] + fn subscribed_connection_events_from_raw_json() { + let json = r#"{"type":"subscribed_connection_events"}"#; + let resp: ManagerResponse = serde_json::from_str(json).unwrap(); + assert!(matches!(resp, ManagerResponse::SubscribedConnectionEvents)); + } + + #[test] + fn subscribed_connection_events_ignores_extra_fields_due_to_serde_unit_variant_limitation() { + // NOTE: serde's `deny_unknown_fields` with `#[serde(tag = "type")]` does NOT + // reject extra fields on unit variants -- the tag is consumed and remaining + // fields are silently ignored. This test documents this known serde behavior. + let json = r#"{"type":"subscribed_connection_events","extra":1}"#; + let result = serde_json::from_str::(json); + assert!(matches!( + result.unwrap(), + ManagerResponse::SubscribedConnectionEvents + )); + } + + // --------------------------------------------------------------- + // ReconnectInitiated serde round-trip + // --------------------------------------------------------------- + + #[test] + fn reconnect_initiated_serde_round_trip() { + let resp = ManagerResponse::ReconnectInitiated { id: 55 }; + let json = serde_json::to_string(&resp).unwrap(); + let val: serde_json::Value = serde_json::from_str(&json).unwrap(); + assert_eq!(val["type"], "reconnect_initiated"); + assert_eq!(val["id"], 55); + + let restored: ManagerResponse = serde_json::from_str(&json).unwrap(); + match restored { + ManagerResponse::ReconnectInitiated { id } => assert_eq!(id, 55), + other => panic!("Expected ReconnectInitiated, got {other:?}"), + } + } + + #[test] + fn reconnect_initiated_from_raw_json() { + let json = r#"{"type":"reconnect_initiated","id":100}"#; + let resp: ManagerResponse = serde_json::from_str(json).unwrap(); + match resp { + ManagerResponse::ReconnectInitiated { id } => assert_eq!(id, 100), + other => panic!("Expected ReconnectInitiated, got {other:?}"), + } + } + + #[test] + fn reconnect_initiated_should_reject_missing_id() { + let json = r#"{"type":"reconnect_initiated"}"#; + let result = serde_json::from_str::(json); + assert!(result.is_err()); + } + + #[test] + fn reconnect_initiated_should_reject_extra_fields() { + let json = r#"{"type":"reconnect_initiated","id":1,"extra":"nope"}"#; + let result = serde_json::from_str::(json); + assert!(result.is_err()); + } + + // --------------------------------------------------------------- + // ConnectionStateChanged additional edge cases + // --------------------------------------------------------------- + + #[test] + fn connection_state_changed_should_reject_extra_fields() { + let json = + r#"{"type":"connection_state_changed","id":1,"state":"connected","extra":"nope"}"#; + let result = serde_json::from_str::(json); + assert!(result.is_err()); + } + + #[test] + fn connection_state_changed_should_reject_capitalized_state() { + // ConnectionState uses snake_case, so "Connected" should fail + let json = r#"{"type":"connection_state_changed","id":1,"state":"Connected"}"#; + let result = serde_json::from_str::(json); + assert!(result.is_err()); + } + + #[test] + fn reconnect_initiated_should_reject_wrong_id_type() { + let json = r#"{"type":"reconnect_initiated","id":"not_a_number"}"#; + let result = serde_json::from_str::(json); + assert!(result.is_err()); + } + + #[test] + fn connection_state_changed_should_reject_wrong_id_type() { + let json = r#"{"type":"connection_state_changed","id":"abc","state":"connected"}"#; + let result = serde_json::from_str::(json); + assert!(result.is_err()); + } + + // --------------------------------------------------------------- + // Wrong-case type tag rejection + // --------------------------------------------------------------- + + #[test] + fn connection_state_changed_should_reject_wrong_case_type() { + let json = r#"{"type":"ConnectionStateChanged","id":1,"state":"connected"}"#; + let result = serde_json::from_str::(json); + assert!(result.is_err()); + } + + #[test] + fn subscribed_connection_events_should_reject_wrong_case_type() { + let json = r#"{"type":"SubscribedConnectionEvents"}"#; + let result = serde_json::from_str::(json); + assert!(result.is_err()); + } + + #[test] + fn reconnect_initiated_should_reject_wrong_case_type() { + let json = r#"{"type":"ReconnectInitiated","id":1}"#; + let result = serde_json::from_str::(json); + assert!(result.is_err()); + } + + // --------------------------------------------------------------- + // Boundary value round-trips + // --------------------------------------------------------------- + + #[test] + fn connection_state_changed_should_preserve_boundary_id_values() { + for id in [0u32, 1, u32::MAX] { + let resp = ManagerResponse::ConnectionStateChanged { + id, + state: ConnectionState::Connected, + }; + let json = serde_json::to_string(&resp).unwrap(); + let restored: ManagerResponse = serde_json::from_str(&json).unwrap(); + match restored { + ManagerResponse::ConnectionStateChanged { + id: restored_id, + state, + } => { + assert_eq!(restored_id, id, "id {id} should survive round-trip"); + assert_eq!(state, ConnectionState::Connected); + } + other => panic!("Expected ConnectionStateChanged, got {other:?}"), + } + } + } + + #[test] + fn reconnect_initiated_should_preserve_boundary_id_values() { + for id in [0u32, 1, u32::MAX] { + let resp = ManagerResponse::ReconnectInitiated { id }; + let json = serde_json::to_string(&resp).unwrap(); + let restored: ManagerResponse = serde_json::from_str(&json).unwrap(); + match restored { + ManagerResponse::ReconnectInitiated { id: restored_id } => { + assert_eq!(restored_id, id, "id {id} should survive round-trip"); + } + other => panic!("Expected ReconnectInitiated, got {other:?}"), + } + } + } +} diff --git a/distant-core/src/net/manager/server.rs b/distant-core/src/net/manager/server.rs index 71ed73955..64be105c5 100644 --- a/distant-core/src/net/manager/server.rs +++ b/distant-core/src/net/manager/server.rs @@ -1,18 +1,22 @@ use std::collections::HashMap; +use std::future::Future; use std::io; +use std::pin::Pin; use std::sync::Arc; -use crate::auth::msg::AuthenticationResponse; use log::*; -use tokio::sync::{RwLock, oneshot}; +use tokio::sync::{RwLock, broadcast, mpsc, oneshot}; +use crate::auth::Authenticator; +use crate::auth::msg::*; +use crate::net::client::ConnectionState; use crate::net::common::{ConnectionId, Map}; use crate::net::manager::{ ConnectionInfo, ConnectionList, ManagerAuthenticationId, ManagerChannelId, ManagerRequest, ManagerResponse, SemVer, }; use crate::net::server::{RequestCtx, Server, ServerHandler}; -use crate::plugin::extract_scheme; +use crate::plugin::{Plugin, extract_scheme}; mod authentication; pub use authentication::*; @@ -32,12 +36,22 @@ pub struct ManagerServer { /// enabling us to cancel the tasks on demand channels: RwLock>, - /// Mapping of connection id -> connection - connections: RwLock>, + /// Mapping of connection id -> connection. + /// Wrapped in `Arc` so the background reconnection task can access connections + /// without holding a borrow on `ManagerServer`. + connections: Arc>>, /// Mapping of auth id -> callback registry: Arc>>>, + + /// Channel for sending connection death notifications from monitor tasks. + /// Each [`ManagerConnection`] spawned by this server receives a clone to report + /// when its underlying transport disconnects. + death_tx: mpsc::UnboundedSender, + + /// Broadcast channel for sending connection state change events to subscribed clients. + event_tx: broadcast::Sender, } impl ManagerServer { @@ -45,11 +59,42 @@ impl ManagerServer { /// methods. The provided `config` will be used to configure the launch and connect handlers /// for the server as well as provide other defaults. pub fn new(config: Config) -> Server { + let (death_tx, mut death_rx) = mpsc::unbounded_channel(); + let (event_tx, _event_rx) = broadcast::channel(16); + let connections = Arc::new(RwLock::new(HashMap::new())); + + // Spawn a background task that handles connection deaths. + // When a connection dies, this task orchestrates reconnection using the + // plugin's reconnect() method and retry strategy. + { + let connections = Arc::clone(&connections); + let event_tx = event_tx.clone(); + let death_tx = death_tx.clone(); + let plugins = config.plugins.clone(); + let fallback_scheme = config.connect_fallback_scheme.clone(); + tokio::spawn(async move { + while let Some(id) = death_rx.recv().await { + warn!("[Conn {id}] Connection death detected by manager"); + handle_reconnection( + id, + &connections, + &plugins, + &fallback_scheme, + &death_tx, + &event_tx, + ) + .await; + } + }); + } + Server::new().handler(Self { config, channels: RwLock::new(HashMap::new()), - connections: RwLock::new(HashMap::new()), + connections, registry: Arc::new(RwLock::new(HashMap::new())), + death_tx, + event_tx, }) } @@ -123,8 +168,13 @@ impl ManagerServer { .connect(raw_destination, &options, &mut authenticator) .await?; - let connection = - ManagerConnection::spawn(raw_destination.to_string(), options, client).await?; + let connection = ManagerConnection::spawn( + raw_destination.to_string(), + options, + client, + Some(self.death_tx.clone()), + ) + .await?; let id = connection.id; self.connections.write().await.insert(id, connection); Ok(id) @@ -192,6 +242,235 @@ impl ManagerServer { } } +/// An [`Authenticator`] that fails all interactive authentication challenges. +/// +/// Used during background reconnection where no user is available to answer +/// prompts. Plugins that rely solely on key files or ssh-agent will never +/// invoke the challenge/verify methods, so reconnection succeeds silently. +/// If the server requires interactive auth, reconnection fails immediately. +struct NonInteractiveAuthenticator; + +impl Authenticator for NonInteractiveAuthenticator { + fn initialize<'a>( + &'a mut self, + initialization: Initialization, + ) -> Pin> + Send + 'a>> { + // Accept whatever methods the server offers — let the plugin decide + Box::pin(async move { + Ok(InitializationResponse { + methods: initialization.methods, + }) + }) + } + + fn challenge<'a>( + &'a mut self, + _challenge: Challenge, + ) -> Pin> + Send + 'a>> { + Box::pin(async { + Err(io::Error::new( + io::ErrorKind::PermissionDenied, + "non-interactive reconnection cannot answer authentication challenges", + )) + }) + } + + fn verify<'a>( + &'a mut self, + _verification: Verification, + ) -> Pin> + Send + 'a>> { + // Auto-accept host verification during reconnection (already verified on first connect) + Box::pin(async { Ok(VerificationResponse { valid: true }) }) + } + + fn info<'a>( + &'a mut self, + _info: Info, + ) -> Pin> + Send + 'a>> { + Box::pin(async { Ok(()) }) + } + + fn error<'a>( + &'a mut self, + _error: Error, + ) -> Pin> + Send + 'a>> { + Box::pin(async { Ok(()) }) + } + + fn start_method<'a>( + &'a mut self, + _start_method: StartMethod, + ) -> Pin> + Send + 'a>> { + Box::pin(async { Ok(()) }) + } + + fn finished<'a>(&'a mut self) -> Pin> + Send + 'a>> { + Box::pin(async { Ok(()) }) + } +} + +/// Sends a [`ConnectionState`] change notification through the event broadcast channel. +fn notify_state_change( + event_tx: &broadcast::Sender, + id: ConnectionId, + state: ConnectionState, +) { + let _ = event_tx.send(ManagerResponse::ConnectionStateChanged { id, state }); +} + +/// Orchestrates reconnection for the connection with the given `id`. +/// +/// Steps: +/// 1. Read connection info (destination, options) and look up the plugin by scheme. +/// 2. Check if the plugin supports reconnection (`reconnect_strategy != Fail`). +/// 3. Broadcast `Reconnecting` state to subscribers. +/// 4. Execute the plugin's `reconnect()` in a retry loop using the strategy. +/// 5. On success: hot-swap the old connection via `replace_client()`, broadcast `Connected`. +/// 6. On failure: broadcast `Disconnected`. +async fn handle_reconnection( + id: ConnectionId, + connections: &RwLock>, + plugins: &HashMap>, + fallback_scheme: &str, + death_tx: &mpsc::UnboundedSender, + event_tx: &broadcast::Sender, +) { + // Step 1: Read connection info without holding the lock across await points + let (destination, options) = { + let conns = connections.read().await; + match conns.get(&id) { + Some(conn) => (conn.destination.clone(), conn.options.clone()), + None => { + warn!("[Conn {id}] Reconnection aborted: connection not found"); + return; + } + } + }; + + // Check if reconnection is disabled for this connection + if options.get("no_reconnect").is_some_and(|v| v == "true") { + info!("[Conn {id}] Reconnection disabled (--no-reconnect)"); + notify_state_change(event_tx, id, ConnectionState::Disconnected); + return; + } + + // Look up the plugin by scheme + let scheme = match extract_scheme(&destination) { + Some(scheme) => scheme.to_lowercase(), + None => fallback_scheme.to_lowercase(), + }; + + let plugin = match plugins.get(&scheme) { + Some(plugin) => Arc::clone(plugin), + None => { + error!("[Conn {id}] Reconnection aborted: no plugin for scheme '{scheme}'"); + notify_state_change(event_tx, id, ConnectionState::Disconnected); + return; + } + }; + + // Step 2: Check reconnect strategy + let strategy = plugin.reconnect_strategy(); + if strategy.is_fail() { + info!("[Conn {id}] Plugin '{scheme}' does not support reconnection"); + notify_state_change(event_tx, id, ConnectionState::Disconnected); + return; + } + + // Step 3: Broadcast Reconnecting + info!("[Conn {id}] Starting reconnection via plugin '{scheme}'"); + notify_state_change(event_tx, id, ConnectionState::Reconnecting); + + // Step 4: Retry loop using ReconnectStrategy's delay logic + let mut previous_sleep = None; + let mut current_sleep = strategy.initial_sleep_duration(); + let mut retries_remaining = strategy.max_retries(); + let timeout = strategy.timeout(); + let max_duration = strategy.max_duration(); + + let mut last_err: Option = None; + + while retries_remaining.is_none() || retries_remaining > Some(0) { + let mut authenticator = NonInteractiveAuthenticator; + + let result = match timeout { + Some(t) => { + match tokio::time::timeout( + t, + plugin.reconnect(&destination, &options, &mut authenticator), + ) + .await + { + Ok(r) => r, + Err(elapsed) => Err(elapsed.into()), + } + } + None => { + plugin + .reconnect(&destination, &options, &mut authenticator) + .await + } + }; + + match result { + Ok(new_client) => { + // Step 5: Hot-swap the connection + let mut conns = connections.write().await; + if let Some(conn) = conns.get_mut(&id) { + match conn + .replace_client(new_client, Some(death_tx.clone())) + .await + { + Ok(()) => { + info!("[Conn {id}] Reconnection succeeded"); + notify_state_change(event_tx, id, ConnectionState::Connected); + return; + } + Err(e) => { + error!("[Conn {id}] Failed to replace client after reconnect: {e}"); + last_err = Some(e); + } + } + } else { + warn!("[Conn {id}] Connection removed during reconnection"); + return; + } + } + Err(e) => { + debug!("[Conn {id}] Reconnect attempt failed: {e}"); + last_err = Some(e); + } + } + + // Decrement remaining retries if we have a limit + if let Some(remaining) = retries_remaining.as_mut() + && *remaining > 0 + { + *remaining -= 1; + } + + // Sleep before next attempt + tokio::time::sleep(current_sleep).await; + + // Update sleep duration using the strategy's backoff logic + let next_sleep = strategy.adjust_sleep(previous_sleep, current_sleep); + previous_sleep = Some(current_sleep); + current_sleep = if let Some(duration) = max_duration { + std::cmp::min(next_sleep, duration) + } else { + next_sleep + }; + } + + // Step 6: All retries exhausted + let err_msg = last_err + .as_ref() + .map(|e| e.to_string()) + .unwrap_or_else(|| "unknown error".to_string()); + error!("[Conn {id}] Reconnection failed after all retries: {err_msg}"); + notify_state_change(event_tx, id, ConnectionState::Disconnected); +} + impl ServerHandler for ManagerServer { type Request = ManagerRequest; type Response = ManagerResponse; @@ -360,6 +639,48 @@ impl ServerHandler for ManagerServer { Err(x) => ManagerResponse::from(x), } } + ManagerRequest::SubscribeConnectionEvents => { + let mut event_rx = self.event_tx.subscribe(); + let reply_clone = reply.clone(); + tokio::spawn(async move { + while let Ok(event) = event_rx.recv().await { + if reply_clone.send(event).is_err() { + break; + } + } + }); + ManagerResponse::SubscribedConnectionEvents + } + ManagerRequest::Reconnect { id } => { + // Verify the connection exists before initiating reconnection + let exists = self.connections.read().await.contains_key(&id); + if !exists { + ManagerResponse::from(io::Error::new( + io::ErrorKind::NotConnected, + "No connection found", + )) + } else { + info!("[Conn {id}] Manual reconnection requested"); + // Spawn reconnection in the background so the response is immediate + let connections = Arc::clone(&self.connections); + let plugins = self.config.plugins.clone(); + let fallback_scheme = self.config.connect_fallback_scheme.clone(); + let death_tx = self.death_tx.clone(); + let event_tx = self.event_tx.clone(); + tokio::spawn(async move { + handle_reconnection( + id, + &connections, + &plugins, + &fallback_scheme, + &death_tx, + &event_tx, + ) + .await; + }); + ManagerResponse::ReconnectInitiated { id } + } + } }; if let Err(x) = reply.send(response) { @@ -372,7 +693,7 @@ impl ServerHandler for ManagerServer { mod tests { use std::pin::Pin; - use tokio::sync::mpsc; + use tokio::sync::{broadcast, mpsc}; use super::*; use crate::auth::Authenticator; @@ -471,11 +792,16 @@ mod tests { registry: Arc::clone(®istry), }; + let (death_tx, _death_rx) = mpsc::unbounded_channel(); + let (event_tx, _event_rx) = broadcast::channel(16); + let server = ManagerServer { config, channels: RwLock::new(HashMap::new()), - connections: RwLock::new(HashMap::new()), + connections: Arc::new(RwLock::new(HashMap::new())), registry, + death_tx, + event_tx, }; (server, authenticator) @@ -610,6 +936,7 @@ mod tests { "scheme://host", "key=value".parse().unwrap(), detached_untyped_client(), + None, ) .await .unwrap(); @@ -643,6 +970,7 @@ mod tests { "scheme://host", "key=value".parse().unwrap(), detached_untyped_client(), + None, ) .await .unwrap(); @@ -653,6 +981,7 @@ mod tests { "other://host2", "key=value".parse().unwrap(), detached_untyped_client(), + None, ) .await .unwrap(); @@ -680,6 +1009,7 @@ mod tests { "scheme://host", "key=value".parse().unwrap(), detached_untyped_client(), + None, ) .await .unwrap(); @@ -691,4 +1021,147 @@ mod tests { let lock = server.connections.read().await; assert!(!lock.contains_key(&id), "Connection still exists"); } + + // --------------------------------------------------------------- + // NonInteractiveAuthenticator tests + // --------------------------------------------------------------- + + #[tokio::test] + async fn non_interactive_authenticator_initialize_should_echo_methods() { + let mut auth = NonInteractiveAuthenticator; + let init = crate::auth::msg::Initialization { + methods: vec!["publickey".to_string(), "keyboard-interactive".to_string()], + }; + let resp = auth.initialize(init).await.unwrap(); + assert_eq!( + resp.methods, + vec!["publickey".to_string(), "keyboard-interactive".to_string()] + ); + } + + #[tokio::test] + async fn non_interactive_authenticator_initialize_should_echo_empty_methods() { + let mut auth = NonInteractiveAuthenticator; + let init = crate::auth::msg::Initialization { methods: vec![] }; + let resp = auth.initialize(init).await.unwrap(); + assert!(resp.methods.is_empty()); + } + + #[tokio::test] + async fn non_interactive_authenticator_initialize_should_echo_single_method() { + let mut auth = NonInteractiveAuthenticator; + let init = crate::auth::msg::Initialization { + methods: vec!["none".to_string()], + }; + let resp = auth.initialize(init).await.unwrap(); + assert_eq!(resp.methods, vec!["none".to_string()]); + } + + #[tokio::test] + async fn non_interactive_authenticator_challenge_should_return_permission_denied() { + let mut auth = NonInteractiveAuthenticator; + let challenge = crate::auth::msg::Challenge { + questions: vec![crate::auth::msg::Question::new("Enter password")], + options: std::collections::HashMap::new(), + }; + let err = auth.challenge(challenge).await.unwrap_err(); + assert_eq!(err.kind(), io::ErrorKind::PermissionDenied); + } + + #[tokio::test] + async fn non_interactive_authenticator_challenge_should_have_descriptive_error_message() { + let mut auth = NonInteractiveAuthenticator; + let challenge = crate::auth::msg::Challenge { + questions: vec![], + options: std::collections::HashMap::new(), + }; + let err = auth.challenge(challenge).await.unwrap_err(); + assert!( + err.to_string() + .contains("non-interactive reconnection cannot answer authentication challenges"), + "Error message was: {}", + err + ); + } + + #[tokio::test] + async fn non_interactive_authenticator_verify_should_return_valid_true() { + let mut auth = NonInteractiveAuthenticator; + let verification = crate::auth::msg::Verification { + kind: crate::auth::msg::VerificationKind::Host, + text: "ssh-ed25519 AAAA...".to_string(), + }; + let resp = auth.verify(verification).await.unwrap(); + assert!(resp.valid); + } + + #[tokio::test] + async fn non_interactive_authenticator_verify_should_return_valid_for_unknown_kind() { + let mut auth = NonInteractiveAuthenticator; + let verification = crate::auth::msg::Verification { + kind: crate::auth::msg::VerificationKind::Unknown, + text: "something".to_string(), + }; + let resp = auth.verify(verification).await.unwrap(); + assert!(resp.valid); + } + + #[tokio::test] + async fn non_interactive_authenticator_info_should_succeed() { + let mut auth = NonInteractiveAuthenticator; + let info = crate::auth::msg::Info { + text: "Connecting to host...".to_string(), + }; + auth.info(info).await.unwrap(); + } + + #[tokio::test] + async fn non_interactive_authenticator_error_should_succeed() { + let mut auth = NonInteractiveAuthenticator; + let error = crate::auth::msg::Error::fatal("auth failed"); + auth.error(error).await.unwrap(); + } + + #[tokio::test] + async fn non_interactive_authenticator_start_method_should_succeed() { + let mut auth = NonInteractiveAuthenticator; + let start = crate::auth::msg::StartMethod { + method: "publickey".to_string(), + }; + auth.start_method(start).await.unwrap(); + } + + #[tokio::test] + async fn non_interactive_authenticator_finished_should_succeed() { + let mut auth = NonInteractiveAuthenticator; + auth.finished().await.unwrap(); + } + + // --------------------------------------------------------------- + // notify_state_change tests + // --------------------------------------------------------------- + + #[tokio::test] + async fn notify_state_change_should_broadcast_connection_state_changed() { + let (event_tx, mut event_rx) = broadcast::channel(16); + let id: ConnectionId = 42; + + notify_state_change(&event_tx, id, ConnectionState::Reconnecting); + + let msg = event_rx.recv().await.unwrap(); + match msg { + ManagerResponse::ConnectionStateChanged { id: recv_id, state } => { + assert_eq!(recv_id, 42); + assert_eq!(state, ConnectionState::Reconnecting); + } + other => panic!("Expected ConnectionStateChanged, got {other:?}"), + } + } + + #[tokio::test] + async fn notify_state_change_should_not_panic_with_no_subscribers() { + let (event_tx, _) = broadcast::channel::(16); + // Drop the receiver before sending -- should not panic + notify_state_change(&event_tx, 1, ConnectionState::Disconnected); + } } diff --git a/distant-core/src/net/manager/server/connection.rs b/distant-core/src/net/manager/server/connection.rs index 44e0a3581..f24098698 100644 --- a/distant-core/src/net/manager/server/connection.rs +++ b/distant-core/src/net/manager/server/connection.rs @@ -5,12 +5,12 @@ use log::*; use tokio::sync::{mpsc, oneshot}; use tokio::task::JoinHandle; -use crate::net::client::{Mailbox, UntypedClient}; +use crate::net::client::{ConnectionWatcher, Mailbox, UntypedClient}; use crate::net::common::{ConnectionId, Map, UntypedRequest, UntypedResponse}; use crate::net::manager::data::{ManagerChannelId, ManagerResponse}; use crate::net::server::ServerReply; -/// Represents a connection a distant manager has with some distant-compatible server +/// Represents a connection a distant manager has with some distant-compatible server. pub struct ManagerConnection { pub id: ConnectionId, /// Raw destination string as provided by the user (e.g. `"docker://ubuntu:22.04"`). @@ -21,6 +21,10 @@ pub struct ManagerConnection { action_task: JoinHandle<()>, request_task: JoinHandle<()>, response_task: JoinHandle<()>, + + /// Optional task that monitors the underlying connection health and sends + /// a death notification when the connection transitions to `Disconnected`. + monitor_task: Option>, } #[derive(Clone)] @@ -60,10 +64,15 @@ impl ManagerChannel { } impl ManagerConnection { + /// Spawns a new manager connection wrapping the given [`UntypedClient`]. + /// + /// If `death_tx` is provided, a background monitor task will watch the client's connection + /// health and send the connection ID through the channel when the connection dies. pub async fn spawn( destination: impl Into, options: Map, mut client: UntypedClient, + death_tx: Option>, ) -> io::Result { let destination = destination.into(); let connection_id = rand::random(); @@ -75,6 +84,9 @@ impl ManagerConnection { // never triggering! client.shutdown_on_drop(true); + // Clone the connection watcher before moving the client into tasks + let watcher = client.clone_connection_watcher(); + let (request_tx, request_rx) = mpsc::unbounded_channel(); let action_task = tokio::spawn(action_task(connection_id, rx, request_tx)); let response_task = tokio::spawn(response_task( @@ -84,6 +96,10 @@ impl ManagerConnection { )); let request_task = tokio::spawn(request_task(connection_id, client, request_rx)); + // Spawn a monitor task if a death notification channel was provided + let monitor_task = + death_tx.map(|dtx| tokio::spawn(connection_monitor(connection_id, watcher, dtx))); + Ok(Self { id: connection_id, destination, @@ -92,9 +108,66 @@ impl ManagerConnection { action_task, request_task, response_task, + monitor_task, }) } + /// Replaces the underlying client with a new one, aborting old tasks and + /// respawning them with the same [`ConnectionId`]. + /// + /// **Existing channels are invalidated** — the old action task is aborted + /// and a new one is spawned, so any [`ManagerChannel`] handles obtained + /// before this call will fail on subsequent sends. Callers must re-open + /// channels after replacement. + /// + /// If `death_tx` is provided, a new connection monitor task is spawned. + /// + /// # Errors + /// + /// Returns an error if the default mailbox cannot be assigned on the new client. + pub async fn replace_client( + &mut self, + mut client: UntypedClient, + death_tx: Option>, + ) -> io::Result<()> { + let id = self.id; + debug!("[Conn {id}] Replacing client — aborting old tasks"); + + // Abort old tasks (action_task is NOT aborted — channels live there) + self.request_task.abort(); + self.response_task.abort(); + if let Some(ref task) = self.monitor_task { + task.abort(); + } + + // Configure the new client + client.shutdown_on_drop(true); + + // Clone watcher before moving the client + let watcher = client.clone_connection_watcher(); + + // Set up new request and response tasks using the existing action tx + let (request_tx, request_rx) = mpsc::unbounded_channel(); + let mailbox = client.assign_default_mailbox(100).await?; + + self.response_task = tokio::spawn(response_task(id, mailbox, self.tx.clone())); + self.request_task = tokio::spawn(request_task(id, client, request_rx)); + + // Abort the old action task and respawn with the new request_tx. + // Channel registrations start fresh — existing ManagerChannel handles + // hold a clone of the OLD self.tx and will fail on next send. + self.action_task.abort(); + let (tx, rx) = mpsc::unbounded_channel(); + self.action_task = tokio::spawn(action_task(id, rx, request_tx)); + self.tx = tx; + + // Spawn a new monitor task if requested + self.monitor_task = death_tx.map(|dtx| tokio::spawn(connection_monitor(id, watcher, dtx))); + + info!("[Conn {id}] Client replaced successfully"); + Ok(()) + } + pub fn open_channel(&self, reply: ServerReply) -> io::Result { let channel_id = rand::random(); self.tx @@ -139,6 +212,9 @@ impl ManagerConnection { self.action_task.abort(); self.request_task.abort(); self.response_task.abort(); + if let Some(ref task) = self.monitor_task { + task.abort(); + } } } @@ -184,6 +260,27 @@ impl fmt::Debug for Action { } } +/// Watches a connection's health state and sends a death notification when the +/// connection transitions to [`ConnectionState::Disconnected`]. +/// +/// If the watcher channel closes (sender dropped), the connection is also considered dead. +async fn connection_monitor( + id: ConnectionId, + mut watcher: ConnectionWatcher, + death_tx: mpsc::UnboundedSender, +) { + while let Some(state) = watcher.next().await { + if state.is_disconnected() { + info!("[Conn {id}] Connection died, notifying manager"); + let _ = death_tx.send(id); + return; + } + } + // Watcher channel closed (sender dropped) — connection is dead + debug!("[Conn {id}] Connection watcher closed"); + let _ = death_tx.send(id); +} + /// Internal task to process outgoing [`UntypedRequest`]s. async fn request_task( id: ConnectionId, @@ -287,8 +384,7 @@ async fn action_task( #[cfg(test)] mod tests { - //! Tests for ManagerChannel (send, close, clone), ManagerConnection (spawn, open_channel, - //! channel_ids, close/unregister, abort), and Action Debug formatting. + use std::time::Duration; use super::*; use crate::net::client::UntypedClient; @@ -371,7 +467,7 @@ mod tests { let dest = "scheme://host".to_string(); let opts: Map = "key=value".parse().unwrap(); - let conn = ManagerConnection::spawn(dest.clone(), opts.clone(), client) + let conn = ManagerConnection::spawn(dest.clone(), opts.clone(), client, None) .await .unwrap(); @@ -388,7 +484,9 @@ mod tests { let dest = "scheme://host".to_string(); let opts: Map = Map::new(); - let conn = ManagerConnection::spawn(dest, opts, client).await.unwrap(); + let conn = ManagerConnection::spawn(dest, opts, client, None) + .await + .unwrap(); let (reply_tx, _reply_rx) = mpsc::unbounded_channel(); let reply = ServerReply { @@ -407,7 +505,9 @@ mod tests { let dest = "scheme://host".to_string(); let opts: Map = Map::new(); - let conn = ManagerConnection::spawn(dest, opts, client).await.unwrap(); + let conn = ManagerConnection::spawn(dest, opts, client, None) + .await + .unwrap(); let (reply_tx, _reply_rx) = mpsc::unbounded_channel(); let reply = ServerReply { @@ -431,7 +531,9 @@ mod tests { let dest = "scheme://host".to_string(); let opts: Map = Map::new(); - let conn = ManagerConnection::spawn(dest, opts, client).await.unwrap(); + let conn = ManagerConnection::spawn(dest, opts, client, None) + .await + .unwrap(); let ids = conn.channel_ids().await.unwrap(); assert!(ids.is_empty()); @@ -443,7 +545,9 @@ mod tests { let dest = "scheme://host".to_string(); let opts: Map = Map::new(); - let conn = ManagerConnection::spawn(dest, opts, client).await.unwrap(); + let conn = ManagerConnection::spawn(dest, opts, client, None) + .await + .unwrap(); let mut channel_ids = Vec::new(); for _ in 0..3 { @@ -472,7 +576,9 @@ mod tests { let dest = "scheme://host".to_string(); let opts: Map = Map::new(); - let conn = ManagerConnection::spawn(dest, opts, client).await.unwrap(); + let conn = ManagerConnection::spawn(dest, opts, client, None) + .await + .unwrap(); let (reply_tx, _reply_rx) = mpsc::unbounded_channel(); let reply = ServerReply { @@ -503,7 +609,9 @@ mod tests { let dest = "scheme://host".to_string(); let opts: Map = Map::new(); - let conn = ManagerConnection::spawn(dest, opts, client).await.unwrap(); + let conn = ManagerConnection::spawn(dest, opts, client, None) + .await + .unwrap(); conn.abort(); // After abort, channel_ids should fail because the action task is aborted @@ -518,7 +626,9 @@ mod tests { let dest = "scheme://host".to_string(); let opts: Map = Map::new(); - let conn = ManagerConnection::spawn(dest, opts, client).await.unwrap(); + let conn = ManagerConnection::spawn(dest, opts, client, None) + .await + .unwrap(); conn.abort(); // Give time for abort to take effect @@ -542,6 +652,124 @@ mod tests { } } + // ---- Connection Monitor ---- + + #[test_log::test(tokio::test)] + async fn connection_monitor_should_send_death_on_disconnect() { + let (client, server_conn) = make_untyped_client(); + let watcher = client.clone_connection_watcher(); + let connection_id: ConnectionId = 12345; + + let (death_tx, mut death_rx) = mpsc::unbounded_channel(); + + // Spawn the monitor in the background + let _monitor = tokio::spawn(connection_monitor(connection_id, watcher, death_tx)); + + // Drop the server side of the connection to trigger disconnect. + // The client's event loop will detect the broken transport, attempt + // reconnection (which fails immediately with the default Fail strategy), + // and transition to Disconnected. + drop(server_conn); + // Also drop the client so the watcher task can observe the state change + // before the client is fully cleaned up. Actually, the client task runs + // independently, so the watcher should see Disconnected once the task + // completes its reconnect failure path. + drop(client); + + // Wait for the death notification with a timeout + let received_id = tokio::time::timeout(Duration::from_secs(5), death_rx.recv()) + .await + .expect("timed out waiting for death notification") + .expect("death channel closed without sending"); + + assert_eq!(received_id, connection_id); + } + + #[test_log::test(tokio::test)] + async fn connection_monitor_should_send_death_when_watcher_closes() { + // Create a client with shutdown_on_drop=true so dropping it aborts the + // internal task immediately, which drops the watch::Sender without first + // sending a Disconnected state. This exercises the fallback path in + // connection_monitor where watcher.next() returns None. + let (mut client, _server_conn) = make_untyped_client(); + client.shutdown_on_drop(true); + let watcher = client.clone_connection_watcher(); + let connection_id: ConnectionId = 67890; + + let (death_tx, mut death_rx) = mpsc::unbounded_channel(); + + // Spawn the monitor in the background + let _monitor = tokio::spawn(connection_monitor(connection_id, watcher, death_tx)); + + // Drop the client. Because shutdown_on_drop is true, this aborts the + // internal task, dropping the watch sender. The server connection is kept + // alive so the client task has no reason to send Disconnected before abort. + drop(client); + + let received_id = tokio::time::timeout(Duration::from_secs(5), death_rx.recv()) + .await + .expect("timed out waiting for death notification") + .expect("death channel closed without sending"); + + assert_eq!(received_id, connection_id); + } + + #[test_log::test(tokio::test)] + async fn spawn_with_death_tx_should_notify_on_client_drop() { + let (client, server_conn) = make_untyped_client(); + let (death_tx, mut death_rx) = mpsc::unbounded_channel(); + + let conn = ManagerConnection::spawn("scheme://host", Map::new(), client, Some(death_tx)) + .await + .unwrap(); + + let connection_id = conn.id; + + // Drop the server side to trigger disconnection in the underlying transport. + // The client event loop will fail reconnection and transition to Disconnected, + // which the monitor task observes and sends through death_tx. + drop(server_conn); + + let received_id = tokio::time::timeout(Duration::from_secs(5), death_rx.recv()) + .await + .expect("timed out waiting for death notification") + .expect("death channel closed without sending"); + + assert_eq!(received_id, connection_id); + + // Clean up the connection to abort its tasks + conn.abort(); + } + + #[test_log::test(tokio::test)] + async fn spawn_without_death_tx_should_not_have_monitor_task() { + let (client, _server_conn) = make_untyped_client(); + + let conn = ManagerConnection::spawn("scheme://host", Map::new(), client, None) + .await + .unwrap(); + + assert!( + conn.monitor_task.is_none(), + "monitor_task should be None when no death_tx is provided" + ); + } + + #[test_log::test(tokio::test)] + async fn spawn_with_death_tx_should_have_monitor_task() { + let (client, _server_conn) = make_untyped_client(); + let (death_tx, _death_rx) = mpsc::unbounded_channel(); + + let conn = ManagerConnection::spawn("scheme://host", Map::new(), client, Some(death_tx)) + .await + .unwrap(); + + assert!( + conn.monitor_task.is_some(), + "monitor_task should be Some when death_tx is provided" + ); + } + // ---- Action Debug ---- #[test] @@ -595,4 +823,199 @@ mod tests { let debug = format!("{action:?}"); assert_eq!(debug, "Action::Write { id: 7, .. }"); } + + // ---- replace_client ---- + + #[test_log::test(tokio::test)] + async fn replace_client_should_preserve_connection_id() { + let (client, _server) = make_untyped_client(); + let conn = ManagerConnection::spawn("scheme://host", Map::new(), client, None) + .await + .unwrap(); + let original_id = conn.id; + + // Build a new client to replace with + let (new_client, _new_server) = make_untyped_client(); + + let mut conn = conn; + conn.replace_client(new_client, None).await.unwrap(); + + assert_eq!(conn.id, original_id); + } + + #[test_log::test(tokio::test)] + async fn replace_client_should_preserve_destination_and_options() { + let (client, _server) = make_untyped_client(); + let opts: Map = "key=value".parse().unwrap(); + let dest = "scheme://host".to_string(); + let conn = ManagerConnection::spawn(dest.clone(), opts.clone(), client, None) + .await + .unwrap(); + + let (new_client, _new_server) = make_untyped_client(); + + let mut conn = conn; + conn.replace_client(new_client, None).await.unwrap(); + + assert_eq!(conn.destination, dest); + assert_eq!(conn.options, opts); + } + + #[test_log::test(tokio::test)] + async fn replace_client_should_allow_new_channels_after_replacement() { + let (client, _server) = make_untyped_client(); + let mut conn = ManagerConnection::spawn("scheme://host", Map::new(), client, None) + .await + .unwrap(); + + // Replace with a new client + let (new_client, _new_server) = make_untyped_client(); + conn.replace_client(new_client, None).await.unwrap(); + + // Give the new action task time to start + tokio::time::sleep(Duration::from_millis(50)).await; + + // Open a channel on the replacement + let (reply_tx, _reply_rx) = mpsc::unbounded_channel(); + let reply = ServerReply { + origin_id: "test".to_string(), + tx: reply_tx, + }; + let channel = conn.open_channel(reply).unwrap(); + let channel_id = channel.id(); + + // Wait for registration + tokio::time::sleep(Duration::from_millis(50)).await; + + let ids = conn.channel_ids().await.unwrap(); + assert!( + ids.contains(&channel_id), + "New channel should be registered after replace_client" + ); + } + + #[test_log::test(tokio::test)] + async fn replace_client_with_death_tx_should_have_monitor_task() { + let (client, _server) = make_untyped_client(); + let mut conn = ManagerConnection::spawn("scheme://host", Map::new(), client, None) + .await + .unwrap(); + + assert!( + conn.monitor_task.is_none(), + "Initial spawn without death_tx should have no monitor" + ); + + let (death_tx, _death_rx) = mpsc::unbounded_channel(); + let (new_client, _new_server) = make_untyped_client(); + conn.replace_client(new_client, Some(death_tx)) + .await + .unwrap(); + + assert!( + conn.monitor_task.is_some(), + "After replace_client with death_tx, monitor should be Some" + ); + } + + #[test_log::test(tokio::test)] + async fn replace_client_without_death_tx_should_not_have_monitor_task() { + let (client, _server) = make_untyped_client(); + let (death_tx, _death_rx) = mpsc::unbounded_channel(); + let mut conn = + ManagerConnection::spawn("scheme://host", Map::new(), client, Some(death_tx)) + .await + .unwrap(); + + assert!( + conn.monitor_task.is_some(), + "Initial spawn with death_tx should have monitor" + ); + + let (new_client, _new_server) = make_untyped_client(); + conn.replace_client(new_client, None).await.unwrap(); + + assert!( + conn.monitor_task.is_none(), + "After replace_client without death_tx, monitor should be None" + ); + } + + #[test_log::test(tokio::test)] + async fn replace_client_should_start_with_empty_channel_registrations() { + let (client, _server) = make_untyped_client(); + let mut conn = ManagerConnection::spawn("scheme://host", Map::new(), client, None) + .await + .unwrap(); + + // Register a channel on the old action task + let (reply_tx, _reply_rx) = mpsc::unbounded_channel(); + let reply = ServerReply { + origin_id: "test".to_string(), + tx: reply_tx, + }; + let _channel = conn.open_channel(reply).unwrap(); + tokio::time::sleep(Duration::from_millis(50)).await; + + let ids_before = conn.channel_ids().await.unwrap(); + assert_eq!(ids_before.len(), 1); + + // Replace the client -- this replaces the action task, so registrations reset + let (new_client, _new_server) = make_untyped_client(); + conn.replace_client(new_client, None).await.unwrap(); + + // Give the new action task time to start + tokio::time::sleep(Duration::from_millis(50)).await; + + let ids_after = conn.channel_ids().await.unwrap(); + assert!( + ids_after.is_empty(), + "Channel registrations should be empty after replace_client, but found: {ids_after:?}" + ); + } + + #[test_log::test(tokio::test)] + async fn replace_client_death_tx_should_notify_on_new_client_disconnect() { + let (client, _server) = make_untyped_client(); + let mut conn = ManagerConnection::spawn("scheme://host", Map::new(), client, None) + .await + .unwrap(); + let conn_id = conn.id; + + let (death_tx, mut death_rx) = mpsc::unbounded_channel(); + let (new_client, new_server) = make_untyped_client(); + conn.replace_client(new_client, Some(death_tx)) + .await + .unwrap(); + + // Drop the server side to trigger disconnect detection on the new client + drop(new_server); + + let received_id = tokio::time::timeout(Duration::from_secs(5), death_rx.recv()) + .await + .expect("timed out waiting for death notification") + .expect("death channel closed without sending"); + + assert_eq!(received_id, conn_id); + } + + #[test_log::test(tokio::test)] + async fn replace_client_should_propagate_error_when_client_task_is_dead() { + let (client, _server) = make_untyped_client(); + let mut conn = ManagerConnection::spawn("scheme://host", Map::new(), client, None) + .await + .unwrap(); + + // Create a new client and abort its internal task so the post office + // is dropped. This causes assign_default_mailbox to fail with + // NotConnected because the Weak cannot be upgraded. + let (mut dead_client, _dead_server) = make_untyped_client(); + dead_client.shutdown_on_drop(true); + dead_client.abort(); + // Give the task time to actually stop + tokio::time::sleep(Duration::from_millis(50)).await; + + let err = conn.replace_client(dead_client, None).await.unwrap_err(); + assert_eq!(err.kind(), io::ErrorKind::NotConnected); + } } diff --git a/distant-core/src/net/server.rs b/distant-core/src/net/server.rs index 8075d834a..1b501818c 100644 --- a/distant-core/src/net/server.rs +++ b/distant-core/src/net/server.rs @@ -240,6 +240,7 @@ where .shutdown_timer(Arc::downgrade(&timer)) .sleep_duration(config.connection_sleep) .heartbeat_duration(config.connection_heartbeat) + .max_heartbeat_failures(config.max_heartbeat_failures) .verifier(Arc::downgrade(&verifier)) .version(version.clone()) .spawn(), diff --git a/distant-core/src/net/server/config.rs b/distant-core/src/net/server/config.rs index a64a54488..ed9ebe21d 100644 --- a/distant-core/src/net/server/config.rs +++ b/distant-core/src/net/server/config.rs @@ -7,6 +7,11 @@ use serde::{Deserialize, Serialize}; const DEFAULT_CONNECTION_SLEEP: Duration = Duration::from_millis(1); const DEFAULT_HEARTBEAT_DURATION: Duration = Duration::from_secs(5); +const DEFAULT_MAX_HEARTBEAT_FAILURES: u32 = 3; + +fn default_max_heartbeat_failures() -> u32 { + DEFAULT_MAX_HEARTBEAT_FAILURES +} /// Represents a general-purpose set of properties tied with a server instance #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -19,6 +24,11 @@ pub struct ServerConfig { /// Rules for how a server will shutdown automatically pub shutdown: Shutdown, + + /// Maximum consecutive heartbeat write failures before the connection is terminated. + /// A value of 0 means heartbeat failures are never escalated. + #[serde(default = "default_max_heartbeat_failures")] + pub max_heartbeat_failures: u32, } impl Default for ServerConfig { @@ -27,6 +37,7 @@ impl Default for ServerConfig { connection_sleep: DEFAULT_CONNECTION_SLEEP, connection_heartbeat: DEFAULT_HEARTBEAT_DURATION, shutdown: Default::default(), + max_heartbeat_failures: DEFAULT_MAX_HEARTBEAT_FAILURES, } } } @@ -158,6 +169,12 @@ mod tests { assert_eq!(config.shutdown, Shutdown::Never); } + #[test] + fn server_config_default_has_expected_max_heartbeat_failures() { + let config = ServerConfig::default(); + assert_eq!(config.max_heartbeat_failures, 3); + } + // ---- ServerConfig serde round-trip ---- #[test] @@ -174,6 +191,7 @@ mod tests { connection_sleep: Duration::from_millis(50), connection_heartbeat: Duration::from_secs(10), shutdown: Shutdown::After(Duration::from_secs(30)), + ..Default::default() }; let serialized = serde_json::to_string(&config).unwrap(); let deserialized: ServerConfig = serde_json::from_str(&serialized).unwrap(); @@ -186,10 +204,47 @@ mod tests { connection_sleep: Duration::from_millis(10), connection_heartbeat: Duration::from_secs(3), shutdown: Shutdown::Lonely(Duration::from_secs(60)), + ..Default::default() + }; + let serialized = serde_json::to_string(&config).unwrap(); + let deserialized: ServerConfig = serde_json::from_str(&serialized).unwrap(); + assert_eq!(config, deserialized); + } + + #[test] + fn server_config_should_serialize_and_deserialize_with_custom_max_heartbeat_failures() { + let config = ServerConfig { + max_heartbeat_failures: 10, + ..Default::default() + }; + let serialized = serde_json::to_string(&config).unwrap(); + let deserialized: ServerConfig = serde_json::from_str(&serialized).unwrap(); + assert_eq!(config, deserialized); + assert_eq!(deserialized.max_heartbeat_failures, 10); + } + + #[test] + fn server_config_should_deserialize_missing_max_heartbeat_failures_as_default() { + // Simulate a JSON payload from an older version that lacks the field + let json = r#"{ + "connection_sleep":{"secs":0,"nanos":1000000}, + "connection_heartbeat":{"secs":5,"nanos":0}, + "shutdown":"never" + }"#; + let config: ServerConfig = serde_json::from_str(json).unwrap(); + assert_eq!(config.max_heartbeat_failures, 3); + } + + #[test] + fn server_config_should_serialize_and_deserialize_with_max_heartbeat_failures_zero() { + let config = ServerConfig { + max_heartbeat_failures: 0, + ..Default::default() }; let serialized = serde_json::to_string(&config).unwrap(); let deserialized: ServerConfig = serde_json::from_str(&serialized).unwrap(); assert_eq!(config, deserialized); + assert_eq!(deserialized.max_heartbeat_failures, 0); } // ---- Shutdown Display ---- diff --git a/distant-core/src/net/server/connection.rs b/distant-core/src/net/server/connection.rs index 5fefb3e25..9b68bfc20 100644 --- a/distant-core/src/net/server/connection.rs +++ b/distant-core/src/net/server/connection.rs @@ -25,6 +25,9 @@ const SLEEP_DURATION: Duration = Duration::from_millis(1); /// Minimum time between heartbeats to communicate to the client connection. const MINIMUM_HEARTBEAT_DURATION: Duration = Duration::from_secs(5); +/// Default maximum consecutive heartbeat write failures before terminating the connection. +const DEFAULT_MAX_HEARTBEAT_FAILURES: u32 = 3; + /// Represents an individual connection on the server. pub(super) struct ConnectionTask(JoinHandle>); @@ -64,6 +67,7 @@ pub(super) struct ConnectionTaskBuilder { shutdown_timer: Weak>, sleep_duration: Duration, heartbeat_duration: Duration, + max_heartbeat_failures: u32, verifier: Weak, version: Version, } @@ -80,6 +84,7 @@ impl ConnectionTaskBuilder<(), (), ()> { shutdown_timer: Weak::new(), sleep_duration: SLEEP_DURATION, heartbeat_duration: MINIMUM_HEARTBEAT_DURATION, + max_heartbeat_failures: DEFAULT_MAX_HEARTBEAT_FAILURES, verifier: Weak::new(), version: Version::default(), } @@ -97,6 +102,7 @@ impl ConnectionTaskBuilder { shutdown_timer: self.shutdown_timer, sleep_duration: self.sleep_duration, heartbeat_duration: self.heartbeat_duration, + max_heartbeat_failures: self.max_heartbeat_failures, verifier: self.verifier, version: self.version, } @@ -112,6 +118,7 @@ impl ConnectionTaskBuilder { shutdown_timer: self.shutdown_timer, sleep_duration: self.sleep_duration, heartbeat_duration: self.heartbeat_duration, + max_heartbeat_failures: self.max_heartbeat_failures, verifier: self.verifier, version: self.version, } @@ -127,6 +134,7 @@ impl ConnectionTaskBuilder { shutdown_timer: self.shutdown_timer, sleep_duration: self.sleep_duration, heartbeat_duration: self.heartbeat_duration, + max_heartbeat_failures: self.max_heartbeat_failures, verifier: self.verifier, version: self.version, } @@ -142,6 +150,7 @@ impl ConnectionTaskBuilder { shutdown_timer: self.shutdown_timer, sleep_duration: self.sleep_duration, heartbeat_duration: self.heartbeat_duration, + max_heartbeat_failures: self.max_heartbeat_failures, verifier: self.verifier, version: self.version, } @@ -157,6 +166,7 @@ impl ConnectionTaskBuilder { shutdown_timer: self.shutdown_timer, sleep_duration: self.sleep_duration, heartbeat_duration: self.heartbeat_duration, + max_heartbeat_failures: self.max_heartbeat_failures, verifier: self.verifier, version: self.version, } @@ -175,6 +185,7 @@ impl ConnectionTaskBuilder { shutdown_timer, sleep_duration: self.sleep_duration, heartbeat_duration: self.heartbeat_duration, + max_heartbeat_failures: self.max_heartbeat_failures, verifier: self.verifier, version: self.version, } @@ -190,6 +201,7 @@ impl ConnectionTaskBuilder { shutdown_timer: self.shutdown_timer, sleep_duration, heartbeat_duration: self.heartbeat_duration, + max_heartbeat_failures: self.max_heartbeat_failures, verifier: self.verifier, version: self.version, } @@ -208,6 +220,26 @@ impl ConnectionTaskBuilder { shutdown_timer: self.shutdown_timer, sleep_duration: self.sleep_duration, heartbeat_duration, + max_heartbeat_failures: self.max_heartbeat_failures, + verifier: self.verifier, + version: self.version, + } + } + + pub fn max_heartbeat_failures( + self, + max_heartbeat_failures: u32, + ) -> ConnectionTaskBuilder { + ConnectionTaskBuilder { + handler: self.handler, + state: self.state, + keychain: self.keychain, + transport: self.transport, + shutdown: self.shutdown, + shutdown_timer: self.shutdown_timer, + sleep_duration: self.sleep_duration, + heartbeat_duration: self.heartbeat_duration, + max_heartbeat_failures, verifier: self.verifier, version: self.version, } @@ -223,6 +255,7 @@ impl ConnectionTaskBuilder { shutdown_timer: self.shutdown_timer, sleep_duration: self.sleep_duration, heartbeat_duration: self.heartbeat_duration, + max_heartbeat_failures: self.max_heartbeat_failures, verifier, version: self.version, } @@ -238,6 +271,7 @@ impl ConnectionTaskBuilder { shutdown_timer: self.shutdown_timer, sleep_duration: self.sleep_duration, heartbeat_duration: self.heartbeat_duration, + max_heartbeat_failures: self.max_heartbeat_failures, verifier: self.verifier, version, } @@ -265,6 +299,7 @@ where shutdown_timer, sleep_duration, heartbeat_duration, + max_heartbeat_failures, verifier, version, } = self; @@ -459,6 +494,7 @@ where } let mut last_heartbeat = Instant::now(); + let mut consecutive_heartbeat_failures: u32 = 0; // Restore our connection's channels if we have them, otherwise make new ones let (tx, mut rx) = match state.connections.write().await.remove(&id) { @@ -557,9 +593,23 @@ where if last_heartbeat.elapsed() >= heartbeat_duration { trace!("[Conn {id}] Sending heartbeat via empty frame"); match connection.try_write_frame(Frame::empty()) { - Ok(()) => (), + Ok(()) => { + consecutive_heartbeat_failures = 0; + } Err(x) if x.kind() == io::ErrorKind::WouldBlock => write_blocked = true, - Err(x) => error!("[Conn {id}] Send failed: {x}"), + Err(x) => { + consecutive_heartbeat_failures += 1; + error!( + "[Conn {id}] Heartbeat send failed ({consecutive_heartbeat_failures}/{max_heartbeat_failures}): {x}" + ); + if max_heartbeat_failures > 0 + && consecutive_heartbeat_failures >= max_heartbeat_failures + { + terminate_connection!(@error(tx, rx) + "[Conn {id}] Terminated after {consecutive_heartbeat_failures} consecutive heartbeat failures" + ); + } + } } last_heartbeat = Instant::now(); } @@ -580,7 +630,9 @@ where match response.to_vec() { Ok(data) => match connection.try_write_frame(data) { - Ok(()) => (), + Ok(()) => { + consecutive_heartbeat_failures = 0; + } Err(x) if x.kind() == io::ErrorKind::WouldBlock => write_blocked = true, Err(x) => error!("[Conn {id}] Send failed: {x}"), }, diff --git a/distant-core/src/net/server/ref.rs b/distant-core/src/net/server/ref.rs index bee325b66..a5c130d34 100644 --- a/distant-core/src/net/server/ref.rs +++ b/distant-core/src/net/server/ref.rs @@ -5,17 +5,19 @@ use std::task::{Context, Poll}; use tokio::sync::broadcast; use tokio::task::{JoinError, JoinHandle}; -/// Represents a reference to a server +/// Represents a reference to a server. pub struct ServerRef { pub(crate) shutdown: broadcast::Sender<()>, pub(crate) task: JoinHandle<()>, } impl ServerRef { + /// Returns `true` if the server task has completed. pub fn is_finished(&self) -> bool { self.task.is_finished() } + /// Sends a shutdown signal to the server, causing it to terminate gracefully. pub fn shutdown(&self) { let _ = self.shutdown.send(()); } @@ -24,6 +26,33 @@ impl ServerRef { pub fn subscribe_shutdown(&self) -> broadcast::Receiver<()> { self.shutdown.subscribe() } + + /// Returns a lightweight handle that can trigger server shutdown. + /// + /// Unlike [`ServerRef`] itself, [`ShutdownSender`] is [`Clone`] and [`Send`], + /// making it suitable for passing to background health-monitoring tasks that + /// need to shut down the server when a backend dies. + pub fn shutdown_sender(&self) -> ShutdownSender { + ShutdownSender { + sender: self.shutdown.clone(), + } + } +} + +/// A lightweight, cloneable handle for triggering server shutdown. +/// +/// Obtained via [`ServerRef::shutdown_sender`]. Calling [`shutdown`](Self::shutdown) +/// sends the same signal as [`ServerRef::shutdown`]. +#[derive(Clone)] +pub struct ShutdownSender { + sender: broadcast::Sender<()>, +} + +impl ShutdownSender { + /// Sends the shutdown signal to the associated server. + pub fn shutdown(&self) { + let _ = self.sender.send(()); + } } impl Future for ServerRef { @@ -48,3 +77,120 @@ mod windows; #[cfg(windows)] pub use windows::*; + +#[cfg(test)] +mod tests { + use super::*; + + fn make_server_ref() -> ServerRef { + let (shutdown, _) = broadcast::channel(1); + let task = tokio::spawn(async {}); + ServerRef { shutdown, task } + } + + // --------------------------------------------------------------- + // ShutdownSender + // --------------------------------------------------------------- + + #[test_log::test(tokio::test)] + async fn shutdown_sender_triggers_receiver() { + let (tx, _) = broadcast::channel::<()>(1); + let mut rx = tx.subscribe(); + let sender = ShutdownSender { sender: tx }; + + sender.shutdown(); + + let result = rx.recv().await; + assert_eq!(result.unwrap(), ()); + } + + #[test_log::test(tokio::test)] + async fn shutdown_sender_clone_triggers_same_channel() { + let (tx, _) = broadcast::channel::<()>(1); + let mut rx = tx.subscribe(); + let sender = ShutdownSender { sender: tx }; + let cloned = sender.clone(); + + // Use the clone to send + cloned.shutdown(); + + let result = rx.recv().await; + assert_eq!(result.unwrap(), ()); + } + + #[test_log::test(tokio::test)] + async fn shutdown_sender_original_and_clone_both_work() { + let (tx, _) = broadcast::channel::<()>(2); + let mut rx = tx.subscribe(); + let sender = ShutdownSender { sender: tx }; + let cloned = sender.clone(); + + sender.shutdown(); + let first = rx.recv().await; + assert_eq!(first.unwrap(), ()); + + cloned.shutdown(); + let second = rx.recv().await; + assert_eq!(second.unwrap(), ()); + } + + #[test_log::test(tokio::test)] + async fn shutdown_sender_does_not_panic_when_no_receivers() { + let (tx, _) = broadcast::channel::<()>(1); + let sender = ShutdownSender { sender: tx }; + // No receiver subscribed — shutdown should not panic + sender.shutdown(); + } + + // --------------------------------------------------------------- + // ServerRef::shutdown_sender + // --------------------------------------------------------------- + + #[test_log::test(tokio::test)] + async fn server_ref_shutdown_sender_returns_working_sender() { + let server_ref = make_server_ref(); + let mut rx = server_ref.subscribe_shutdown(); + let sender = server_ref.shutdown_sender(); + + sender.shutdown(); + + let result = rx.recv().await; + assert_eq!(result.unwrap(), ()); + } + + #[test_log::test(tokio::test)] + async fn server_ref_shutdown_sender_shares_channel_with_server_ref() { + let server_ref = make_server_ref(); + let sender = server_ref.shutdown_sender(); + let mut rx = server_ref.subscribe_shutdown(); + + // Shutting down via the sender should be equivalent to server_ref.shutdown() + sender.shutdown(); + + let result = rx.recv().await; + assert_eq!(result.unwrap(), ()); + } + + #[test_log::test(tokio::test)] + async fn server_ref_shutdown_via_sender_stops_server_task() { + // Create a server ref with a long-running task that listens for shutdown + let (shutdown_tx, _) = broadcast::channel(1); + let mut shutdown_rx = shutdown_tx.subscribe(); + let task = tokio::spawn(async move { + let _ = shutdown_rx.recv().await; + }); + let server_ref = ServerRef { + shutdown: shutdown_tx, + task, + }; + + assert!(!server_ref.is_finished()); + + let sender = server_ref.shutdown_sender(); + sender.shutdown(); + + // Wait for the task to notice and finish + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + assert!(server_ref.is_finished()); + } +} diff --git a/distant-core/src/plugin/mod.rs b/distant-core/src/plugin/mod.rs index 5d4a4c4a2..b128fd8c2 100644 --- a/distant-core/src/plugin/mod.rs +++ b/distant-core/src/plugin/mod.rs @@ -3,7 +3,7 @@ use std::io; use std::pin::Pin; use crate::auth::Authenticator; -use crate::net::client::UntypedClient; +use crate::net::client::{ReconnectStrategy, UntypedClient}; use crate::net::common::{Destination, Map}; mod process; @@ -52,6 +52,34 @@ pub trait Plugin: Send + Sync { )) }) } + + /// Attempt to re-establish a previously connected session. + /// + /// Called by the manager when a connection dies. Receives the same + /// destination and options from the original `connect()` call. The + /// default returns `Unsupported` (no reconnection capability). + fn reconnect<'a>( + &'a self, + _raw_destination: &'a str, + _options: &'a Map, + _authenticator: &'a mut dyn Authenticator, + ) -> Pin> + Send + 'a>> { + Box::pin(async { + Err(io::Error::new( + io::ErrorKind::Unsupported, + "reconnect not supported", + )) + }) + } + + /// Reconnection retry strategy for this plugin. + /// + /// Returns the strategy the manager should use when orchestrating + /// reconnection attempts. The default is `Fail` (no automatic + /// reconnection). Plugins override this to specify backoff behavior. + fn reconnect_strategy(&self) -> ReconnectStrategy { + ReconnectStrategy::Fail + } } /// Parses a raw destination string into a core [`Destination`]. @@ -256,6 +284,62 @@ mod tests { assert_eq!(Arc::strong_count(&plugin), 3); } + // ----------------------------------------------------------------------- + // reconnect() default implementation + // ----------------------------------------------------------------------- + + #[test(tokio::test)] + async fn default_reconnect_returns_unsupported_error() { + let plugin = MockPlugin::new("test"); + let options = Map::new(); + let mut auth = TestAuthenticator::default(); + + let result = plugin + .reconnect("ssh://localhost", &options, &mut auth) + .await; + assert!(result.is_err()); + + let err = result.unwrap_err(); + assert_eq!(err.kind(), io::ErrorKind::Unsupported); + assert_eq!(err.to_string(), "reconnect not supported"); + } + + // ----------------------------------------------------------------------- + // reconnect_strategy() default implementation + // ----------------------------------------------------------------------- + + #[test] + fn default_reconnect_strategy_returns_fail() { + let plugin = MockPlugin::new("test"); + let strategy = plugin.reconnect_strategy(); + assert!(strategy.is_fail()); + } + + // ----------------------------------------------------------------------- + // reconnect() / reconnect_strategy() via Arc + // ----------------------------------------------------------------------- + + #[test(tokio::test)] + async fn arc_dyn_plugin_reconnect_returns_unsupported() { + let plugin: Arc = Arc::new(MockPlugin::new("arctest")); + let options = Map::new(); + let mut auth = TestAuthenticator::default(); + + let result = plugin.reconnect("ssh://host", &options, &mut auth).await; + assert!(result.is_err()); + + let err = result.unwrap_err(); + assert_eq!(err.kind(), io::ErrorKind::Unsupported); + assert_eq!(err.to_string(), "reconnect not supported"); + } + + #[test] + fn arc_dyn_plugin_reconnect_strategy_returns_fail() { + let plugin: Arc = Arc::new(MockPlugin::new("arctest")); + let strategy = plugin.reconnect_strategy(); + assert!(strategy.is_fail()); + } + // ----------------------------------------------------------------------- // parse_destination helper // ----------------------------------------------------------------------- diff --git a/distant-docker/src/lib.rs b/distant-docker/src/lib.rs index c4d152ea3..202d2567b 100644 --- a/distant-docker/src/lib.rs +++ b/distant-docker/src/lib.rs @@ -23,6 +23,7 @@ #![allow(clippy::manual_async_fn)] use std::io; +use std::time::Duration; use bollard::Docker as BollardDocker; use bollard::models::ContainerCreateBody; @@ -33,7 +34,7 @@ use bollard::query_parameters::{ use distant_core::net::auth::{DummyAuthHandler, Verifier}; use distant_core::net::client::{Client as NetClient, ClientConfig}; use distant_core::net::common::{InmemoryTransport, OneshotListener}; -use distant_core::net::server::{Server, ServerRef}; +use distant_core::net::server::{Server, ServerRef, ShutdownSender}; use distant_core::{ApiServerHandler, Client}; use futures::StreamExt; use log::*; @@ -358,7 +359,9 @@ impl Docker { /// Converts this Docker connection into a distant [`Client`]. /// /// Creates an in-memory server/client pair where the server side is backed by [`DockerApi`]. - /// If `auto_remove` is enabled, the container is stopped and removed when the server task ends. + /// If `auto_remove` is enabled, the container is stopped and removed when the server shuts + /// down. A health monitor task is spawned to detect Docker daemon or container death and + /// trigger server shutdown, which causes the client to see a disconnect. pub async fn into_distant_client(self) -> io::Result { let auto_remove = self.auto_remove; let cleanup_client = if auto_remove { @@ -367,6 +370,8 @@ impl Docker { None }; let container_name = self.container.clone(); + let health_client = self.client.clone(); + let health_container = self.container.clone(); let api = DockerApi::new(self.client.clone(), self.container, self.opts).await; @@ -376,22 +381,35 @@ impl Docker { .handler(ApiServerHandler::new(api)) .verifier(Verifier::none()); - tokio::spawn(async move { - let _ = server.start(OneshotListener::from_value(t2)); + let server_ref = server + .start(OneshotListener::from_value(t2)) + .map_err(io::Error::other)?; - if let Some(client) = cleanup_client { + // Spawn cleanup task if auto_remove is set + if let Some(client) = cleanup_client { + let mut shutdown_rx = server_ref.subscribe_shutdown(); + let cleanup_container = container_name.clone(); + tokio::spawn(async move { + let _ = shutdown_rx.recv().await; info!( "Auto-removing container '{}' after server shutdown", - container_name + cleanup_container ); - if let Err(e) = Self::stop_and_remove(&client, &container_name).await { + if let Err(e) = Self::stop_and_remove(&client, &cleanup_container).await { warn!( "Failed to auto-remove container '{}': {}", - container_name, e + cleanup_container, e ); } - } - }); + }); + } + + // Spawn health monitor that detects Docker daemon/container death + tokio::spawn(Self::docker_health_monitor( + health_client, + health_container, + server_ref.shutdown_sender(), + )); let client = NetClient::build() .auth_handler(DummyAuthHandler) @@ -407,7 +425,8 @@ impl Docker { /// Converts this Docker connection into a pair of distant client and server ref. /// /// If `auto_remove` is enabled, the container is stopped and removed when the server - /// shuts down. + /// shuts down. A health monitor task is spawned to detect Docker daemon or container + /// death and trigger server shutdown. pub async fn into_distant_pair(self) -> io::Result<(Client, ServerRef)> { let auto_remove = self.auto_remove; let cleanup_client = if auto_remove { @@ -416,6 +435,8 @@ impl Docker { None }; let container_name = self.container.clone(); + let health_client = self.client.clone(); + let health_container = self.container.clone(); let api = DockerApi::new(self.client, self.container, self.opts).await; @@ -432,21 +453,29 @@ impl Docker { // Spawn cleanup task that waits for server shutdown signal if let Some(client) = cleanup_client { let mut shutdown_rx = server_ref.subscribe_shutdown(); + let cleanup_container = container_name; tokio::spawn(async move { let _ = shutdown_rx.recv().await; info!( "Auto-removing container '{}' after server shutdown", - container_name + cleanup_container ); - if let Err(e) = Self::stop_and_remove(&client, &container_name).await { + if let Err(e) = Self::stop_and_remove(&client, &cleanup_container).await { warn!( "Failed to auto-remove container '{}': {}", - container_name, e + cleanup_container, e ); } }); } + // Spawn health monitor that detects Docker daemon/container death + tokio::spawn(Self::docker_health_monitor( + health_client, + health_container, + server_ref.shutdown_sender(), + )); + let client = NetClient::build() .auth_handler(DummyAuthHandler) .config(ClientConfig::default()) @@ -463,6 +492,59 @@ impl Docker { &self.container } + /// Monitors Docker daemon and container health, triggering server shutdown on failure. + /// + /// Checks every 5 seconds: + /// 1. Docker daemon responsiveness via ping + /// 2. Container running state via inspect + /// + /// When either check fails, the server shutdown signal is sent, which drops + /// the in-memory transport and causes the client to see a disconnect. + async fn docker_health_monitor( + client: DockerClient, + container: String, + shutdown: ShutdownSender, + ) { + let mut interval = tokio::time::interval(Duration::from_secs(5)); + loop { + interval.tick().await; + + // Check daemon responsiveness + if client.ping().await.is_err() { + warn!( + "Docker daemon unreachable, triggering server shutdown for container '{container}'" + ); + shutdown.shutdown(); + return; + } + + // Check container state + match client + .inner() + .inspect_container(&container, None::) + .await + { + Ok(info) => { + let running = info.state.as_ref().and_then(|s| s.running).unwrap_or(false); + if !running { + warn!( + "Container '{container}' is no longer running, triggering server shutdown" + ); + shutdown.shutdown(); + return; + } + } + Err(e) => { + warn!( + "Cannot inspect container '{container}': {e}, triggering server shutdown" + ); + shutdown.shutdown(); + return; + } + } + } + } + /// Creates a Docker client from the provided options. fn create_client(opts: &DockerOpts) -> io::Result { match &opts.docker_host { diff --git a/distant-docker/src/plugin.rs b/distant-docker/src/plugin.rs index 3a14825a3..769b5ae45 100644 --- a/distant-docker/src/plugin.rs +++ b/distant-docker/src/plugin.rs @@ -7,10 +7,11 @@ use std::future::Future; use std::io; use std::pin::Pin; +use std::time::Duration; use distant_core::Plugin; use distant_core::auth::Authenticator; -use distant_core::net::client::UntypedClient; +use distant_core::net::client::{ReconnectStrategy, UntypedClient}; use distant_core::net::common::{Destination, Map}; use log::*; @@ -80,6 +81,28 @@ impl Plugin for DockerPlugin { }) }) } + + fn reconnect<'a>( + &'a self, + raw_destination: &'a str, + options: &'a Map, + authenticator: &'a mut dyn Authenticator, + ) -> Pin> + Send + 'a>> { + // Re-establish connection to the Docker daemon and verify the container + // is still running. If the container was stopped, connect() will fail + // (container state check in Docker::connect). + self.connect(raw_destination, options, authenticator) + } + + fn reconnect_strategy(&self) -> ReconnectStrategy { + ReconnectStrategy::ExponentialBackoff { + base: Duration::from_secs(1), + factor: 2.0, + max_duration: Some(Duration::from_secs(60)), + max_retries: Some(10), + timeout: Some(Duration::from_secs(30)), + } + } } /// Parse Docker-specific options from the options map. @@ -103,3 +126,34 @@ fn parse_docker_opts(options: &Map) -> DockerOpts { .cloned(), } } + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use distant_core::Plugin; + + use super::*; + + // ------------------------------------------------------- + // DockerPlugin::name + // ------------------------------------------------------- + #[test] + fn docker_plugin_name_is_docker() { + let plugin = DockerPlugin; + assert_eq!(Plugin::name(&plugin), "docker"); + } + + // ------------------------------------------------------- + // DockerPlugin::reconnect_strategy + // ------------------------------------------------------- + #[test] + fn reconnect_strategy_returns_exponential_backoff() { + let plugin = DockerPlugin; + let strategy = Plugin::reconnect_strategy(&plugin); + assert!(strategy.is_exponential_backoff()); + assert_eq!(strategy.max_retries(), Some(10)); + assert_eq!(strategy.max_duration(), Some(Duration::from_secs(60))); + assert_eq!(strategy.timeout(), Some(Duration::from_secs(30))); + } +} diff --git a/distant-host/src/plugin.rs b/distant-host/src/plugin.rs index 596f66f2f..67a34c1f0 100644 --- a/distant-host/src/plugin.rs +++ b/distant-host/src/plugin.rs @@ -322,6 +322,25 @@ impl Plugin for HostPlugin { } }) } + + fn reconnect<'a>( + &'a self, + raw_destination: &'a str, + options: &'a Map, + authenticator: &'a mut dyn Authenticator, + ) -> Pin> + Send + 'a>> { + self.connect(raw_destination, options, authenticator) + } + + fn reconnect_strategy(&self) -> ReconnectStrategy { + ReconnectStrategy::ExponentialBackoff { + base: Duration::from_secs(2), + factor: 2.0, + max_duration: Some(Duration::from_secs(30)), + max_retries: Some(3), + timeout: Some(Duration::from_secs(60)), + } + } } #[cfg(test)] @@ -558,4 +577,17 @@ mod tests { .unwrap_or_else(|| "any".to_string()); assert_eq!(bind_server, "any"); } + + // ------------------------------------------------------- + // HostPlugin::reconnect_strategy + // ------------------------------------------------------- + #[test] + fn host_plugin_reconnect_strategy_returns_exponential_backoff() { + let plugin = HostPlugin::new(); + let strategy = Plugin::reconnect_strategy(&plugin); + assert!(strategy.is_exponential_backoff()); + assert_eq!(strategy.max_retries(), Some(3)); + assert_eq!(strategy.max_duration(), Some(Duration::from_secs(30))); + assert_eq!(strategy.timeout(), Some(Duration::from_secs(60))); + } } diff --git a/distant-ssh/src/api.rs b/distant-ssh/src/api.rs index 29b85b504..f74be7b84 100644 --- a/distant-ssh/src/api.rs +++ b/distant-ssh/src/api.rs @@ -52,6 +52,14 @@ impl SshApi { } } + /// Returns `true` if the underlying SSH session has been closed. + /// + /// This checks whether the russh connection task has terminated, which happens + /// when the SSH connection drops or the remote end disconnects. + pub fn is_session_closed(&self) -> bool { + self.pool.is_closed() + } + /// Get or create the cached SFTP session via the channel pool. async fn get_sftp(&self) -> io::Result { self.pool.sftp().await diff --git a/distant-ssh/src/lib.rs b/distant-ssh/src/lib.rs index 29e9459fb..5b7d1da6a 100644 --- a/distant-ssh/src/lib.rs +++ b/distant-ssh/src/lib.rs @@ -19,7 +19,7 @@ use std::time::Duration; use distant_core::net::auth::{AuthHandlerMap, DummyAuthHandler, Verifier}; use distant_core::net::client::{Client as NetClient, ClientConfig}; use distant_core::net::common::{InmemoryTransport, OneshotListener, Version}; -use distant_core::net::server::{Server, ServerRef}; +use distant_core::net::server::{Server, ServerRef, ShutdownSender}; use distant_core::protocol::PROTOCOL_VERSION; use distant_core::{ApiServerHandler, Client, Credentials}; use log::*; @@ -958,20 +958,30 @@ impl Ssh { Ok(family) } - /// Converts into a distant client + /// Converts into a distant client. + /// + /// Creates an in-memory server/client pair where the server side is backed by [`SshApi`]. + /// A health monitor task is spawned to detect SSH session death and trigger server shutdown, + /// which causes the client to see a disconnect. pub async fn into_distant_client(self) -> io::Result { let family = self.detect_family().await?; - let api = SshApi::new(self.pool, family, self.user.clone()); + let api = Arc::new(SshApi::new(self.pool, family, self.user.clone())); let (t1, t2) = InmemoryTransport::pair(100); let server = Server::new() - .handler(ApiServerHandler::new(api)) + .handler(ApiServerHandler::from_arc(Arc::clone(&api))) .verifier(Verifier::none()); - tokio::spawn(async move { - let _ = server.start(OneshotListener::from_value(t2)); - }); + let server_ref = server + .start(OneshotListener::from_value(t2)) + .map_err(io::Error::other)?; + + // Spawn health monitor that detects SSH session death + tokio::spawn(Self::ssh_health_monitor( + Arc::clone(&api), + server_ref.shutdown_sender(), + )); let client = NetClient::build() .auth_handler(DummyAuthHandler) @@ -984,21 +994,29 @@ impl Ssh { Ok(client) } - /// Converts into a pair of distant client and server ref + /// Converts into a pair of distant client and server ref. + /// + /// A health monitor task is spawned to detect SSH session death and trigger server shutdown. pub async fn into_distant_pair(self) -> io::Result<(Client, ServerRef)> { let family = self.detect_family().await?; - let api = SshApi::new(self.pool, family, self.user.clone()); + let api = Arc::new(SshApi::new(self.pool, family, self.user.clone())); let (t1, t2) = InmemoryTransport::pair(100); let server = Server::new() - .handler(ApiServerHandler::new(api)) + .handler(ApiServerHandler::from_arc(Arc::clone(&api))) .verifier(Verifier::none()); let server_ref = server .start(OneshotListener::from_value(t2)) .map_err(io::Error::other)?; + // Spawn health monitor that detects SSH session death + tokio::spawn(Self::ssh_health_monitor( + Arc::clone(&api), + server_ref.shutdown_sender(), + )); + let client = NetClient::build() .auth_handler(DummyAuthHandler) .config(ClientConfig::default()) @@ -1010,6 +1028,24 @@ impl Ssh { Ok((client, server_ref)) } + /// Monitors the SSH session health and triggers server shutdown when the session closes. + /// + /// Polls every 2 seconds. When the underlying russh connection task terminates + /// (e.g., the remote end disconnects or the network drops), the health monitor + /// sends a shutdown signal, which drops the in-memory transport and causes the + /// client to see a disconnect. + async fn ssh_health_monitor(api: Arc, shutdown: ShutdownSender) { + let mut interval = tokio::time::interval(Duration::from_secs(2)); + loop { + interval.tick().await; + if api.is_session_closed() { + warn!("SSH session closed, triggering server shutdown for reconnection"); + shutdown.shutdown(); + return; + } + } + } + /// Consume [`Ssh`] and launch a distant server on the remote machine, returning credentials /// for connecting to the launched server. pub async fn launch(self, opts: LaunchOpts) -> io::Result { diff --git a/distant-ssh/src/plugin.rs b/distant-ssh/src/plugin.rs index 6b53e8d05..4a149686d 100644 --- a/distant-ssh/src/plugin.rs +++ b/distant-ssh/src/plugin.rs @@ -8,11 +8,12 @@ use std::future::Future; use std::io; use std::path::PathBuf; use std::pin::Pin; +use std::time::Duration; use distant_core::Plugin; use distant_core::auth::Authenticator; use distant_core::auth::msg::*; -use distant_core::net::client::UntypedClient; +use distant_core::net::client::{ReconnectStrategy, UntypedClient}; use distant_core::net::common::{Destination, Map}; use log::*; use tokio::sync::Mutex; @@ -89,6 +90,27 @@ impl Plugin for SshPlugin { ssh.launch(opts).await?.try_to_destination() }) } + + fn reconnect<'a>( + &'a self, + raw_destination: &'a str, + options: &'a Map, + authenticator: &'a mut dyn Authenticator, + ) -> Pin> + Send + 'a>> { + // russh doesn't support session resumption. Must establish a new SSH + // session with full authentication (key files or ssh-agent). + self.connect(raw_destination, options, authenticator) + } + + fn reconnect_strategy(&self) -> ReconnectStrategy { + ReconnectStrategy::ExponentialBackoff { + base: Duration::from_secs(2), + factor: 2.0, + max_duration: Some(Duration::from_secs(30)), + max_retries: Some(5), + timeout: Some(Duration::from_secs(30)), + } + } } /// Adapter that bridges distant's [`Authenticator`] protocol with SSH authentication events. @@ -244,3 +266,34 @@ async fn load_ssh(destination: &Destination, options: &Map) -> io::Result io::Error { io::Error::new(io::ErrorKind::InvalidInput, format!("Invalid {label}")) } + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use distant_core::Plugin; + + use super::*; + + // ------------------------------------------------------- + // SshPlugin::name + // ------------------------------------------------------- + #[test] + fn ssh_plugin_name_is_ssh() { + let plugin = SshPlugin; + assert_eq!(Plugin::name(&plugin), "ssh"); + } + + // ------------------------------------------------------- + // SshPlugin::reconnect_strategy + // ------------------------------------------------------- + #[test] + fn reconnect_strategy_returns_exponential_backoff() { + let plugin = SshPlugin; + let strategy = Plugin::reconnect_strategy(&plugin); + assert!(strategy.is_exponential_backoff()); + assert_eq!(strategy.max_retries(), Some(5)); + assert_eq!(strategy.max_duration(), Some(Duration::from_secs(30))); + assert_eq!(strategy.timeout(), Some(Duration::from_secs(30))); + } +} diff --git a/distant-ssh/src/pool.rs b/distant-ssh/src/pool.rs index 06faf8849..5130ee898 100644 --- a/distant-ssh/src/pool.rs +++ b/distant-ssh/src/pool.rs @@ -183,6 +183,14 @@ impl ChannelPool { inner.open_count = inner.open_count.saturating_sub(1); } + /// Returns `true` if the underlying SSH connection has been closed. + /// + /// Delegates to russh's `Handle::is_closed()`, which returns true when + /// the connection task has terminated. + pub fn is_closed(&self) -> bool { + self.handle.is_closed() + } + /// Returns the server's channel limit, if it has been discovered. /// /// The limit is discovered when the first `channel_open_session` call fails, diff --git a/src/cli/commands/client.rs b/src/cli/commands/client.rs index 255913f71..8c9b98b3f 100644 --- a/src/cli/commands/client.rs +++ b/src/cli/commands/client.rs @@ -27,7 +27,8 @@ use dialoguer::theme::ColorfulTheme; use crate::cli::common::{ Cache, JsonAuthHandler, MsgReceiver, MsgSender, PromptAuthHandler, Ui, connect_to_manager, - format_connection, try_connect as try_connect_no_autostart, + format_connection, subscribe_and_display_connection_events, + try_connect as try_connect_no_autostart, }; use crate::constants::MAX_PIPE_CHUNK_SIZE; use crate::options::{ @@ -67,9 +68,13 @@ async fn async_run(cmd: ClientSubcommand, quiet: bool) -> CliResult { destination, format, network, - options, + mut options, new, + no_reconnect, } => { + if no_reconnect { + options.insert("no_reconnect".to_string(), "true".to_string()); + } debug!("Connecting to manager"); let mut client = connect_to_manager(format, network, &ui).await?; @@ -200,7 +205,11 @@ async fn async_run(cmd: ClientSubcommand, quiet: bool) -> CliResult { format, network, mut options, + no_reconnect, } => { + if no_reconnect { + options.insert("no_reconnect".to_string(), "true".to_string()); + } debug!("Connecting to manager"); let mut client = connect_to_manager(format, network, &ui).await?; @@ -339,6 +348,8 @@ async fn async_run(cmd: ClientSubcommand, quiet: bool) -> CliResult { Is it running? Start it with: distant manager listen --daemon", )?; + subscribe_and_display_connection_events(&mut client, Format::Json).await; + let mut cache = read_cache(&cache).await; let connection_id = use_or_lookup_connection_id(&mut cache, connection, &mut client).await?; @@ -467,6 +478,8 @@ async fn async_run(cmd: ClientSubcommand, quiet: bool) -> CliResult { debug!("Connecting to manager"); let mut client = connect_to_manager(Format::Shell, network, &ui).await?; + subscribe_and_display_connection_events(&mut client, Format::Shell).await; + let mut cache = read_cache(&cache).await; let connection_id = use_or_lookup_connection_id(&mut cache, connection, &mut client).await?; @@ -511,6 +524,8 @@ async fn async_run(cmd: ClientSubcommand, quiet: bool) -> CliResult { debug!("Connecting to manager"); let mut client = connect_to_manager(Format::Shell, network, &ui).await?; + subscribe_and_display_connection_events(&mut client, Format::Shell).await; + let mut cache = read_cache(&cache).await; let connection_id = use_or_lookup_connection_id(&mut cache, connection, &mut client).await?; @@ -1140,17 +1155,23 @@ async fn async_run(cmd: ClientSubcommand, quiet: bool) -> CliResult { ClientSubcommand::Ssh { cache, destination, - options, + mut options, network, current_dir, environment, predict, new, + no_reconnect, cmd, } => { + if no_reconnect { + options.insert("no_reconnect".to_string(), "true".to_string()); + } debug!("Connecting to manager (auto-start enabled)"); let mut client = connect_to_manager(Format::Shell, network, &ui).await?; + subscribe_and_display_connection_events(&mut client, Format::Shell).await; + // Ensure destination has ssh:// scheme let destination = ensure_scheme(&destination, "ssh"); @@ -1558,6 +1579,28 @@ async fn async_run(cmd: ClientSubcommand, quiet: bool) -> CliResult { } } } + ClientSubcommand::Reconnect { + id, + format, + network, + .. + } => { + debug!("Connecting to manager"); + let mut client = connect_to_manager(format, network, &ui).await?; + + debug!("Requesting reconnection for connection {}", id); + client + .reconnect(id) + .await + .with_context(|| format!("Failed to initiate reconnection for connection {id}"))?; + + match format { + Format::Json => println!("{}", json!({"type": "reconnect_initiated", "id": id})), + Format::Shell => { + ui.success(&format!("Reconnection initiated for connection {id}")); + } + } + } } Ok(()) diff --git a/src/cli/commands/server.rs b/src/cli/commands/server.rs index 6b4ac65f0..90afda629 100644 --- a/src/cli/commands/server.rs +++ b/src/cli/commands/server.rs @@ -108,6 +108,8 @@ async fn async_run(cmd: ServerSubcommand, _is_forked: bool) -> CliResult { current_dir, watch, daemon: _, + heartbeat_interval, + max_heartbeat_failures, key_from_stdin, output_to_local_pipe, } => { @@ -156,6 +158,8 @@ async fn async_run(cmd: ServerSubcommand, _is_forked: bool) -> CliResult { let server = Server::tcp() .config(NetServerConfig { shutdown: shutdown.into_inner(), + connection_heartbeat: std::time::Duration::from_secs(heartbeat_interval), + max_heartbeat_failures, ..Default::default() }) .handler(handler) diff --git a/src/cli/common/client.rs b/src/cli/common/client.rs index 162f9ea2f..da981cb26 100644 --- a/src/cli/common/client.rs +++ b/src/cli/common/client.rs @@ -7,7 +7,7 @@ use anyhow::Context; use distant_core::net::auth::msg::*; use distant_core::net::auth::{AuthHandler, AuthMethodHandler}; use distant_core::net::client::{Client as NetClient, ClientConfig, ReconnectStrategy}; -use distant_core::net::manager::{ManagerClient, PROTOCOL_VERSION}; +use distant_core::net::manager::{ManagerClient, ManagerResponse, PROTOCOL_VERSION}; use log::*; use crate::cli::common::ui::{Spinner, Ui}; @@ -487,6 +487,51 @@ pub async fn try_connect( } } +/// Subscribe to connection state change events and spawn a background task +/// that displays them to the user. +/// +/// Subscribes to the manager's connection event stream. Events are printed +/// to stderr in shell format or to stdout as JSON, matching the project's +/// output conventions (`ui.rs:10`). The background task runs until the +/// mailbox closes (i.e., the manager connection drops). +/// +/// Subscription failures are logged but do not fail the caller, since event +/// display is best-effort and should not block the primary command. +pub async fn subscribe_and_display_connection_events(client: &mut ManagerClient, format: Format) { + match client.subscribe_connection_events().await { + Ok(mut mailbox) => { + tokio::spawn(async move { + while let Some(res) = mailbox.next().await { + match res.payload { + ManagerResponse::ConnectionStateChanged { id, state } => match format { + Format::Shell => { + eprintln!("[distant] Connection {id}: {state}"); + } + Format::Json => { + println!( + "{}", + serde_json::json!({ + "type": "connection_state", + "id": id, + "state": state.to_string() + }) + ); + } + }, + _ => { + trace!("Ignoring non-state-change event on subscription mailbox"); + } + } + } + trace!("Connection event subscription mailbox closed"); + }); + } + Err(err) => { + debug!("Failed to subscribe to connection events: {err}"); + } + } +} + #[cfg(test)] mod tests { //! Tests for `Client` construction and auth handler swapping, diff --git a/src/options.rs b/src/options.rs index 3673da91b..6de46e979 100644 --- a/src/options.rs +++ b/src/options.rs @@ -240,6 +240,9 @@ impl Options { ClientSubcommand::Select { network, .. } => { network.merge(config.client.network); } + ClientSubcommand::Reconnect { network, .. } => { + network.merge(config.client.network); + } } } DistantSubcommand::Generate(_) => { @@ -421,6 +424,10 @@ pub enum ClientSubcommand { #[clap(long)] new: bool, + /// Disable automatic reconnection on connection loss + #[clap(long)] + no_reconnect: bool, + /// Destination URI (e.g. `ssh://user@host:22`, `docker://ubuntu:22.04`) destination: String, }, @@ -516,6 +523,10 @@ pub enum ClientSubcommand { #[clap(short, long, default_value_t, value_enum)] format: Format, + /// Disable automatic reconnection on connection loss + #[clap(long)] + no_reconnect: bool, + /// Destination URI (e.g. `ssh://user@host:22`, `docker://ubuntu:22.04`) destination: String, }, @@ -707,6 +718,10 @@ pub enum ClientSubcommand { #[clap(long)] new: bool, + /// Disable automatic reconnection on connection loss + #[clap(long)] + no_reconnect: bool, + /// Destination URI (e.g. `ssh://user@host:22`, `docker://ubuntu:22.04`) destination: String, @@ -780,6 +795,27 @@ pub enum ClientSubcommand { )] cache: PathBuf, }, + + /// Manually trigger reconnection for a connection + Reconnect { + /// Connection ID to reconnect + id: ConnectionId, + + #[clap(short, long, default_value_t, value_enum)] + format: Format, + + #[clap(flatten)] + network: NetworkSettings, + + /// Location to store cached data + #[clap( + long, + value_hint = ValueHint::FilePath, + value_parser, + default_value = CACHE_FILE_PATH_STR.as_str() + )] + cache: PathBuf, + }, } impl ClientSubcommand { @@ -799,6 +835,7 @@ impl ClientSubcommand { Self::Version { cache, .. } => cache.as_path(), Self::Kill { cache, .. } => cache.as_path(), Self::Select { cache, .. } => cache.as_path(), + Self::Reconnect { cache, .. } => cache.as_path(), } } @@ -818,6 +855,7 @@ impl ClientSubcommand { Self::Version { network, .. } => network, Self::Kill { network, .. } => network, Self::Select { network, .. } => network, + Self::Reconnect { network, .. } => network, } } @@ -839,6 +877,7 @@ impl ClientSubcommand { Self::Version { format, .. } => *format, Self::Kill { format, .. } => *format, Self::Select { format, .. } => *format, + Self::Reconnect { format, .. } => *format, } } } @@ -1442,6 +1481,15 @@ pub enum ServerSubcommand { #[clap(long)] daemon: bool, + /// Heartbeat interval in seconds (default: 5) + #[clap(long, default_value = "5")] + heartbeat_interval: u64, + + /// Maximum consecutive heartbeat failures before the connection is terminated (default: 3). + /// A value of 0 means heartbeat failures are never escalated. + #[clap(long, default_value = "3")] + max_heartbeat_failures: u32, + #[clap(flatten)] watch: ServerListenWatchOptions, @@ -1789,6 +1837,7 @@ mod tests { }, format: Format::Json, new: false, + no_reconnect: false, destination: "test://destination".to_string(), }), }; @@ -1829,6 +1878,7 @@ mod tests { }, format: Format::Json, new: false, + no_reconnect: false, destination: "test://destination".to_string(), }), } @@ -1853,6 +1903,7 @@ mod tests { }, format: Format::Json, new: false, + no_reconnect: false, destination: "test://destination".to_string(), }), }; @@ -1893,6 +1944,7 @@ mod tests { }, format: Format::Json, new: false, + no_reconnect: false, destination: "test://destination".to_string(), }), } @@ -1919,6 +1971,7 @@ mod tests { windows_pipe: None, }, format: Format::Json, + no_reconnect: false, destination: "test://destination".to_string(), }), }; @@ -1970,6 +2023,7 @@ mod tests { windows_pipe: Some(String::from("config-windows-pipe")), }, format: Format::Json, + no_reconnect: false, destination: "test://destination".to_string(), }), } @@ -1996,6 +2050,7 @@ mod tests { windows_pipe: Some(String::from("cli-windows-pipe")), }, format: Format::Json, + no_reconnect: false, destination: "test://destination".to_string(), }), }; @@ -2047,6 +2102,7 @@ mod tests { windows_pipe: Some(String::from("cli-windows-pipe")), }, format: Format::Json, + no_reconnect: false, destination: "test://destination".to_string(), }), } @@ -4255,6 +4311,120 @@ mod tests { ); } + #[test] + fn distant_reconnect_should_support_merging_with_config() { + let mut options = Options { + quiet: false, + config_path: None, + logging: LoggingSettings { + log_file: None, + log_level: None, + }, + command: DistantSubcommand::Client(ClientSubcommand::Reconnect { + id: 1, + format: Format::Shell, + cache: PathBuf::new(), + network: NetworkSettings { + unix_socket: None, + windows_pipe: None, + }, + }), + }; + + options.merge(Config { + client: ClientConfig { + logging: LoggingSettings { + log_file: Some(PathBuf::from("config-log-file")), + log_level: Some(LogLevel::Trace), + }, + network: NetworkSettings { + unix_socket: Some(PathBuf::from("config-unix-socket")), + windows_pipe: Some(String::from("config-windows-pipe")), + }, + ..Default::default() + }, + ..Default::default() + }); + + assert_eq!( + options, + Options { + quiet: false, + config_path: None, + logging: LoggingSettings { + log_file: Some(PathBuf::from("config-log-file")), + log_level: Some(LogLevel::Trace), + }, + command: DistantSubcommand::Client(ClientSubcommand::Reconnect { + id: 1, + format: Format::Shell, + cache: PathBuf::new(), + network: NetworkSettings { + unix_socket: Some(PathBuf::from("config-unix-socket")), + windows_pipe: Some(String::from("config-windows-pipe")), + }, + }), + } + ); + } + + #[test] + fn distant_reconnect_should_prioritize_explicit_cli_options_when_merging() { + let mut options = Options { + quiet: false, + config_path: None, + logging: LoggingSettings { + log_file: Some(PathBuf::from("cli-log-file")), + log_level: Some(LogLevel::Info), + }, + command: DistantSubcommand::Client(ClientSubcommand::Reconnect { + id: 42, + format: Format::Json, + cache: PathBuf::new(), + network: NetworkSettings { + unix_socket: Some(PathBuf::from("cli-unix-socket")), + windows_pipe: Some(String::from("cli-windows-pipe")), + }, + }), + }; + + options.merge(Config { + client: ClientConfig { + logging: LoggingSettings { + log_file: Some(PathBuf::from("config-log-file")), + log_level: Some(LogLevel::Trace), + }, + network: NetworkSettings { + unix_socket: Some(PathBuf::from("config-unix-socket")), + windows_pipe: Some(String::from("config-windows-pipe")), + }, + ..Default::default() + }, + ..Default::default() + }); + + assert_eq!( + options, + Options { + quiet: false, + config_path: None, + logging: LoggingSettings { + log_file: Some(PathBuf::from("cli-log-file")), + log_level: Some(LogLevel::Info), + }, + command: DistantSubcommand::Client(ClientSubcommand::Reconnect { + id: 42, + format: Format::Json, + cache: PathBuf::new(), + network: NetworkSettings { + unix_socket: Some(PathBuf::from("cli-unix-socket")), + windows_pipe: Some(String::from("cli-windows-pipe")), + }, + }), + } + ); + } + #[test] fn distant_manager_listen_should_support_merging_with_config() { let mut options = Options { @@ -4495,6 +4665,8 @@ mod tests { watch_debounce_tick_rate: None, }, daemon: false, + heartbeat_interval: 5, + max_heartbeat_failures: 3, key_from_stdin: false, output_to_local_pipe: None, }), @@ -4547,6 +4719,8 @@ mod tests { watch_debounce_tick_rate: Some(Seconds::from(300u32)), }, daemon: false, + heartbeat_interval: 5, + max_heartbeat_failures: 3, key_from_stdin: false, output_to_local_pipe: None, }), @@ -4578,6 +4752,8 @@ mod tests { watch_debounce_tick_rate: Some(Seconds::from(30u32)), }, daemon: false, + heartbeat_interval: 5, + max_heartbeat_failures: 3, key_from_stdin: false, output_to_local_pipe: None, }), @@ -4630,6 +4806,8 @@ mod tests { watch_debounce_tick_rate: Some(Seconds::from(30u32)), }, daemon: false, + heartbeat_interval: 5, + max_heartbeat_failures: 3, key_from_stdin: false, output_to_local_pipe: None, }), @@ -4718,6 +4896,120 @@ mod tests { } } + #[test] + fn distant_reconnect_should_parse_with_id() { + let options = Options::try_parse_from(["distant", "reconnect", "42"]).unwrap(); + match options.command { + DistantSubcommand::Client(ClientSubcommand::Reconnect { id, .. }) => { + assert_eq!(id, 42); + } + other => panic!("Expected Reconnect with id=42, got {other:?}"), + } + } + + #[test] + fn distant_reconnect_should_require_id() { + assert!(Options::try_parse_from(["distant", "reconnect"]).is_err()); + } + + #[test] + fn distant_reconnect_should_parse_with_format_json() { + let options = + Options::try_parse_from(["distant", "reconnect", "--format", "json", "10"]).unwrap(); + match options.command { + DistantSubcommand::Client(ClientSubcommand::Reconnect { id, format, .. }) => { + assert_eq!(id, 10); + assert_eq!(format, Format::Json); + } + other => panic!("Expected Reconnect with json format, got {other:?}"), + } + } + + #[test] + fn distant_reconnect_should_reject_non_numeric_id() { + assert!(Options::try_parse_from(["distant", "reconnect", "abc"]).is_err()); + } + + #[test] + fn distant_reconnect_should_default_to_shell_format() { + let options = Options::try_parse_from(["distant", "reconnect", "7"]).unwrap(); + match options.command { + DistantSubcommand::Client(ClientSubcommand::Reconnect { id, format, .. }) => { + assert_eq!(id, 7); + assert_eq!(format, Format::Shell); + } + other => panic!("Expected Reconnect with default shell format, got {other:?}"), + } + } + + #[test] + fn distant_reconnect_should_parse_with_unix_socket() { + let options = Options::try_parse_from([ + "distant", + "reconnect", + "--unix-socket", + "/tmp/test.sock", + "3", + ]) + .unwrap(); + match options.command { + DistantSubcommand::Client(ClientSubcommand::Reconnect { id, network, .. }) => { + assert_eq!(id, 3); + assert_eq!(network.unix_socket, Some(PathBuf::from("/tmp/test.sock"))); + } + other => panic!("Expected Reconnect with unix socket, got {other:?}"), + } + } + + #[test] + fn distant_reconnect_should_parse_with_custom_cache_path() { + let options = + Options::try_parse_from(["distant", "reconnect", "--cache", "/custom/cache", "5"]) + .unwrap(); + match options.command { + DistantSubcommand::Client(ClientSubcommand::Reconnect { id, cache, .. }) => { + assert_eq!(id, 5); + assert_eq!(cache, PathBuf::from("/custom/cache")); + } + other => panic!("Expected Reconnect with custom cache, got {other:?}"), + } + } + + #[test] + fn distant_reconnect_should_parse_with_all_options() { + let options = Options::try_parse_from([ + "distant", + "reconnect", + "--format", + "json", + "--unix-socket", + "/tmp/mgr.sock", + "--cache", + "/my/cache", + "99", + ]) + .unwrap(); + match options.command { + DistantSubcommand::Client(ClientSubcommand::Reconnect { + id, + format, + network, + cache, + }) => { + assert_eq!(id, 99); + assert_eq!(format, Format::Json); + assert_eq!(network.unix_socket, Some(PathBuf::from("/tmp/mgr.sock"))); + assert_eq!(cache, PathBuf::from("/my/cache")); + } + other => panic!("Expected Reconnect with all options, got {other:?}"), + } + } + + #[test] + fn distant_reconnect_should_reject_negative_id() { + assert!(Options::try_parse_from(["distant", "reconnect", "-1"]).is_err()); + } + #[test] fn distant_manager_list_should_not_parse() { assert!(Options::try_parse_from(["distant", "manager", "list"]).is_err()); @@ -4795,6 +5087,7 @@ mod tests { environment: Default::default(), predict: PredictMode::Adaptive, new: false, + no_reconnect: false, destination: "test://host".to_string(), cmd: None, }; @@ -4819,6 +5112,7 @@ mod tests { network: NetworkSettings::default(), format: Format::Json, new: false, + no_reconnect: false, destination: "test://host".to_string(), }; assert!(cmd.format().is_json()); @@ -4834,6 +5128,7 @@ mod tests { options: Default::default(), network: NetworkSettings::default(), format: Format::Shell, + no_reconnect: false, destination: "test://host".to_string(), }; assert_eq!(cmd.format(), Format::Shell); @@ -4883,6 +5178,17 @@ mod tests { assert!(cmd.format().is_json()); } + #[test] + fn format_reconnect_returns_specified_format() { + let cmd = ClientSubcommand::Reconnect { + id: 1, + format: Format::Json, + network: NetworkSettings::default(), + cache: PathBuf::new(), + }; + assert!(cmd.format().is_json()); + } + #[test] fn format_filesystem_returns_shell() { let cmd = ClientSubcommand::FileSystem(ClientFileSystemSubcommand::Copy { @@ -4917,6 +5223,8 @@ mod tests { shutdown: Value::Default(distant_core::net::server::Shutdown::Never), current_dir: None, daemon: false, + heartbeat_interval: 5, + max_heartbeat_failures: 3, watch: ServerListenWatchOptions { watch_polling: false, watch_poll_interval: None, @@ -4975,6 +5283,7 @@ mod tests { network: net.clone(), format: Format::Shell, new: false, + no_reconnect: false, destination: "test://host".to_string(), }, ClientSubcommand::Launch { @@ -4985,6 +5294,7 @@ mod tests { options: Default::default(), network: net.clone(), format: Format::Shell, + no_reconnect: false, destination: "test://host".to_string(), }, ClientSubcommand::Shell { @@ -5029,6 +5339,7 @@ mod tests { environment: Default::default(), predict: PredictMode::Adaptive, new: false, + no_reconnect: false, destination: "test://host".to_string(), cmd: None, }, @@ -5050,6 +5361,12 @@ mod tests { network: net.clone(), cache: cache.clone(), }, + ClientSubcommand::Reconnect { + id: 1, + format: Format::Shell, + network: net.clone(), + cache: cache.clone(), + }, ]; for cmd in &cases { @@ -5205,6 +5522,7 @@ mod tests { environment: Default::default(), predict: PredictMode::Adaptive, new: false, + no_reconnect: false, destination: "test://host".to_string(), cmd: None, }; @@ -5235,6 +5553,14 @@ mod tests { }; assert_eq!(cmd.network_settings(), &net); + let cmd = ClientSubcommand::Reconnect { + id: 1, + format: Format::Shell, + network: net.clone(), + cache: PathBuf::new(), + }; + assert_eq!(cmd.network_settings(), &net); + // FileSystem wrapper let cmd = ClientSubcommand::FileSystem(ClientFileSystemSubcommand::Copy { cache: PathBuf::new(), @@ -5466,6 +5792,8 @@ mod tests { shutdown: Value::Default(distant_core::net::server::Shutdown::Never), current_dir: None, daemon: false, + heartbeat_interval: 5, + max_heartbeat_failures: 3, watch: ServerListenWatchOptions { watch_polling: false, watch_poll_interval: None, @@ -5503,6 +5831,7 @@ mod tests { environment: Default::default(), predict: PredictMode::Adaptive, new: false, + no_reconnect: false, destination: "test://host".to_string(), cmd: None, }), @@ -5546,6 +5875,7 @@ mod tests { environment: Default::default(), predict: PredictMode::Adaptive, new: false, + no_reconnect: false, destination: "test://host".to_string(), cmd: None, }), @@ -5574,6 +5904,7 @@ mod tests { environment: Default::default(), predict: PredictMode::Adaptive, new: false, + no_reconnect: false, destination: "test://host".to_string(), cmd: None, }), @@ -5617,6 +5948,7 @@ mod tests { environment: Default::default(), predict: PredictMode::Adaptive, new: false, + no_reconnect: false, destination: "test://host".to_string(), cmd: None, }), @@ -5879,4 +6211,255 @@ mod tests { assert!(!client.is_manager()); assert!(!client.is_generate()); } + + // ------------------------------------------------------- + // --no-reconnect flag CLI parsing tests + // ------------------------------------------------------- + + #[test] + fn distant_connect_should_default_no_reconnect_to_false() { + let options = + Options::try_parse_from(["distant", "connect", "test://destination"]).unwrap(); + match options.command { + DistantSubcommand::Client(ClientSubcommand::Connect { no_reconnect, .. }) => { + assert!(!no_reconnect); + } + other => panic!("Expected Connect, got {other:?}"), + } + } + + #[test] + fn distant_connect_should_parse_no_reconnect_flag() { + let options = + Options::try_parse_from(["distant", "connect", "--no-reconnect", "test://destination"]) + .unwrap(); + match options.command { + DistantSubcommand::Client(ClientSubcommand::Connect { no_reconnect, .. }) => { + assert!(no_reconnect); + } + other => panic!("Expected Connect with --no-reconnect, got {other:?}"), + } + } + + #[test] + fn distant_launch_should_default_no_reconnect_to_false() { + let options = Options::try_parse_from(["distant", "launch", "test://destination"]).unwrap(); + match options.command { + DistantSubcommand::Client(ClientSubcommand::Launch { no_reconnect, .. }) => { + assert!(!no_reconnect); + } + other => panic!("Expected Launch, got {other:?}"), + } + } + + #[test] + fn distant_launch_should_parse_no_reconnect_flag() { + let options = + Options::try_parse_from(["distant", "launch", "--no-reconnect", "test://destination"]) + .unwrap(); + match options.command { + DistantSubcommand::Client(ClientSubcommand::Launch { no_reconnect, .. }) => { + assert!(no_reconnect); + } + other => panic!("Expected Launch with --no-reconnect, got {other:?}"), + } + } + + #[cfg(feature = "ssh")] + #[test] + fn distant_ssh_should_default_no_reconnect_to_false() { + let options = Options::try_parse_from(["distant", "ssh", "user@host"]).unwrap(); + match options.command { + DistantSubcommand::Client(ClientSubcommand::Ssh { no_reconnect, .. }) => { + assert!(!no_reconnect); + } + other => panic!("Expected Ssh, got {other:?}"), + } + } + + #[cfg(feature = "ssh")] + #[test] + fn distant_ssh_should_parse_no_reconnect_flag() { + let options = + Options::try_parse_from(["distant", "ssh", "--no-reconnect", "user@host"]).unwrap(); + match options.command { + DistantSubcommand::Client(ClientSubcommand::Ssh { no_reconnect, .. }) => { + assert!(no_reconnect); + } + other => panic!("Expected Ssh with --no-reconnect, got {other:?}"), + } + } + + // ------------------------------------------------------- + // --heartbeat-interval and --max-heartbeat-failures CLI parsing tests + // ------------------------------------------------------- + + #[cfg(feature = "host")] + #[test] + fn distant_server_listen_should_default_heartbeat_interval_to_5() { + let options = Options::try_parse_from(["distant", "server", "listen"]).unwrap(); + match options.command { + DistantSubcommand::Server(ServerSubcommand::Listen { + heartbeat_interval, .. + }) => { + assert_eq!(heartbeat_interval, 5); + } + other => panic!("Expected Server Listen, got {other:?}"), + } + } + + #[cfg(feature = "host")] + #[test] + fn distant_server_listen_should_parse_custom_heartbeat_interval() { + let options = + Options::try_parse_from(["distant", "server", "listen", "--heartbeat-interval", "10"]) + .unwrap(); + match options.command { + DistantSubcommand::Server(ServerSubcommand::Listen { + heartbeat_interval, .. + }) => { + assert_eq!(heartbeat_interval, 10); + } + other => panic!("Expected Server Listen with heartbeat_interval=10, got {other:?}"), + } + } + + #[cfg(feature = "host")] + #[test] + fn distant_server_listen_should_default_max_heartbeat_failures_to_3() { + let options = Options::try_parse_from(["distant", "server", "listen"]).unwrap(); + match options.command { + DistantSubcommand::Server(ServerSubcommand::Listen { + max_heartbeat_failures, + .. + }) => { + assert_eq!(max_heartbeat_failures, 3); + } + other => panic!("Expected Server Listen, got {other:?}"), + } + } + + #[cfg(feature = "host")] + #[test] + fn distant_server_listen_should_parse_custom_max_heartbeat_failures() { + let options = Options::try_parse_from([ + "distant", + "server", + "listen", + "--max-heartbeat-failures", + "10", + ]) + .unwrap(); + match options.command { + DistantSubcommand::Server(ServerSubcommand::Listen { + max_heartbeat_failures, + .. + }) => { + assert_eq!(max_heartbeat_failures, 10); + } + other => { + panic!("Expected Server Listen with max_heartbeat_failures=10, got {other:?}") + } + } + } + + #[cfg(feature = "host")] + #[test] + fn distant_server_listen_should_accept_zero_max_heartbeat_failures() { + let options = Options::try_parse_from([ + "distant", + "server", + "listen", + "--max-heartbeat-failures", + "0", + ]) + .unwrap(); + match options.command { + DistantSubcommand::Server(ServerSubcommand::Listen { + max_heartbeat_failures, + .. + }) => { + assert_eq!(max_heartbeat_failures, 0); + } + other => { + panic!("Expected Server Listen with max_heartbeat_failures=0, got {other:?}") + } + } + } + + #[cfg(feature = "host")] + #[test] + fn distant_server_listen_should_parse_heartbeat_and_max_failures_together() { + let options = Options::try_parse_from([ + "distant", + "server", + "listen", + "--heartbeat-interval", + "15", + "--max-heartbeat-failures", + "7", + ]) + .unwrap(); + match options.command { + DistantSubcommand::Server(ServerSubcommand::Listen { + heartbeat_interval, + max_heartbeat_failures, + .. + }) => { + assert_eq!(heartbeat_interval, 15); + assert_eq!(max_heartbeat_failures, 7); + } + other => panic!( + "Expected Server Listen with heartbeat_interval=15 and max_heartbeat_failures=7, got {other:?}" + ), + } + } + + #[cfg(feature = "host")] + #[test] + fn distant_server_listen_should_reject_non_numeric_heartbeat_interval() { + assert!( + Options::try_parse_from(["distant", "server", "listen", "--heartbeat-interval", "abc"]) + .is_err() + ); + } + + #[cfg(feature = "host")] + #[test] + fn distant_server_listen_should_reject_negative_heartbeat_interval() { + assert!( + Options::try_parse_from(["distant", "server", "listen", "--heartbeat-interval", "-1"]) + .is_err() + ); + } + + #[cfg(feature = "host")] + #[test] + fn distant_server_listen_should_reject_non_numeric_max_heartbeat_failures() { + assert!( + Options::try_parse_from([ + "distant", + "server", + "listen", + "--max-heartbeat-failures", + "abc" + ]) + .is_err() + ); + } + + #[cfg(feature = "host")] + #[test] + fn distant_server_listen_should_reject_negative_max_heartbeat_failures() { + assert!( + Options::try_parse_from([ + "distant", + "server", + "listen", + "--max-heartbeat-failures", + "-1" + ]) + .is_err() + ); + } }