Skip to content

fix(filesync): replace suspend detector with app-level heartbeat #23

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
guydols merged 1 commit into from
Sep 21, 2026
Merged

fix(filesync): replace suspend detector with app-level heartbeat #23

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 56 additions & 6 deletions crates/filesync/src/client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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();
Expand All @@ -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");
Expand All @@ -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());
Expand All @@ -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",
Expand Down Expand Up @@ -320,7 +338,7 @@ impl Client {
credential: None,
})?;

match conn.recv()? {
match recv_with_heartbeat(&conn)? {
Message::Hello {
node_id,
protocol_version,
Expand Down Expand Up @@ -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");
Expand Down Expand Up @@ -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");
Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -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<Message> {
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<SyncEngine>,
conn: Arc<Connection>,
Expand All @@ -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();
Expand Down Expand Up @@ -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)) {
Expand Down
74 changes: 12 additions & 62 deletions crates/filesync/src/gui/manager.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<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) {
break;
Expand All @@ -140,8 +95,6 @@ fn session_loop(
continue;
}

resume_pending.store(false, Ordering::SeqCst);

{
let mut s = state.write();
s.status = ConnectionStatus::Connecting;
Expand All @@ -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();
Expand All @@ -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) {
Expand All @@ -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;
}

Expand All @@ -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));
}
}
Expand Down
1 change: 0 additions & 1 deletion crates/filesync/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
7 changes: 6 additions & 1 deletion crates/filesync/src/protocol.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<Vec<u8>> {
Expand Down
Loading
Loading