diff --git a/aw-sync/Cargo.toml b/aw-sync/Cargo.toml index 900c304c..9035940b 100644 --- a/aw-sync/Cargo.toml +++ b/aw-sync/Cargo.toml @@ -17,7 +17,7 @@ path = "src/main.rs" log = "0.4" toml = "0.8" chrono = { version = "0.4", features = ["serde"] } -serde = "1.0" +serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" reqwest = { version = "0.12", features = ["json", "blocking"] } dirs = "6" diff --git a/aw-sync/README.md b/aw-sync/README.md index 69cebfc5..6819b3de 100644 --- a/aw-sync/README.md +++ b/aw-sync/README.md @@ -32,6 +32,7 @@ aw-sync daemon --mode pull aw-sync sync --start-date "2024-01-01" # Doctor: why is pull empty / which peers exist in the folder? +# Also prints the last persisted pass (peers, events, skips). aw-sync status ``` diff --git a/aw-sync/src/android.rs b/aw-sync/src/android.rs index c0b27d53..cb25b894 100644 --- a/aw-sync/src/android.rs +++ b/aw-sync/src/android.rs @@ -274,12 +274,9 @@ pub extern "C" fn Java_net_activitywatch_android_SyncInterface_syncPullAll( jni_guard(&mut env, "syncPullAll", |env| { let result: Result = (|| { let client = get_client(port)?; - pull_all(&client).map_err(|e| format!("Sync pull failed: {}", e))?; - Ok(json!({ - "success": true, - "message": "Successfully pulled from all hosts" - }) - .to_string()) + let report = pull_all(&client).map_err(|e| format!("Sync pull failed: {}", e))?; + crate::report::persist_last_report_warn(&report); + Ok(report.to_jni_json()) })(); sync_result_to_jstring(env, "syncPullAll", result) }) @@ -301,13 +298,10 @@ pub extern "C" fn Java_net_activitywatch_android_SyncInterface_syncPull( .map_err(|e| format!("Failed to get hostname string: {}", e))? .into(); - pull(&hostname_str, &client).map_err(|e| format!("Sync pull failed: {}", e))?; - - Ok(json!({ - "success": true, - "message": format!("Successfully pulled from host: {}", hostname_str) - }) - .to_string()) + let report = + pull(&hostname_str, &client).map_err(|e| format!("Sync pull failed: {}", e))?; + crate::report::persist_last_report_warn(&report); + Ok(report.to_jni_json()) })(); sync_result_to_jstring(env, "syncPull", result) }) @@ -328,13 +322,10 @@ pub extern "C" fn Java_net_activitywatch_android_SyncInterface_syncPush( .map_err(|e| format!("Failed to get hostname: {}", e))? .into(); let client = get_client(port)?; - push_with_hostname(&client, &hostname_str) + let report = push_with_hostname(&client, &hostname_str) .map_err(|e| format!("Sync push failed: {}", e))?; - Ok(json!({ - "success": true, - "message": "Successfully pushed local data" - }) - .to_string()) + crate::report::persist_last_report_warn(&report); + Ok(report.to_jni_json()) })(); sync_result_to_jstring(env, "syncPush", result) }) @@ -356,16 +347,22 @@ pub extern "C" fn Java_net_activitywatch_android_SyncInterface_syncBoth( .into(); let client = get_client(port)?; - pull_all(&client).map_err(|e| format!("Pull phase failed: {}", e))?; - - push_with_hostname(&client, &hostname_str) - .map_err(|e| format!("Push phase failed: {}", e))?; + let mut report = pull_all(&client).map_err(|e| format!("Pull phase failed: {}", e))?; - Ok(json!({ - "success": true, - "message": "Successfully completed full sync" - }) - .to_string()) + // Persist the pull result even if push fails, so the report on + // disk reflects the work the pass actually did. + match push_with_hostname(&client, &hostname_str) { + Ok(push_report) => report.merge(push_report), + Err(e) => { + report.record_push_failure(&e); + report.finish(); + crate::report::persist_last_report_warn(&report); + return Err(format!("Push phase failed: {}", e)); + } + } + report.finish(); + crate::report::persist_last_report_warn(&report); + Ok(report.to_jni_json()) })(); sync_result_to_jstring(env, "syncBoth", result) }) diff --git a/aw-sync/src/lib.rs b/aw-sync/src/lib.rs index 7cd0a570..d165dceb 100644 --- a/aw-sync/src/lib.rs +++ b/aw-sync/src/lib.rs @@ -4,6 +4,12 @@ extern crate chrono; extern crate serde; extern crate serde_json; +mod report; +pub use report::{ + last_report_path, load_last_report, persist_last_report, persist_last_report_warn, + BucketReport, PeerOutcome, PeerReport, SyncMode, SyncReport, +}; + mod sync; pub use sync::create_datastore; pub use sync::sync_datastores; diff --git a/aw-sync/src/main.rs b/aw-sync/src/main.rs index c8433d1b..dae3608c 100644 --- a/aw-sync/src/main.rs +++ b/aw-sync/src/main.rs @@ -26,6 +26,7 @@ use aw_client_rust::blocking::AwClient; mod accessmethod; mod dirs; +mod report; mod status; mod sync; mod sync_wrapper; @@ -137,6 +138,7 @@ enum Commands { /// /// 3-level peers come from the same `RemoteDb` walker `pull_all` uses; /// 2-level leftovers and unrecognised entries sit on top of that list. + /// Also prints the last persisted `SyncReport` (what the previous pass did). /// Does not create staging files. Status {}, } @@ -268,26 +270,57 @@ fn main() -> Result<(), Box> { start: start_date, }; - sync::sync_run(&client, &sync_spec, mode.unwrap_or(sync::SyncMode::Both))? + let report = + sync::sync_run(&client, &sync_spec, mode.unwrap_or(sync::SyncMode::Both))?; + info!("{}", report.summary_message()); + aw_sync_persist(&report); } else { // Simple host-based sync mode (backwards compatibility) + let mut report = sync::SyncReport::new(sync::SyncMode::Both); // Pull match host { Some(hosts) => { for host in hosts.iter() { info!("Pulling from host: {}", host); - sync_wrapper::pull(host, &client)?; + // A later host's `?` must not drop earlier hosts + // from last-sync-report.json. Same contract as the + // push-failure path below: persist the aggregate, + // then propagate. + match sync_wrapper::pull(host, &client) { + Ok(one) => report.merge(one), + Err(e) => { + report.record_pull_failure(host, &e); + report.finish(); + info!("{}", report.summary_message()); + aw_sync_persist(&report); + return Err(e); + } + } } } None => { info!("Pulling from all hosts"); - sync_wrapper::pull_all(&client)?; + report.merge(sync_wrapper::pull_all(&client)?); } } - // Push + // Push. On failure the pull phase already did real work: + // persist what was pulled before propagating, so + // `aw-sync status` shows the pull, not just the push. info!("Pushing local data"); - sync_wrapper::push(&client)? + match sync_wrapper::push(&client) { + Ok(push_report) => report.merge(push_report), + Err(e) => { + report.record_push_failure(&e); + report.finish(); + info!("{}", report.summary_message()); + aw_sync_persist(&report); + return Err(e); + } + } + report.finish(); + info!("{}", report.summary_message()); + aw_sync_persist(&report); } } @@ -337,9 +370,15 @@ fn daemon( }; loop { - if let Err(e) = sync::sync_run(client, &sync_spec, mode) { - error!("Error during sync cycle: {}", e); - return Err(e); + match sync::sync_run(client, &sync_spec, mode) { + Ok(report) => { + info!("{}", report.summary_message()); + aw_sync_persist(&report); + } + Err(e) => { + error!("Error during sync cycle: {}", e); + return Err(e); + } } info!("Sync pass done, sleeping for 5 minutes"); @@ -357,3 +396,7 @@ fn daemon( Ok(()) } + +fn aw_sync_persist(report: &sync::SyncReport) { + crate::report::persist_last_report_warn(report); +} diff --git a/aw-sync/src/report.rs b/aw-sync/src/report.rs new file mode 100644 index 00000000..d651c540 --- /dev/null +++ b/aw-sync/src/report.rs @@ -0,0 +1,695 @@ +//! What a sync pass did. +//! +//! A pass used to return `()` on success, so JNI, `aw-sync status`, and the +//! daemon all had to invent a story (a fixed string, a boolean, or silence). +//! `SyncReport` is that story, persisted locally — never in the Syncthing +//! folder, which would leak one device's last pass onto every peer. + +use std::error::Error; +use std::fmt; +use std::fs; +use std::path::{Path, PathBuf}; + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; + +#[cfg(feature = "cli")] +use clap::ValueEnum; + +#[derive(PartialEq, Eq, Copy, Clone, Debug, Serialize, Deserialize)] +#[cfg_attr(feature = "cli", derive(ValueEnum))] +#[serde(rename_all = "lowercase")] +pub enum SyncMode { + Push, + Pull, + Both, +} + +impl SyncMode { + pub fn as_str(self) -> &'static str { + match self { + SyncMode::Push => "push", + SyncMode::Pull => "pull", + SyncMode::Both => "both", + } + } +} + +/// Facts about one completed (or failed-and-aborted) sync pass. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct SyncReport { + pub started: DateTime, + pub finished: DateTime, + pub mode: SyncMode, + /// One entry per peer considered on pull, including skipped and failed. + pub peers: Vec, + /// Buckets written to this device's staging db on push. + pub pushed: Vec, + /// Discovery / layout diagnostics from this pass. Empty-peer pulls must + /// still leave a record (ActivityWatch/aw-server-rust#682 / #695). + #[serde(default)] + pub warnings: Vec, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct PeerReport { + pub device_id: String, + pub hostname: String, + pub path: PathBuf, + pub outcome: PeerOutcome, + pub buckets: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "lowercase")] +pub enum PeerOutcome { + Imported, + Skipped { reason: String }, + Failed { error: String }, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct BucketReport { + pub bucket_id: String, + pub events_new: i64, + pub resumed_at: Option>, +} + +impl SyncReport { + pub fn new(mode: SyncMode) -> Self { + let now = Utc::now(); + Self { + started: now, + finished: now, + mode, + peers: vec![], + pushed: vec![], + warnings: vec![], + } + } + + pub fn finish(&mut self) { + self.finished = Utc::now(); + } + + /// Fold another pass into this one (pull then push, or per-peer pull_all). + pub fn merge(&mut self, other: SyncReport) { + self.started = self.started.min(other.started); + self.finished = self.finished.max(other.finished); + if self.mode != other.mode { + self.mode = SyncMode::Both; + } + self.peers.extend(other.peers); + self.pushed.extend(other.pushed); + self.warnings.extend(other.warnings); + } + + /// Log and keep discovery warnings on the report so a later `status` can + /// show why a pass imported nobody. + pub fn capture_warnings(&mut self, warnings: impl IntoIterator) { + for line in warnings { + warn!("{line}"); + self.warnings.push(line); + } + } + + /// Record a push abort so the persisted report is not a silent success. + /// + /// The caller still returns the `Err`. Without this, `aw-sync status` shows + /// an empty pass (no peers, no pushed buckets, no warnings) and looks like + /// a no-op rather than a failure. + pub fn record_push_failure(&mut self, err: impl fmt::Display) { + self.warnings.push(format!("push failed: {err}")); + } + + /// Record a per-host pull abort on an aggregate report. + /// + /// Simple host-based sync (`aw-sync sync --host a --host b`) merges each + /// host's report. If a later host fails, the `?` would drop earlier hosts + /// from `last-sync-report.json`. The caller records this, persists, then + /// returns the `Err`. + pub fn record_pull_failure(&mut self, host: &str, err: impl fmt::Display) { + self.warnings + .push(format!("pull from {host} failed: {err}")); + } + + pub fn events_new(&self) -> i64 { + self.peers + .iter() + .flat_map(|p| p.buckets.iter()) + .map(|b| b.events_new) + .sum::() + + self.pushed.iter().map(|b| b.events_new).sum::() + } + + pub fn events_pulled(&self) -> i64 { + self.peers + .iter() + .flat_map(|p| p.buckets.iter()) + .map(|b| b.events_new) + .sum() + } + + pub fn events_pushed(&self) -> i64 { + self.pushed.iter().map(|b| b.events_new).sum() + } + + pub fn peers_imported(&self) -> usize { + self.peers + .iter() + .filter(|p| matches!(p.outcome, PeerOutcome::Imported)) + .count() + } + + pub fn peers_skipped(&self) -> usize { + self.peers + .iter() + .filter(|p| matches!(p.outcome, PeerOutcome::Skipped { .. })) + .count() + } + + pub fn peers_failed(&self) -> usize { + self.peers + .iter() + .filter(|p| matches!(p.outcome, PeerOutcome::Failed { .. })) + .count() + } + + pub fn summary_message(&self) -> String { + let pulled = self.events_pulled(); + let pushed = self.events_pushed(); + let imported = self.peers_imported(); + let skipped = self.peers_skipped(); + let failed = self.peers_failed(); + let peer_n = self.peers.len(); + let mut msg = match self.mode { + SyncMode::Pull => format!( + "Pulled {pulled} new events from {imported}/{peer_n} peers \ + ({skipped} skipped, {failed} failed)" + ), + SyncMode::Push => format!( + "Pushed {pushed} new events across {} bucket(s)", + self.pushed.len() + ), + SyncMode::Both => format!( + "Synced {pulled} events in from {imported}/{peer_n} peers \ + ({skipped} skipped, {failed} failed); pushed {pushed} events" + ), + }; + // Daemon and JNI log this string. Warnings (push abort, empty-dir + // diagnosis) must ride along or the line looks like a no-op success. + if !self.warnings.is_empty() { + msg.push_str("; "); + msg.push_str(&self.warnings.join("; ")); + } + msg + } + + /// JSON `SyncInterface.performSyncAsync` already parses (`success` + `message`), + /// plus the counts that ActivityWatch/aw-android#274 can render without a + /// JNI-side redesign. + #[cfg(any(target_os = "android", test))] + pub fn to_jni_json(&self) -> String { + serde_json::json!({ + "success": true, + "message": self.summary_message(), + "started": self.started.to_rfc3339(), + "finished": self.finished.to_rfc3339(), + "mode": self.mode, + "events_new": self.events_new(), + "events_pulled": self.events_pulled(), + "events_pushed": self.events_pushed(), + "peers_imported": self.peers_imported(), + "peers_skipped": self.peers_skipped(), + "peers_failed": self.peers_failed(), + "peers": self.peers, + "pushed": self.pushed, + "warnings": self.warnings, + }) + .to_string() + } +} + +impl PeerReport { + pub fn imported( + device_id: String, + hostname: String, + path: PathBuf, + buckets: Vec, + ) -> Self { + Self { + device_id, + hostname, + path, + outcome: PeerOutcome::Imported, + buckets, + } + } + + pub fn skipped(device_id: String, hostname: String, path: PathBuf, reason: String) -> Self { + Self { + device_id, + hostname, + path, + outcome: PeerOutcome::Skipped { reason }, + buckets: vec![], + } + } + + pub fn failed(device_id: String, hostname: String, path: PathBuf, error: String) -> Self { + Self { + device_id, + hostname, + path, + outcome: PeerOutcome::Failed { error }, + buckets: vec![], + } + } +} + +impl fmt::Display for SyncReport { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + writeln!(f, "Last pass")?; + writeln!(f, "---------")?; + writeln!(f, "started: {}", self.started.to_rfc3339())?; + writeln!(f, "finished: {}", self.finished.to_rfc3339())?; + writeln!(f, "mode: {}", self.mode.as_str())?; + writeln!( + f, + "peers: {} imported, {} skipped, {} failed", + self.peers_imported(), + self.peers_skipped(), + self.peers_failed() + )?; + writeln!( + f, + "events: {} pulled, {} pushed ({} total)", + self.events_pulled(), + self.events_pushed(), + self.events_new() + )?; + if self.peers.is_empty() { + writeln!(f, "peers: []")?; + } else { + writeln!(f, "peers:")?; + for peer in &self.peers { + writeln!(f, " {}", format_peer_line(peer))?; + } + } + if !self.warnings.is_empty() { + writeln!(f, "warnings:")?; + for line in &self.warnings { + writeln!(f, " ! {line}")?; + } + } + if !self.pushed.is_empty() { + writeln!(f, "pushed:")?; + for bucket in &self.pushed { + writeln!( + f, + " {} +{} events{}", + bucket.bucket_id, + bucket.events_new, + match bucket.resumed_at { + Some(t) => format!(" resumed {}", t.to_rfc3339()), + None => String::new(), + } + )?; + } + } + Ok(()) + } +} + +fn format_peer_line(peer: &PeerReport) -> String { + let host = if peer.hostname.is_empty() { + "?".to_string() + } else { + peer.hostname.clone() + }; + let did = if peer.device_id.is_empty() { + "?".to_string() + } else { + peer.device_id.clone() + }; + let events: i64 = peer.buckets.iter().map(|b| b.events_new).sum(); + match &peer.outcome { + PeerOutcome::Imported => format!("{host} / {did} imported +{events} events"), + PeerOutcome::Skipped { reason } => { + format!("{host} / {did} skipped {reason}") + } + PeerOutcome::Failed { error } => format!("{host} / {did} failed {error}"), + } +} + +/// Override with `AW_SYNC_LAST_REPORT` (absolute path) in tests. +pub fn last_report_path() -> Result> { + if let Ok(p) = std::env::var("AW_SYNC_LAST_REPORT") { + if !p.is_empty() { + return Ok(PathBuf::from(p)); + } + } + let dir = aw_server::dirs::get_data_dir().map_err(|_| "Could not get data dir")?; + Ok(dir.join("aw-sync").join("last-sync-report.json")) +} + +pub fn persist_last_report(report: &SyncReport) -> Result> { + let path = last_report_path()?; + persist_last_report_to(report, &path)?; + Ok(path) +} + +pub fn persist_last_report_to(report: &SyncReport, path: &Path) -> Result<(), Box> { + if let Some(parent) = path.parent() { + fs::create_dir_all(parent)?; + } + // Unique temp name so overlapping daemon/manual/Android writers cannot + // truncate each other's `.json.tmp`. + let tmp = { + let name = path + .file_name() + .and_then(|s| s.to_str()) + .unwrap_or("last-sync-report.json"); + path.with_file_name(format!( + "{}.tmp.{}-{}", + name, + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0) + )) + }; + fs::write(&tmp, serde_json::to_string_pretty(report)?)?; + fs::rename(&tmp, path)?; + Ok(()) +} + +/// Persist, but a write failure must not fail a pass that already succeeded. +pub fn persist_last_report_warn(report: &SyncReport) { + match persist_last_report(report) { + Ok(path) => debug!("Wrote last sync report to {}", path.display()), + Err(e) => warn!("Failed to persist last sync report: {e}"), + } +} + +pub fn load_last_report() -> Result, Box> { + load_last_report_from(&last_report_path()?) +} + +pub fn load_last_report_from(path: &Path) -> Result, Box> { + if !path.exists() { + return Ok(None); + } + let body = fs::read_to_string(path)?; + Ok(Some(serde_json::from_str(&body)?)) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::{SystemTime, UNIX_EPOCH}; + + fn temp_report_path() -> PathBuf { + std::env::temp_dir().join(format!( + "aw-sync-report-{}-{}.json", + std::process::id(), + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos() + )) + } + + #[test] + fn roundtrip_json_and_counts() { + let mut report = SyncReport::new(SyncMode::Both); + report.peers.push(PeerReport::imported( + "dev-a".into(), + "laptop".into(), + PathBuf::from("/sync/laptop/dev-a/test.db"), + vec![BucketReport { + bucket_id: "aw-watcher-window".into(), + events_new: 10, + resumed_at: None, + }], + )); + report.peers.push(PeerReport { + device_id: "dev-a".into(), + hostname: "old-name".into(), + path: PathBuf::from("/sync/old-name/dev-a/test.db"), + outcome: PeerOutcome::Skipped { + reason: "duplicate device_id dev-a; kept larger db".into(), + }, + buckets: vec![], + }); + report.pushed.push(BucketReport { + bucket_id: "aw-watcher-afk".into(), + events_new: 3, + resumed_at: None, + }); + report.finish(); + + assert_eq!(report.events_pulled(), 10); + assert_eq!(report.events_pushed(), 3); + assert_eq!(report.events_new(), 13); + assert_eq!(report.peers_imported(), 1); + assert_eq!(report.peers_skipped(), 1); + assert_eq!(report.peers_failed(), 0); + + let path = temp_report_path(); + persist_last_report_to(&report, &path).unwrap(); + let loaded = load_last_report_from(&path).unwrap().unwrap(); + let _ = fs::remove_file(&path); + assert_eq!(loaded.peers_imported(), 1); + assert_eq!(loaded.peers_skipped(), 1); + assert_eq!(loaded.pushed[0].events_new, 3); + assert!(loaded.summary_message().contains("10 events")); + let json = loaded.to_jni_json(); + assert!(json.contains("\"success\":true")); + assert!(json.contains("\"peers_skipped\":1")); + } + + #[test] + fn load_missing_is_none() { + let path = temp_report_path(); + let _ = fs::remove_file(&path); + assert!(load_last_report_from(&path).unwrap().is_none()); + } + + #[test] + fn display_includes_skipped_reason() { + let mut report = SyncReport::new(SyncMode::Pull); + report.peers.push(PeerReport { + device_id: "abc".into(), + hostname: "phone".into(), + path: PathBuf::from("/sync/phone/abc/test.db"), + outcome: PeerOutcome::Skipped { + reason: "duplicate device_id abc; kept larger db".into(), + }, + buckets: vec![], + }); + let text = report.to_string(); + assert!(text.contains("Last pass")); + assert!(text.contains("skipped duplicate device_id abc")); + } + + /// ActivityWatch/aw-server-rust#682 / #695: a pass that finds zero peers + /// must still produce a report saying so — `peers: []` plus the discovery + /// warnings. Logging them and discarding the report was the silent failure. + #[test] + fn zero_peer_pass_reports_empty_peers_and_captures_discovery_warnings() { + let root = std::env::temp_dir().join(format!( + "aw-sync-zero-peers-{}-{}", + std::process::id(), + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + fs::create_dir_all(&root).unwrap(); + + let mut report = SyncReport::new(SyncMode::Pull); + report.capture_warnings(crate::util::pull_discovery_warnings( + &root, + Some("local-device"), + &[], + )); + report.finish(); + + assert!( + report.peers.is_empty(), + "zero remotes → peers: [], got {:?}", + report.peers + ); + assert!( + report + .warnings + .iter() + .any(|w| w.contains("Found 0 remote db files")), + "discovery warnings must live on the report, got {:?}", + report.warnings + ); + assert!( + report + .warnings + .iter() + .any(|w| w.contains("sync directory is empty")), + "empty-dir diagnosis must be captured: {:?}", + report.warnings + ); + + let path = temp_report_path(); + persist_last_report_to(&report, &path).unwrap(); + let body = fs::read_to_string(&path).unwrap(); + let loaded = load_last_report_from(&path).unwrap().unwrap(); + let _ = fs::remove_file(&path); + let _ = fs::remove_dir_all(&root); + + assert!(loaded.peers.is_empty()); + assert!( + body.contains("\"peers\": []"), + "persisted JSON must say peers: [], got {body}" + ); + assert!( + loaded + .warnings + .iter() + .any(|w| w.contains("Found 0 remote db files")), + "loaded report dropped discovery warnings: {:?}", + loaded.warnings + ); + let text = loaded.to_string(); + assert!(text.contains("peers: []"), "display: {text}"); + assert!( + text.contains("Found 0 remote db files"), + "display must surface the warning: {text}" + ); + let json = loaded.to_jni_json(); + assert!(json.contains("\"peers\":[]") || json.contains("\"peers\": []")); + assert!(json.contains("Found 0 remote db files")); + } + + /// A push abort must leave a warning on disk. The #695 contract is that + /// a failed pass is visible afterwards; an empty report looks like success. + #[test] + fn push_failure_persisted_report_is_not_a_silent_success() { + let mut report = SyncReport::new(SyncMode::Push); + report.record_push_failure("disk full"); + report.finish(); + + assert!( + report + .warnings + .iter() + .any(|w| w == "push failed: disk full"), + "push abort must be on the report, got {:?}", + report.warnings + ); + + let path = temp_report_path(); + persist_last_report_to(&report, &path).unwrap(); + let body = fs::read_to_string(&path).unwrap(); + let loaded = load_last_report_from(&path).unwrap().unwrap(); + let _ = fs::remove_file(&path); + + assert!(loaded.peers.is_empty()); + assert!(loaded.pushed.is_empty()); + assert!( + loaded + .warnings + .iter() + .any(|w| w.contains("push failed: disk full")), + "loaded report dropped the push failure: {:?}", + loaded.warnings + ); + assert!( + body.contains("push failed: disk full"), + "persisted JSON must name the push failure, got {body}" + ); + let text = loaded.to_string(); + assert!( + text.contains("push failed: disk full"), + "display must surface the push failure: {text}" + ); + assert!( + loaded.summary_message().contains("push failed: disk full"), + "daemon/JNI summary must not look like success: {}", + loaded.summary_message() + ); + } + + /// A later host's pull abort must keep earlier hosts on disk. The + /// `--host a --host b` path merges into one report; dropping it on `?` + /// would leave only the failing host's inner persist. + #[test] + fn later_host_pull_failure_keeps_earlier_hosts_on_disk() { + let mut report = SyncReport::new(SyncMode::Both); + report.peers.push(PeerReport::imported( + "dev-a".into(), + "host-a".into(), + PathBuf::from("/sync/host-a/dev-a/test.db"), + vec![BucketReport { + bucket_id: "aw-watcher-window".into(), + events_new: 7, + resumed_at: None, + }], + )); + report.record_pull_failure("host-b", "no db found"); + report.finish(); + + assert!( + report + .warnings + .iter() + .any(|w| w == "pull from host-b failed: no db found"), + "pull abort must be on the report, got {:?}", + report.warnings + ); + assert_eq!(report.peers_imported(), 1); + assert!( + report + .summary_message() + .contains("pull from host-b failed: no db found"), + "daemon summary must name the failed host: {}", + report.summary_message() + ); + + let path = temp_report_path(); + persist_last_report_to(&report, &path).unwrap(); + let loaded = load_last_report_from(&path).unwrap().unwrap(); + let _ = fs::remove_file(&path); + + assert_eq!(loaded.peers_imported(), 1); + assert_eq!(loaded.peers[0].hostname, "host-a"); + assert_eq!(loaded.events_pulled(), 7); + assert!( + loaded + .warnings + .iter() + .any(|w| w.contains("pull from host-b failed: no db found")), + "loaded report dropped the later-host failure: {:?}", + loaded.warnings + ); + } + + #[test] + fn old_report_without_warnings_field_still_loads() { + let path = temp_report_path(); + fs::write( + &path, + r#"{ + "started": "2026-09-16T08:00:00Z", + "finished": "2026-09-16T08:00:01Z", + "mode": "pull", + "peers": [], + "pushed": [] + }"#, + ) + .unwrap(); + let loaded = load_last_report_from(&path).unwrap().unwrap(); + let _ = fs::remove_file(&path); + assert!(loaded.peers.is_empty()); + assert!(loaded.warnings.is_empty()); + } +} diff --git a/aw-sync/src/status.rs b/aw-sync/src/status.rs index 5323c99b..e1d14c6e 100644 --- a/aw-sync/src/status.rs +++ b/aw-sync/src/status.rs @@ -92,6 +92,21 @@ pub fn collect_status( )); out.push('\n'); + match crate::report::load_last_report() { + Ok(Some(report)) => { + out.push_str(&report.to_string()); + out.push('\n'); + } + Ok(None) => { + out.push_str("Last pass: (none — no pass has been persisted yet)\n\n"); + } + Err(e) => { + out.push_str(&format!( + "Last pass: (could not read last-sync-report.json: {e})\n\n" + )); + } + } + if inspected.is_empty() { out.push_str("Entries: (none)\n"); } else { diff --git a/aw-sync/src/sync.rs b/aw-sync/src/sync.rs index c9bab57f..a09b6526 100644 --- a/aw-sync/src/sync.rs +++ b/aw-sync/src/sync.rs @@ -19,18 +19,11 @@ use chrono::{DateTime, Duration, Utc}; use aw_datastore::{Datastore, DatastoreError}; use aw_models::{Bucket, Event}; -#[cfg(feature = "cli")] -use clap::ValueEnum; - use crate::accessmethod::AccessMethod; +use crate::report::{BucketReport, PeerReport}; +use crate::util::find_remotes_nonlocal_selection; -#[derive(PartialEq, Eq, Copy, Clone)] -#[cfg_attr(feature = "cli", derive(ValueEnum))] -pub enum SyncMode { - Push, - Pull, - Both, -} +pub use crate::report::{SyncMode, SyncReport}; #[derive(Debug)] pub struct SyncSpec { @@ -58,12 +51,17 @@ impl Default for SyncSpec { } } -/// Performs a single sync pass +/// Performs a single sync pass and returns what it did. +/// +/// A `Ok` report is complete. On error, a partial report (failed peer, any +/// peers already imported) is persisted so `aw-sync status` can show the +/// failure without the process having to stay alive. pub fn sync_run( client: &AwClient, sync_spec: &SyncSpec, mode: SyncMode, -) -> Result<(), Box> { +) -> Result> { + let mut report = SyncReport::new(mode); let info = client.get_info()?; // FIXME: Here it is assumed that the device_id for the local server is the one used by @@ -78,11 +76,12 @@ pub fn sync_run( // "each device only writes files it owns" invariant (see // ActivityWatch/aw-server-rust#682). let ds_localremote = maybe_setup_local_remote(sync_spec.path.as_path(), device_id, mode)?; - let remote_dbfiles = crate::util::find_remotes_nonlocal( + let selection = find_remotes_nonlocal_selection( sync_spec.path.as_path(), device_id, sync_spec.path_db.as_ref(), )?; + let remote_dbfiles: Vec<_> = selection.selected.iter().map(|d| d.path.clone()).collect(); // Log if remotes found // TODO: Only log remotes of interest @@ -95,26 +94,55 @@ pub fn sync_run( } // Finding zero peers in a configured sync dir is the interesting case — - // ActivityWatch/aw-server-rust#684. Do not stay silent. + // ActivityWatch/aw-server-rust#682 / #695. Do not stay silent: keep the + // warnings on the report, not just in the log. if mode == SyncMode::Pull || mode == SyncMode::Both { - for line in crate::util::pull_discovery_warnings( + report.capture_warnings(crate::util::pull_discovery_warnings( sync_spec.path.as_path(), - device_id, + Some(device_id), &remote_dbfiles, - ) { - warn!("{line}"); + )); + for skipped in &selection.skipped { + report.peers.push(PeerReport::skipped( + skipped.db.device_id.clone(), + skipped.db.hostname.clone(), + skipped.db.path.clone(), + skipped.reason.clone(), + )); } } // Peer files are opened read-only: never migrate, never flip WAL - // (ActivityWatch/aw-server-rust#693). Version mismatch is skipped, not fatal. + // (ActivityWatch/aw-server-rust#693). Version mismatch is skipped, not fatal — + // but it must be *visible*: record it on the report so status/JNI do not + // present an all-incompatible pass as a clean empty one. let mut ds_remotes = Vec::new(); - for path in &remote_dbfiles { - match open_peer_datastore(path) { - Ok(Some(ds)) => ds_remotes.push(ds), - Ok(None) => {} + for db in &selection.selected { + match open_peer_datastore(&db.path) { + Ok(OpenedPeer::Ready(ds)) => ds_remotes.push((db.clone(), ds)), + Ok(OpenedPeer::Incompatible(msg)) => { + if mode == SyncMode::Pull || mode == SyncMode::Both { + report.peers.push(PeerReport::skipped( + db.device_id.clone(), + db.hostname.clone(), + db.path.clone(), + format!("incompatible database version: {msg}"), + )); + } + } Err(e) => { - warn!("Failed to open remote db {}: {e}", path.display()); + warn!("Failed to open remote db {}: {e}", db.path.display()); + if mode == SyncMode::Pull || mode == SyncMode::Both { + report.peers.push(PeerReport::failed( + db.device_id.clone(), + db.hostname.clone(), + db.path.clone(), + e.clone(), + )); + report.finish(); + crate::report::persist_last_report_warn(&report); + } + close_opened_datastores(&ds_remotes, &ds_localremote); return Err(e.into()); } } @@ -124,31 +152,54 @@ pub fn sync_run( info!( "Found {} remote datastores: {:?}", ds_remotes.len(), - ds_remotes + ds_remotes.iter().map(|(_, ds)| ds).collect::>() ); } // Pull if mode == SyncMode::Pull || mode == SyncMode::Both { info!("Pulling..."); - for ds_from in &ds_remotes { - sync_datastores(ds_from, client, false, None, sync_spec)?; + for (db, ds_from) in &ds_remotes { + match sync_datastores(ds_from, client, false, None, sync_spec) { + Ok(buckets) => report.peers.push(PeerReport::imported( + db.device_id.clone(), + db.hostname.clone(), + db.path.clone(), + buckets, + )), + Err(e) => { + report.peers.push(PeerReport::failed( + db.device_id.clone(), + db.hostname.clone(), + db.path.clone(), + e.clone(), + )); + report.finish(); + crate::report::persist_last_report_warn(&report); + close_opened_datastores(&ds_remotes, &ds_localremote); + return Err(e.into()); + } + } } } // Push local server buckets to sync folder - if let Some(ds_localremote) = &ds_localremote { + if let Some(ds_local) = &ds_localremote { info!("Pushing..."); - sync_datastores(client, ds_localremote, true, Some(device_id), sync_spec)?; + match sync_datastores(client, ds_local, true, Some(device_id), sync_spec) { + Ok(buckets) => report.pushed = buckets, + Err(e) => { + report.record_push_failure(&e); + report.finish(); + crate::report::persist_last_report_warn(&report); + close_opened_datastores(&ds_remotes, &ds_localremote); + return Err(e.into()); + } + } } // Close open database connections - for ds_from in &ds_remotes { - ds_from.close(); - } - if let Some(ds_localremote) = &ds_localremote { - ds_localremote.close(); - } + close_opened_datastores(&ds_remotes, &ds_localremote); // Dropping also works to close the database connections, weirdly enough. // Probably because once the database is dropped, the thread will stop, @@ -159,7 +210,24 @@ pub fn sync_run( // NOTE: Will fail if db connections not closed (as it will open them again) //list_buckets(&client, sync_spec.path.as_path()); - Ok(()) + report.finish(); + Ok(report) +} + +/// Stop datastore worker threads. Drop alone does not wait for the sqlite +/// lock; the success path already called `close()` for that reason. Error +/// returns must do the same or a long-lived daemon can leak connections +/// across failed passes. +fn close_opened_datastores( + ds_remotes: &[(crate::util::RemoteDb, Datastore)], + ds_localremote: &Option, +) { + for (_, ds_from) in ds_remotes { + ds_from.close(); + } + if let Some(ds) = ds_localremote { + ds.close(); + } } #[allow(dead_code)] @@ -177,7 +245,7 @@ pub fn list_buckets(client: &AwClient) -> Result<(), Box> { let mut ds_remotes = Vec::new(); for path in &remote_dbfiles { - if let Some(ds) = open_peer_datastore(path)? { + if let OpenedPeer::Ready(ds) = open_peer_datastore(path)? { ds_remotes.push(ds); } } @@ -240,15 +308,21 @@ pub fn create_datastore(path: &Path) -> Result { /// Open a *peer* database for pull: read-only, no migration, no WAL sidecars. /// -/// Returns `Ok(None)` when `user_version` does not match this binary so the -/// caller can skip that peer and keep walking (ActivityWatch/aw-server-rust#693). -fn open_peer_datastore(path: &Path) -> Result, String> { +/// Returns `OpenedPeer::Incompatible` when `user_version` does not match this +/// binary so the caller can skip that peer and keep walking, keeping the +/// version-mismatch reason for reporting (ActivityWatch/aw-server-rust#693). +enum OpenedPeer { + Ready(Datastore), + Incompatible(String), +} + +fn open_peer_datastore(path: &Path) -> Result { let pathstr = utf8_db_path(path)?; match Datastore::open_read_only(pathstr.to_string()) { - Ok(ds) => Ok(Some(ds)), + Ok(ds) => Ok(OpenedPeer::Ready(ds)), Err(DatastoreError::OldDbVersion(msg)) => { warn!("Skipping peer db {}: {msg}", path.display()); - Ok(None) + Ok(OpenedPeer::Incompatible(msg)) } Err(e) => Err(format!( "Failed to open remote db {}: {e:?}", @@ -375,7 +449,7 @@ pub fn sync_datastores( is_push: bool, src_did: Option<&str>, sync_spec: &SyncSpec, -) -> Result<(), String> { +) -> Result, String> { // FIXME: "-synced" should only be appended when synced to the local database, not to the // staging area for local buckets. info!("Syncing {:?} to {:?}", ds_from, ds_to); @@ -448,12 +522,13 @@ pub fn sync_datastores( // Sync buckets in order of most recently updated buckets_from.sort_by_key(|b| b.metadata.end); + let mut buckets = Vec::with_capacity(buckets_from.len()); for bucket_from in buckets_from { let bucket_to = get_or_create_sync_bucket(&bucket_from, ds_to, is_push)?; - sync_one(ds_from, ds_to, bucket_from, bucket_to, sync_spec)?; + buckets.push(sync_one(ds_from, ds_to, bucket_from, bucket_to, sync_spec)?); } - Ok(()) + Ok(buckets) } /// Syncs a single bucket from one datastore to another @@ -463,7 +538,7 @@ fn sync_one( bucket_from: Bucket, bucket_to: Bucket, sync_spec: &SyncSpec, -) -> Result<(), String> { +) -> Result { let eventcount_to_old = ds_to.get_event_count(bucket_to.id.as_str())?; info!(" ⟳ Syncing bucket '{}'", bucket_to.id); @@ -628,7 +703,11 @@ fn sync_one( } } - Ok(()) + Ok(BucketReport { + bucket_id: bucket_to.id, + events_new: new_events_count, + resumed_at: resume_sync_at, + }) } fn log_buckets(ds: &dyn AccessMethod) -> Result<(), String> { diff --git a/aw-sync/src/sync_wrapper.rs b/aw-sync/src/sync_wrapper.rs index c5f438a6..b14d2542 100644 --- a/aw-sync/src/sync_wrapper.rs +++ b/aw-sync/src/sync_wrapper.rs @@ -2,32 +2,68 @@ use std::error::Error; use std::fs; use std::path::Path; -use crate::sync::{sync_run, SyncMode, SyncSpec}; +use crate::report::{PeerReport, SyncMode, SyncReport}; +use crate::sync::{sync_run, SyncSpec}; use aw_client_rust::blocking::AwClient; -pub fn pull_all(client: &AwClient) -> Result<(), Box> { +pub fn pull_all(client: &AwClient) -> Result> { let sync_root = crate::dirs::get_sync_dir().map_err(|_| "Could not get sync dir")?; let dbs = crate::util::list_remote_dbs(&sync_root)?; - let selected = crate::util::select_remote_dbs_by_device_id(dbs); - if selected.is_empty() { + let selection = crate::util::select_remote_dbs_detailed(dbs); + let mut report = SyncReport::new(SyncMode::Pull); + for skipped in &selection.skipped { + report.peers.push(PeerReport::skipped( + skipped.db.device_id.clone(), + skipped.db.hostname.clone(), + skipped.db.path.clone(), + skipped.reason.clone(), + )); + } + let found: Vec<_> = selection.selected.iter().map(|d| d.path.clone()).collect(); + // No get_info() here: reqwest's client timeout is 120s, and an empty-dir + // pass (#682) must stay a local filesystem check. `None` is unknown, not + // `Some("")` — empty string does not unclassify entries, and leftover + // 2-level own staging then shows up as "peer db(s) that pull did not select". + report.capture_warnings(crate::util::pull_discovery_warnings( + &sync_root, None, &found, + )); + if selection.selected.is_empty() { info!("No remote databases found in {:?}", sync_root); - return Ok(()); + report.finish(); + return Ok(report); } info!( "Pulling {} remote database(s): {:?}", - selected.len(), - selected + selection.selected.len(), + selection + .selected .iter() .map(|d| d.path.display().to_string()) .collect::>() ); - for remote in selected { - pull_db(client, &remote.hostname, &remote.path)?; + for remote in selection.selected { + match pull_db(client, &remote.hostname, &remote.path) { + Ok(one) => report.merge(one), + Err(e) => { + report.peers.push(PeerReport::failed( + remote.device_id, + remote.hostname, + remote.path, + e.to_string(), + )); + report.finish(); + // Persist the aggregate (earlier peers + this failure), not + // only the phase-local report from the failing `sync_run`. + crate::report::persist_last_report_warn(&report); + return Err(e); + } + } } - Ok(()) + report.finish(); + Ok(report) } -pub fn pull(host: &str, client: &AwClient) -> Result<(), Box> { +pub fn pull(host: &str, client: &AwClient) -> Result> { // Path to the sync folder // Sync folder is structured ./{hostname}/{device_id}/test.db let sync_root_dir = crate::dirs::get_sync_dir().map_err(|_| "Could not get sync dir")?; @@ -61,7 +97,7 @@ pub fn pull(host: &str, client: &AwClient) -> Result<(), Box> { pull_db(client, host, &db.path()) } -fn pull_db(client: &AwClient, host: &str, db_path: &Path) -> Result<(), Box> { +fn pull_db(client: &AwClient, host: &str, db_path: &Path) -> Result> { client.wait_for_start()?; let sync_root_dir = crate::dirs::get_sync_dir().map_err(|_| "Could not get sync dir")?; let sync_dir = sync_root_dir.join(host); @@ -71,15 +107,14 @@ fn pull_db(client: &AwClient, host: &str, db_path: &Path) -> Result<(), Box Result<(), Box> { +pub fn push(client: &AwClient) -> Result> { push_with_hostname(client, &client.hostname) } -pub fn push_with_hostname(client: &AwClient, hostname: &str) -> Result<(), Box> { +pub fn push_with_hostname(client: &AwClient, hostname: &str) -> Result> { let sync_dir = crate::dirs::get_sync_dir() .map_err(|_| "Could not get sync dir")? .join(hostname); @@ -90,7 +125,5 @@ pub fn push_with_hostname(client: &AwClient, hostname: &str) -> Result<(), Box std::io::Result Ok(dbs) } +/// A remote that `select_remote_dbs_detailed` chose not to import. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct SkippedRemote { + pub db: RemoteDb, + pub reason: String, +} + +/// Selected remotes plus the duplicates that were dropped. +#[derive(Debug, Clone, Default)] +pub(crate) struct RemoteSelection { + pub selected: Vec, + pub skipped: Vec, +} + /// Keep one database per `device_id`, preferring the largest file. /// /// A hostname change (or sanitization) can leave the same device writing under @@ -361,27 +408,43 @@ pub(crate) fn list_remote_dbs(sync_root: &Path) -> std::io::Result /// bucket hostname, so both land in the same destination bucket, and resume /// then silently drops the older history. See ActivityWatch/aw-server-rust#683. pub(crate) fn select_remote_dbs_by_device_id(dbs: Vec) -> Vec { + select_remote_dbs_detailed(dbs).selected +} + +/// Same collapse as [`select_remote_dbs_by_device_id`], but keeps the losers +/// so a `SyncReport` can record `PeerOutcome::Skipped` instead of a log line. +pub(crate) fn select_remote_dbs_detailed(dbs: Vec) -> RemoteSelection { let mut by_device: HashMap> = HashMap::new(); for db in dbs { by_device.entry(db.device_id.clone()).or_default().push(db); } let mut selected = Vec::with_capacity(by_device.len()); + let mut skipped = Vec::new(); for (device_id, mut group) in by_device { group.sort_by(|a, b| b.size.cmp(&a.size).then_with(|| a.path.cmp(&b.path))); let mut group = group.into_iter(); let winner = group.next().expect("device_id group is non-empty"); - let skipped: Vec = group - .map(|d| format!("{} ({} bytes)", d.path.display(), d.size)) - .collect(); - if !skipped.is_empty() { + let losers: Vec = group.collect(); + if !losers.is_empty() { + let skip_paths: Vec = losers + .iter() + .map(|d| format!("{} ({} bytes)", d.path.display(), d.size)) + .collect(); warn!( "device_id {device_id} appears under {} folders; using largest {} ({} bytes), skipping: {:?}", - skipped.len() + 1, + losers.len() + 1, winner.path.display(), winner.size, - skipped + skip_paths ); + for db in losers { + let reason = format!( + "duplicate device_id {device_id}; kept larger {}", + winner.path.display() + ); + skipped.push(SkippedRemote { db, reason }); + } } selected.push(winner); } @@ -391,7 +454,8 @@ pub(crate) fn select_remote_dbs_by_device_id(dbs: Vec) -> Vec, ) -> std::io::Result> { + Ok( + find_remotes_nonlocal_selection(sync_directory, device_id, sync_db)? + .selected + .into_iter() + .map(|d| d.path) + .collect(), + ) +} + +/// Same as [`find_remotes_nonlocal`], plus the duplicate-device_id skips. +pub(crate) fn find_remotes_nonlocal_selection( + sync_directory: &Path, + device_id: &str, + sync_db: Option<&PathBuf>, +) -> std::io::Result { let remotes_all = find_remotes(sync_directory)?; let filtered: Vec = remotes_all .into_iter() @@ -450,11 +529,20 @@ pub fn find_remotes_nonlocal( } }) .collect(); - Ok(select_db_paths_by_device_id(filtered)) + Ok(paths_to_remote_selection(filtered)) } /// Collapse `{…}/{device_id}/*.db` paths to the largest file per device_id. +#[cfg(test)] fn select_db_paths_by_device_id(paths: Vec) -> Vec { + paths_to_remote_selection(paths) + .selected + .into_iter() + .map(|d| d.path) + .collect() +} + +fn paths_to_remote_selection(paths: Vec) -> RemoteSelection { let dbs: Vec = paths .into_iter() .filter_map(|path| { @@ -475,10 +563,7 @@ fn select_db_paths_by_device_id(paths: Vec) -> Vec { }) }) .collect(); - select_remote_dbs_by_device_id(dbs) - .into_iter() - .map(|d| d.path) - .collect() + select_remote_dbs_detailed(dbs) } /// How a database sits in the sync folder. @@ -800,12 +885,19 @@ pub fn format_bytes(n: u64) -> String { } /// Warn-lines for a pull that found no remotes, or that missed classified peers. +/// +/// `local_device_id` is `None` when the local server was not contacted +/// (`pull_all` must not `get_info()` — that call has a 120s HTTP timeout). +/// Empty string is treated as unknown: `Some("")` is not unclassified, it +/// makes `path.contains("")` true and leftover 2-level own staging look like +/// a peer that pull failed to select. pub fn pull_discovery_warnings( sync_directory: &Path, - local_device_id: &str, + local_device_id: Option<&str>, found_remotes: &[PathBuf], ) -> Vec { - let scan = match scan_sync_dir(sync_directory, Some(local_device_id)) { + let local_device_id = local_device_id.filter(|s| !s.is_empty()); + let scan = match scan_sync_dir(sync_directory, local_device_id) { Ok(entries) => entries, Err(e) => { return vec![format!( @@ -820,7 +912,24 @@ pub fn pull_discovery_warnings( let missed: Vec<&SyncDirEntry> = scan .iter() .filter(|e| { - e.kind == SyncEntryKind::Peer && e.db_path.as_ref().is_some_and(|p| !found.contains(p)) + if e.kind != SyncEntryKind::Peer { + return false; + } + let Some(path) = e.db_path.as_ref() else { + return false; + }; + if found.contains(path) { + return false; + } + // Without a local id, 2-level leftovers cannot be told from own + // staging (`{sync_root}/{device_id}/test.db`). `pull_all` also + // never selects 2-level, so calling them "unselected peers" is + // a false diagnostic. Known local id keeps the mixed-layout + // warning for a real 2-level *peer*. + if local_device_id.is_none() && e.layout == Some(SyncLayout::TwoLevel) { + return false; + } + true }) .collect(); @@ -1031,7 +1140,7 @@ mod scan_tests { "skipped duplicate must use the pull selector: {skipped:?}" ); - let warnings = pull_discovery_warnings(&root, local, &[]); + let warnings = pull_discovery_warnings(&root, Some(local), &[]); let joined = warnings.join("\n"); assert!( joined.contains("Found 0 remote db files"), @@ -1100,7 +1209,7 @@ mod scan_tests { assert!(peer_entry.not_visible_to_daemon.is_none()); let found = vec![root.join(peer).join("test.db")]; - let warnings = pull_discovery_warnings(&root, local, &found); + let warnings = pull_discovery_warnings(&root, Some(local), &found); assert!( warnings .iter() @@ -1111,6 +1220,38 @@ mod scan_tests { let _ = fs::remove_dir_all(root); } + #[test] + fn unknown_local_id_does_not_report_two_level_leftover_as_unselected_peer() { + // Mixed layout: leftover 2-level own staging (pre-#685) plus a + // 3-level peer that pull_all selected. Passing "" used to classify + // the leftover as Peer and warn "pull did not select". + let root = temp_sync_dir(); + let local = "local-device"; + let peer = "peer-device"; + touch_db(&root.join(local).join("test.db"), 64); + let peer_db = root.join("other-host").join(peer).join("test.db"); + touch_db(&peer_db, 273); + let found = vec![peer_db]; + + let unknown = pull_discovery_warnings(&root, None, &found).join("\n"); + assert!( + !unknown.contains("pull did not select"), + "unknown local id must not call leftover 2-level own staging a missed peer: {unknown}" + ); + let empty_str = pull_discovery_warnings(&root, Some(""), &found).join("\n"); + assert!( + !empty_str.contains("pull did not select"), + "empty string is unknown, not a match-everything id: {empty_str}" + ); + let known = pull_discovery_warnings(&root, Some(local), &found).join("\n"); + assert!( + !known.contains("pull did not select"), + "known local id classifies 2-level own as OwnStaging: {known}" + ); + + let _ = fs::remove_dir_all(root); + } + #[cfg(feature = "cli")] #[test] fn inspect_sync_db_reads_hostname_and_newest_event() { diff --git a/aw-sync/tests/sync.rs b/aw-sync/tests/sync.rs index 9d24695a..add2937b 100644 --- a/aw-sync/tests/sync.rs +++ b/aw-sync/tests/sync.rs @@ -222,6 +222,39 @@ mod sync_tests { assert!(buckets_src.len() == buckets_dest.len()); } + #[test] + fn test_sync_datastores_returns_new_event_counts() { + let state = init_teststate(); + let bucket_id = create_bucket(&state.ds_src, 0); + create_events(&state.ds_src, &bucket_id, 3); + + let reports = aw_sync::sync_datastores( + &state.ds_src, + &state.ds_dest, + false, + None, + &SyncSpec::default(), + ) + .unwrap(); + assert_eq!(reports.len(), 1); + assert_eq!(reports[0].events_new, 3); + assert!( + reports[0].bucket_id.contains("-synced-from-"), + "pull destination id should carry provenance, got {}", + reports[0].bucket_id + ); + + let second = aw_sync::sync_datastores( + &state.ds_src, + &state.ds_dest, + false, + None, + &SyncSpec::default(), + ) + .unwrap(); + assert_eq!(second[0].events_new, 0, "second pull must be a no-op"); + } + /// On pull (is_push=false), the destination bucket should have $aw.sync.origin set to /// the source bucket's hostname. On push (is_push=true), $aw.sync.origin must NOT be /// written to the staging copy.