diff --git a/crates/filesync/src/client.rs b/crates/filesync/src/client.rs index a7e71dd..5c61913 100644 --- a/crates/filesync/src/client.rs +++ b/crates/filesync/src/client.rs @@ -13,6 +13,7 @@ use crate::watcher::{self, FsEvent}; use bytehive_core::MessageBus; use crossbeam_channel::{bounded, Receiver, RecvTimeoutError, TryRecvError}; use log::{debug, error, info, warn}; +use parking_lot::Mutex; use std::io; use std::net::TcpStream; @@ -33,6 +34,23 @@ pub struct Client { tls_config: Arc, gui_state: Option, awaiting_approval: Arc, + /// The connection currently in use by an in-flight `session()` call, if + /// any. This lets `shutdown()` force-close a live connection (e.g. after + /// the system resumes from suspend) instead of only preventing *future* + /// connection attempts. + active_conn: Arc>>>, +} + +/// RAII guard that clears the client's `active_conn` slot when a `session()` +/// call finishes, regardless of which return path is taken. +struct ActiveConnGuard<'a> { + slot: &'a Mutex>>, +} + +impl<'a> Drop for ActiveConnGuard<'a> { + fn drop(&mut self) { + self.slot.lock().take(); + } } impl Client { @@ -57,6 +75,7 @@ impl Client { tls_config, gui_state: None, awaiting_approval: Arc::new(AtomicBool::new(false)), + active_conn: Arc::new(Mutex::new(None)), } } @@ -82,6 +101,7 @@ impl Client { tls_config, gui_state: None, awaiting_approval: Arc::new(AtomicBool::new(false)), + active_conn: Arc::new(Mutex::new(None)), } } @@ -111,6 +131,7 @@ impl Client { tls_config, gui_state, awaiting_approval: Arc::new(AtomicBool::new(false)), + active_conn: Arc::new(Mutex::new(None)), } } @@ -121,6 +142,14 @@ impl Client { pub fn shutdown(&self) { debug!("filesync client: shutdown requested"); self.stopped.store(true, Ordering::SeqCst); + // Force-close any connection that a blocked `session()` call is + // currently using. This unblocks the recv/send loops the same way a + // normal network drop would, so the caller's reconnect logic kicks in + // immediately instead of waiting on a possibly-stale socket. + if let Some(conn) = self.active_conn.lock().clone() { + debug!("filesync client: forcing active connection closed"); + conn.shutdown(); + } } pub fn is_awaiting_approval(&self) -> bool { @@ -193,6 +222,11 @@ impl Client { } pub fn session(&self) -> io::Result<()> { + if self.stopped.load(Ordering::SeqCst) { + debug!("filesync session: shutdown already requested, not connecting"); + return Ok(()); + } + self.engine.clear_in_progress(); debug!("filesync session: cleared any in-progress large-file state"); @@ -231,6 +265,14 @@ impl Client { ); debug!("filesync session: TLS 1.3 handshake complete"); + // Make this connection reachable from `shutdown()` for the rest of the + // session so a forced shutdown (e.g. suspend/resume) can close it even + // while we're blocked in the receive/send loops below. + *self.active_conn.lock() = Some(conn.clone()); + let _active_conn_guard = ActiveConnGuard { + slot: &self.active_conn, + }; + { let known_servers_path = self.identity_dir.join("known_servers.toml"); let mut ks = KnownServers::load_or_create(&known_servers_path); diff --git a/crates/filesync/src/gui/manager.rs b/crates/filesync/src/gui/manager.rs index f88b6b5..ee10cd3 100644 --- a/crates/filesync/src/gui/manager.rs +++ b/crates/filesync/src/gui/manager.rs @@ -5,6 +5,7 @@ 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 std::path::PathBuf; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; @@ -85,7 +86,49 @@ fn session_loop( } } - let mut suspend_detector = SuspendDetector::new(); + // 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) { @@ -94,10 +137,11 @@ fn session_loop( if paused.load(Ordering::SeqCst) { thread::sleep(Duration::from_millis(250)); - let _ = suspend_detector.check_for_resume(); continue; } + resume_pending.store(false, Ordering::SeqCst); + { let mut s = state.write(); s.status = ConnectionStatus::Connecting; @@ -106,12 +150,14 @@ fn session_loop( let identity_dir: PathBuf = GuiConfig::config_dir().join("filesync"); - let client = Client::new_standalone( + let client = Arc::new(Client::new_standalone( engine.clone(), cfg.server_addr.clone(), identity_dir, Some(state.clone()), - ); + )); + + *active_client.lock() = Some(client.clone()); match client.session() { Ok(()) => { @@ -127,6 +173,8 @@ fn session_loop( } } + *active_client.lock() = None; + refresh_manifest_stats(&engine, &state); if stopped.load(Ordering::SeqCst) { @@ -136,20 +184,26 @@ 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. + continue; + } + let deadline = Instant::now() + Duration::from_secs(5); while Instant::now() < deadline { if stopped.load(Ordering::SeqCst) || paused.load(Ordering::SeqCst) { break; } - if suspend_detector.check_for_resume() { + if resume_pending.swap(false, Ordering::SeqCst) { state .write() .log_event("System resumed from suspend — reconnecting immediately."); break; } - thread::sleep(Duration::from_millis(1000)); + thread::sleep(Duration::from_millis(200)); } }