diff --git a/crates/filesync/src/client.rs b/crates/filesync/src/client.rs index 0d2b42b..0306037 100644 --- a/crates/filesync/src/client.rs +++ b/crates/filesync/src/client.rs @@ -7,8 +7,8 @@ use crate::ledger::{DeletionLedger, Tombstone}; use crate::manifest; use crate::protocol::*; use crate::sync_engine::{ConflictInfo, SyncEngine}; +use crate::transport::{Connection, HEARTBEAT_POLL_SECS, is_heartbeat_timeout}; use crate::timestamp_id; -use crate::transport::Connection; use crate::watcher::{self, FsEvent}; use bytehive_core::MessageBus; @@ -138,6 +138,12 @@ impl Client { pub fn shutdown(&self) { debug!("filesync client: shutdown requested"); self.stopped.store(true, Ordering::SeqCst); + self.force_disconnect(); + } + + /// Force-close the active connection without marking the client stopped, + /// so the `run()` loop can reconnect immediately (used on resume). + pub fn force_disconnect(&self) { if let Some(conn) = self.active_conn.lock().clone() { debug!("filesync client: forcing active connection closed"); conn.shutdown(); @@ -158,6 +164,11 @@ impl Client { break; } info!("filesync: connecting to {} …", self.server_addr); + // Heartbeat fast-path: a heartbeat timeout means the peer is + // (probably) gone or we just resumed from suspend — the wall-clock + // gap already proved liveness is lost, so reconnect immediately + // instead of waiting out the backoff. + let mut skip_backoff = false; match self.session() { Ok(()) => { info!("filesync: session ended cleanly"); @@ -183,6 +194,10 @@ impl Client { ); // Back off a long time; the admin needs to explicitly re-allow. backoff = Duration::from_secs(300); + } else if is_heartbeat_timeout(&e) { + info!("filesync: heartbeat timeout — reconnecting immediately"); + backoff = Duration::from_secs(1); + skip_backoff = true; } else { error!("filesync: session error: {e}"); debug!("filesync: session error kind={:?}", e.kind()); @@ -193,6 +208,9 @@ impl Client { debug!("filesync: run loop: stop flag set after session, exiting"); break; } + if skip_backoff { + continue; + } info!("filesync: reconnecting in {} s …", backoff.as_secs()); debug!( "filesync: backoff={} ms before next connection attempt", @@ -320,7 +338,7 @@ impl Client { credential: None, })?; - match conn.recv()? { + match recv_with_heartbeat(&conn)? { Message::Hello { node_id, protocol_version, @@ -392,7 +410,7 @@ impl Client { } debug!("filesync session: waiting for server ManifestExchange"); - let remote = match conn.recv()? { + let remote = match recv_with_heartbeat(&conn)? { Message::ManifestExchange(m) => m, other => { error!("filesync session: expected ManifestExchange from server"); @@ -477,7 +495,7 @@ impl Client { // ---- Deletion-ledger exchange (symmetric order: server sent first, // we reply — avoids both-wait deadlock) ---- debug!("filesync session: waiting for server LedgerExchange"); - let peer_ledger_entries = match conn.recv()? { + let peer_ledger_entries = match recv_with_heartbeat(&conn)? { Message::LedgerExchange { entries } => entries, other => { error!("filesync session: expected LedgerExchange from server"); @@ -536,7 +554,7 @@ impl Client { debug!("filesync session: entering initial-sync receive loop"); loop { - match conn.recv()? { + match recv_with_heartbeat(&conn)? { Message::Bundle(b) => { let n_files = b.files.iter().filter(|f| !f.metadata.is_dir).count(); let n_dirs = b.files.iter().filter(|f| f.metadata.is_dir).count(); @@ -882,6 +900,26 @@ impl Client { } } +/// Blocking recv with heartbeat maintenance: polls with a 2 s timeout so +/// Ping is sent while idle and a dead peer / suspend-resume gap trips the +/// heartbeat timeout (returned as Err) instead of hanging forever. +fn recv_with_heartbeat(conn: &Connection) -> io::Result { + loop { + match conn.recv_timeout(Duration::from_secs(HEARTBEAT_POLL_SECS)) { + Ok(msg) => return Ok(msg), + Err(e) + if e.kind() == io::ErrorKind::TimedOut + || e.kind() == io::ErrorKind::WouldBlock => + { + // Idle window: send Ping if due, trip timeout if peer is + // stale (suspend recovery lands here on next check). + conn.heartbeat_maintenance()?; + } + Err(e) => return Err(e), + } + } +} + fn recv_loop( engine: Arc, conn: Arc, @@ -891,7 +929,7 @@ fn recv_loop( let prefix = "filesync recv"; debug!("{prefix}: loop started, waiting for incremental messages from server"); loop { - match conn.recv() { + match recv_with_heartbeat(&conn) { Ok(Message::Bundle(b)) => { let n_files = b.files.iter().filter(|f| !f.metadata.is_dir).count(); let n_dirs = b.files.iter().filter(|f| f.metadata.is_dir).count(); @@ -1116,6 +1154,18 @@ fn send_loop( Err(TryRecvError::Empty) => {} } + // Heartbeat tick (already wakes every DEBOUNCE_MS): sends Ping when + // idle, breaks the loop on timeout so the session tears down and + // run() reconnects + rescans. + if let Err(e) = conn.heartbeat_maintenance() { + if is_heartbeat_timeout(&e) { + error!("filesync send: heartbeat timeout ({e}), tearing down session"); + } else { + error!("filesync send: heartbeat maintenance failed ({e}), tearing down session"); + } + break; + } + pending.periodic_rescan(&engine, "filesync"); match fs_rx.recv_timeout(Duration::from_millis(DEBOUNCE_MS)) { diff --git a/crates/filesync/src/gui/manager.rs b/crates/filesync/src/gui/manager.rs index ee10cd3..edd349d 100644 --- a/crates/filesync/src/gui/manager.rs +++ b/crates/filesync/src/gui/manager.rs @@ -2,10 +2,9 @@ use crate::client::Client; use crate::exclusions::{ExclusionConfig, Exclusions}; use crate::gui::config::GuiConfig; use crate::gui::state::{ConnectionStatus, SharedState}; -use crate::suspend_detector::SuspendDetector; use crate::sync_engine::SyncEngine; use crate::timestamp_id; -use parking_lot::Mutex; +use crate::transport::is_heartbeat_timeout; use std::path::PathBuf; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; @@ -86,50 +85,6 @@ fn session_loop( } } - // Tracks whichever `Client` is currently connecting/connected, so the - // suspend-monitor thread below can force it to disconnect. `None` while - // we're idling between sessions (paused, or waiting to reconnect). - let active_client: Arc>>> = Arc::new(Mutex::new(None)); - // Set by the suspend-monitor thread whenever it detects a resume, so the - // reconnect-backoff loop below can skip its wait and retry immediately. - let resume_pending = Arc::new(AtomicBool::new(false)); - - // Runs for the lifetime of this sync session (independent of connect/ - // reconnect cycles) so that suspend/resume is detected promptly even - // while a session is actively connected — not just between attempts. - // On resume it force-closes whatever connection is active, which makes - // the main loop below reconnect and rescan exactly like the startup - // sequence. - { - let state = state.clone(); - let stopped = stopped.clone(); - let active_client = active_client.clone(); - let resume_pending = resume_pending.clone(); - thread::Builder::new() - .name("suspend-monitor".into()) - .spawn(move || { - let mut suspend_detector = SuspendDetector::new(); - loop { - if stopped.load(Ordering::SeqCst) { - break; - } - - if suspend_detector.check_for_resume() { - state - .write() - .log_event("System resumed from suspend — reconnecting."); - resume_pending.store(true, Ordering::SeqCst); - if let Some(client) = active_client.lock().clone() { - client.shutdown(); - } - } - - thread::sleep(Duration::from_millis(1000)); - } - }) - .expect("spawn suspend-monitor"); - } - loop { if stopped.load(Ordering::SeqCst) { break; @@ -140,8 +95,6 @@ fn session_loop( continue; } - resume_pending.store(false, Ordering::SeqCst); - { let mut s = state.write(); s.status = ConnectionStatus::Connecting; @@ -157,8 +110,10 @@ fn session_loop( Some(state.clone()), )); - *active_client.lock() = Some(client.clone()); - + // Heartbeat fast-path: a heartbeat timeout means liveness is already + // proven lost (dead peer or suspend/resume gap) — skip the 5 s wait + // once and reconnect immediately. + let mut skip_wait = false; match client.session() { Ok(()) => { let mut s = state.write(); @@ -167,14 +122,18 @@ fn session_loop( } Err(e) => { let msg = e.to_string(); + if is_heartbeat_timeout(&e) { + state.write().log_event( + "Heartbeat timeout — reconnecting immediately.", + ); + skip_wait = true; + } let mut s = state.write(); s.status = ConnectionStatus::Error(msg.clone()); s.log_event(format!("Connection error: {msg}")); } } - *active_client.lock() = None; - refresh_manifest_stats(&engine, &state); if stopped.load(Ordering::SeqCst) { @@ -184,9 +143,7 @@ fn session_loop( continue; } - if resume_pending.swap(false, Ordering::SeqCst) { - // The suspend-monitor already forced this disconnect; reconnect - // right away instead of waiting out the usual backoff. + if skip_wait { continue; } @@ -196,13 +153,6 @@ fn session_loop( break; } - if resume_pending.swap(false, Ordering::SeqCst) { - state - .write() - .log_event("System resumed from suspend — reconnecting immediately."); - break; - } - thread::sleep(Duration::from_millis(200)); } } diff --git a/crates/filesync/src/lib.rs b/crates/filesync/src/lib.rs index 4622f14..828b28d 100644 --- a/crates/filesync/src/lib.rs +++ b/crates/filesync/src/lib.rs @@ -10,7 +10,6 @@ pub mod manifest; pub mod manifest_cache; pub mod protocol; pub mod server; -pub mod suspend_detector; pub mod sync_engine; pub mod transport; pub mod watcher; diff --git a/crates/filesync/src/protocol.rs b/crates/filesync/src/protocol.rs index 6c17cb9..eee3d1e 100644 --- a/crates/filesync/src/protocol.rs +++ b/crates/filesync/src/protocol.rs @@ -8,7 +8,7 @@ pub const BUNDLE_MAX_FILES: usize = 500; pub const LARGE_FILE_THRESHOLD: u64 = 8 * 1024 * 1024; pub const FILE_CHUNK_SIZE: usize = 8 * 1024 * 1024; pub const MAX_FRAME_BYTES: usize = 32 * 1024 * 1024; -pub const PROTOCOL_VERSION: u32 = 8; +pub const PROTOCOL_VERSION: u32 = 9; pub const DEBOUNCE_MS: u64 = 200; pub const FILE_STABILITY_MS: u64 = 500; pub const FILE_CHANGE_COALESCE_MS: u64 = 100; @@ -129,6 +129,11 @@ pub enum Message { available_bytes: u64, required_bytes: u64, }, + // App-level heartbeat (PROTOCOL_VERSION 9). Appended at END so existing + // bincode discriminants are unchanged. Ping/Pong are swallowed in + // transport::io_loop and never reach app recv loops. + Ping, + Pong, } pub fn serialise_message(msg: &Message) -> io::Result> { diff --git a/crates/filesync/src/server.rs b/crates/filesync/src/server.rs index 388c4f5..85b8e49 100644 --- a/crates/filesync/src/server.rs +++ b/crates/filesync/src/server.rs @@ -6,7 +6,7 @@ use crate::ledger::{DeletionLedger, Tombstone}; use crate::manifest; use crate::protocol::*; use crate::sync_engine::SyncEngine; -use crate::transport::{Connection, Frame}; +use crate::transport::{Connection, Frame, HEARTBEAT_POLL_SECS}; use crate::watcher::{self, FsEvent}; use bytehive_core::MessageBus; @@ -193,6 +193,26 @@ impl Server { } } +/// Blocking recv with heartbeat maintenance: polls with a 2 s timeout so +/// Ping is sent while idle and a dead peer / suspend-resume gap trips the +/// heartbeat timeout (returned as Err) instead of hanging forever. +fn recv_with_heartbeat(conn: &Connection) -> io::Result { + loop { + match conn.recv_timeout(Duration::from_secs(HEARTBEAT_POLL_SECS)) { + Ok(msg) => return Ok(msg), + Err(e) + if e.kind() == io::ErrorKind::TimedOut + || e.kind() == io::ErrorKind::WouldBlock => + { + // Idle window: send Ping if due, trip timeout if peer is + // stale (suspend recovery lands here on next check). + conn.heartbeat_maintenance()?; + } + Err(e) => return Err(e), + } + } +} + fn handle_client( stream: TcpStream, addr: String, @@ -228,7 +248,7 @@ fn handle_client( active_conns, }; - let client_id = match conn.recv()? { + let client_id = match recv_with_heartbeat(&conn)? { Message::Hello { node_id, protocol_version, @@ -343,7 +363,7 @@ fn handle_client( conn.send(&Message::ManifestExchange(local.clone()))?; debug!("filesync: sent ManifestExchange to {client_id}"); - let remote = match conn.recv()? { + let remote = match recv_with_heartbeat(&conn)? { Message::ManifestExchange(m) => m, Message::InsufficientDiskSpace { available_bytes, @@ -381,7 +401,7 @@ fn handle_client( entries: engine.ledger_entries(), })?; debug!("filesync: sent LedgerExchange to {client_id}"); - let peer_ledger_entries = match conn.recv()? { + let peer_ledger_entries = match recv_with_heartbeat(&conn)? { Message::LedgerExchange { entries } => entries, _ => { return Err(io::Error::new( @@ -485,7 +505,7 @@ fn handle_client( let sync_start = Instant::now(); loop { - match conn.recv()? { + match recv_with_heartbeat(&conn)? { Message::Bundle(b) => { let n_files = b.files.iter().filter(|f| !f.metadata.is_dir).count(); let b_bytes: u64 = b.files.iter().map(|f| f.metadata.size).sum(); @@ -731,11 +751,22 @@ fn handle_client( let conn_fwd = conn.clone(); let fwd_handle = thread::Builder::new() .name(format!("srv-bcast-fwd-{client_id}")) - .spawn(move || { - while let Ok(Some(frame)) = bcast_rx.recv() { - if conn_fwd.send_frame(frame).is_err() { - break; + .spawn(move || loop { + match bcast_rx.recv_timeout(Duration::from_secs(HEARTBEAT_POLL_SECS)) { + Ok(Some(frame)) => { + if conn_fwd.send_frame(frame).is_err() { + break; + } + } + Ok(None) => break, + Err(crossbeam_channel::RecvTimeoutError::Timeout) => { + // Idle window: send Ping if due; break so the peer is + // cleaned up when the heartbeat times out. + if conn_fwd.heartbeat_maintenance().is_err() { + break; + } } + Err(crossbeam_channel::RecvTimeoutError::Disconnected) => break, } })?; @@ -778,7 +809,7 @@ fn client_recv_loop( ) { debug!("filesync: recv loop started for {client_id}"); loop { - match conn.recv() { + match recv_with_heartbeat(&conn) { Ok(Message::Bundle(b)) => { let n_files = b.files.iter().filter(|f| !f.metadata.is_dir).count(); let b_bytes: u64 = b.files.iter().map(|f| f.metadata.size).sum(); diff --git a/crates/filesync/src/suspend_detector.rs b/crates/filesync/src/suspend_detector.rs deleted file mode 100644 index a6901db..0000000 --- a/crates/filesync/src/suspend_detector.rs +++ /dev/null @@ -1,69 +0,0 @@ -use std::fs; -use std::time::{Duration, Instant}; - -const SUSPEND_THRESHOLD_SECS: u64 = 5; - -pub struct SuspendDetector { - last_check: Instant, - last_uptime: Duration, -} - -impl SuspendDetector { - pub fn new() -> Self { - let uptime = read_system_uptime().unwrap_or(Duration::ZERO); - Self { - last_check: Instant::now(), - last_uptime: uptime, - } - } - - pub fn check_for_resume(&mut self) -> bool { - let now = Instant::now(); - let wall_elapsed = now.duration_since(self.last_check); - - let current_uptime = match read_system_uptime() { - Ok(uptime) => uptime, - Err(_) => { - self.last_check = now; - return false; - } - }; - - let uptime_elapsed = current_uptime.saturating_sub(self.last_uptime); - let suspended = wall_elapsed.as_secs() > uptime_elapsed.as_secs() + SUSPEND_THRESHOLD_SECS; - - self.last_check = now; - self.last_uptime = current_uptime; - - suspended - } -} - -impl Default for SuspendDetector { - fn default() -> Self { - Self::new() - } -} - -#[cfg(target_os = "linux")] -fn read_system_uptime() -> std::io::Result { - let contents = fs::read_to_string("/proc/uptime")?; - let uptime_str = contents - .split_whitespace() - .next() - .ok_or_else(|| std::io::Error::new(std::io::ErrorKind::InvalidData, "empty uptime"))?; - - let uptime_secs: f64 = uptime_str - .parse() - .map_err(|_| std::io::Error::new(std::io::ErrorKind::InvalidData, "invalid uptime"))?; - - Ok(Duration::from_secs_f64(uptime_secs)) -} - -#[cfg(not(target_os = "linux"))] -fn read_system_uptime() -> std::io::Result { - Err(std::io::Error::new( - std::io::ErrorKind::Unsupported, - "system uptime reading not supported on this platform", - )) -} diff --git a/crates/filesync/src/transport.rs b/crates/filesync/src/transport.rs index ae95df6..0368dff 100644 --- a/crates/filesync/src/transport.rs +++ b/crates/filesync/src/transport.rs @@ -7,12 +7,30 @@ use std::io::{self, Read, Write}; use std::net::TcpStream; use std::sync::Arc; use std::thread; -use std::time::Duration; +use std::time::{Duration, SystemTime}; pub type Frame = Arc>; + +/// True when a session error came from heartbeat liveness (stale peer or +/// suspend/resume gap): TimedOut kind, or a message mentioning "heartbeat". +/// Used for the reconnect fast-path (skip backoff once). +pub fn is_heartbeat_timeout(e: &io::Error) -> bool { + e.kind() == io::ErrorKind::TimedOut || e.to_string().contains("heartbeat") +} const READ_POLL_MS: u64 = 5; const RECV_CHANNEL_DEPTH: usize = 1024; +/// App-level heartbeat: send Ping after this long without any outbound +/// traffic, treat the peer as dead after this long without ANY inbound +/// traffic. 30 s timeout tolerates one missed ping + slow disk; suspend +/// recovery is still fast because the wall-clock gap trips the timeout on +/// the next maintenance check after resume. +pub const HEARTBEAT_INTERVAL_SECS: u64 = 15; +pub const HEARTBEAT_TIMEOUT_SECS: u64 = 30; +/// Blocking recv loops poll with this timeout so heartbeat maintenance runs +/// even while idle (also keeps suspend-resume detection latency ≤ 2 s). +pub const HEARTBEAT_POLL_SECS: u64 = 2; + enum TlsStream { Server(StreamOwned), Client(StreamOwned), @@ -138,6 +156,7 @@ fn io_loop( mut tls: TlsStream, recv_tx: Sender>, send_rx: Receiver>, + last_recv_wall: Arc>, ) { debug!( "tls-io: thread started (send_depth={SEND_QUEUE_DEPTH} recv_depth={RECV_CHANNEL_DEPTH})" @@ -169,9 +188,37 @@ fn io_loop( match reader.poll(&mut tls) { Ok(Some(msg)) => { - if recv_tx.send(Ok(msg)).is_err() { - debug!("tls-io: recv_tx consumer gone, exiting"); - return; + // Wall-clock liveness: ANY inbound frame (data, Ping, Pong) + // proves the peer path is alive. Instant can't be used here — + // it pauses during suspend so it can't detect resume gaps. + *last_recv_wall.lock() = SystemTime::now(); + match msg { + // Auto-reply in the io thread: zero app churn, no recv + // loop match arms need changing. Swallow both so the app + // only ever sees data messages. + Message::Ping => { + match protocol::serialise_message(&Message::Pong) { + Ok(frame) => { + if let Err(e) = + tls.write_all(&frame).and_then(|_| tls.flush()) + { + debug!("tls-io: pong write error (kind={:?}): {e}", e.kind()); + let _ = recv_tx.send(Err(e)); + return; + } + } + Err(e) => { + debug!("tls-io: pong serialise error: {e}"); + } + } + } + Message::Pong => {} + msg => { + if recv_tx.send(Ok(msg)).is_err() { + debug!("tls-io: recv_tx consumer gone, exiting"); + return; + } + } } } Ok(None) => {} @@ -187,6 +234,8 @@ fn io_loop( pub struct Connection { recv_rx: Mutex>>, send_tx: Sender>, + last_send_wall: Mutex, + last_recv_wall: Arc>, pub peer_cert: Option>, } @@ -237,9 +286,12 @@ impl Connection { let (recv_tx, recv_rx) = bounded::>(RECV_CHANNEL_DEPTH); let (send_tx, send_rx) = bounded::>(SEND_QUEUE_DEPTH); + let now = SystemTime::now(); + let last_recv_wall = Arc::new(Mutex::new(now)); + let last_recv_wall_io = last_recv_wall.clone(); thread::Builder::new() .name("tls-io".into()) - .spawn(move || io_loop(tls, recv_tx, send_rx)) + .spawn(move || io_loop(tls, recv_tx, send_rx, last_recv_wall_io)) .expect("spawn tls-io thread"); debug!( "tls: io-loop thread spawned (recv_channel={RECV_CHANNEL_DEPTH} send_channel={SEND_QUEUE_DEPTH})" @@ -248,10 +300,16 @@ impl Connection { Ok(Self { recv_rx: Mutex::new(recv_rx), send_tx, + last_send_wall: Mutex::new(now), + last_recv_wall, peer_cert, }) } + fn record_send(&self) { + *self.last_send_wall.lock() = SystemTime::now(); + } + pub fn send(&self, msg: &Message) -> io::Result<()> { let frame = Arc::new(protocol::serialise_message(msg)?); self.send_frame(frame) @@ -274,14 +332,22 @@ impl Connection { frame.len() ); } - self.send_tx + let r = self + .send_tx .send(Some(frame)) - .map_err(|_| io::Error::new(io::ErrorKind::BrokenPipe, "connection closed")) + .map_err(|_| io::Error::new(io::ErrorKind::BrokenPipe, "connection closed")); + if r.is_ok() { + self.record_send(); + } + r } pub fn try_send_frame(&self, frame: Frame) -> bool { match self.send_tx.try_send(Some(frame)) { - Ok(_) => true, + Ok(_) => { + self.record_send(); + true + } Err(TrySendError::Full(_)) => { warn!( "tls-io: try_send_frame dropped — send queue full ({}/{})", @@ -308,6 +374,65 @@ impl Connection { } } + /// Blocking recv with a timeout. Maps channel disconnect to + /// ConnectionAborted (same as `recv`) and channel timeout to TimedOut + /// (== WouldBlock-family via `is_timeout`-style handling at call sites). + pub fn recv_timeout(&self, t: Duration) -> io::Result { + match self.recv_rx.lock().recv_timeout(t) { + Ok(Ok(msg)) => Ok(msg), + Ok(Err(e)) => Err(e), + Err(crossbeam_channel::RecvTimeoutError::Timeout) => Err(io::Error::new( + io::ErrorKind::TimedOut, + "recv timeout", + )), + Err(crossbeam_channel::RecvTimeoutError::Disconnected) => Err(io::Error::new( + io::ErrorKind::ConnectionAborted, + "connection closed", + )), + } + } + + /// Heartbeat maintenance: send Ping when this side has been idle for + /// >= HEARTBEAT_INTERVAL_SECS, and fail with TimedOut when nothing has been + /// received for >= HEARTBEAT_TIMEOUT_SECS (covers dead peers AND + /// suspend/resume gaps — SystemTime keeps advancing across suspend + /// while Instant pauses, so the gap trips the timeout on next check). + /// Backwards clock jumps re-baseline instead of false-tripping. + /// Only short parking_lot critical sections — never blocks on io_loop + /// (io_loop only takes last_recv_wall briefly too). + pub fn heartbeat_maintenance(&self) -> io::Result<()> { + let now = SystemTime::now(); + let (since_send, since_recv) = { + let last_send = *self.last_send_wall.lock(); + let last_recv = *self.last_recv_wall.lock(); + match (now.duration_since(last_send), now.duration_since(last_recv)) { + (Ok(s), Ok(r)) => (Some(s), Some(r)), + // Clock jumped backwards: re-baseline both, don't trip. + _ => { + *self.last_send_wall.lock() = now; + *self.last_recv_wall.lock() = now; + return Ok(()); + } + } + }; + let since_send = since_send.unwrap(); + let since_recv = since_recv.unwrap(); + if since_send >= Duration::from_secs(HEARTBEAT_INTERVAL_SECS) { + // send() records last_send on success. + self.send(&Message::Ping)?; + } + if since_recv >= Duration::from_secs(HEARTBEAT_TIMEOUT_SECS) { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + format!( + "heartbeat timeout: no data for {} s", + since_recv.as_secs() + ), + )); + } + Ok(()) + } + pub fn shutdown(&self) { let _ = self.send_tx.send(None); } diff --git a/crates/filesync/tests/test_deletion_ledger.rs b/crates/filesync/tests/test_deletion_ledger.rs index 3489c6c..a735b5a 100644 --- a/crates/filesync/tests/test_deletion_ledger.rs +++ b/crates/filesync/tests/test_deletion_ledger.rs @@ -46,7 +46,7 @@ fn sorted_entries(engine: &SyncEngine) -> Vec<(PathBuf, u64, String, Option