From 32b28ee7b52904c33d55ff31f94606cd298f357f Mon Sep 17 00:00:00 2001 From: Bob Date: Fri, 18 Sep 2026 16:22:40 +0000 Subject: [PATCH 1/4] feat(aw-sync): add dedupe subcommand for pre-#713 -synced-from- duplicates aw-sync sync/daemon pull passes before #711/#713 re-imported the resume-boundary event on every pass, so -synced-from- buckets that pulled before that fix carry hundreds of exact-duplicate copies of the same event. Add `aw-sync dedupe [--bucket ...] [--dry-run]`: for every -synced-from- bucket (optionally filtered), collapse events with identical (timestamp, duration, data) down to the lowest-id (first-imported) copy, reporting counts per bucket. Never touches first-hand (non-synced) buckets. Android staging db cleanup (the phone's own test.db before push) is a separate follow-up on the aw-android side. Refs #717 Git-Session-Id: eb7a54b0-3d37-51f0-acba-6c4a02daea1e --- aw-sync/README.md | 7 ++ aw-sync/src/dedupe.rs | 192 ++++++++++++++++++++++++++++++++++++++++++ aw-sync/src/main.rs | 20 +++++ aw-sync/src/sync.rs | 2 +- 4 files changed, 220 insertions(+), 1 deletion(-) create mode 100644 aw-sync/src/dedupe.rs diff --git a/aw-sync/README.md b/aw-sync/README.md index fd1d3985..f2cc8f9c 100644 --- a/aw-sync/README.md +++ b/aw-sync/README.md @@ -23,6 +23,13 @@ aw-sync sync # Doctor: why is pull empty / which peers exist in the folder? # Also prints the last persisted pass (peers, events, skips). aw-sync status + +# One-off cleanup: collapse exact-duplicate events that accumulated in +# `-synced-from-` buckets from pull passes before #713. Never touches a +# host's own first-hand buckets. +aw-sync dedupe --dry-run # report only +aw-sync dedupe # delete the extras (lowest-id copy of each is kept) +aw-sync dedupe --bucket aw-watcher-window-synced-from-host # restrict to one bucket ``` `aw-sync` / `aw-sync daemon` currently **does not pull** from the 3-level `{hostname}/{device_id}/` layout that Android and `aw-sync sync` write. The daemon stages `{device_id}/test.db` at the sync-folder root and only walks two directory levels, so those peers are invisible ([#682](https://github.com/ActivityWatch/aw-server-rust/issues/682)). Until that switch lands, use `aw-sync sync` (or a systemd/cron timer around it) — not the daemon. diff --git a/aw-sync/src/dedupe.rs b/aw-sync/src/dedupe.rs new file mode 100644 index 00000000..0860cf9f --- /dev/null +++ b/aw-sync/src/dedupe.rs @@ -0,0 +1,192 @@ +//! `aw-sync dedupe` — one-off cleanup of exact-duplicate events that +//! accumulated in `-synced-from-` buckets before #713 stopped new pull +//! passes from re-importing the resume-boundary event. +//! +//! `-synced-from-` is a reserved marker in aw-sync's own ID grammar (see +//! `sync::is_synced_bucket`); this only ever touches buckets carrying it, so +//! a user's own local (non-synced) buckets are never candidates. + +use std::collections::hash_map::Entry; +use std::collections::HashMap; +use std::error::Error; +use std::io::{self, Write}; + +use aw_client_rust::blocking::AwClient; +use aw_models::Event; + +/// Among `events`, return the ids of every event that is an exact duplicate +/// (same timestamp, duration and data — `Event`'s own `PartialEq`) of an +/// event with a lower id. The lowest-id copy of each group is kept. +/// +/// `get_events` has no ordering guarantee callers should rely on for this +/// (it's newest-first, clipped by query range), so this sorts by id itself +/// rather than trusting call order. +fn find_duplicate_ids(events: &[Event]) -> Vec { + let mut sorted: Vec<&Event> = events.iter().collect(); + sorted.sort_by_key(|e| e.id.unwrap_or(i64::MAX)); + + let mut seen: HashMap<(i64, i64, String), ()> = HashMap::new(); + let mut duplicate_ids = Vec::new(); + for event in sorted { + let Some(id) = event.id else { continue }; + let key = ( + event.timestamp.timestamp_nanos_opt().unwrap_or(0), + event.duration.num_nanoseconds().unwrap_or(0), + serde_json::to_string(&event.data).unwrap_or_default(), + ); + match seen.entry(key) { + Entry::Occupied(_) => duplicate_ids.push(id), + Entry::Vacant(e) => { + e.insert(()); + } + } + } + duplicate_ids +} + +pub struct BucketDedupeResult { + pub bucket_id: String, + pub total_events: usize, + pub duplicate_events: usize, +} + +pub fn run_dedupe( + client: &AwClient, + buckets_filter: Option>, + dry_run: bool, +) -> Result<(), Box> { + let buckets = client.get_buckets().map_err(|e| e.to_string())?; + let mut targets: Vec = buckets + .values() + .filter(|b| crate::sync::is_synced_bucket(b)) + .map(|b| b.id.clone()) + .collect(); + if let Some(filter) = &buckets_filter { + targets.retain(|id| filter.contains(id)); + for wanted in filter { + match buckets.get(wanted) { + None => warn!("--bucket {wanted}: no such bucket, skipping"), + Some(b) if !crate::sync::is_synced_bucket(b) => warn!( + "--bucket {wanted}: not a -synced-from- bucket, skipping (dedupe only touches synced buckets)" + ), + Some(_) => {} + } + } + } + targets.sort(); + + if targets.is_empty() { + info!("No -synced-from- buckets found, nothing to dedupe"); + return Ok(()); + } + + let mut results = Vec::new(); + for bucket_id in targets { + let events = client.get_events(&bucket_id, None, None, None)?; + let total_events = events.len(); + let duplicate_ids = find_duplicate_ids(&events); + let duplicate_events = duplicate_ids.len(); + + if duplicate_events > 0 && !dry_run { + for id in duplicate_ids { + client.delete_event(&bucket_id, id)?; + } + } + + results.push(BucketDedupeResult { + bucket_id, + total_events, + duplicate_events, + }); + } + + let mut stdout = io::stdout().lock(); + let total_duplicates: usize = results.iter().map(|r| r.duplicate_events).sum(); + for r in &results { + if r.duplicate_events > 0 { + writeln!( + stdout, + "{}: {} event(s), {} duplicate(s){}", + r.bucket_id, + r.total_events, + r.duplicate_events, + if dry_run { + " [dry-run, not deleted]" + } else { + "" + } + )?; + } + } + if total_duplicates == 0 { + writeln!(stdout, "No duplicates found in {} bucket(s)", results.len())?; + } else if dry_run { + writeln!( + stdout, + "Total: {total_duplicates} duplicate(s) found across {} bucket(s) [dry-run, nothing deleted — re-run without --dry-run to delete]", + results.len() + )?; + } else { + writeln!( + stdout, + "Total: {total_duplicates} duplicate(s) deleted across {} bucket(s)", + results.len() + )?; + } + + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use aw_models::Event; + use chrono::{Duration, TimeZone, Utc}; + use serde_json::Map; + + fn event(id: i64, ts_secs: i64, dur_secs: i64, data: &str) -> Event { + let mut map = Map::new(); + map.insert("k".into(), serde_json::Value::String(data.into())); + Event { + id: Some(id), + timestamp: Utc.timestamp_opt(ts_secs, 0).unwrap(), + duration: Duration::seconds(dur_secs), + data: map, + } + } + + #[test] + fn keeps_lowest_id_of_each_duplicate_group() { + let events = vec![ + event(1, 100, 10, "a"), + event(2, 100, 10, "a"), // dup of 1 + event(3, 200, 10, "a"), // distinct timestamp + event(4, 100, 10, "a"), // dup of 1 + event(5, 100, 10, "b"), // distinct data + ]; + let mut dups = find_duplicate_ids(&events); + dups.sort(); + assert_eq!(dups, vec![2, 4]); + } + + #[test] + fn unaffected_by_input_order() { + // get_events returns newest-first; dedupe must still keep the + // lowest id (the original, first-imported copy) regardless of the + // order events arrive in. + let events = vec![ + event(4, 100, 10, "a"), + event(2, 100, 10, "a"), + event(1, 100, 10, "a"), + ]; + let mut dups = find_duplicate_ids(&events); + dups.sort(); + assert_eq!(dups, vec![2, 4]); + } + + #[test] + fn no_duplicates_returns_empty() { + let events = vec![event(1, 100, 10, "a"), event(2, 200, 10, "a")]; + assert!(find_duplicate_ids(&events).is_empty()); + } +} diff --git a/aw-sync/src/main.rs b/aw-sync/src/main.rs index d69cf192..158ae599 100644 --- a/aw-sync/src/main.rs +++ b/aw-sync/src/main.rs @@ -25,6 +25,7 @@ use std::time::Duration; use aw_client_rust::blocking::AwClient; mod accessmethod; +mod dedupe; mod dirs; mod report; mod status; @@ -142,6 +143,24 @@ enum Commands { /// Also prints the last persisted `SyncReport` (what the previous pass did). /// Does not create staging files. Status {}, + /// One-off cleanup: collapse exact-duplicate events in `-synced-from-` + /// buckets (accumulated by pre-#713 pull passes that re-imported the + /// resume-boundary event on every run). + /// + /// Only ever touches buckets carrying the `-synced-from-` marker; a + /// host's own first-hand buckets are never candidates. Duplicates are + /// exact matches on (timestamp, duration, data); the lowest-id (first + /// imported) copy of each group is kept. + Dedupe { + /// Restrict to specific bucket id(s), comma separated. + /// By default, every `-synced-from-` bucket is checked. + #[clap(long, value_parser=parse_list)] + bucket: Option>, + + /// Report duplicate counts without deleting anything. + #[clap(long)] + dry_run: bool, + }, } fn parse_start_date(arg: &str) -> Result, chrono::ParseError> { @@ -357,6 +376,7 @@ fn main() -> Result<(), Box> { // List all buckets Commands::List {} => sync::list_buckets(&client)?, Commands::Status {} => status::run_status(&client, &opts.host, port, &profile)?, + Commands::Dedupe { bucket, dry_run } => dedupe::run_dedupe(&client, bucket, dry_run)?, } // Needed to give the datastores some time to commit before program is shut down. diff --git a/aw-sync/src/sync.rs b/aw-sync/src/sync.rs index 9024a35a..43302776 100644 --- a/aw-sync/src/sync.rs +++ b/aw-sync/src/sync.rs @@ -699,7 +699,7 @@ const EDIT_RECONCILE_LOOKBACK: Duration = Duration::days(7); /// ID truncated on import, independently of this check. Treating it as a marker /// therefore adds no new failure mode. Issue #649 tracks moving provenance to /// bucket metadata, which removes the dependency on the ID string entirely. -fn is_synced_bucket(bucket: &Bucket) -> bool { +pub(crate) fn is_synced_bucket(bucket: &Bucket) -> bool { bucket.id.contains("-synced-from-") } From 4c40f27d67815e12e21d2376713bbc3956cdb793 Mon Sep 17 00:00:00 2001 From: Bob Date: Fri, 18 Sep 2026 17:23:37 +0000 Subject: [PATCH 2/4] =?UTF-8?q?fix(aw-sync/dedupe):=20address=20review=20f?= =?UTF-8?q?eedback=20=E2=80=94=20paginated=20fetch,=20progress=20log,=20RE?= =?UTF-8?q?ADME=20caveat?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Replace unbounded get_events with paginated fetch (PAGE_SIZE=5000, boundary-safe same-timestamp pop) so large buckets (1.5M events) don't OOM or hang the dry-run - Log per-bucket deletion progress at start and every 10k deletes - Add --help note that large cleanups take a long time (individual HTTP deletes) - README: add caveat that HTTP timestamp precision is ms, not ns — two genuinely distinct events with identical (timestamp-ms, duration, data) within the same millisecond would be collapsed (practically impossible for real AW data) - Fix test assertion: paginated test expected [10,11] but lowest-id is keeper, so duplicates are [11,12]; fix comments to match Closes the two practical asks from ErikBjare's review on #718. Git-Session-Id: c85b --- aw-sync/README.md | 14 ++++- aw-sync/src/dedupe.rs | 136 ++++++++++++++++++++++++++++++++++++++++-- aw-sync/src/main.rs | 13 +++- 3 files changed, 154 insertions(+), 9 deletions(-) diff --git a/aw-sync/README.md b/aw-sync/README.md index f2cc8f9c..08e6ad45 100644 --- a/aw-sync/README.md +++ b/aw-sync/README.md @@ -27,9 +27,17 @@ aw-sync status # One-off cleanup: collapse exact-duplicate events that accumulated in # `-synced-from-` buckets from pull passes before #713. Never touches a # host's own first-hand buckets. -aw-sync dedupe --dry-run # report only -aw-sync dedupe # delete the extras (lowest-id copy of each is kept) -aw-sync dedupe --bucket aw-watcher-window-synced-from-host # restrict to one bucket +aw-sync dedupe --dry-run # report only, across every -synced-from- bucket +aw-sync dedupe --bucket aw-watcher-window-synced-from-host # delete the extras in one bucket (lowest-id copy is kept) +# Deleting from every -synced-from- bucket at once (no --bucket) requires --dry-run +# review first: -synced-from- is an ID convention, not a reserved bucket type, so +# `--bucket` makes the buckets you're deleting from an explicit, reviewed list. +# +# Caveat: deduplication compares timestamps at nanosecond precision in the local +# database, but aw-server's HTTP API returns timestamps at millisecond precision. +# Two genuinely distinct events with identical (timestamp-ms, duration, data) in +# the same millisecond would be collapsed. This is practically impossible for +# real activity data, but note it before running on a live instance. ``` `aw-sync` / `aw-sync daemon` currently **does not pull** from the 3-level `{hostname}/{device_id}/` layout that Android and `aw-sync sync` write. The daemon stages `{device_id}/test.db` at the sync-folder root and only walks two directory levels, so those peers are invisible ([#682](https://github.com/ActivityWatch/aw-server-rust/issues/682)). Until that switch lands, use `aw-sync sync` (or a systemd/cron timer around it) — not the daemon. diff --git a/aw-sync/src/dedupe.rs b/aw-sync/src/dedupe.rs index 0860cf9f..c0e8c0c7 100644 --- a/aw-sync/src/dedupe.rs +++ b/aw-sync/src/dedupe.rs @@ -13,6 +13,16 @@ use std::io::{self, Write}; use aw_client_rust::blocking::AwClient; use aw_models::Event; +use chrono::{DateTime, Duration, Utc}; + +/// Bounded page size for `dedupe_bucket_paginated`'s fetch loop, so a +/// long-lived synced bucket can't be pulled into memory in one unbounded +/// `get_events` call. Mirrors the pagination shape in `sync::sync_one` +/// (that module's `BATCH_SIZE` is private, so this is a separate constant). +#[cfg(not(test))] +const PAGE_SIZE: usize = 5000; +#[cfg(test)] +const PAGE_SIZE: usize = 3; /// Among `events`, return the ids of every event that is an exact duplicate /// (same timestamp, duration and data — `Event`'s own `PartialEq`) of an @@ -50,6 +60,59 @@ pub struct BucketDedupeResult { pub duplicate_events: usize, } +/// Fetch `bucket_id`'s events via `fetch_page` in bounded pages (newest-first, +/// walking the `end` boundary backwards) and return the total event count and +/// duplicate ids, without ever holding the whole bucket in memory at once. +/// +/// A duplicate group's members all share the same `(timestamp, duration, +/// data)`, so as long as a page never splits a run of same-timestamp events, +/// every duplicate group is fully contained within one page and +/// `find_duplicate_ids` can be applied per page independently — no +/// across-page bookkeeping needed. Boundary handling mirrors `sync::sync_one`. +fn dedupe_bucket_paginated(mut fetch_page: F) -> Result<(usize, Vec), Box> +where + F: FnMut(Option>) -> Result, Box>, +{ + let mut total_events = 0usize; + let mut duplicate_ids = Vec::new(); + let mut fetch_end: Option> = None; + + loop { + let mut page = fetch_page(fetch_end)?; + if page.is_empty() { + break; + } + let is_last_page = page.len() < PAGE_SIZE; + + if !is_last_page { + // page is newest-first; page.last() = oldest event in this full page. + // Never split a run of same-timestamp events across two pages, or a + // duplicate group could be cut in half with its older half missed. + let boundary_ts = page.last().unwrap().timestamp; + let newest_ts = page.first().unwrap().timestamp; + if newest_ts != boundary_ts { + while page.last().is_some_and(|e| e.timestamp == boundary_ts) { + page.pop(); + } + fetch_end = Some(boundary_ts); + } else { + // Whole page shares one timestamp; can't split further without + // an unbounded query. Not expected for real AW event data. + fetch_end = Some(boundary_ts - Duration::nanoseconds(1)); + } + } + + total_events += page.len(); + duplicate_ids.extend(find_duplicate_ids(&page)); + + if is_last_page { + break; + } + } + + Ok((total_events, duplicate_ids)) +} + pub fn run_dedupe( client: &AwClient, buckets_filter: Option>, @@ -72,6 +135,19 @@ pub fn run_dedupe( Some(_) => {} } } + } else if !dry_run { + // `-synced-from-` is an ID convention aw-sync itself reserves, but + // bucket creation doesn't enforce that reservation (ActivityWatch/aw-server-rust#649 + // tracks moving provenance to bucket metadata instead of the ID string) — a + // hand-created bucket could coincidentally match it and get deleted from + // unattended. Require the caller to have reviewed a --dry-run report and + // named buckets explicitly before deleting from every match at once. + return Err( + "Refusing to delete from every -synced-from- bucket without --bucket: run with \ + --dry-run first, review the per-bucket counts, then re-run with \ + --bucket naming the buckets you confirmed." + .into(), + ); } targets.sort(); @@ -82,14 +158,23 @@ pub fn run_dedupe( let mut results = Vec::new(); for bucket_id in targets { - let events = client.get_events(&bucket_id, None, None, None)?; - let total_events = events.len(); - let duplicate_ids = find_duplicate_ids(&events); + let (total_events, duplicate_ids) = dedupe_bucket_paginated(|end| { + client + .get_events(&bucket_id, None, end, Some(PAGE_SIZE as u64)) + .map_err(|e| -> Box { e.into() }) + })?; let duplicate_events = duplicate_ids.len(); if duplicate_events > 0 && !dry_run { - for id in duplicate_ids { + info!( + "Deleting from {}: {} duplicate(s)...", + bucket_id, duplicate_events + ); + for (i, id) in duplicate_ids.into_iter().enumerate() { client.delete_event(&bucket_id, id)?; + if (i + 1) % 10_000 == 0 { + info!(" {}/{} deleted", i + 1, duplicate_events); + } } } @@ -189,4 +274,47 @@ mod tests { let events = vec![event(1, 100, 10, "a"), event(2, 200, 10, "a")]; assert!(find_duplicate_ids(&events).is_empty()); } + + /// A fake `get_events(end)`: returns events with `timestamp <= end` + /// (or all, if `end` is None), newest-first — same contract as the real + /// client, so `dedupe_bucket_paginated` can be exercised without a server. + fn fake_fetch_page( + all: &[Event], + end: Option>, + ) -> Result, Box> { + let mut page: Vec = all + .iter() + .filter(|e| end.is_none_or(|end| e.timestamp <= end)) + .cloned() + .collect(); + page.sort_by_key(|e| std::cmp::Reverse(e.id)); + page.truncate(PAGE_SIZE); + Ok(page) + } + + #[test] + fn paginated_matches_single_shot_across_page_boundaries() { + // PAGE_SIZE is 3 under #[cfg(test)]. 7 events, forcing 3 pages, with a + // duplicate group (ids 11 and 12, both dups of the lowest-id copy 10) + // that straddles where a naive page cut would land — the boundary-safe + // pop must keep the whole tied run together in one page. + let all = vec![ + event(20, 700, 10, "g"), + event(19, 600, 10, "f"), + event(18, 500, 10, "e"), + event(12, 400, 10, "a"), // dup of 10 (lowest id is keeper) + event(11, 400, 10, "a"), // dup of 10 + event(10, 400, 10, "a"), // keeper: lowest id in the group + event(1, 100, 10, "z"), + ]; + let (total, mut dups) = dedupe_bucket_paginated(|end| fake_fetch_page(&all, end)).unwrap(); + dups.sort(); + assert_eq!(total, all.len()); + assert_eq!(dups, vec![11, 12]); + + // Must match the unbounded single-shot result exactly. + let mut single_shot = find_duplicate_ids(&all); + single_shot.sort(); + assert_eq!(dups, single_shot); + } } diff --git a/aw-sync/src/main.rs b/aw-sync/src/main.rs index 158ae599..5f77f730 100644 --- a/aw-sync/src/main.rs +++ b/aw-sync/src/main.rs @@ -151,9 +151,18 @@ enum Commands { /// host's own first-hand buckets are never candidates. Duplicates are /// exact matches on (timestamp, duration, data); the lowest-id (first /// imported) copy of each group is kept. + /// + /// `-synced-from-` is an ID convention, not a bucket type bucket creation + /// reserves (#649 tracks moving provenance to metadata instead), so + /// deleting from every matching bucket unattended is refused: without + /// `--dry-run`, `--bucket` is required, naming buckets you reviewed. + /// + /// Note: each duplicate is removed with an individual HTTP call, so + /// cleaning up tens of thousands of duplicates can take a long time. Dedupe { - /// Restrict to specific bucket id(s), comma separated. - /// By default, every `-synced-from-` bucket is checked. + /// Restrict to specific bucket id(s), comma separated. Every + /// `-synced-from-` bucket is checked with `--dry-run`; deleting + /// requires naming buckets here explicitly. #[clap(long, value_parser=parse_list)] bucket: Option>, From 262e3569ee8994375d311f343222a45c6e23b494 Mon Sep 17 00:00:00 2001 From: Bob Date: Fri, 18 Sep 2026 18:09:08 +0000 Subject: [PATCH 3/4] fix(aw-sync/dedupe): carry boundary events to avoid server duration-clipping MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P1 — Boundary Fetch Changes Durations (Greptile #4049200869): When the pagination boundary fell mid-timestamp-tie, re-fetching those events with end=boundary_ts caused the server to clip their durations to zero (the server clips event end-times at the query boundary). This made two distinct events with the same timestamp but different durations appear identical, risking permanent deletion of a non-duplicate event. Fix: pop boundary events into a `carry` buffer and advance fetch_end to boundary_ts - 1ns. The carried events (with their true, unclipped durations) are prepended to the next iteration's batch alongside newly fetched older events. No server re-fetch with a clipping end-time is needed. P1 — Large Timestamp Ties Are Skipped (Greptile #4049200874): When an entire page shares one timestamp, remaining events at that timestamp beyond PAGE_SIZE can't be fetched without offset support. Added a warning so operators know a second pass may be needed in this (unlikely) case. P2 — Timestamp Precision Warning Was Incorrect (Greptile #4049200879): The README caveat incorrectly stated the HTTP API returns ms-precision timestamps. The server retains ns precision through storage, serialization and deserialization. Updated the caveat to describe the actual dedup key (exact timestamp + duration + data match) without the inaccurate ms claim. Also: update fake_fetch_page in tests to simulate server-side duration clipping so the regression test can catch this class of bug. New test `carry_preserves_duration_across_page_boundary` exercises the P1 fix directly — two events with the same timestamp but different durations must not be flagged as duplicates across a page boundary. Git-Session-Id: 2515863d-bf36-563a-a4a0-130e48543412 --- aw-sync/README.md | 9 ++-- aw-sync/src/dedupe.rs | 103 +++++++++++++++++++++++++++++++++++++----- 2 files changed, 95 insertions(+), 17 deletions(-) diff --git a/aw-sync/README.md b/aw-sync/README.md index 08e6ad45..91fe083e 100644 --- a/aw-sync/README.md +++ b/aw-sync/README.md @@ -33,11 +33,10 @@ aw-sync dedupe --bucket aw-watcher-window-synced-from-host # delete the extras # review first: -synced-from- is an ID convention, not a reserved bucket type, so # `--bucket` makes the buckets you're deleting from an explicit, reviewed list. # -# Caveat: deduplication compares timestamps at nanosecond precision in the local -# database, but aw-server's HTTP API returns timestamps at millisecond precision. -# Two genuinely distinct events with identical (timestamp-ms, duration, data) in -# the same millisecond would be collapsed. This is practically impossible for -# real activity data, but note it before running on a live instance. +# Caveat: deduplication compares (timestamp, duration, data) for exact equality. +# Two genuinely distinct events that happen to share all three fields would be +# collapsed — one copy deleted. This is practically impossible for real activity +# data, but note it before running on a live instance. ``` `aw-sync` / `aw-sync daemon` currently **does not pull** from the 3-level `{hostname}/{device_id}/` layout that Android and `aw-sync sync` write. The daemon stages `{device_id}/test.db` at the sync-folder root and only walks two directory levels, so those peers are invisible ([#682](https://github.com/ActivityWatch/aw-server-rust/issues/682)). Until that switch lands, use `aw-sync sync` (or a systemd/cron timer around it) — not the daemon. diff --git a/aw-sync/src/dedupe.rs b/aw-sync/src/dedupe.rs index c0e8c0c7..48b595e5 100644 --- a/aw-sync/src/dedupe.rs +++ b/aw-sync/src/dedupe.rs @@ -68,7 +68,12 @@ pub struct BucketDedupeResult { /// data)`, so as long as a page never splits a run of same-timestamp events, /// every duplicate group is fully contained within one page and /// `find_duplicate_ids` can be applied per page independently — no -/// across-page bookkeeping needed. Boundary handling mirrors `sync::sync_one`. +/// across-page bookkeeping needed. +/// +/// Boundary events are carried forward into the next page's batch rather than +/// re-fetched with `end = boundary_ts`, because the server clips event +/// durations at the `end` boundary — events starting exactly at `boundary_ts` +/// would have their durations zeroed, producing false duplicate keys. fn dedupe_bucket_paginated(mut fetch_page: F) -> Result<(usize, Vec), Box> where F: FnMut(Option>) -> Result, Box>, @@ -76,28 +81,53 @@ where let mut total_events = 0usize; let mut duplicate_ids = Vec::new(); let mut fetch_end: Option> = None; + // Boundary events popped from the end of a full page, carried into the + // next iteration so they're processed alongside any remaining events at + // the same timestamp — without going through a server re-fetch that would + // clip their durations. + let mut carry: Vec = Vec::new(); loop { - let mut page = fetch_page(fetch_end)?; + let fetched = fetch_page(fetch_end)?; + let is_last_page = fetched.len() < PAGE_SIZE; + + // Combine carried boundary events (newer) with the newly fetched page + // (older). Carry is always newer: it came from the bottom of the + // previous full page, so timestamps in carry >= timestamps in fetched. + let mut page = carry; + carry = Vec::new(); + page.extend(fetched); + if page.is_empty() { break; } - let is_last_page = page.len() < PAGE_SIZE; if !is_last_page { - // page is newest-first; page.last() = oldest event in this full page. + // page is newest-first; page.last() = oldest event in this batch. // Never split a run of same-timestamp events across two pages, or a // duplicate group could be cut in half with its older half missed. let boundary_ts = page.last().unwrap().timestamp; let newest_ts = page.first().unwrap().timestamp; if newest_ts != boundary_ts { + // Pop all events at boundary_ts into carry. Advance to one ns + // before boundary_ts so the next server fetch excludes that + // timestamp — carried events already hold the true-duration + // copies from this page. while page.last().is_some_and(|e| e.timestamp == boundary_ts) { - page.pop(); + carry.push(page.pop().unwrap()); } - fetch_end = Some(boundary_ts); + fetch_end = Some(boundary_ts - Duration::nanoseconds(1)); } else { - // Whole page shares one timestamp; can't split further without - // an unbounded query. Not expected for real AW event data. + // Entire batch shares one timestamp. Advance past it. + // If more than PAGE_SIZE events share this timestamp, those + // beyond what fit in this page are not deduped in this pass. + // This is not expected for normal AW event data; a subsequent + // run will catch any residual duplicates. + warn!( + "Entire page shares timestamp {}; if more events exist at \ + this timestamp they will not be deduped in this pass", + boundary_ts + ); fetch_end = Some(boundary_ts - Duration::nanoseconds(1)); } } @@ -105,7 +135,7 @@ where total_events += page.len(); duplicate_ids.extend(find_duplicate_ids(&page)); - if is_last_page { + if is_last_page && carry.is_empty() { break; } } @@ -276,8 +306,10 @@ mod tests { } /// A fake `get_events(end)`: returns events with `timestamp <= end` - /// (or all, if `end` is None), newest-first — same contract as the real - /// client, so `dedupe_bucket_paginated` can be exercised without a server. + /// (or all, if `end` is None), newest-first, with server-side duration + /// clipping applied — same contract as the real client. Events that start + /// at exactly `end` have their duration clipped to zero (the real server + /// clips event end-times at the query boundary). fn fake_fetch_page( all: &[Event], end: Option>, @@ -285,13 +317,60 @@ mod tests { let mut page: Vec = all .iter() .filter(|e| end.is_none_or(|end| e.timestamp <= end)) - .cloned() + .map(|e| { + // Simulate server-side duration clipping: events whose + // duration extends past `end` are clipped to fit. + if let Some(end) = end { + let event_end = e.timestamp + e.duration; + if event_end > end { + let clipped_dur = end - e.timestamp; + return Event { + duration: if clipped_dur < Duration::zero() { + Duration::zero() + } else { + clipped_dur + }, + ..e.clone() + }; + } + } + e.clone() + }) .collect(); page.sort_by_key(|e| std::cmp::Reverse(e.id)); page.truncate(PAGE_SIZE); Ok(page) } + /// Regression test for the duration-clipping P1 bug: a boundary event's + /// duration must not be changed by server-side clipping when the event is + /// carried into the next batch. Two events with same timestamp but + /// different durations are distinct; neither should be flagged as a + /// duplicate of the other. + #[test] + fn carry_preserves_duration_across_page_boundary() { + // PAGE_SIZE=3. Events at ts=200 straddle a page boundary: + // page 1 fetched (end=None): ids [20, 11, 10] — ts=200 events fill the boundary + // Without the fix, id=10 and id=11 would be re-fetched with end=200ns, + // clipping their durations to 0 and making them look like duplicates + // of id=9 (which also has ts=200 and dur=0 in the clipped view). + // + // The two events at ts=200 have DIFFERENT durations (5s vs 10s) — they + // are genuinely distinct and must not be deleted. + let all = vec![ + event(20, 500, 10, "z"), + event(11, 200, 10, "a"), // distinct: dur=10s — must NOT be flagged + event(10, 200, 5, "a"), // distinct: dur=5s — must NOT be flagged + event(9, 100, 10, "b"), + ]; + let (total, dups) = dedupe_bucket_paginated(|end| fake_fetch_page(&all, end)).unwrap(); + assert_eq!(total, all.len()); + assert!( + dups.is_empty(), + "expected no duplicates (all events are distinct), got: {dups:?}" + ); + } + #[test] fn paginated_matches_single_shot_across_page_boundaries() { // PAGE_SIZE is 3 under #[cfg(test)]. 7 events, forcing 3 pages, with a From de0c8a54dbad09047e242ab24b02485e7b9587b0 Mon Sep 17 00:00:00 2001 From: Bob Date: Fri, 18 Sep 2026 19:14:46 +0000 Subject: [PATCH 4/4] fix(aw-sync/dedupe): fetch each timestamp run in full so page boundaries can't split duplicate groups A full page cut through a duplicate run left the remainder unfetched (next page started at cut_ts - 1ns), so runs straddling a page edge were silently under-cleaned. Cut pages at the PAGE_SIZE-th event's timestamp and grow the fetch limit until an older event proves the run complete; also covers runs larger than a page. The unit-test fake now follows the real server ordering/limit/clipping. Refusal for unnamed buckets moves after the empty-target check so a fresh install gets a clean no-op. Git-Session-Id: 6f002d6d-0117-5806-a4c0-de3cc45f0101 --- aw-sync/src/dedupe.rs | 267 +++++++++++++++++++++++------------------- 1 file changed, 144 insertions(+), 123 deletions(-) diff --git a/aw-sync/src/dedupe.rs b/aw-sync/src/dedupe.rs index 48b595e5..e29ba9cc 100644 --- a/aw-sync/src/dedupe.rs +++ b/aw-sync/src/dedupe.rs @@ -60,82 +60,57 @@ pub struct BucketDedupeResult { pub duplicate_events: usize, } -/// Fetch `bucket_id`'s events via `fetch_page` in bounded pages (newest-first, -/// walking the `end` boundary backwards) and return the total event count and +/// Fetch a bucket's events via `fetch_page(end, limit)` in bounded pages +/// (newest-first, walking `end` backwards) and return the total event count and /// duplicate ids, without ever holding the whole bucket in memory at once. /// /// A duplicate group's members all share the same `(timestamp, duration, -/// data)`, so as long as a page never splits a run of same-timestamp events, -/// every duplicate group is fully contained within one page and +/// data)`, so as long as every page contains each timestamp's run *in full*, /// `find_duplicate_ids` can be applied per page independently — no /// across-page bookkeeping needed. /// -/// Boundary events are carried forward into the next page's batch rather than -/// re-fetched with `end = boundary_ts`, because the server clips event -/// durations at the `end` boundary — events starting exactly at `boundary_ts` -/// would have their durations zeroed, producing false duplicate keys. +/// A page is cut at the timestamp of its `PAGE_SIZE`-th event (`cut_ts`). +/// The run at `cut_ts` may continue past the fetched events, so the page is +/// only accepted once the fetch also contains an event strictly older than +/// `cut_ts` (proving the run is complete) or has exhausted the bucket; until +/// then the limit is doubled and the same window re-fetched. Only events at or +/// after `cut_ts` are processed; the next page starts at `cut_ts - 1ns`. +/// +/// Runs are never re-fetched through `end = cut_ts`: the server clips event +/// durations at the `end` boundary, which would zero the durations of events +/// starting exactly there and make distinct events look like duplicates. fn dedupe_bucket_paginated(mut fetch_page: F) -> Result<(usize, Vec), Box> where - F: FnMut(Option>) -> Result, Box>, + F: FnMut(Option>, usize) -> Result, Box>, { let mut total_events = 0usize; let mut duplicate_ids = Vec::new(); let mut fetch_end: Option> = None; - // Boundary events popped from the end of a full page, carried into the - // next iteration so they're processed alongside any remaining events at - // the same timestamp — without going through a server re-fetch that would - // clip their durations. - let mut carry: Vec = Vec::new(); loop { - let fetched = fetch_page(fetch_end)?; - let is_last_page = fetched.len() < PAGE_SIZE; - - // Combine carried boundary events (newer) with the newly fetched page - // (older). Carry is always newer: it came from the bottom of the - // previous full page, so timestamps in carry >= timestamps in fetched. - let mut page = carry; - carry = Vec::new(); - page.extend(fetched); - - if page.is_empty() { - break; - } - - if !is_last_page { - // page is newest-first; page.last() = oldest event in this batch. - // Never split a run of same-timestamp events across two pages, or a - // duplicate group could be cut in half with its older half missed. - let boundary_ts = page.last().unwrap().timestamp; - let newest_ts = page.first().unwrap().timestamp; - if newest_ts != boundary_ts { - // Pop all events at boundary_ts into carry. Advance to one ns - // before boundary_ts so the next server fetch excludes that - // timestamp — carried events already hold the true-duration - // copies from this page. - while page.last().is_some_and(|e| e.timestamp == boundary_ts) { - carry.push(page.pop().unwrap()); - } - fetch_end = Some(boundary_ts - Duration::nanoseconds(1)); - } else { - // Entire batch shares one timestamp. Advance past it. - // If more than PAGE_SIZE events share this timestamp, those - // beyond what fit in this page are not deduped in this pass. - // This is not expected for normal AW event data; a subsequent - // run will catch any residual duplicates. - warn!( - "Entire page shares timestamp {}; if more events exist at \ - this timestamp they will not be deduped in this pass", - boundary_ts - ); - fetch_end = Some(boundary_ts - Duration::nanoseconds(1)); + let mut limit = PAGE_SIZE + 1; + let mut cut_ts: Option> = None; + let (mut page, exhausted) = loop { + let page = fetch_page(fetch_end, limit)?; + if page.len() < limit { + break (page, true); + } + let cut = *cut_ts.get_or_insert(page[PAGE_SIZE - 1].timestamp); + if page[page.len() - 1].timestamp < cut { + break (page, false); } + limit *= 2; + }; + + if let Some(cut) = cut_ts.filter(|_| !exhausted) { + page.retain(|e| e.timestamp >= cut); + fetch_end = Some(cut - Duration::nanoseconds(1)); } total_events += page.len(); duplicate_ids.extend(find_duplicate_ids(&page)); - if is_last_page && carry.is_empty() { + if exhausted { break; } } @@ -165,7 +140,15 @@ pub fn run_dedupe( Some(_) => {} } } - } else if !dry_run { + } + targets.sort(); + + if targets.is_empty() { + info!("No -synced-from- buckets found, nothing to dedupe"); + return Ok(()); + } + + if buckets_filter.is_none() && !dry_run { // `-synced-from-` is an ID convention aw-sync itself reserves, but // bucket creation doesn't enforce that reservation (ActivityWatch/aw-server-rust#649 // tracks moving provenance to bucket metadata instead of the ID string) — a @@ -179,18 +162,12 @@ pub fn run_dedupe( .into(), ); } - targets.sort(); - - if targets.is_empty() { - info!("No -synced-from- buckets found, nothing to dedupe"); - return Ok(()); - } let mut results = Vec::new(); for bucket_id in targets { - let (total_events, duplicate_ids) = dedupe_bucket_paginated(|end| { + let (total_events, duplicate_ids) = dedupe_bucket_paginated(|end, limit| { client - .get_events(&bucket_id, None, end, Some(PAGE_SIZE as u64)) + .get_events(&bucket_id, None, end, Some(limit as u64)) .map_err(|e| -> Box { e.into() }) })?; let duplicate_events = duplicate_ids.len(); @@ -305,95 +282,139 @@ mod tests { assert!(find_duplicate_ids(&events).is_empty()); } - /// A fake `get_events(end)`: returns events with `timestamp <= end` - /// (or all, if `end` is None), newest-first, with server-side duration - /// clipping applied — same contract as the real client. Events that start - /// at exactly `end` have their duration clipped to zero (the real server - /// clips event end-times at the query boundary). + /// A fake `get_events(end, limit)` following the real server contract: + /// events with `timestamp <= end` (all, if `end` is None), ordered + /// `starttime DESC, endtime ASC, id ASC`, truncated to `limit`, with + /// durations clipped at `end` (an event starting exactly at `end` comes + /// back with a zero duration). fn fake_fetch_page( all: &[Event], end: Option>, + limit: usize, ) -> Result, Box> { let mut page: Vec = all .iter() .filter(|e| end.is_none_or(|end| e.timestamp <= end)) - .map(|e| { - // Simulate server-side duration clipping: events whose - // duration extends past `end` are clipped to fit. - if let Some(end) = end { - let event_end = e.timestamp + e.duration; - if event_end > end { - let clipped_dur = end - e.timestamp; - return Event { - duration: if clipped_dur < Duration::zero() { - Duration::zero() - } else { - clipped_dur - }, - ..e.clone() - }; - } - } - e.clone() + .map(|e| match end { + Some(end) if e.timestamp + e.duration > end => Event { + duration: end - e.timestamp, + ..e.clone() + }, + _ => e.clone(), }) .collect(); - page.sort_by_key(|e| std::cmp::Reverse(e.id)); - page.truncate(PAGE_SIZE); + page.sort_by_key(|e| { + ( + std::cmp::Reverse(e.timestamp), + e.timestamp + e.duration, + e.id, + ) + }); + page.truncate(limit); Ok(page) } - /// Regression test for the duration-clipping P1 bug: a boundary event's - /// duration must not be changed by server-side clipping when the event is - /// carried into the next batch. Two events with same timestamp but - /// different durations are distinct; neither should be flagged as a - /// duplicate of the other. + fn paginated(all: &[Event]) -> (usize, Vec) { + let (total, mut dups) = + dedupe_bucket_paginated(|end, limit| fake_fetch_page(all, end, limit)).unwrap(); + dups.sort(); + (total, dups) + } + + fn single_shot(all: &[Event]) -> Vec { + let mut dups = find_duplicate_ids(all); + dups.sort(); + dups + } + + /// Two events with the same timestamp and data but different durations are + /// distinct and must never be collapsed by a page boundary landing between + /// them (re-fetching with `end = ts` would clip both to zero duration). #[test] - fn carry_preserves_duration_across_page_boundary() { - // PAGE_SIZE=3. Events at ts=200 straddle a page boundary: - // page 1 fetched (end=None): ids [20, 11, 10] — ts=200 events fill the boundary - // Without the fix, id=10 and id=11 would be re-fetched with end=200ns, - // clipping their durations to 0 and making them look like duplicates - // of id=9 (which also has ts=200 and dur=0 in the clipped view). - // - // The two events at ts=200 have DIFFERENT durations (5s vs 10s) — they - // are genuinely distinct and must not be deleted. + fn distinct_durations_survive_page_boundary() { let all = vec![ event(20, 500, 10, "z"), - event(11, 200, 10, "a"), // distinct: dur=10s — must NOT be flagged - event(10, 200, 5, "a"), // distinct: dur=5s — must NOT be flagged + event(11, 200, 10, "a"), + event(10, 200, 5, "a"), event(9, 100, 10, "b"), ]; - let (total, dups) = dedupe_bucket_paginated(|end| fake_fetch_page(&all, end)).unwrap(); + let (total, dups) = paginated(&all); assert_eq!(total, all.len()); - assert!( - dups.is_empty(), - "expected no duplicates (all events are distinct), got: {dups:?}" - ); + assert!(dups.is_empty(), "all events are distinct, got: {dups:?}"); } #[test] fn paginated_matches_single_shot_across_page_boundaries() { - // PAGE_SIZE is 3 under #[cfg(test)]. 7 events, forcing 3 pages, with a - // duplicate group (ids 11 and 12, both dups of the lowest-id copy 10) - // that straddles where a naive page cut would land — the boundary-safe - // pop must keep the whole tied run together in one page. + // PAGE_SIZE is 3 under #[cfg(test)]: 7 events force multiple pages, + // with a duplicate group at ts=400 sitting on a page boundary. let all = vec![ event(20, 700, 10, "g"), event(19, 600, 10, "f"), event(18, 500, 10, "e"), - event(12, 400, 10, "a"), // dup of 10 (lowest id is keeper) - event(11, 400, 10, "a"), // dup of 10 + event(12, 400, 10, "a"), + event(11, 400, 10, "a"), event(10, 400, 10, "a"), // keeper: lowest id in the group event(1, 100, 10, "z"), ]; - let (total, mut dups) = dedupe_bucket_paginated(|end| fake_fetch_page(&all, end)).unwrap(); - dups.sort(); + let (total, dups) = paginated(&all); assert_eq!(total, all.len()); assert_eq!(dups, vec![11, 12]); + assert_eq!(dups, single_shot(&all)); + } - // Must match the unbounded single-shot result exactly. - let mut single_shot = find_duplicate_ids(&all); - single_shot.sort(); - assert_eq!(dups, single_shot); + /// A duplicate run that starts inside a page and continues past its end + /// must still be found in full (it used to be cut at the page boundary, + /// leaving the remainder undetected). + #[test] + fn run_straddling_page_boundary_is_found_in_full() { + let mut all = vec![event(100, 900, 10, "new1"), event(101, 800, 10, "new2")]; + all.extend((1..=5).map(|id| event(id, 500, 10, "dup"))); + all.push(event(50, 100, 10, "old")); + let (total, dups) = paginated(&all); + assert_eq!(total, all.len()); + assert_eq!(dups, vec![2, 3, 4, 5]); + } + + /// More same-timestamp events than fit in one page (the reviewer's + /// "large tie" case): every extra copy must still be found. + #[test] + fn run_larger_than_a_page_is_found_in_full() { + let all: Vec = (1..=10).map(|id| event(id, 500, 10, "dup")).collect(); + let (total, dups) = paginated(&all); + assert_eq!(total, 10); + assert_eq!(dups, (2..=10).collect::>()); + } + + #[test] + fn empty_bucket_is_fine() { + assert_eq!(paginated(&[]), (0, vec![])); + } + + /// Randomised layouts (runs of varying length landing on every possible + /// page offset) must agree exactly with the unbounded single-shot result. + #[test] + fn paginated_matches_single_shot_on_random_layouts() { + let mut state: u64 = 0x2545_f491_4f6c_dd1d; + let mut next = |n: u64| { + state = state + .wrapping_mul(6364136223846793005) + .wrapping_add(1442695040888963407); + (state >> 33) % n + }; + for _ in 0..200 { + let n_events = next(40) as usize; + let mut all = Vec::new(); + for id in 1..=n_events as i64 { + // Spaced 1000s apart with durations <= 10s: no event ever + // crosses a page cut, so server clipping can't alter keys. + let ts = 1000 * (1 + next(8) as i64); + let dur = 1 + next(3) as i64 * 5; + let data = ["a", "b"][next(2) as usize]; + all.push(event(id, ts, dur, data)); + } + let (total, dups) = paginated(&all); + assert_eq!(total, all.len()); + assert_eq!(dups, single_shot(&all), "layout: {all:?}"); + } } }