Force reconnect on suspend/resume during active sync session

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<rustls::ClientConfig>,
     gui_state: Option<SharedState>,
     awaiting_approval: Arc<AtomicBool>,
+    /// 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<Mutex<Option<Arc<Connection>>>>,
+}
+
+/// 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<Option<Arc<Connection>>>,
+}
+
+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);

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<Mutex<Option<Arc<Client>>>> = 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));
         }
     }