From 3cf04345f0e7bd6a16230835887baaa05ee5d3db Mon Sep 17 00:00:00 2001 From: Bob Date: Wed, 16 Sep 2026 08:28:27 +0000 Subject: [PATCH 1/9] feat(aw-sync): return and persist a SyncReport from each pass A successful pass used to return (). JNI, aw-sync status, and the daemon had nothing to report. SyncReport carries per-peer/per-bucket counts, including duplicate-device_id skips, and is persisted locally (not in the Syncthing folder). Closes #695 Git-Session-Id: 94e08188-0a05-5fed-b8c0-fd327ae1f803 --- aw-sync/Cargo.toml | 2 +- aw-sync/README.md | 1 + aw-sync/src/android.rs | 46 ++-- aw-sync/src/lib.rs | 6 + aw-sync/src/main.rs | 33 ++- aw-sync/src/report.rs | 429 ++++++++++++++++++++++++++++++++++++ aw-sync/src/status.rs | 15 ++ aw-sync/src/sync.rs | 104 ++++++--- aw-sync/src/sync_wrapper.rs | 60 +++-- aw-sync/src/util.rs | 109 ++++++++- aw-sync/tests/sync.rs | 33 +++ 11 files changed, 743 insertions(+), 95 deletions(-) create mode 100644 aw-sync/src/report.rs 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..fc3c3283 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,15 @@ 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()) + report.merge( + push_with_hostname(&client, &hostname_str) + .map_err(|e| 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..b8563f5e 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,33 @@ 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)?; + report.merge(sync_wrapper::pull(host, &client)?); } } None => { info!("Pulling from all hosts"); - sync_wrapper::pull_all(&client)?; + report.merge(sync_wrapper::pull_all(&client)?); } } // Push info!("Pushing local data"); - sync_wrapper::push(&client)? + report.merge(sync_wrapper::push(&client)?); + report.finish(); + info!("{}", report.summary_message()); + aw_sync_persist(&report); } } @@ -337,9 +346,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 +372,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..f52b0a3b --- /dev/null +++ b/aw-sync/src/report.rs @@ -0,0 +1,429 @@ +//! 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, +} + +#[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![], + } + } + + 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); + } + + 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(); + 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" + ), + } + } + + /// 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, + }) + .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:")?; + for peer in &self.peers { + writeln!(f, " {}", format_peer_line(peer))?; + } + } + 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)?; + } + let tmp = path.with_extension("json.tmp"); + 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")); + } +} 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..a620d40d 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 @@ -104,17 +103,35 @@ pub fn sync_run( ) { 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. let mut ds_remotes = Vec::new(); - for path in &remote_dbfiles { - match open_peer_datastore(path) { - Ok(Some(ds)) => ds_remotes.push(ds), + for db in &selection.selected { + match open_peer_datastore(&db.path) { + Ok(Some(ds)) => ds_remotes.push((db.clone(), ds)), Ok(None) => {} 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); + } return Err(e.into()); } } @@ -124,26 +141,51 @@ 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); + return Err(e.into()); + } + } } } // Push local server buckets to sync folder if let Some(ds_localremote) = &ds_localremote { info!("Pushing..."); - sync_datastores(client, ds_localremote, true, Some(device_id), sync_spec)?; + match sync_datastores(client, ds_localremote, true, Some(device_id), sync_spec) { + Ok(buckets) => report.pushed = buckets, + Err(e) => { + report.finish(); + crate::report::persist_last_report_warn(&report); + return Err(e.into()); + } + } } // Close open database connections - for ds_from in &ds_remotes { + for (_, ds_from) in &ds_remotes { ds_from.close(); } if let Some(ds_localremote) = &ds_localremote { @@ -159,7 +201,8 @@ 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) } #[allow(dead_code)] @@ -375,7 +418,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 +491,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 +507,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 +672,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..a83a259d 100644 --- a/aw-sync/src/sync_wrapper.rs +++ b/aw-sync/src/sync_wrapper.rs @@ -2,32 +2,57 @@ 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(), + )); + } + 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(); + 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 +86,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 +96,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 +114,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. 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. From 564569e9d932456067df3c9024694597601224bc Mon Sep 17 00:00:00 2001 From: Bob Date: Wed, 16 Sep 2026 08:42:26 +0000 Subject: [PATCH 2/9] feat(aw-sync): persist discovery warnings on zero-peer SyncReport The #682 failure mode was silence: a pass that found nobody logged warnings and threw them away. Capture them on the report so `peers: []` plus the diagnosis survive in last-sync-report.json. Also persist pull_all's aggregate on a later-peer failure, and use a unique temp file for the atomic write. Git-Session-Id: 2a2cd4fe-793a-5f03-bfbc-7da3c8ff9738 --- aw-sync/src/report.rs | 140 +++++++++++++++++++++++++++++++++++- aw-sync/src/sync.rs | 9 ++- aw-sync/src/sync_wrapper.rs | 9 +++ 3 files changed, 151 insertions(+), 7 deletions(-) diff --git a/aw-sync/src/report.rs b/aw-sync/src/report.rs index f52b0a3b..e22f785c 100644 --- a/aw-sync/src/report.rs +++ b/aw-sync/src/report.rs @@ -45,6 +45,10 @@ pub struct SyncReport { 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)] @@ -80,6 +84,7 @@ impl SyncReport { mode, peers: vec![], pushed: vec![], + warnings: vec![], } } @@ -96,6 +101,16 @@ impl SyncReport { } 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); + } } pub fn events_new(&self) -> i64 { @@ -182,6 +197,7 @@ impl SyncReport { "peers_failed": self.peers_failed(), "peers": self.peers, "pushed": self.pushed, + "warnings": self.warnings, }) .to_string() } @@ -245,12 +261,20 @@ impl fmt::Display for SyncReport { self.events_pushed(), self.events_new() )?; - if !self.peers.is_empty() { + 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 { @@ -312,7 +336,23 @@ pub fn persist_last_report_to(report: &SyncReport, path: &Path) -> Result<(), Bo if let Some(parent) = path.parent() { fs::create_dir_all(parent)?; } - let tmp = path.with_extension("json.tmp"); + // 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(()) @@ -426,4 +466,100 @@ mod tests { 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, + "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")); + } + + #[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/sync.rs b/aw-sync/src/sync.rs index a620d40d..cd7cc0cf 100644 --- a/aw-sync/src/sync.rs +++ b/aw-sync/src/sync.rs @@ -94,15 +94,14 @@ 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, &remote_dbfiles, - ) { - warn!("{line}"); - } + )); for skipped in &selection.skipped { report.peers.push(PeerReport::skipped( skipped.db.device_id.clone(), diff --git a/aw-sync/src/sync_wrapper.rs b/aw-sync/src/sync_wrapper.rs index a83a259d..9abdbc79 100644 --- a/aw-sync/src/sync_wrapper.rs +++ b/aw-sync/src/sync_wrapper.rs @@ -19,6 +19,12 @@ pub fn pull_all(client: &AwClient) -> Result> { skipped.reason.clone(), )); } + let found: Vec<_> = selection.selected.iter().map(|d| d.path.clone()).collect(); + // No get_info() here: an empty-dir pass must stay a local filesystem + // check (the #682 case). Own-vs-peer classification is optional for + // the warning text; "" leaves entries unclassified rather than + // hanging if the server is down. + report.capture_warnings(crate::util::pull_discovery_warnings(&sync_root, "", &found)); if selection.selected.is_empty() { info!("No remote databases found in {:?}", sync_root); report.finish(); @@ -44,6 +50,9 @@ pub fn pull_all(client: &AwClient) -> Result> { 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); } } From 0c48acbfdfeaea815274a1d8946b29f9dd47eba0 Mon Sep 17 00:00:00 2001 From: Bob Date: Wed, 16 Sep 2026 12:17:37 +0000 Subject: [PATCH 3/9] fix(aw-sync): persist pull-phase report when push fails MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The simple host-based sync path and the Android syncBoth JNI entry both did `report.merge(push(...)?)` — the `?` discarded the merged pull report when push returned Err, so last-sync-report.json (and the JNI error payload) reflected only the push-phase report from inside sync_run. Persist the pull-phase result with a push warning before propagating. Rebased onto master (#698 landed); conflict in sync_run resolved by keeping maybe_setup_local_remote + selection-based remote discovery. Git-Session-Id: 2a2cd4fe-793a-5f03-bfbc-7da3c8ff9738 --- aw-sync/src/android.rs | 15 +++++++++++---- aw-sync/src/main.rs | 15 +++++++++++++-- 2 files changed, 24 insertions(+), 6 deletions(-) diff --git a/aw-sync/src/android.rs b/aw-sync/src/android.rs index fc3c3283..416c4d6d 100644 --- a/aw-sync/src/android.rs +++ b/aw-sync/src/android.rs @@ -349,10 +349,17 @@ pub extern "C" fn Java_net_activitywatch_android_SyncInterface_syncBoth( let mut report = pull_all(&client).map_err(|e| format!("Pull phase failed: {}", e))?; - report.merge( - push_with_hostname(&client, &hostname_str) - .map_err(|e| format!("Push phase failed: {}", e))?, - ); + // 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.warnings.push(format!("push failed: {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()) diff --git a/aw-sync/src/main.rs b/aw-sync/src/main.rs index b8563f5e..25a3a3b6 100644 --- a/aw-sync/src/main.rs +++ b/aw-sync/src/main.rs @@ -291,9 +291,20 @@ fn main() -> Result<(), Box> { } } - // 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"); - report.merge(sync_wrapper::push(&client)?); + match sync_wrapper::push(&client) { + Ok(push_report) => report.merge(push_report), + Err(e) => { + report.warnings.push(format!("push failed: {e}")); + report.finish(); + info!("{}", report.summary_message()); + aw_sync_persist(&report); + return Err(e); + } + } report.finish(); info!("{}", report.summary_message()); aw_sync_persist(&report); From fae8838c9db0f8229b590e26c28cd2d6ee458554 Mon Sep 17 00:00:00 2001 From: Bob Date: Wed, 16 Sep 2026 12:56:17 +0000 Subject: [PATCH 4/9] fix(aw-sync): record push failures on the persisted SyncReport The inner sync_run push-error path finished and persisted without a warning, so last-sync-report.json looked like a no-op success. Same string the simple-path and JNI callers already used. Git-Session-Id: 01a0aa45-67f2-7c32-9c50-447c12587192 --- aw-sync/src/android.rs | 2 +- aw-sync/src/main.rs | 2 +- aw-sync/src/report.rs | 53 ++++++++++++++++++++++++++++++++++++++++++ aw-sync/src/sync.rs | 1 + 4 files changed, 56 insertions(+), 2 deletions(-) diff --git a/aw-sync/src/android.rs b/aw-sync/src/android.rs index 416c4d6d..cb25b894 100644 --- a/aw-sync/src/android.rs +++ b/aw-sync/src/android.rs @@ -354,7 +354,7 @@ pub extern "C" fn Java_net_activitywatch_android_SyncInterface_syncBoth( match push_with_hostname(&client, &hostname_str) { Ok(push_report) => report.merge(push_report), Err(e) => { - report.warnings.push(format!("push failed: {e}")); + report.record_push_failure(&e); report.finish(); crate::report::persist_last_report_warn(&report); return Err(format!("Push phase failed: {}", e)); diff --git a/aw-sync/src/main.rs b/aw-sync/src/main.rs index 25a3a3b6..b7634f95 100644 --- a/aw-sync/src/main.rs +++ b/aw-sync/src/main.rs @@ -298,7 +298,7 @@ fn main() -> Result<(), Box> { match sync_wrapper::push(&client) { Ok(push_report) => report.merge(push_report), Err(e) => { - report.warnings.push(format!("push failed: {e}")); + report.record_push_failure(&e); report.finish(); info!("{}", report.summary_message()); aw_sync_persist(&report); diff --git a/aw-sync/src/report.rs b/aw-sync/src/report.rs index e22f785c..a0f246ba 100644 --- a/aw-sync/src/report.rs +++ b/aw-sync/src/report.rs @@ -113,6 +113,15 @@ impl SyncReport { } } + /// 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}")); + } + pub fn events_new(&self) -> i64 { self.peers .iter() @@ -543,6 +552,50 @@ mod tests { 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}" + ); + } + #[test] fn old_report_without_warnings_field_still_loads() { let path = temp_report_path(); diff --git a/aw-sync/src/sync.rs b/aw-sync/src/sync.rs index cd7cc0cf..c1992790 100644 --- a/aw-sync/src/sync.rs +++ b/aw-sync/src/sync.rs @@ -176,6 +176,7 @@ pub fn sync_run( match sync_datastores(client, ds_localremote, 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); return Err(e.into()); From cd3ea9b806388073279386cb77cf5a502d766a02 Mon Sep 17 00:00:00 2001 From: Bob Date: Wed, 16 Sep 2026 14:19:54 +0000 Subject: [PATCH 5/9] fix(aw-sync): surface warnings in SyncReport summary_message MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Daemon and JNI log this string. A push abort already lived in `warnings` and the Display impl, but summary_message still said "Pushed 0 new events" — a success-like line for a failed pass. Git-Session-Id: 01a0aa92-68eb-7062-9ad7-aca64833f8b1 --- aw-sync/src/report.rs | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/aw-sync/src/report.rs b/aw-sync/src/report.rs index a0f246ba..2ba26d18 100644 --- a/aw-sync/src/report.rs +++ b/aw-sync/src/report.rs @@ -171,7 +171,7 @@ impl SyncReport { let skipped = self.peers_skipped(); let failed = self.peers_failed(); let peer_n = self.peers.len(); - match self.mode { + let mut msg = match self.mode { SyncMode::Pull => format!( "Pulled {pulled} new events from {imported}/{peer_n} peers \ ({skipped} skipped, {failed} failed)" @@ -184,7 +184,14 @@ impl SyncReport { "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`), @@ -594,6 +601,11 @@ mod tests { 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() + ); } #[test] From 421029bee0da63de493c068301bc6943a8dbdf32 Mon Sep 17 00:00:00 2001 From: Bob Date: Wed, 16 Sep 2026 15:16:16 +0000 Subject: [PATCH 6/9] fix(aw-sync): treat empty local device_id as unknown in discovery warnings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pull_all passed "" into pull_discovery_warnings, which wrapped it as Some(""). That does not unclassify entries: leftover 2-level own staging became a Peer and showed up as "peer db(s) that pull did not select". Pass None instead, and skip 2-level leftovers from the missed-peer warning when the local id is unknown. Do not call get_info() here — reqwest's timeout is 120s; empty-dir diagnosis must stay a filesystem check (ActivityWatch/aw-server-rust#682). Git-Session-Id: 01a0aabd-c1af-70b2-807a-abd6ce4640cd --- aw-sync/src/report.rs | 2 +- aw-sync/src/sync.rs | 2 +- aw-sync/src/sync_wrapper.rs | 12 ++++--- aw-sync/src/util.rs | 66 ++++++++++++++++++++++++++++++++++--- 4 files changed, 70 insertions(+), 12 deletions(-) diff --git a/aw-sync/src/report.rs b/aw-sync/src/report.rs index 2ba26d18..8100906f 100644 --- a/aw-sync/src/report.rs +++ b/aw-sync/src/report.rs @@ -501,7 +501,7 @@ mod tests { let mut report = SyncReport::new(SyncMode::Pull); report.capture_warnings(crate::util::pull_discovery_warnings( &root, - "local-device", + Some("local-device"), &[], )); report.finish(); diff --git a/aw-sync/src/sync.rs b/aw-sync/src/sync.rs index c1992790..b225a8b5 100644 --- a/aw-sync/src/sync.rs +++ b/aw-sync/src/sync.rs @@ -99,7 +99,7 @@ pub fn sync_run( if mode == SyncMode::Pull || mode == SyncMode::Both { report.capture_warnings(crate::util::pull_discovery_warnings( sync_spec.path.as_path(), - device_id, + Some(device_id), &remote_dbfiles, )); for skipped in &selection.skipped { diff --git a/aw-sync/src/sync_wrapper.rs b/aw-sync/src/sync_wrapper.rs index 9abdbc79..b14d2542 100644 --- a/aw-sync/src/sync_wrapper.rs +++ b/aw-sync/src/sync_wrapper.rs @@ -20,11 +20,13 @@ pub fn pull_all(client: &AwClient) -> Result> { )); } let found: Vec<_> = selection.selected.iter().map(|d| d.path.clone()).collect(); - // No get_info() here: an empty-dir pass must stay a local filesystem - // check (the #682 case). Own-vs-peer classification is optional for - // the warning text; "" leaves entries unclassified rather than - // hanging if the server is down. - report.capture_warnings(crate::util::pull_discovery_warnings(&sync_root, "", &found)); + // 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); report.finish(); diff --git a/aw-sync/src/util.rs b/aw-sync/src/util.rs index 76e294c9..71ff00ab 100644 --- a/aw-sync/src/util.rs +++ b/aw-sync/src/util.rs @@ -885,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!( @@ -905,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(); @@ -1116,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"), @@ -1185,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() @@ -1196,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() { From f8374cd2f0f409e42383627623c046a32ee868b0 Mon Sep 17 00:00:00 2001 From: Bob Date: Wed, 16 Sep 2026 15:28:02 +0000 Subject: [PATCH 7/9] fix(aw-sync): close datastore workers on sync_run error returns Datastore::close() stops the sqlite worker thread. The success path already did that; the new persist-then-return error paths skipped it, so a long-lived daemon could leak connections across failed passes. Git-Session-Id: 01a0aabd-c1af-70b2-807a-abd6ce4640cd --- aw-sync/src/sync.rs | 30 ++++++++++++++++++++++-------- 1 file changed, 22 insertions(+), 8 deletions(-) diff --git a/aw-sync/src/sync.rs b/aw-sync/src/sync.rs index b225a8b5..fdfabeee 100644 --- a/aw-sync/src/sync.rs +++ b/aw-sync/src/sync.rs @@ -131,6 +131,7 @@ pub fn sync_run( report.finish(); crate::report::persist_last_report_warn(&report); } + close_opened_datastores(&ds_remotes, &ds_localremote); return Err(e.into()); } } @@ -164,6 +165,7 @@ pub fn sync_run( )); report.finish(); crate::report::persist_last_report_warn(&report); + close_opened_datastores(&ds_remotes, &ds_localremote); return Err(e.into()); } } @@ -171,26 +173,22 @@ pub fn sync_run( } // Push local server buckets to sync folder - if let Some(ds_localremote) = &ds_localremote { + if let Some(ds_local) = &ds_localremote { info!("Pushing..."); - match 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, @@ -205,6 +203,22 @@ pub fn sync_run( 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)] pub fn list_buckets(client: &AwClient) -> Result<(), Box> { let sync_directory = crate::dirs::get_sync_dir().map_err(|_| "Could not get sync dir")?; From e4e92ecb53ab1cb54d2c5d9272c9b9a4f8192ed5 Mon Sep 17 00:00:00 2001 From: Bob Date: Wed, 16 Sep 2026 16:21:50 +0000 Subject: [PATCH 8/9] fix(aw-sync): record incompatible-version peers as skipped on the SyncReport MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit open_peer_datastore returned Ok(None) on a user_version mismatch, so a peer rejected for an incompatible schema silently vanished from the report — an all-incompatible pass looked like a clean empty one in status and JNI (Greptile P1 on ActivityWatch/aw-server-rust#699). The mismatch reason is now carried back and each incompatible peer is recorded as skipped. Git-Session-Id: b1d605fd-5f81-5e6b-9293-7353ee952e7c --- aw-sync/src/sync.rs | 35 ++++++++++++++++++++++++++--------- 1 file changed, 26 insertions(+), 9 deletions(-) diff --git a/aw-sync/src/sync.rs b/aw-sync/src/sync.rs index fdfabeee..a09b6526 100644 --- a/aw-sync/src/sync.rs +++ b/aw-sync/src/sync.rs @@ -113,12 +113,23 @@ pub fn sync_run( } // 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 db in &selection.selected { match open_peer_datastore(&db.path) { - Ok(Some(ds)) => ds_remotes.push((db.clone(), ds)), - Ok(None) => {} + 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}", db.path.display()); if mode == SyncMode::Pull || mode == SyncMode::Both { @@ -234,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); } } @@ -297,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:?}", From 2fab2f1452bf0fcffdc2623464fe2676d439890b Mon Sep 17 00:00:00 2001 From: Bob Date: Wed, 16 Sep 2026 18:05:46 +0000 Subject: [PATCH 9/9] fix(aw-sync): persist aggregate report when a later host pull fails MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Simple host-based sync merges per-host pull reports. A later host's `?` dropped earlier hosts from last-sync-report.json, leaving only the failing host's inner persist. Catch, record, persist the aggregate, then propagate — same contract as the push-failure path. Git-Session-Id: ee496982-751b-50ab-a42d-742b7c76cf44 --- aw-sync/src/main.rs | 15 +++++++++- aw-sync/src/report.rs | 65 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 79 insertions(+), 1 deletion(-) diff --git a/aw-sync/src/main.rs b/aw-sync/src/main.rs index b7634f95..dae3608c 100644 --- a/aw-sync/src/main.rs +++ b/aw-sync/src/main.rs @@ -282,7 +282,20 @@ fn main() -> Result<(), Box> { Some(hosts) => { for host in hosts.iter() { info!("Pulling from host: {}", host); - report.merge(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 => { diff --git a/aw-sync/src/report.rs b/aw-sync/src/report.rs index 8100906f..d651c540 100644 --- a/aw-sync/src/report.rs +++ b/aw-sync/src/report.rs @@ -122,6 +122,17 @@ impl SyncReport { 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() @@ -608,6 +619,60 @@ mod tests { ); } + /// 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();