diff --git a/crates/filesync/src/client.rs b/crates/filesync/src/client.rs index 31188d5..a7e71dd 100644 --- a/crates/filesync/src/client.rs +++ b/crates/filesync/src/client.rs @@ -1,11 +1,11 @@ use crate::cert_fingerprint; use crate::common::{self, LargeFileEndOutcome, PendingChanges}; use crate::gui::state::ConnectionStatus; -use crate::gui::state::SharedState; +use crate::gui::state::{ConflictKind, SharedState}; use crate::known_hosts::KnownServers; use crate::manifest; use crate::protocol::*; -use crate::sync_engine::SyncEngine; +use crate::sync_engine::{ConflictInfo, SyncEngine}; use crate::timestamp_id; use crate::transport::Connection; use crate::watcher::{self, FsEvent}; @@ -16,11 +16,11 @@ use log::{debug, error, info, warn}; use std::io; use std::net::TcpStream; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::thread; -use std::time::{Duration, Instant}; +use std::time::{Duration, Instant, SystemTime}; pub use crate::common::count_manifest; @@ -499,7 +499,9 @@ impl Client { b.bundle_id, n_files, n_dirs, bundle_bytes, bytes_received + bundle_bytes ); - let n = self.engine.apply_bundle(&b)?.written; + let apply_result = self.engine.apply_bundle(&b)?; + let n = apply_result.written; + push_gui_conflicts(&self.gui_state, &self.engine, &apply_result.conflicts); for fd in &b.files { if fd.metadata.is_dir { dirs_received += 1; @@ -592,6 +594,7 @@ impl Client { (conflict copy: {:?})", ci.conflict_copy_path ); + push_gui_conflicts(&self.gui_state, &self.engine, &[ci]); } crate::sync_engine::FinishResult::MissingChunks(indices) => { warn!( @@ -748,11 +751,12 @@ impl Client { let eng_r = self.engine.clone(); let conn_r = conn.clone(); let bus_r = self.bus.clone(); + let gui_state_r = self.gui_state.clone(); debug!("filesync session: spawning recv-loop thread"); let recv_handle = thread::Builder::new() .name("recv-srv".into()) .spawn(move || { - recv_loop(eng_r, conn_r, bus_r); + recv_loop(eng_r, conn_r, bus_r, gui_state_r); let _ = shut_tx.send(()); })?; @@ -763,6 +767,7 @@ impl Client { fs_rx, shut_rx, self.bus.clone(), + self.gui_state.clone(), ); debug!("filesync session: send-loop returned, shutting down connection"); @@ -773,7 +778,12 @@ impl Client { } } -fn recv_loop(engine: Arc, conn: Arc, bus: Option>) { +fn recv_loop( + engine: Arc, + conn: Arc, + bus: Option>, + gui_state: Option, +) { let prefix = "filesync recv"; debug!("{prefix}: loop started, waiting for incremental messages from server"); loop { @@ -786,8 +796,17 @@ fn recv_loop(engine: Arc, conn: Arc, bus: Option { + if let Some(ref gs) = gui_state { + let mut s = gs.write(); + s.begin_sync_activity(); + s.files_received += applied.files_count as u64; + s.bytes_received += applied.bytes; + } + push_gui_conflicts(&gui_state, &engine, &applied.conflicts); + } + Err(e) => error!("{prefix}: apply_bundle: {e}"), } // Send acknowledgment for this bundle @@ -832,6 +851,9 @@ fn recv_loop(engine: Arc, conn: Arc, bus: Option, conn: Arc, bus: Option, conn: Arc, bus: Option { debug!("{prefix}: LargeFileEnd committed {path:?}"); + if let Some(ref gs) = gui_state { + let mut s = gs.write(); + s.begin_sync_activity(); + s.files_received += 1; + } // Note: For large files, we don't have the metadata here to get the sequence number // The acknowledgment would need to be handled differently for large files } + Ok(LargeFileEndOutcome::CommittedWithConflict(ci)) => { + debug!("{prefix}: LargeFileEnd committed with conflict {path:?}"); + if let Some(ref gs) = gui_state { + let mut s = gs.write(); + s.begin_sync_activity(); + s.files_received += 1; + } + push_gui_conflicts(&gui_state, &engine, &[ci]); + } Err(e) => error!("{prefix}: large_file_end: {e}"), } } @@ -884,6 +925,9 @@ fn recv_loop(engine: Arc, conn: Arc, bus: Option { debug!("{prefix}: Rename {from:?} → {to:?}"); @@ -892,6 +936,9 @@ fn recv_loop(engine: Arc, conn: Arc, bus: Option { let kind = e.kind(); @@ -940,6 +987,7 @@ fn send_loop( fs_rx: Receiver, shutdown: Receiver<()>, bus: Option>, + gui_state: Option, ) { debug!("filesync send: loop started"); let mut pending = PendingChanges::new(); @@ -1000,7 +1048,7 @@ fn send_loop( pending.renames.len() ); flush_count += 1; - if let Err(e) = flush_to_server(&engine, &conn, &mut pending, &bus) { + if let Err(e) = flush_to_server(&engine, &conn, &mut pending, &bus, &gui_state) { error!("filesync send: flush #{flush_count} error: {e}"); debug!( "filesync send: flush error kind={:?}, breaking out of send-loop", @@ -1020,6 +1068,7 @@ fn flush_to_server( conn: &Arc, pending: &mut PendingChanges, bus: &Option>, + gui_state: &Option, ) -> io::Result<()> { let renames = pending.take_renames(); if !renames.is_empty() { @@ -1031,6 +1080,9 @@ fn flush_to_server( from: from.clone(), to: to.clone(), })?; + if let Some(ref gs) = gui_state { + gs.write().begin_sync_activity(); + } if let Some(ref bus) = bus { bus.publish( "filesync", @@ -1048,7 +1100,7 @@ fn flush_to_server( let ready = pending.take_ready(); if !ready.is_empty() { debug!("filesync send: flushing {} ready path(s)", ready.len()); - send_paths_to_server(engine, conn, bus, ready)?; + send_paths_to_server(engine, conn, bus, gui_state, ready)?; } let stable = pending.take_stable_changes(engine); @@ -1057,7 +1109,7 @@ fn flush_to_server( "filesync send: flushing {} stable-change path(s)", stable.len() ); - send_paths_to_server(engine, conn, bus, stable)?; + send_paths_to_server(engine, conn, bus, gui_state, stable)?; } let (paths, delete_count) = pending.take_deletes(engine.root()); @@ -1068,6 +1120,9 @@ fn flush_to_server( delete_count ); conn.send(&Message::Delete { paths })?; + if let Some(ref gs) = gui_state { + gs.write().begin_sync_activity(); + } if let Some(ref bus) = bus { bus.publish( @@ -1085,10 +1140,51 @@ fn flush_to_server( Ok(()) } +fn push_gui_conflicts( + gui_state: &Option, + engine: &SyncEngine, + conflicts: &[ConflictInfo], +) { + if conflicts.is_empty() { + return; + } + let Some(gs) = gui_state else { return }; + + let root = engine.root(); + let mut s = gs.write(); + for ci in conflicts { + let filename = ci + .original_path + .file_name() + .map(|n| n.to_string_lossy().into_owned()) + .unwrap_or_else(|| ci.original_path.to_string_lossy().into_owned()); + let folder_path = root + .join(&ci.original_path) + .parent() + .map(|p| p.to_string_lossy().into_owned()) + .unwrap_or_else(|| root.to_string_lossy().into_owned()); + let remote_modified = format_modified(&root.join(&ci.original_path)); + let local_modified = format_modified(&root.join(&ci.conflict_copy_path)); + + s.push_conflict( + filename, + folder_path, + local_modified, + remote_modified, + ConflictKind::BothModified, + ); + s.log_event(format!( + "Conflict detected: {:?} (local copy saved as {:?})", + ci.original_path, ci.conflict_copy_path + )); + } +} + fn send_paths_to_server( engine: &Arc, conn: &Arc, bus: &Option>, + gui_state: &Option, paths: Vec, ) -> io::Result<()> { let manifest = engine.get_manifest(); @@ -1108,11 +1204,20 @@ fn send_paths_to_server( files_count, bytes_sent ); + if let Some(ref gs) = gui_state { + gs.write().begin_sync_activity(); + } engine.send_paths(&paths, conn)?; debug!( "filesync send: send_paths complete — {} file(s) {} B", files_count, bytes_sent ); + if let Some(ref gs) = gui_state { + let mut s = gs.write(); + s.begin_sync_activity(); + s.files_sent += files_count as u64; + s.bytes_sent += bytes_sent; + } if files_count > 0 { if let Some(ref bus) = bus { @@ -1130,3 +1235,145 @@ fn send_paths_to_server( } Ok(()) } + +fn format_modified(path: &Path) -> String { + match std::fs::metadata(path).and_then(|m| m.modified()) { + Ok(t) => format_unix_secs( + t.duration_since(SystemTime::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0), + ), + Err(_) => "unknown".to_string(), + } +} + +fn format_unix_secs(secs: u64) -> String { + let days = (secs / 86_400) as i64; + let rem = secs % 86_400; + let hour = rem / 3600; + let minute = (rem % 3600) / 60; + + let z = days + 719_468; + let era = if z >= 0 { z } else { z - 146_096 } / 146_097; + let doe = z - era * 146_097; + let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146096) / 365; + let y = yoe + era * 400; + let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); + let mp = (5 * doy + 2) / 153; + let d = doy - (153 * mp + 2) / 5 + 1; + let m = if mp < 10 { mp + 3 } else { mp - 9 }; + let y = if m <= 2 { y + 1 } else { y }; + + format!("{y:04}-{m:02}-{d:02} {hour:02}:{minute:02}") +} + +#[cfg(test)] +mod format_tests { + use super::format_unix_secs; + + #[test] + fn epoch_formats_correctly() { + assert_eq!(format_unix_secs(0), "1970-01-01 00:00"); + } + + #[test] + fn known_date_formats_correctly() { + assert_eq!(format_unix_secs(1_717_236_000), "2024-06-01 10:00"); + } +} + +#[cfg(test)] +mod push_gui_conflicts_tests { + use super::push_gui_conflicts; + use crate::exclusions::{ExclusionConfig, Exclusions}; + use crate::gui::state::new_shared_state; + use crate::protocol::{FileBundle, FileData, FileMetadata}; + use crate::sync_engine::SyncEngine; + use std::fs; + use std::path::PathBuf; + use std::sync::Arc; + + fn tmp_dir(label: &str) -> PathBuf { + let d = std::env::temp_dir().join(format!( + "filesync_push_gui_conflicts_{label}_{:x}", + crate::timestamp_id() + )); + fs::create_dir_all(&d).unwrap(); + d + } + + fn make_engine(root: PathBuf) -> SyncEngine { + let ex = Arc::new(Exclusions::compile(&ExclusionConfig::default())); + SyncEngine::new(root, "test-node".to_string(), ex) + } + + fn file_bundle(rel: &str, content: &[u8]) -> FileBundle { + let hash: [u8; 32] = blake3::hash(content).into(); + FileBundle { + files: vec![FileData { + metadata: FileMetadata { + change_sequence: 0, + rel_path: PathBuf::from(rel), + size: content.len() as u64, + hash, + modified_ms: 1_000, + is_dir: false, + }, + content: content.to_vec(), + }], + bundle_id: 1, + } + } + + /// Reproduces the exact root cause of the "GUI can't show conflicts" bug: + /// a `ConflictInfo` produced by `apply_bundle` must end up visible in + /// `SyncSnapshot.conflicts` once routed through `push_gui_conflicts`. + #[test] + fn conflict_from_apply_bundle_is_visible_in_gui_snapshot() { + let dir = tmp_dir("visible"); + let engine = make_engine(dir.clone()); + + // Establish the ancestor state, then simulate an offline local edit + // (bypassing apply_bundle, exactly like a real filesystem edit while + // disconnected), then apply a genuinely different incoming version. + engine + .apply_bundle(&file_bundle("shared.txt", b"version A")) + .unwrap(); + fs::write(dir.join("shared.txt"), b"version B (local)").unwrap(); + let result = engine + .apply_bundle(&file_bundle("shared.txt", b"version C (remote)")) + .unwrap(); + assert_eq!(result.conflicts.len(), 1, "expected exactly one conflict"); + + let gui_state = new_shared_state(); + push_gui_conflicts(&Some(gui_state.clone()), &engine, &result.conflicts); + + let snap = gui_state.read().clone(); + assert_eq!(snap.conflicts.len(), 1); + assert_eq!(snap.conflicts[0].filename, "shared.txt"); + + fs::remove_dir_all(&dir).ok(); + } + + #[test] + fn no_gui_state_does_not_panic() { + let dir = tmp_dir("no_gui"); + let engine = make_engine(dir.clone()); + engine.apply_bundle(&file_bundle("f.txt", b"A")).unwrap(); + fs::write(dir.join("f.txt"), b"B").unwrap(); + let result = engine.apply_bundle(&file_bundle("f.txt", b"C")).unwrap(); + // Must be a no-op, not a panic, when gui_state is None. + push_gui_conflicts(&None, &engine, &result.conflicts); + fs::remove_dir_all(&dir).ok(); + } + + #[test] + fn empty_conflicts_does_not_touch_gui_state() { + let dir = tmp_dir("empty"); + let engine = make_engine(dir.clone()); + let gui_state = new_shared_state(); + push_gui_conflicts(&Some(gui_state.clone()), &engine, &[]); + assert!(gui_state.read().conflicts.is_empty()); + fs::remove_dir_all(&dir).ok(); + } +} diff --git a/crates/filesync/src/common.rs b/crates/filesync/src/common.rs index e326717..82aa81c 100644 --- a/crates/filesync/src/common.rs +++ b/crates/filesync/src/common.rs @@ -237,6 +237,7 @@ pub enum ChunkOutcome { pub enum LargeFileEndOutcome { Committed, + CommittedWithConflict(ConflictInfo), MissingChunks(Vec), } @@ -408,7 +409,7 @@ pub fn handle_recv_large_file_end( }), ); } - Ok(LargeFileEndOutcome::Committed) + Ok(LargeFileEndOutcome::CommittedWithConflict(ci)) } FinishResult::MissingChunks(indices) => { warn!( diff --git a/crates/filesync/src/gui/app.rs b/crates/filesync/src/gui/app.rs index dc8826e..e91520f 100644 --- a/crates/filesync/src/gui/app.rs +++ b/crates/filesync/src/gui/app.rs @@ -88,6 +88,7 @@ struct DashboardState { log_expanded: bool, file_tree: Vec, active_tab: SideTab, + had_conflicts: bool, last_tree_refresh: Instant, } @@ -159,6 +160,7 @@ impl FileSyncGui { log_expanded: false, file_tree, active_tab: SideTab::Stats, + had_conflicts: false, last_tree_refresh: Instant::now(), }) } @@ -315,6 +317,7 @@ impl FileSyncGui { log_expanded: false, file_tree, active_tab: SideTab::Stats, + had_conflicts: false, last_tree_refresh: Instant::now(), }); } @@ -333,14 +336,15 @@ impl FileSyncGui { } if let Screen::Dashboard(d) = &mut self.screen { + d.state + .write() + .end_sync_activity_if_quiet(std::time::Duration::from_millis(1500)); d.snapshot = d.state.read().clone(); - // Auto-switch to Conflicts tab when conflicts appear. - if !d.snapshot.conflicts.is_empty() && d.active_tab == SideTab::Stats { + let now_has_conflicts = !d.snapshot.conflicts.is_empty(); + if now_has_conflicts && !d.had_conflicts { d.active_tab = SideTab::Conflicts; - } else if d.snapshot.conflicts.is_empty() && d.active_tab == SideTab::Conflicts - { - d.active_tab = SideTab::Stats; } + d.had_conflicts = now_has_conflicts; // Periodically refresh the file tree from disk. if d.last_tree_refresh.elapsed() >= std::time::Duration::from_secs(5) { d.file_tree = refresh_file_tree(&d.file_tree, &d.config.sync_root); @@ -498,7 +502,6 @@ impl FileSyncGui { } } - fn view_setup(s: &SetupState) -> Element<'_, Message> { let step_num = match s.step { SetupStep::Folder => 1u8, @@ -790,7 +793,6 @@ fn view_setup_review(s: &SetupState) -> Element<'_, Message> { setup_card(inner.into()) } - fn view_dashboard(d: &DashboardState) -> Element<'_, Message> { let snap = &d.snapshot; let is_paused = d.manager.is_paused(); @@ -843,7 +845,6 @@ fn main_content(d: &DashboardState) -> Element<'_, Message> { .into() } - fn toggle_expanded(nodes: &mut Vec, id: usize) { for node in nodes.iter_mut() { if node.id == id && node.is_dir { @@ -965,7 +966,6 @@ fn refresh_file_tree(old_tree: &[FileNode], root: &std::path::Path) -> Vec String { s.replace('\\', "\\\\").replace('"', "\\\"") @@ -1004,7 +1004,6 @@ exclude_regex = [] ) } - fn setup_card(content: Element) -> Element { container(content) .width(Length::Fill) diff --git a/crates/filesync/src/gui/components/header.rs b/crates/filesync/src/gui/components/header.rs index 8f29dab..d8f0579 100644 --- a/crates/filesync/src/gui/components/header.rs +++ b/crates/filesync/src/gui/components/header.rs @@ -62,6 +62,7 @@ fn connection_badge(status: &ConnectionStatus) -> Element<'_, Message> { match status { ConnectionStatus::Idle => (theme::GREEN, theme::green_text), ConnectionStatus::InitialSync => (theme::AMBER, theme::amber_text), + ConnectionStatus::Syncing => (theme::AMBER, theme::amber_text), ConnectionStatus::Connecting => (theme::YELLOW, theme::yellow_text), ConnectionStatus::AwaitingApproval => (theme::YELLOW, theme::yellow_text), ConnectionStatus::Paused => (theme::YELLOW, theme::yellow_text), @@ -111,6 +112,11 @@ mod tests { let _ = super::view(&ConnectionStatus::InitialSync); } + #[test] + fn view_syncing_does_not_panic() { + let _ = super::view(&ConnectionStatus::Syncing); + } + #[test] fn view_idle_does_not_panic() { let _ = super::view(&ConnectionStatus::Idle); diff --git a/crates/filesync/src/gui/components/status_panel.rs b/crates/filesync/src/gui/components/status_panel.rs index e2626d4..4fbd78f 100644 --- a/crates/filesync/src/gui/components/status_panel.rs +++ b/crates/filesync/src/gui/components/status_panel.rs @@ -10,19 +10,31 @@ use crate::gui::theme; pub fn view(snap: &SyncSnapshot, is_paused: bool) -> Element<'_, Message> { let status_label = text(snap.status.label()).size(13).style(theme::secondary); - let right_detail: Element = if matches!(snap.status, ConnectionStatus::InitialSync) { + let is_transferring = matches!( + snap.status, + ConnectionStatus::InitialSync | ConnectionStatus::Syncing + ); + + let right_detail: Element = if is_transferring { let transferred = snap.bytes_sent + snap.bytes_received; + let files = snap.files_sent + snap.files_received; if snap.transfer_total > 0 { let pct = (transferred as f32 / snap.transfer_total as f32 * 100.0) as u32; - text(format!("{} ({pct}%)", fmt_bytes(transferred))) - .size(12) - .style(theme::muted) - .into() + text(format!( + "{files} file(s) \u{b7} {} ({pct}%)", + fmt_bytes(transferred) + )) + .size(12) + .style(theme::muted) + .into() } else { - text(format!("{} transferred", fmt_bytes(transferred))) - .size(12) - .style(theme::muted) - .into() + text(format!( + "{files} file(s) \u{b7} {} transferred", + fmt_bytes(transferred) + )) + .size(12) + .style(theme::muted) + .into() } } else { Space::new().width(0).into() @@ -68,7 +80,7 @@ pub fn view(snap: &SyncSnapshot, is_paused: bool) -> Element<'_, Message> { ] .align_y(Alignment::Center); - let progress: Element = if matches!(snap.status, ConnectionStatus::InitialSync) { + let progress: Element = if is_transferring { let transferred = snap.bytes_sent + snap.bytes_received; let fraction = if snap.transfer_total > 0 { (transferred as f32 / snap.transfer_total as f32).clamp(0.0, 1.0) @@ -99,6 +111,7 @@ fn status_indicator_dot(status: &ConnectionStatus) -> Element<'static, Message> let color = match status { ConnectionStatus::Idle => theme::GREEN, ConnectionStatus::InitialSync => theme::AMBER, + ConnectionStatus::Syncing => theme::AMBER, ConnectionStatus::Connecting => theme::AMBER, ConnectionStatus::Paused => theme::YELLOW, ConnectionStatus::AwaitingApproval => theme::YELLOW, @@ -202,7 +215,6 @@ mod tests { use super::fmt_bytes; use crate::gui::state::{ConnectionStatus, SyncSnapshot}; - #[test] fn fmt_bytes_zero() { assert_eq!(fmt_bytes(0), "0 B"); @@ -243,7 +255,6 @@ mod tests { assert_eq!(fmt_bytes(2_147_483_648), "2.00 GiB"); } - #[test] fn view_disconnected_not_paused_does_not_panic() { let _ = super::view(&SyncSnapshot::default(), false); @@ -264,6 +275,16 @@ mod tests { let _ = super::view(&snap, false); } + #[test] + fn view_syncing_no_total_does_not_panic() { + let mut snap = SyncSnapshot::default(); + snap.status = ConnectionStatus::Syncing; + snap.transfer_total = 0; + snap.files_received = 3; + snap.bytes_received = 4096; + let _ = super::view(&snap, false); + } + #[test] fn view_initial_sync_with_total_does_not_panic() { let mut snap = SyncSnapshot::default(); diff --git a/crates/filesync/src/gui/state.rs b/crates/filesync/src/gui/state.rs index 349f940..406856f 100644 --- a/crates/filesync/src/gui/state.rs +++ b/crates/filesync/src/gui/state.rs @@ -7,6 +7,10 @@ pub enum ConnectionStatus { Disconnected, Connecting, InitialSync, + /// Actively exchanging an incremental (post-initial-sync) batch of + /// changes with the peer — files being sent/received in the background + /// while the connection is otherwise idle. + Syncing, Idle, Paused, /// The server has received the connection but an administrator has not yet @@ -22,6 +26,7 @@ impl ConnectionStatus { Self::Disconnected => "Disconnected", Self::Connecting => "Connecting…", Self::InitialSync => "Initial sync…", + Self::Syncing => "Syncing…", Self::Idle => "Connected", Self::Paused => "Paused", Self::AwaitingApproval => "Awaiting approval…", @@ -33,6 +38,7 @@ impl ConnectionStatus { match self { Self::Idle => [34, 197, 94, 255], Self::InitialSync => [59, 130, 246, 255], + Self::Syncing => [14, 165, 233, 255], Self::Connecting => [245, 158, 11, 255], Self::Paused => [168, 85, 247, 255], Self::AwaitingApproval => [251, 191, 36, 255], @@ -191,6 +197,11 @@ pub struct SyncSnapshot { pub transfer_total: u64, + /// Timestamp of the most recent incremental (live) sync activity — + /// used to detect when a live-sync batch has gone quiet so the UI can + /// drop back from `Syncing` to `Idle`. + pub last_activity: Option, + pub conflicts: Vec, pub last_connected: Option, @@ -209,6 +220,7 @@ impl Default for SyncSnapshot { files_received: 0, bytes_received: 0, transfer_total: 0, + last_activity: None, conflicts: Vec::new(), last_connected: None, log: EventLog::new(), @@ -220,6 +232,65 @@ impl SyncSnapshot { pub fn log_event(&mut self, msg: impl Into) { self.log.push(msg); } + + /// Marks the start (or continuation) of a live incremental-sync batch. + /// Transitions the status to `Syncing` and, if this is the start of a + /// new batch (status was previously `Idle`), resets the transfer + /// counters so the UI reflects only this batch's progress. + pub fn begin_sync_activity(&mut self) { + if self.status == ConnectionStatus::Idle { + self.files_sent = 0; + self.bytes_sent = 0; + self.files_received = 0; + self.bytes_received = 0; + self.transfer_total = 0; + self.status = ConnectionStatus::Syncing; + } + self.last_activity = Some(Instant::now()); + } + + /// Called periodically (e.g. on a UI tick) to drop back from `Syncing` + /// to `Idle` once no live-sync activity has been observed for + /// `idle_after`. + pub fn end_sync_activity_if_quiet(&mut self, idle_after: std::time::Duration) { + if self.status == ConnectionStatus::Syncing { + let quiet = self + .last_activity + .map(|t| t.elapsed() >= idle_after) + .unwrap_or(true); + if quiet { + self.status = ConnectionStatus::Idle; + self.last_connected = Some(Instant::now()); + } + } + } + + /// Adds a new conflict entry with a freshly-allocated, unique id. + pub fn push_conflict( + &mut self, + filename: String, + folder_path: String, + local_modified: String, + remote_modified: String, + kind: ConflictKind, + ) { + self.conflicts.push(Conflict { + id: next_conflict_id(), + filename, + folder_path, + local_modified, + remote_modified, + kind, + }); + } +} + +static NEXT_CONFLICT_ID: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(1); + +/// Allocates a fresh, process-unique conflict id (used so conflicts raised +/// from different code paths never collide). +pub fn next_conflict_id() -> usize { + NEXT_CONFLICT_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed) } pub type SharedState = Arc>; @@ -288,6 +359,7 @@ mod tests { assert_eq!(ConnectionStatus::Disconnected.label(), "Disconnected"); assert_eq!(ConnectionStatus::Connecting.label(), "Connecting…"); assert_eq!(ConnectionStatus::InitialSync.label(), "Initial sync…"); + assert_eq!(ConnectionStatus::Syncing.label(), "Syncing…"); assert_eq!(ConnectionStatus::Idle.label(), "Connected"); assert_eq!(ConnectionStatus::Paused.label(), "Paused"); assert_eq!( @@ -305,6 +377,7 @@ mod tests { let statuses = [ ConnectionStatus::Idle, ConnectionStatus::InitialSync, + ConnectionStatus::Syncing, ConnectionStatus::Connecting, ConnectionStatus::Paused, ConnectionStatus::AwaitingApproval, @@ -327,6 +400,7 @@ mod tests { let statuses = [ ConnectionStatus::Idle, ConnectionStatus::InitialSync, + ConnectionStatus::Syncing, ConnectionStatus::Connecting, ConnectionStatus::Paused, ConnectionStatus::AwaitingApproval, @@ -385,6 +459,7 @@ mod tests { assert_eq!(snap.bytes_received, 0); assert_eq!(snap.transfer_total, 0); assert!(snap.last_connected.is_none()); + assert!(snap.last_activity.is_none()); assert!(snap.log.entries().is_empty()); } @@ -397,6 +472,68 @@ mod tests { assert_eq!(snap.log.entries()[0], "connected"); } + #[test] + fn begin_sync_activity_transitions_from_idle_and_resets_counters() { + let mut snap = SyncSnapshot::default(); + snap.status = ConnectionStatus::Idle; + snap.bytes_sent = 500; + snap.files_received = 3; + snap.begin_sync_activity(); + assert_eq!(snap.status, ConnectionStatus::Syncing); + assert_eq!(snap.bytes_sent, 0); + assert_eq!(snap.files_received, 0); + assert!(snap.last_activity.is_some()); + } + + #[test] + fn begin_sync_activity_does_not_reset_counters_mid_batch() { + let mut snap = SyncSnapshot::default(); + snap.status = ConnectionStatus::Syncing; + snap.bytes_received = 42; + snap.begin_sync_activity(); + assert_eq!(snap.status, ConnectionStatus::Syncing); + assert_eq!(snap.bytes_received, 42); + } + + #[test] + fn end_sync_activity_if_quiet_returns_to_idle_after_timeout() { + let mut snap = SyncSnapshot::default(); + snap.status = ConnectionStatus::Syncing; + snap.last_activity = Some(Instant::now() - std::time::Duration::from_secs(10)); + snap.end_sync_activity_if_quiet(std::time::Duration::from_millis(100)); + assert_eq!(snap.status, ConnectionStatus::Idle); + } + + #[test] + fn end_sync_activity_if_quiet_stays_syncing_when_recent() { + let mut snap = SyncSnapshot::default(); + snap.status = ConnectionStatus::Syncing; + snap.last_activity = Some(Instant::now()); + snap.end_sync_activity_if_quiet(std::time::Duration::from_secs(5)); + assert_eq!(snap.status, ConnectionStatus::Syncing); + } + + #[test] + fn push_conflict_assigns_unique_ids() { + let mut snap = SyncSnapshot::default(); + snap.push_conflict( + "a.txt".into(), + "/sync".into(), + "t1".into(), + "t2".into(), + ConflictKind::BothModified, + ); + snap.push_conflict( + "b.txt".into(), + "/sync".into(), + "t1".into(), + "t2".into(), + ConflictKind::BothModified, + ); + assert_eq!(snap.conflicts.len(), 2); + assert_ne!(snap.conflicts[0].id, snap.conflicts[1].id); + } + #[test] fn new_shared_state_starts_disconnected() { let state = new_shared_state(); diff --git a/crates/filesync/src/server.rs b/crates/filesync/src/server.rs index 6e1cf07..c9269d7 100644 --- a/crates/filesync/src/server.rs +++ b/crates/filesync/src/server.rs @@ -794,7 +794,8 @@ fn client_recv_loop( match common::handle_recv_large_file_end( &engine, path, final_hash, &client_id, &bus, "filesync", ) { - Ok(LargeFileEndOutcome::Committed) => { + Ok(LargeFileEndOutcome::Committed) + | Ok(LargeFileEndOutcome::CommittedWithConflict(_)) => { broadcast_to_others( &peers, &client_id, diff --git a/crates/filesync/tests/test_client_gui_state.rs b/crates/filesync/tests/test_client_gui_state.rs new file mode 100644 index 0000000..9cd334c --- /dev/null +++ b/crates/filesync/tests/test_client_gui_state.rs @@ -0,0 +1,199 @@ +#![cfg(target_os = "linux")] + +//! Integration coverage for `Client::new_standalone(..., Some(gui_state))`. +//! +//! Every other integration test in this crate exercises the client with +//! `gui_state: None` (see `integration_tests.rs`), leaving the GUI-attached +//! code path — used exclusively by the real `filesync-gui` binary — entirely +//! untested. This test fills that gap by driving a real client/server pair +//! over TCP+TLS with a live `SharedState` attached, and asserts that initial +//! sync actually completes (i.e. does not hang) and leaves the GUI snapshot +//! in a sane, non-stuck state with correct file/byte counters. +//! +//! Conflict-propagation into `SyncSnapshot.conflicts` (the "GUI can't show +//! conflicts" bug) is covered separately by whitebox unit tests in +//! `src/client.rs` (`push_gui_conflicts_tests`), since reliably reproducing a +//! two-sided conflict over a real network race is inherently flaky; the unit +//! tests instead directly exercise the exact function that routes +//! `ConflictInfo` into the GUI state. + +use std::fs; +use std::net::{TcpListener, TcpStream}; +use std::path::PathBuf; +use std::sync::Arc; +use std::thread; +use std::time::{Duration, Instant}; + +use bytehive_core::MessageBus; + +use bytehive_filesync::app::build_server_tls_config; +use bytehive_filesync::client::Client; +use bytehive_filesync::exclusions::{ExclusionConfig, Exclusions}; +use bytehive_filesync::gui::state::{new_shared_state, ConnectionStatus}; +use bytehive_filesync::known_hosts::KnownClients; +use bytehive_filesync::server::Server; +use bytehive_filesync::sync_engine::SyncEngine; +use bytehive_filesync::timestamp_id; + +fn tmp_dir(label: &str) -> PathBuf { + let d = std::env::temp_dir().join(format!("filesync_gui_{label}_{:x}", timestamp_id())); + fs::create_dir_all(&d).unwrap(); + d +} + +fn no_exclusions() -> Arc { + Arc::new(Exclusions::compile(&ExclusionConfig::default())) +} + +fn make_engine(root: PathBuf) -> Arc { + let id = format!("test-{:x}", timestamp_id()); + Arc::new(SyncEngine::new(root, id, no_exclusions())) +} + +struct TestServer { + server: Arc, + port: u16, +} + +impl TestServer { + fn new(dir: PathBuf) -> Self { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + listener.set_nonblocking(true).unwrap(); + let port = listener.local_addr().unwrap().port(); + + let bind = format!("127.0.0.1:{port}"); + let engine = make_engine(dir); + let bus = MessageBus::new(); + let tls_dir = tmp_dir("server_tls"); + let tls = build_server_tls_config(&tls_dir).expect("server TLS config"); + let known_clients = Arc::new(parking_lot::Mutex::new( + KnownClients::load_from_config_permissive(tls_dir.join("config.toml")), + )); + + let server = Arc::new(Server::new(engine, bind, bus, known_clients, tls)); + + let srv = server.clone(); + thread::Builder::new() + .name(format!("gui-test-server:{port}")) + .spawn(move || { + let _ = srv.run_with_listener(listener); + }) + .unwrap(); + + let deadline = Instant::now() + Duration::from_secs(5); + loop { + let addr: std::net::SocketAddr = format!("127.0.0.1:{port}").parse().unwrap(); + if TcpStream::connect_timeout(&addr, Duration::from_millis(50)).is_ok() { + break; + } + if Instant::now() >= deadline { + panic!("server on port {port} did not become ready within 5 s"); + } + thread::sleep(Duration::from_millis(10)); + } + thread::sleep(Duration::from_millis(200)); + + TestServer { server, port } + } + + fn addr(&self) -> String { + format!("127.0.0.1:{}", self.port) + } +} + +impl Drop for TestServer { + fn drop(&mut self) { + self.server.shutdown(); + thread::sleep(Duration::from_millis(100)); + } +} + +/// Runs `client.session()` on a background thread, waits (with timeout) for +/// `condition` to become true, then shuts the server down to unblock the +/// client's live-sync loop and joins the thread. +fn run_session( + client: Client, + server: &Arc, + timeout: Duration, + condition: impl Fn() -> bool, +) -> bool { + let handle = thread::Builder::new() + .name("gui-test-session".into()) + .spawn(move || client.session()) + .unwrap(); + + let deadline = Instant::now() + timeout; + let mut satisfied = false; + while Instant::now() < deadline { + if condition() { + satisfied = true; + break; + } + thread::sleep(Duration::from_millis(50)); + } + if !satisfied { + satisfied = condition(); + } + + server.shutdown(); + let _ = handle.join(); + satisfied +} + +#[test] +fn gui_state_reaches_idle_after_initial_sync_with_no_hang() { + let srv_dir = tmp_dir("idle_srv"); + let cli_dir = tmp_dir("idle_cli"); + + fs::write(srv_dir.join("hello.txt"), b"hello from server").unwrap(); + + let srv = TestServer::new(srv_dir.clone()); + let gui_state = new_shared_state(); + + let identity_dir = cli_dir.join(".identity"); + let cli_engine = make_engine(cli_dir.clone()); + let client = Client::new_standalone( + cli_engine, + srv.addr(), + identity_dir, + Some(gui_state.clone()), + ); + + let cli_dir_check = cli_dir.clone(); + let gui_state_check = gui_state.clone(); + let done = run_session(client, &srv.server, Duration::from_secs(10), move || { + cli_dir_check.join("hello.txt").exists() + && !matches!( + gui_state_check.read().status, + ConnectionStatus::InitialSync | ConnectionStatus::Connecting + ) + }); + + assert!( + done, + "initial sync with gui_state attached must complete (this is the reported freeze scenario)" + ); + + let snap = gui_state.read().clone(); + assert_eq!( + fs::read(cli_dir.join("hello.txt")).unwrap(), + b"hello from server" + ); + assert!( + matches!( + snap.status, + ConnectionStatus::Idle | ConnectionStatus::Syncing + ), + "expected Idle/Syncing after a successful initial sync, got {:?}", + snap.status + ); + assert_eq!(snap.files_received, 1, "expected exactly one file received"); + assert!(snap.bytes_received > 0, "expected non-zero bytes_received"); + assert!( + snap.conflicts.is_empty(), + "no conflicts should have been recorded for a plain sync" + ); + + fs::remove_dir_all(&srv_dir).ok(); + fs::remove_dir_all(&cli_dir).ok(); +}