From dd68e0e5ba432c8ea3903cf92f53a97ac7f9757c Mon Sep 17 00:00:00 2001 From: Bob Date: Tue, 15 Sep 2026 21:30:02 +0000 Subject: [PATCH 1/2] feat(aw-sync): add `status` doctor command and fail-loud empty-pull warnings `aw-sync daemon` can walk a configured sync dir and stay silent when it finds nothing to pull, which is indistinguishable from a working setup. This does not change which remotes are pulled (ActivityWatch/aw-server-rust#682 / #685). It makes the miss diagnosable: - classify both 2-level (`{device_id}/*.db`) and 3-level (`{hostname}/{device_id}/*.db`) layouts without opening sqlite - warn! on pull when zero remotes are found, with skip reasons - `aw-sync status` prints every entry, inspects peer dbs read-only (no WAL sidecars), and flags duplicate device_id, hostname mismatch, unpublished staging, and unimported peers ActivityWatch/aw-server-rust#684 Git-Session-Id: 375e1ec2-04d0-5884-9beb-cc5cd9c704c6 --- Cargo.lock | 1 + aw-sync/Cargo.toml | 3 +- aw-sync/README.md | 3 + aw-sync/src/lib.rs | 4 + aw-sync/src/main.rs | 7 + aw-sync/src/status.rs | 244 ++++++++++++++++++ aw-sync/src/sync.rs | 27 +- aw-sync/src/util.rs | 559 ++++++++++++++++++++++++++++++++++++++++++ 8 files changed, 842 insertions(+), 6 deletions(-) create mode 100644 aw-sync/src/status.rs diff --git a/Cargo.lock b/Cargo.lock index a08dea78..89dc44ff 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -317,6 +317,7 @@ dependencies = [ "log", "openssl", "reqwest", + "rusqlite", "serde", "serde_json", "toml", diff --git a/aw-sync/Cargo.toml b/aw-sync/Cargo.toml index 74d825da..900c304c 100644 --- a/aw-sync/Cargo.toml +++ b/aw-sync/Cargo.toml @@ -26,6 +26,7 @@ gethostname = "0.4.3" # CLI-only dependencies (optional) clap = { version = "4.1", features = ["derive"], optional = true } ctrlc = { version = "3.4.5", optional = true } +rusqlite = { version = "0.30", features = ["bundled"], optional = true } aw-server = { path = "../aw-server" } aw-models = { path = "../aw-models" } @@ -41,4 +42,4 @@ android_logger = "0.13" [features] default = ["cli"] -cli = ["clap", "ctrlc"] +cli = ["clap", "ctrlc", "dep:rusqlite"] diff --git a/aw-sync/README.md b/aw-sync/README.md index e1d19071..69cebfc5 100644 --- a/aw-sync/README.md +++ b/aw-sync/README.md @@ -30,6 +30,9 @@ aw-sync daemon --mode pull # Sync all buckets once and exit aw-sync sync --start-date "2024-01-01" + +# Doctor: why is pull empty / which peers exist in the folder? +aw-sync status ``` For more options, see `aw-sync --help`. Some notable options: diff --git a/aw-sync/src/lib.rs b/aw-sync/src/lib.rs index 31bae33c..7cd0a570 100644 --- a/aw-sync/src/lib.rs +++ b/aw-sync/src/lib.rs @@ -18,6 +18,10 @@ mod accessmethod; pub use accessmethod::AccessMethod; mod dirs; +#[cfg(feature = "cli")] +mod status; +#[cfg(feature = "cli")] +pub use status::run_status; mod util; #[cfg(target_os = "android")] diff --git a/aw-sync/src/main.rs b/aw-sync/src/main.rs index 604b723b..6de26fb9 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 status; mod sync; mod sync_wrapper; mod util; @@ -132,6 +133,11 @@ enum Commands { }, /// List buckets and their sync status. List {}, + /// Doctor: classify every entry in the sync folder and say why pull is empty. + /// + /// Walks both the 2-level (`{device_id}/*.db`) and 3-level + /// (`{hostname}/{device_id}/*.db`) layouts. Does not create staging files. + Status {}, } fn parse_start_date(arg: &str) -> Result, chrono::ParseError> { @@ -286,6 +292,7 @@ fn main() -> Result<(), Box> { // List all buckets Commands::List {} => sync::list_buckets(&client)?, + Commands::Status {} => status::run_status(&client, &opts.host, port, &profile)?, } // Needed to give the datastores some time to commit before program is shut down. diff --git a/aw-sync/src/status.rs b/aw-sync/src/status.rs new file mode 100644 index 00000000..3feabf82 --- /dev/null +++ b/aw-sync/src/status.rs @@ -0,0 +1,244 @@ +//! `aw-sync status` — a doctor command for the sync folder. +//! +//! Read-only: never creates staging databases. Complements `list_buckets`, +//! which only sees remotes the 2-level daemon walk already finds. + +use std::collections::{BTreeMap, HashSet}; +use std::error::Error; +use std::io::{self, Write}; + +use aw_client_rust::blocking::AwClient; +use chrono::{DateTime, Utc}; + +use crate::util::{ + inspect_sync_db, scan_sync_dir, DbInspect, SyncDirEntry, SyncEntryKind, SyncLayout, +}; + +pub fn run_status( + client: &AwClient, + host: &str, + port: u16, + profile: &str, +) -> Result<(), Box> { + let report = collect_status(client, host, port, profile)?; + let mut stdout = io::stdout().lock(); + write!(stdout, "{report}")?; + Ok(()) +} + +pub fn collect_status( + client: &AwClient, + host: &str, + port: u16, + profile: &str, +) -> Result> { + let sync_dir = crate::dirs::get_sync_dir()?; + let server_info = client.get_info().ok(); + let local_hostname = server_info + .as_ref() + .map(|i| i.hostname.clone()) + .unwrap_or_else(|| client.hostname.clone()); + let device_id = server_info.as_ref().map(|i| i.device_id.clone()); + let server_ok = server_info.is_some(); + + let local_buckets = if server_ok { + client.get_buckets().unwrap_or_default() + } else { + Default::default() + }; + let imported_origins: HashSet = local_buckets + .keys() + .filter_map(|id| { + id.split("-synced-from-") + .nth(1) + .filter(|s| !s.is_empty()) + .map(str::to_string) + }) + .chain(local_buckets.values().filter_map(|b| { + b.data + .get("$aw.sync.origin") + .and_then(|v| v.as_str()) + .map(str::to_string) + })) + .collect(); + let local_newest = local_buckets + .values() + .filter_map(|b| b.metadata.end.or(b.last_updated)) + .max(); + + let mut inspected: Vec<(SyncDirEntry, Option>)> = Vec::new(); + for entry in scan_sync_dir(&sync_dir, device_id.as_deref())? { + let peek = entry.db_path.as_ref().map(|p| inspect_sync_db(p)); + inspected.push((entry, peek)); + } + + let mut out = String::new(); + out.push_str("aw-sync status\n"); + out.push_str("==============\n"); + out.push_str(&format!("sync dir: {}\n", sync_dir.display())); + out.push_str(&format!("profile: {profile}\n")); + out.push_str(&format!("local hostname: {local_hostname}\n")); + match &device_id { + Some(id) => out.push_str(&format!("device_id: {id}\n")), + None => out.push_str("device_id: (unknown — could not reach local server)\n"), + } + out.push_str(&format!( + "server: {host}:{port} {}\n", + if server_ok { + "reachable" + } else { + "UNREACHABLE" + } + )); + out.push('\n'); + + if inspected.is_empty() { + out.push_str("Entries: (none)\n"); + } else { + out.push_str("Entries:\n"); + for (entry, peek) in &inspected { + out.push_str(&entry.diagnostic_line()); + out.push('\n'); + if let Some(result) = peek { + match result { + Ok(info) => out.push_str(&format_inspect(info, &imported_origins)), + Err(e) => out.push_str(&format!(" inspect: {e}\n")), + } + } + } + } + + let warnings = collect_warnings(&inspected, local_newest.as_ref(), &imported_origins); + out.push('\n'); + if warnings.is_empty() { + out.push_str("Warnings: none\n"); + } else { + out.push_str("Warnings:\n"); + for w in warnings { + out.push_str(&format!(" ! {w}\n")); + } + } + + Ok(out) +} + +fn format_inspect(info: &DbInspect, imported_origins: &HashSet) -> String { + let mut s = String::new(); + if let Some(host) = &info.hostname { + s.push_str(&format!(" hostname (buckets): {host}\n")); + } + s.push_str(&format!( + " buckets: {} events: {} newest: {}\n", + info.bucket_count, + info.event_count, + info.newest_event + .map(|t| t.to_rfc3339()) + .unwrap_or_else(|| "-".to_string()) + )); + if let Some(host) = &info.hostname { + let imported = imported_origins.contains(host); + s.push_str(&format!( + " imported locally: {}\n", + if imported { "yes" } else { "no" } + )); + } + s +} + +fn collect_warnings( + inspected: &[(SyncDirEntry, Option>)], + local_newest: Option<&DateTime>, + imported_origins: &HashSet, +) -> Vec { + let mut warnings = Vec::new(); + + let has_two = inspected + .iter() + .any(|(e, _)| e.layout == Some(SyncLayout::TwoLevel) && e.db_path.is_some()); + let has_three = inspected + .iter() + .any(|(e, _)| e.layout == Some(SyncLayout::ThreeLevel) && e.db_path.is_some()); + if has_two && has_three { + warnings.push( + "sync folder has both 2-level ({device_id}/*.db) and 3-level \ + ({hostname}/{device_id}/*.db) databases; new daemon pushes should use the \ + 3-level host layout (ActivityWatch/aw-server-rust#682 / #685)" + .to_string(), + ); + } + + let mut by_device: BTreeMap> = BTreeMap::new(); + for (entry, _) in inspected { + if let (Some(did), Some(_)) = (&entry.device_id, &entry.db_path) { + by_device.entry(did.clone()).or_default().push(entry); + } + } + for (did, group) in &by_device { + if group.len() > 1 { + let folders: Vec = group + .iter() + .map(|e| { + e.hostname_folder + .clone() + .unwrap_or_else(|| e.path.display().to_string()) + }) + .collect(); + warnings.push(format!( + "device_id {did} appears under {} folders: {} — pulling both can truncate history (ActivityWatch/aw-server-rust#683)", + group.len(), + folders.join(", ") + )); + } + } + + for (entry, peek) in inspected { + let Some(Ok(info)) = peek else { continue }; + if let (Some(folder), Some(host)) = (&entry.hostname_folder, &info.hostname) { + if folder != host { + warnings.push(format!( + "folder name '{folder}' ≠ bucket hostname '{host}' ({})", + entry.path.display() + )); + } + } + if entry.kind == SyncEntryKind::Peer { + if let Some(host) = &info.hostname { + if !imported_origins.contains(host) { + warnings.push(format!( + "peer {} ({}) has not been imported locally", + entry.device_id.as_deref().unwrap_or("?"), + host + )); + } + } + } + if entry.kind == SyncEntryKind::OwnStaging { + if let (Some(staging_newest), Some(local)) = (info.newest_event, local_newest) { + if staging_newest < *local { + warnings.push(format!( + "own staging newest event ({}) is older than local server newest ({}) — this device may not be publishing", + staging_newest.to_rfc3339(), + local.to_rfc3339() + )); + } + } + } + if info.buckets.iter().any(|b| b.id.contains("-synced-from-")) { + warnings.push(format!( + "{} still contains re-exported …-synced-from-… buckets (leftover from before ActivityWatch/aw-server-rust#648)", + entry.path.display() + )); + } + } + + if inspected + .iter() + .any(|(e, _)| e.db_path.as_ref().is_some_and(|p| p.ends_with("test.db"))) + { + warnings.push( + "staging databases are still named test.db — leftover test scaffolding in a directory users are told to inspect".to_string(), + ); + } + + warnings +} diff --git a/aw-sync/src/sync.rs b/aw-sync/src/sync.rs index eabcce06..31b7e9d0 100644 --- a/aw-sync/src/sync.rs +++ b/aw-sync/src/sync.rs @@ -89,12 +89,29 @@ 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. + if mode == SyncMode::Pull || mode == SyncMode::Both { + for line in crate::util::pull_discovery_warnings( + sync_spec.path.as_path(), + device_id, + &remote_dbfiles, + ) { + warn!("{line}"); + } + } + // TODO: Check for compatible remote db version before opening - let ds_remotes: Vec = remote_dbfiles - .iter() - .map(|p| p.as_path()) - .map(create_datastore) - .collect::, _>>()?; + let mut ds_remotes = Vec::new(); + for path in &remote_dbfiles { + match create_datastore(path) { + Ok(ds) => ds_remotes.push(ds), + Err(e) => { + warn!("Failed to open remote db {}: {e}", path.display()); + return Err(e.into()); + } + } + } if !ds_remotes.is_empty() { info!( diff --git a/aw-sync/src/util.rs b/aw-sync/src/util.rs index 1caf5950..5649289a 100644 --- a/aw-sync/src/util.rs +++ b/aw-sync/src/util.rs @@ -480,3 +480,562 @@ fn select_db_paths_by_device_id(paths: Vec) -> Vec { .map(|d| d.path) .collect() } + +/// How a database sits in the sync folder. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SyncLayout { + TwoLevel, + ThreeLevel, +} + +impl SyncLayout { + pub fn as_str(self) -> &'static str { + match self { + SyncLayout::TwoLevel => "2-level", + SyncLayout::ThreeLevel => "3-level", + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SyncEntryKind { + Peer, + OwnStaging, + Unrecognised, +} + +impl SyncEntryKind { + pub fn as_str(self) -> &'static str { + match self { + SyncEntryKind::Peer => "peer", + SyncEntryKind::OwnStaging => "own-staging", + SyncEntryKind::Unrecognised => "unrecognised", + } + } +} + +/// One filesystem entry in the sync directory, classified without opening sqlite. +#[derive(Debug, Clone)] +pub struct SyncDirEntry { + pub path: PathBuf, + pub db_path: Option, + pub db_size: Option, + pub layout: Option, + pub kind: SyncEntryKind, + pub hostname_folder: Option, + pub device_id: Option, + /// Why this path is not a pull remote (own staging, junk, empty folder). + pub not_visible_to_daemon: Option, +} + +impl SyncDirEntry { + pub fn diagnostic_line(&self) -> String { + let size = self + .db_size + .map(format_bytes) + .unwrap_or_else(|| "-".to_string()); + let layout = self.layout.map(|l| l.as_str()).unwrap_or("-"); + let mut line = format!( + " [{:<12} {layout}] {} {size}", + self.kind.as_str(), + self.path.display() + ); + if let Some(reason) = &self.not_visible_to_daemon { + line.push_str(&format!("\n not pulled: {reason}")); + } + line + } +} + +/// Walk both known sync-folder layouts and classify every entry. +/// +/// Does not open sqlite files and does not create directories. `local_device_id` +/// is used to tell own staging copies from peers; pass `None` when unknown. +pub fn scan_sync_dir( + sync_directory: &Path, + local_device_id: Option<&str>, +) -> std::io::Result> { + let mut entries = Vec::new(); + if !sync_directory.exists() { + return Ok(entries); + } + + for child in fs::read_dir(sync_directory)? { + let child = child?; + let path = child.path(); + if path.is_file() { + entries.push(file_at_root(path)); + continue; + } + if !path.is_dir() { + continue; + } + classify_top_dir(&path, local_device_id, &mut entries)?; + } + + entries.sort_by(|a, b| a.path.cmp(&b.path)); + Ok(entries) +} + +fn file_at_root(path: PathBuf) -> SyncDirEntry { + let is_db = path.extension().unwrap_or_else(|| OsStr::new("")) == "db"; + let db_size = fs::metadata(&path).ok().map(|m| m.len()); + SyncDirEntry { + db_path: is_db.then(|| path.clone()), + db_size: is_db.then_some(db_size).flatten(), + layout: None, + kind: SyncEntryKind::Unrecognised, + hostname_folder: None, + device_id: None, + not_visible_to_daemon: Some(if is_db { + "database at sync root; expected {device_id}/*.db or {hostname}/{device_id}/*.db" + .to_string() + } else { + "not a database".to_string() + }), + path, + } +} + +fn classify_top_dir( + dir: &Path, + local_device_id: Option<&str>, + entries: &mut Vec, +) -> std::io::Result<()> { + let top_name = file_name_string(dir); + let mut db_files = Vec::new(); + let mut subdirs = Vec::new(); + for child in fs::read_dir(dir)? { + let child = child?; + let path = child.path(); + if path.is_dir() { + subdirs.push(path); + } else if path.extension().unwrap_or_else(|| OsStr::new("")) == "db" { + db_files.push(path); + } + } + + for db in &db_files { + entries.push(db_entry( + db, + SyncLayout::TwoLevel, + None, + top_name.clone(), + local_device_id, + None, + )?); + } + + let mut found_three_level = false; + for sub in &subdirs { + let device_id = file_name_string(sub); + let mut sub_dbs = Vec::new(); + for child in fs::read_dir(sub)? { + let path = child?.path(); + if path.extension().unwrap_or_else(|| OsStr::new("")) == "db" { + sub_dbs.push(path); + } + } + if sub_dbs.is_empty() { + entries.push(SyncDirEntry { + path: sub.clone(), + db_path: None, + db_size: None, + layout: Some(SyncLayout::ThreeLevel), + kind: SyncEntryKind::Unrecognised, + hostname_folder: top_name.clone(), + device_id, + not_visible_to_daemon: Some("hostname/device_id folder with no .db".to_string()), + }); + continue; + } + found_three_level = true; + for db in sub_dbs { + entries.push(db_entry( + &db, + SyncLayout::ThreeLevel, + top_name.clone(), + device_id.clone(), + local_device_id, + None, + )?); + } + } + + if db_files.is_empty() && !found_three_level && subdirs.is_empty() { + entries.push(SyncDirEntry { + path: dir.to_path_buf(), + db_path: None, + db_size: None, + layout: None, + kind: SyncEntryKind::Unrecognised, + hostname_folder: None, + device_id: top_name, + not_visible_to_daemon: Some("directory with no database".to_string()), + }); + } + Ok(()) +} + +fn db_entry( + db: &Path, + layout: SyncLayout, + hostname_folder: Option, + device_id: Option, + local_device_id: Option<&str>, + not_visible_to_daemon: Option, +) -> std::io::Result { + let db_size = fs::metadata(db).ok().map(|m| m.len()); + let own = match (local_device_id, device_id.as_deref()) { + (Some(local), Some(did)) => local == did, + (Some(local), None) => db.to_string_lossy().contains(local), + _ => false, + }; + let mut not_visible = not_visible_to_daemon; + if own && not_visible.is_none() { + not_visible = Some("own device_id, excluded from pull".to_string()); + } + Ok(SyncDirEntry { + path: db.to_path_buf(), + db_path: Some(db.to_path_buf()), + db_size, + layout: Some(layout), + kind: if own { + SyncEntryKind::OwnStaging + } else { + SyncEntryKind::Peer + }, + hostname_folder, + device_id, + not_visible_to_daemon: not_visible, + }) +} + +fn file_name_string(path: &Path) -> Option { + path.file_name() + .and_then(|n| n.to_str()) + .map(str::to_string) +} + +pub fn format_bytes(n: u64) -> String { + const UNITS: [&str; 5] = ["B", "KB", "MB", "GB", "TB"]; + let mut value = n as f64; + let mut unit = 0; + while value >= 1024.0 && unit < UNITS.len() - 1 { + value /= 1024.0; + unit += 1; + } + if unit == 0 { + format!("{n} {}", UNITS[unit]) + } else { + format!("{value:.1} {}", UNITS[unit]) + } +} + +/// Warn-lines for a pull that found no remotes, or that missed classified peers. +pub fn pull_discovery_warnings( + sync_directory: &Path, + local_device_id: &str, + found_remotes: &[PathBuf], +) -> Vec { + let scan = match scan_sync_dir(sync_directory, Some(local_device_id)) { + Ok(entries) => entries, + Err(e) => { + return vec![format!( + "Could not scan sync dir {}: {e}", + sync_directory.display() + )] + } + }; + + let mut lines = Vec::new(); + let found: std::collections::HashSet<&PathBuf> = found_remotes.iter().collect(); + let missed: Vec<&SyncDirEntry> = scan + .iter() + .filter(|e| { + e.kind == SyncEntryKind::Peer && e.db_path.as_ref().is_some_and(|p| !found.contains(p)) + }) + .collect(); + + if found_remotes.is_empty() { + lines.push(format!( + "Found 0 remote db files to pull from {}. \ + Zero peers in a configured sync dir is usually a layout or setup problem, not a no-op.", + sync_directory.display() + )); + if scan.is_empty() { + lines.push( + " (sync directory is empty aside from what this process creates)".to_string(), + ); + } else { + for entry in &scan { + lines.push(entry.diagnostic_line()); + } + } + } else if !missed.is_empty() { + lines.push(format!( + "Found {} peer db(s) that pull did not select:", + missed.len() + )); + for entry in missed { + lines.push(entry.diagnostic_line()); + } + } + + let mut by_device: std::collections::BTreeMap> = + std::collections::BTreeMap::new(); + for entry in scan + .iter() + .filter(|e| e.device_id.is_some() && e.db_path.is_some()) + { + if let Some(did) = &entry.device_id { + by_device.entry(did.clone()).or_default().push(entry); + } + } + for (did, group) in by_device { + if group.len() > 1 { + let folders: Vec = group + .iter() + .map(|e| { + e.hostname_folder + .clone() + .unwrap_or_else(|| e.path.display().to_string()) + }) + .collect(); + lines.push(format!( + "device_id {did} appears under {} folders: {} — pulling both can truncate history (see ActivityWatch/aw-server-rust#683)", + group.len(), + folders.join(", ") + )); + } + } + + lines +} + +/// Read-only peek at a peer sqlite file (bucket hostname, counts, newest event). +/// +/// Opens with `SQLITE_OPEN_READ_ONLY` so a doctor command does not create +/// `-wal`/`-shm` sidecars in a Syncthing folder. +#[cfg(feature = "cli")] +#[derive(Debug, Clone)] +pub struct DbInspect { + pub hostname: Option, + pub bucket_count: usize, + pub event_count: i64, + pub newest_event: Option>, + pub buckets: Vec, +} + +#[cfg(feature = "cli")] +#[derive(Debug, Clone)] +pub struct BucketInspect { + pub id: String, +} + +#[cfg(feature = "cli")] +pub fn inspect_sync_db(path: &Path) -> Result { + use rusqlite::{Connection, OpenFlags}; + + let conn = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY) + .map_err(|e| format!("failed to open {} read-only: {e}", path.display()))?; + + let mut stmt = conn + .prepare( + "SELECT buckets.name, buckets.hostname, + COUNT(events.id), + MAX(events.endtime) + FROM buckets + LEFT JOIN events ON events.bucketrow = buckets.id + GROUP BY buckets.id", + ) + .map_err(|e| format!("schema query failed on {}: {e}", path.display()))?; + + let rows = stmt + .query_map([], |row| { + let id: String = row.get(0)?; + let hostname: String = row.get(1)?; + let events: i64 = row.get(2)?; + let last_ns: Option = row.get(3)?; + Ok((id, hostname, events, last_ns)) + }) + .map_err(|e| e.to_string())?; + + let mut buckets = Vec::new(); + let mut hostname = None; + let mut event_count = 0i64; + let mut newest_ns: Option = None; + for row in rows { + let (id, host, events, last_ns) = row.map_err(|e| e.to_string())?; + event_count += events; + if hostname.is_none() && !host.is_empty() && host != "unknown" { + hostname = Some(host.clone()); + } + if let Some(ns) = last_ns { + newest_ns = Some(newest_ns.map_or(ns, |cur| cur.max(ns))); + } + buckets.push(BucketInspect { id }); + } + + Ok(DbInspect { + hostname, + bucket_count: buckets.len(), + event_count, + newest_event: newest_ns.and_then(ns_to_datetime), + buckets, + }) +} + +#[cfg(feature = "cli")] +fn ns_to_datetime(ns: i64) -> Option> { + let seconds = ns / 1_000_000_000; + let subnanos = (ns % 1_000_000_000) as u32; + chrono::DateTime::from_timestamp(seconds, subnanos) +} + +#[cfg(all(test, not(target_os = "android")))] +mod scan_tests { + use super::*; + use std::fs; + use std::time::{SystemTime, UNIX_EPOCH}; + + fn temp_sync_dir() -> PathBuf { + let path = std::env::temp_dir().join(format!( + "aw-sync-scan-{}-{}", + std::process::id(), + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + fs::create_dir_all(&path).unwrap(); + path + } + + fn touch_db(path: &Path, size: usize) { + fs::create_dir_all(path.parent().unwrap()).unwrap(); + fs::write(path, vec![0u8; size]).unwrap(); + } + + #[test] + fn classifies_both_layouts_and_flags_duplicate_device_id() { + let root = temp_sync_dir(); + let local = "d7bc68e7-aaaa-bbbb-cccc-dddddddddddd"; + let peer = "41662faa-aaaa-bbbb-cccc-dddddddddddd"; + + touch_db(&root.join(local).join("test.db"), 64); + touch_db(&root.join("poco_f8_ultra").join(peer).join("test.db"), 273); + touch_db(&root.join("POCO F8 Ultra").join(peer).join("test.db"), 9); + fs::write(root.join("readme.txt"), "noise").unwrap(); + + let scan = scan_sync_dir(&root, Some(local)).unwrap(); + let kinds: Vec<_> = scan.iter().map(|e| (e.kind, e.layout)).collect(); + assert!( + kinds + .iter() + .any(|(k, l)| *k == SyncEntryKind::OwnStaging && *l == Some(SyncLayout::TwoLevel)), + "own 2-level staging: {kinds:?}" + ); + let three_level_peers: Vec<_> = scan + .iter() + .filter(|e| e.kind == SyncEntryKind::Peer && e.layout == Some(SyncLayout::ThreeLevel)) + .collect(); + assert_eq!(three_level_peers.len(), 2); + assert!(three_level_peers + .iter() + .all(|e| e.not_visible_to_daemon.is_none())); + + let warnings = pull_discovery_warnings(&root, local, &[]); + let joined = warnings.join("\n"); + assert!( + joined.contains("Found 0 remote db files"), + "empty-pull warning: {joined}" + ); + assert!( + joined.contains("3-level"), + "must list 3-level peers in the dump: {joined}" + ); + assert!( + joined.contains(peer), + "must name the duplicated device_id: {joined}" + ); + assert!( + joined.contains("appears under 2 folders"), + "duplicate device_id warning: {joined}" + ); + + let _ = fs::remove_dir_all(root); + } + + #[test] + fn two_level_peer_is_selected_when_found() { + let root = temp_sync_dir(); + let local = "local-device"; + let peer = "peer-device"; + touch_db(&root.join(peer).join("test.db"), 8); + + let scan = scan_sync_dir(&root, Some(local)).unwrap(); + let peer_entry = scan + .iter() + .find(|e| e.kind == SyncEntryKind::Peer) + .expect("peer"); + assert_eq!(peer_entry.layout, Some(SyncLayout::TwoLevel)); + 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); + assert!( + warnings + .iter() + .all(|l| !l.contains("Found 0 remote db files")), + "should not warn about empty remotes when a 2-level peer was found: {warnings:?}" + ); + + let _ = fs::remove_dir_all(root); + } + + #[cfg(feature = "cli")] + #[test] + fn inspect_sync_db_reads_hostname_and_newest_event() { + let root = temp_sync_dir(); + let db_path = root.join("peer").join("test.db"); + fs::create_dir_all(db_path.parent().unwrap()).unwrap(); + + { + use rusqlite::Connection; + let conn = Connection::open(&db_path).unwrap(); + conn.execute_batch( + "CREATE TABLE buckets ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT UNIQUE NOT NULL, + type TEXT NOT NULL, + client TEXT NOT NULL, + hostname TEXT NOT NULL, + created TEXT NOT NULL, + data TEXT NOT NULL DEFAULT '{}' + ); + CREATE TABLE events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + bucketrow INTEGER NOT NULL, + starttime INTEGER NOT NULL, + endtime INTEGER NOT NULL, + data TEXT NOT NULL + ); + INSERT INTO buckets (name, type, client, hostname, created, data) + VALUES ('aw-watcher-android', 'currentwindow', 'aw-android', + 'POCO F8 Ultra', '2021-01-01T00:00:00Z', '{}'); + INSERT INTO events (bucketrow, starttime, endtime, data) + VALUES (1, 1000000000, 2000000000, '{}');", + ) + .unwrap(); + } + + let info = inspect_sync_db(&db_path).unwrap(); + assert_eq!(info.hostname.as_deref(), Some("POCO F8 Ultra")); + assert_eq!(info.bucket_count, 1); + assert_eq!(info.event_count, 1); + assert!(info.newest_event.is_some()); + + let _ = fs::remove_dir_all(root); + } +} From 42d6128a5f74a66c623cd92079859ea3e8625374 Mon Sep 17 00:00:00 2001 From: Bob Date: Wed, 16 Sep 2026 07:21:26 +0000 Subject: [PATCH 2/2] refactor(aw-sync): build status scan on RemoteDb walker `aw-sync status` was a third directory walk beside list_remote_dbs (#686) and collect_db_files (#685). 3-level peers now come from the same list_remote_dbs + select_remote_dbs_by_device_id pair pull_all uses, so duplicate-device_id "not pulled" matches the pull path. Leftover 2-level / unrecognised entries sit on top of that list. ActivityWatch/aw-server-rust#687 Git-Session-Id: eca21d8a-f6da-5096-b28f-783778f4c09f --- aw-sync/src/main.rs | 5 +- aw-sync/src/status.rs | 4 +- aw-sync/src/util.rs | 153 +++++++++++++++++++++++++++++++++++++----- 3 files changed, 140 insertions(+), 22 deletions(-) diff --git a/aw-sync/src/main.rs b/aw-sync/src/main.rs index 6de26fb9..c8433d1b 100644 --- a/aw-sync/src/main.rs +++ b/aw-sync/src/main.rs @@ -135,8 +135,9 @@ enum Commands { List {}, /// Doctor: classify every entry in the sync folder and say why pull is empty. /// - /// Walks both the 2-level (`{device_id}/*.db`) and 3-level - /// (`{hostname}/{device_id}/*.db`) layouts. Does not create staging files. + /// 3-level peers come from the same `RemoteDb` walker `pull_all` uses; + /// 2-level leftovers and unrecognised entries sit on top of that list. + /// Does not create staging files. Status {}, } diff --git a/aw-sync/src/status.rs b/aw-sync/src/status.rs index 3feabf82..5323c99b 100644 --- a/aw-sync/src/status.rs +++ b/aw-sync/src/status.rs @@ -1,7 +1,7 @@ //! `aw-sync status` — a doctor command for the sync folder. //! -//! Read-only: never creates staging databases. Complements `list_buckets`, -//! which only sees remotes the 2-level daemon walk already finds. +//! Read-only: never creates staging databases. 3-level peers are the +//! `RemoteDb` list `pull_all` uses; classification of leftovers sits on top. use std::collections::{BTreeMap, HashSet}; use std::error::Error; diff --git a/aw-sync/src/util.rs b/aw-sync/src/util.rs index 5649289a..29c1a856 100644 --- a/aw-sync/src/util.rs +++ b/aw-sync/src/util.rs @@ -1,4 +1,4 @@ -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::error::Error; use std::ffi::OsStr; use std::fs; @@ -547,7 +547,12 @@ impl SyncDirEntry { } } -/// Walk both known sync-folder layouts and classify every entry. +/// Classify the sync folder for `aw-sync status` and empty-pull warnings. +/// +/// 3-level databases come from [`list_remote_dbs`] + [`select_remote_dbs_by_device_id`] +/// — the same pair `pull_all` uses — so duplicate-`device_id` "not pulled" reasons +/// match the pull path. Status-only overlay on top of that list: 2-level leftovers, +/// unrecognised files/dirs, own-staging vs peer, [`SyncLayout`]. /// /// Does not open sqlite files and does not create directories. `local_device_id` /// is used to tell own staging copies from peers; pass `None` when unknown. @@ -555,28 +560,90 @@ pub fn scan_sync_dir( sync_directory: &Path, local_device_id: Option<&str>, ) -> std::io::Result> { - let mut entries = Vec::new(); - if !sync_directory.exists() { - return Ok(entries); - } + let remotes = list_remote_dbs(sync_directory)?; + let selected = select_remote_dbs_by_device_id(remotes.clone()); + let selected_paths: HashSet = selected.into_iter().map(|d| d.path).collect(); + let remote_paths: HashSet = remotes.iter().map(|d| d.path.clone()).collect(); + let known_hosts: HashSet = remotes.iter().map(|d| d.hostname.clone()).collect(); - for child in fs::read_dir(sync_directory)? { - let child = child?; - let path = child.path(); - if path.is_file() { - entries.push(file_at_root(path)); - continue; - } - if !path.is_dir() { - continue; + let mut entries: Vec = remotes + .into_iter() + .map(|db| remote_db_to_entry(db, local_device_id, &selected_paths)) + .collect(); + + if sync_directory.exists() { + for child in fs::read_dir(sync_directory)? { + let child = child?; + let path = child.path(); + if path.is_file() { + entries.push(file_at_root(path)); + continue; + } + if !path.is_dir() { + continue; + } + let name = file_name_string(&path); + if name.as_ref().is_some_and(|n| known_hosts.contains(n)) { + // Host folder already walked by list_remote_dbs. Only pick + // leftover 2-level *.db files sitting beside device_id dirs. + for file in fs::read_dir(&path)? { + let fp = file?.path(); + if fp.is_file() + && fp.extension().unwrap_or_else(|| OsStr::new("")) == "db" + && !remote_paths.contains(&fp) + { + entries.push(db_entry( + &fp, + SyncLayout::TwoLevel, + None, + name.clone(), + local_device_id, + None, + )?); + } + } + continue; + } + classify_top_dir(&path, local_device_id, &mut entries)?; } - classify_top_dir(&path, local_device_id, &mut entries)?; } entries.sort_by(|a, b| a.path.cmp(&b.path)); Ok(entries) } +fn remote_db_to_entry( + db: RemoteDb, + local_device_id: Option<&str>, + selected_paths: &HashSet, +) -> SyncDirEntry { + let own = local_device_id == Some(db.device_id.as_str()); + let not_visible = if own { + Some("own device_id, excluded from pull".to_string()) + } else if !selected_paths.contains(&db.path) { + Some(format!( + "duplicate device_id {}; pull keeps the largest db only (ActivityWatch/aw-server-rust#683)", + db.device_id + )) + } else { + None + }; + SyncDirEntry { + path: db.path.clone(), + db_path: Some(db.path), + db_size: Some(db.size), + layout: Some(SyncLayout::ThreeLevel), + kind: if own { + SyncEntryKind::OwnStaging + } else { + SyncEntryKind::Peer + }, + hostname_folder: Some(db.hostname), + device_id: Some(db.device_id), + not_visible_to_daemon: not_visible, + } +} + fn file_at_root(path: PathBuf) -> SyncDirEntry { let is_db = path.extension().unwrap_or_else(|| OsStr::new("")) == "db"; let db_size = fs::metadata(&path).ok().map(|m| m.len()); @@ -941,9 +1008,28 @@ mod scan_tests { .filter(|e| e.kind == SyncEntryKind::Peer && e.layout == Some(SyncLayout::ThreeLevel)) .collect(); assert_eq!(three_level_peers.len(), 2); - assert!(three_level_peers + let kept: Vec<_> = three_level_peers .iter() - .all(|e| e.not_visible_to_daemon.is_none())); + .filter(|e| e.not_visible_to_daemon.is_none()) + .collect(); + let skipped: Vec<_> = three_level_peers + .iter() + .filter(|e| e.not_visible_to_daemon.is_some()) + .collect(); + assert_eq!( + kept.len(), + 1, + "pull-equivalent winner: {three_level_peers:?}" + ); + assert_eq!(skipped.len(), 1); + assert_eq!(kept[0].db_size, Some(273)); + assert!( + skipped[0] + .not_visible_to_daemon + .as_ref() + .is_some_and(|s| s.contains("duplicate device_id")), + "skipped duplicate must use the pull selector: {skipped:?}" + ); let warnings = pull_discovery_warnings(&root, local, &[]); let joined = warnings.join("\n"); @@ -967,6 +1053,37 @@ mod scan_tests { let _ = fs::remove_dir_all(root); } + #[test] + fn status_keeps_the_same_duplicate_winner_as_pull() { + let root = temp_sync_dir(); + let local = "local-device"; + let peer = "peer-device"; + touch_db(&root.join("poco_f8_ultra").join(peer).join("test.db"), 273); + touch_db(&root.join("POCO F8 Ultra").join(peer).join("test.db"), 9); + + let remotes = list_remote_dbs(&root).unwrap(); + let selected = select_remote_dbs_by_device_id(remotes); + let winner = selected + .iter() + .find(|d| d.device_id == peer) + .expect("pull selector keeps one db for the peer"); + + let scan = scan_sync_dir(&root, Some(local)).unwrap(); + let kept: Vec<_> = scan + .iter() + .filter(|e| { + e.kind == SyncEntryKind::Peer + && e.layout == Some(SyncLayout::ThreeLevel) + && e.not_visible_to_daemon.is_none() + }) + .collect(); + assert_eq!(kept.len(), 1); + assert_eq!(kept[0].db_path.as_ref(), Some(&winner.path)); + assert_eq!(kept[0].db_size, Some(winner.size)); + + let _ = fs::remove_dir_all(root); + } + #[test] fn two_level_peer_is_selected_when_found() { let root = temp_sync_dir();