From b6d88940b6b208705b7df070862bd8ae7612d4c4 Mon Sep 17 00:00:00 2001 From: Bob Date: Mon, 14 Sep 2026 03:34:07 +0000 Subject: [PATCH 1/8] feat(sync): reconcile owner-originated edits within resume lookback Resume-from-latest-end skipped title edits of already-synced events (ActivityWatch/aw-android#253). Match source and dest by timestamp in the 7 days before the cursor and replace dest rows whose data changed. Run that pass before the incremental copy so a latest-event title edit does not insert a start-clipped duplicate. Git-Session-Id: 0d71 --- aw-sync/README.md | 2 +- aw-sync/src/accessmethod.rs | 12 ++ aw-sync/src/sync.rs | 77 +++++++++ aw-sync/tests/historical_edit.rs | 277 +++++++++++++++++++++++++++++++ 4 files changed, 367 insertions(+), 1 deletion(-) create mode 100644 aw-sync/tests/historical_edit.rs diff --git a/aw-sync/README.md b/aw-sync/README.md index 69cebfc5..ff06c042 100644 --- a/aw-sync/README.md +++ b/aw-sync/README.md @@ -80,7 +80,7 @@ We also avoid having to implement complex features such as conflict resolution, - It doesn't support Android, yet. - It mirrors events to all devices, - If you have a lot of devices you'll get a lot of duplicates, taking up a lot of space and potentially impacting performance. -- It doesn't support modifying/deleting events, yet. +- Event deletion is not synced. Owner-originated edits (same timestamp, new data) are reconciled within 7 days of the destination resume cursor; older edits are skipped to keep peak memory bounded. --- diff --git a/aw-sync/src/accessmethod.rs b/aw-sync/src/accessmethod.rs index be01d782..7160ce04 100644 --- a/aw-sync/src/accessmethod.rs +++ b/aw-sync/src/accessmethod.rs @@ -22,6 +22,7 @@ pub trait AccessMethod: std::fmt::Debug { fn insert_events(&self, bucket_id: &str, events: Vec) -> Result<(), String>; fn get_event_count(&self, bucket_id: &str) -> Result; fn heartbeat(&self, bucket_id: &str, event: Event, duration: f64) -> Result<(), String>; + fn delete_events_by_id(&self, bucket_id: &str, event_ids: Vec) -> Result<(), String>; } /// Every method here returns a `Result`, so a datastore failure must be reported @@ -65,6 +66,11 @@ impl AccessMethod for Datastore { fn get_event_count(&self, bucket_id: &str) -> Result { Datastore::get_event_count(self, bucket_id, None, None).map_err(|e| format!("{e:?}")) } + fn delete_events_by_id(&self, bucket_id: &str, event_ids: Vec) -> Result<(), String> { + Datastore::delete_events_by_id(self, bucket_id, event_ids).map_err(|e| format!("{e:?}"))?; + self.force_commit().unwrap(); + Ok(()) + } } impl AccessMethod for AwClient { @@ -111,4 +117,10 @@ impl AccessMethod for AwClient { fn heartbeat(&self, bucket_id: &str, event: Event, duration: f64) -> Result<(), String> { AwClient::heartbeat(self, bucket_id, &event, duration).map_err(|e| format!("{e:?}")) } + fn delete_events_by_id(&self, bucket_id: &str, event_ids: Vec) -> Result<(), String> { + for event_id in event_ids { + AwClient::delete_event(self, bucket_id, event_id).map_err(|e| e.to_string())?; + } + Ok(()) + } } diff --git a/aw-sync/src/sync.rs b/aw-sync/src/sync.rs index 4abf2aa2..9f3786d9 100644 --- a/aw-sync/src/sync.rs +++ b/aw-sync/src/sync.rs @@ -9,6 +9,7 @@ extern crate chrono; extern crate reqwest; extern crate serde_json; +use std::collections::HashMap; use std::error::Error; use std::fs; use std::path::{Path, PathBuf}; @@ -312,6 +313,11 @@ const BATCH_SIZE: usize = 5000; #[cfg(test)] const BATCH_SIZE: usize = 5; +/// How far before the resume cursor to look for owner-originated edits of +/// already-synced events (ActivityWatch/aw-android#253). Bounded so a full +/// historical bucket is never loaded into memory on Android. +const EDIT_RECONCILE_LOOKBACK: Duration = Duration::days(7); + /// Whether a bucket holds data synced from another host, rather than data /// collected on this host. /// @@ -432,6 +438,75 @@ pub fn sync_datastores( Ok(()) } +/// Replace dest events whose timestamp still exists on the source but whose +/// data or duration changed. +/// +/// WebUI/Android title edits are delete+insert at the same timestamp. The +/// resume cursor starts at the destination's latest event end, so those +/// replacements are outside the incremental fetch window. Under the +/// single-writer model the source is authoritative for its own buckets, so a +/// timestamp match with different data is an owner-originated edit, not a +/// conflict. +/// +/// Must run *before* the incremental copy: a latest-event title edit is also +/// re-fetched as a start-clipped fragment, and heartbeat() refuses to merge +/// different data, which would insert a duplicate unless dest already holds +/// the new payload. +fn reconcile_updated_events( + ds_from: &dyn AccessMethod, + ds_to: &dyn AccessMethod, + bucket_from: &Bucket, + bucket_to: &Bucket, + resume_sync_at: Option>, +) { + let Some(resume) = resume_sync_at else { + return; + }; + let lookback_start = resume - EDIT_RECONCILE_LOOKBACK; + + let source_events = ds_from + .get_events(bucket_from.id.as_str(), Some(lookback_start), None, None) + .unwrap(); + let dest_events = ds_to + .get_events(bucket_to.id.as_str(), Some(lookback_start), None, None) + .unwrap(); + + let dest_by_ts: HashMap, Event> = dest_events + .into_iter() + .filter(|e| e.timestamp >= lookback_start) + .map(|e| (e.timestamp, e)) + .collect(); + + for src in source_events { + if src.timestamp < lookback_start { + continue; + } + let Some(dst) = dest_by_ts.get(&src.timestamp) else { + continue; + }; + if dst.data == src.data && dst.duration == src.duration { + continue; + } + let Some(dst_id) = dst.id else { + warn!( + "Cannot reconcile event at {:?} — dest event has no id", + src.timestamp + ); + continue; + }; + ds_to + .delete_events_by_id(bucket_to.id.as_str(), vec![dst_id]) + .unwrap(); + let ts = src.timestamp; + let mut replacement = src; + replacement.id = None; + ds_to + .insert_events(bucket_to.id.as_str(), vec![replacement]) + .unwrap(); + info!(" ~ Reconciled edited event at {:?}", ts); + } +} + /// Syncs a single bucket from one datastore to another fn sync_one( ds_from: &dyn AccessMethod, @@ -461,6 +536,8 @@ fn sync_one( info!(" + Starting from beginning"); } + reconcile_updated_events(ds_from, ds_to, &bucket_from, &bucket_to, resume_sync_at); + // Fetch events in bounded chunks to avoid OOM on devices with limited RAM (e.g. Android). // get_events returns events in descending order (newest first), so we paginate backwards // using the `end` parameter. Each chunk is written to `ds_to` as soon as it is fetched, diff --git a/aw-sync/tests/historical_edit.rs b/aw-sync/tests/historical_edit.rs new file mode 100644 index 00000000..ed9158b1 --- /dev/null +++ b/aw-sync/tests/historical_edit.rs @@ -0,0 +1,277 @@ +/// Owner-originated historical edits (ActivityWatch/aw-android#253). +/// +/// WebUI/Android title edits are delete+insert at the same timestamp. aw-sync +/// resumes at the destination's latest event end, so those replacements sit +/// outside the fetch window unless a reconcile pass updates matching timestamps. +use chrono::{DateTime, Duration, Utc}; +use serde_json::json; + +use aw_datastore::Datastore; +use aw_models::{Bucket, Event}; +use aw_sync::SyncSpec; + +fn memory_pair() -> (Datastore, Datastore) { + ( + Datastore::new_in_memory(false), + Datastore::new_in_memory(false), + ) +} + +fn create_bucket(ds: &Datastore, id: &str, hostname: &str) -> String { + let bucket_json = format!( + r#"{{ + "id": "{id}", + "type": "currentwindow", + "hostname": "{hostname}", + "client": "test" + }}"# + ); + let bucket: Bucket = serde_json::from_str(&bucket_json).unwrap(); + ds.create_bucket(&bucket).unwrap(); + ds.force_commit().unwrap(); + id.to_string() +} + +fn event_at(ts: DateTime, duration: Duration, title: &str) -> Event { + let mut data = serde_json::Map::new(); + data.insert("app".to_string(), json!("VLC")); + data.insert("title".to_string(), json!(title)); + Event { + id: None, + timestamp: ts, + duration, + data, + } +} + +fn titles(ds: &Datastore, bucket_id: &str) -> Vec<(DateTime, String)> { + let mut events = ds.get_events(bucket_id, None, None, None).unwrap(); + events.sort_by_key(|e| e.timestamp); + events + .into_iter() + .map(|e| { + let title = e + .data + .get("title") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + (e.timestamp, title) + }) + .collect() +} + +/// Match the WebUI/Android edit path: delete the local row, insert a replacement +/// with the same timestamp and duration. +fn edit_title(ds: &Datastore, bucket_id: &str, ts: DateTime, new_title: &str) { + let events = ds.get_events(bucket_id, None, None, None).unwrap(); + let target = events + .iter() + .find(|e| e.timestamp == ts) + .expect("event to edit"); + let id = target.id.expect("source event id"); + ds.delete_events_by_id(bucket_id, vec![id]).unwrap(); + let mut data = target.data.clone(); + data.insert("title".to_string(), json!(new_title)); + ds.insert_events( + bucket_id, + &[Event { + id: None, + timestamp: target.timestamp, + duration: target.duration, + data, + }], + ) + .unwrap(); + ds.force_commit().unwrap(); +} + +fn sync_push(src: &Datastore, dest: &Datastore) { + aw_sync::sync_datastores(src, dest, true, Some("device-phone"), &SyncSpec::default()); +} + +fn sync_pull(src: &Datastore, dest: &Datastore) { + aw_sync::sync_datastores(src, dest, false, None, &SyncSpec::default()); +} + +#[test] +fn historical_title_edit_reaches_staging() { + let (src, dest) = memory_pair(); + let bucket = create_bucket(&src, "aw-watcher-android-test", "phone"); + let t0 = Utc::now() - Duration::hours(2); + let t1 = t0 + Duration::minutes(30); + src.insert_events( + &bucket, + &[ + event_at(t0, Duration::minutes(20), "sanitized.mp4"), + event_at(t1, Duration::minutes(5), "later window"), + ], + ) + .unwrap(); + src.force_commit().unwrap(); + + sync_push(&src, &dest); + assert_eq!( + titles(&dest, &bucket), + vec![ + (t0, "sanitized.mp4".to_string()), + (t1, "later window".to_string()), + ] + ); + + edit_title(&src, &bucket, t0, "Real Video Title"); + sync_push(&src, &dest); + + assert_eq!( + titles(&dest, &bucket), + vec![ + (t0, "Real Video Title".to_string()), + (t1, "later window".to_string()), + ], + "older title edit must update staging without duplicating or dropping the later event" + ); + assert_eq!(dest.get_event_count(&bucket, None, None).unwrap(), 2); +} + +#[test] +fn latest_title_edit_reaches_staging() { + let (src, dest) = memory_pair(); + let bucket = create_bucket(&src, "aw-watcher-android-latest", "phone"); + let t0 = Utc::now() - Duration::hours(1); + let t1 = t0 + Duration::minutes(40); + src.insert_events( + &bucket, + &[ + event_at(t0, Duration::minutes(10), "older"), + event_at(t1, Duration::minutes(15), "sanitized-latest.mp4"), + ], + ) + .unwrap(); + src.force_commit().unwrap(); + + sync_push(&src, &dest); + edit_title(&src, &bucket, t1, "Latest Real Title"); + sync_push(&src, &dest); + + assert_eq!( + titles(&dest, &bucket), + vec![ + (t0, "older".to_string()), + (t1, "Latest Real Title".to_string()), + ] + ); + assert_eq!( + dest.get_event_count(&bucket, None, None).unwrap(), + 2, + "latest-event title edit must not insert a clipped duplicate" + ); +} + +#[test] +fn historical_title_edit_reaches_peer() { + let (src, staging) = memory_pair(); + let peer = Datastore::new_in_memory(false); + let bucket = create_bucket(&src, "aw-watcher-android-peer", "phone"); + let t0 = Utc::now() - Duration::hours(3); + let t1 = t0 + Duration::hours(1); + src.insert_events( + &bucket, + &[ + event_at(t0, Duration::minutes(25), "sanitized.mp4"), + event_at(t1, Duration::minutes(8), "later window"), + ], + ) + .unwrap(); + src.force_commit().unwrap(); + + sync_push(&src, &staging); + sync_pull(&staging, &peer); + + let peer_bucket = format!("{bucket}-synced-from-phone"); + assert_eq!( + titles(&peer, &peer_bucket)[0].1, + "sanitized.mp4", + "precondition: peer imported the original title" + ); + + edit_title(&src, &bucket, t0, "Real Video Title"); + sync_push(&src, &staging); + sync_pull(&staging, &peer); + + assert_eq!( + titles(&staging, &bucket), + vec![ + (t0, "Real Video Title".to_string()), + (t1, "later window".to_string()), + ] + ); + assert_eq!( + titles(&peer, &peer_bucket), + vec![ + (t0, "Real Video Title".to_string()), + (t1, "later window".to_string()), + ], + "peer pull must pick up the owner-originated edit from staging" + ); +} + +#[test] +fn edits_older_than_lookback_are_not_reconciled() { + // Peak-memory bound: only the 7 days before the resume cursor are compared. + let (src, dest) = memory_pair(); + let bucket = create_bucket(&src, "aw-watcher-android-lookback", "phone"); + let t_old = Utc::now() - Duration::days(10); + let t_new = Utc::now() - Duration::minutes(5); + src.insert_events( + &bucket, + &[ + event_at(t_old, Duration::minutes(20), "sanitized-old.mp4"), + event_at(t_new, Duration::minutes(5), "recent window"), + ], + ) + .unwrap(); + src.force_commit().unwrap(); + sync_push(&src, &dest); + edit_title(&src, &bucket, t_old, "Too Old To Reconcile"); + sync_push(&src, &dest); + assert_eq!( + titles(&dest, &bucket), + vec![ + (t_old, "sanitized-old.mp4".to_string()), + (t_new, "recent window".to_string()), + ], + "edits older than 7 days stay on the resume-cursor path" + ); +} + +#[test] +fn wiping_destination_reexports_source_edits() { + // Control: a truly empty destination (internal staging gone) re-reads source + // from the beginning. Deleting only the Android SAF mirror does not do this. + let src = Datastore::new_in_memory(false); + let dest = Datastore::new_in_memory(false); + let bucket = create_bucket(&src, "aw-watcher-android-wipe", "phone"); + let t0 = Utc::now() - Duration::hours(2); + let t1 = t0 + Duration::minutes(30); + src.insert_events( + &bucket, + &[ + event_at(t0, Duration::minutes(20), "sanitized.mp4"), + event_at(t1, Duration::minutes(5), "later window"), + ], + ) + .unwrap(); + src.force_commit().unwrap(); + sync_push(&src, &dest); + edit_title(&src, &bucket, t0, "Real Video Title"); + + let fresh = Datastore::new_in_memory(false); + sync_push(&src, &fresh); + assert_eq!( + titles(&fresh, &bucket), + vec![ + (t0, "Real Video Title".to_string()), + (t1, "later window".to_string()), + ] + ); +} From a80a62ff59c0b6df6687bffc55fb0f456c33289f Mon Sep 17 00:00:00 2001 From: Bob Date: Mon, 14 Sep 2026 03:36:42 +0000 Subject: [PATCH 2/8] fix(sync): match edit reconcile on timestamp and duration A timestamp-only map collapsed two dest events that shared a start time. Identity is (timestamp, duration) so a sibling row is not replaced. Git-Session-Id: 0d71 --- aw-sync/src/sync.rs | 51 ++++++++++++++++++++------------ aw-sync/tests/historical_edit.rs | 50 +++++++++++++++++++++++++++++++ 2 files changed, 82 insertions(+), 19 deletions(-) diff --git a/aw-sync/src/sync.rs b/aw-sync/src/sync.rs index 9f3786d9..fc3a4b39 100644 --- a/aw-sync/src/sync.rs +++ b/aw-sync/src/sync.rs @@ -438,20 +438,29 @@ pub fn sync_datastores( Ok(()) } -/// Replace dest events whose timestamp still exists on the source but whose -/// data or duration changed. +fn event_identity(event: &Event) -> (DateTime, i64) { + ( + event.timestamp, + event.duration.num_nanoseconds().unwrap_or(0), + ) +} + +/// Replace dest events whose timestamp+duration still exist on the source but +/// whose data changed. /// -/// WebUI/Android title edits are delete+insert at the same timestamp. The -/// resume cursor starts at the destination's latest event end, so those -/// replacements are outside the incremental fetch window. Under the +/// WebUI/Android title edits are delete+insert at the same timestamp and +/// duration. The resume cursor starts at the destination's latest event end, +/// so those replacements are outside the incremental fetch window. Under the /// single-writer model the source is authoritative for its own buckets, so a -/// timestamp match with different data is an owner-originated edit, not a -/// conflict. +/// same-identity row with different data is an owner-originated edit, not a +/// conflict. Identity includes duration so two events that share a timestamp +/// are not collapsed into one HashMap slot. /// -/// Must run *before* the incremental copy: a latest-event title edit is also -/// re-fetched as a start-clipped fragment, and heartbeat() refuses to merge -/// different data, which would insert a duplicate unless dest already holds -/// the new payload. +/// Duration-only updates of the live last event stay on the incremental +/// heartbeat path. Must run *before* that copy: a latest-event title edit is +/// also re-fetched as a start-clipped fragment, and heartbeat() refuses to +/// merge different data, which would insert a duplicate unless dest already +/// holds the new payload. fn reconcile_updated_events( ds_from: &dyn AccessMethod, ds_to: &dyn AccessMethod, @@ -471,22 +480,26 @@ fn reconcile_updated_events( .get_events(bucket_to.id.as_str(), Some(lookback_start), None, None) .unwrap(); - let dest_by_ts: HashMap, Event> = dest_events - .into_iter() - .filter(|e| e.timestamp >= lookback_start) - .map(|e| (e.timestamp, e)) - .collect(); + let mut dest_by_identity: HashMap<(DateTime, i64), Vec> = HashMap::new(); + for event in dest_events { + if event.timestamp >= lookback_start { + dest_by_identity + .entry(event_identity(&event)) + .or_default() + .push(event); + } + } for src in source_events { if src.timestamp < lookback_start { continue; } - let Some(dst) = dest_by_ts.get(&src.timestamp) else { + let Some(dsts) = dest_by_identity.get(&event_identity(&src)) else { continue; }; - if dst.data == src.data && dst.duration == src.duration { + let Some(dst) = dsts.iter().find(|dst| dst.data != src.data) else { continue; - } + }; let Some(dst_id) = dst.id else { warn!( "Cannot reconcile event at {:?} — dest event has no id", diff --git a/aw-sync/tests/historical_edit.rs b/aw-sync/tests/historical_edit.rs index ed9158b1..ce3dd6ee 100644 --- a/aw-sync/tests/historical_edit.rs +++ b/aw-sync/tests/historical_edit.rs @@ -215,6 +215,56 @@ fn historical_title_edit_reaches_peer() { ); } +#[test] +fn same_timestamp_sibling_is_not_clobbered() { + let (src, dest) = memory_pair(); + let bucket = create_bucket(&src, "aw-watcher-android-sibling", "phone"); + let t0 = Utc::now() - Duration::hours(2); + let t1 = t0 + Duration::minutes(40); + src.insert_events( + &bucket, + &[ + event_at(t0, Duration::minutes(20), "sanitized.mp4"), + event_at(t0, Duration::minutes(5), "unrelated sibling"), + event_at(t1, Duration::minutes(3), "later window"), + ], + ) + .unwrap(); + src.force_commit().unwrap(); + sync_push(&src, &dest); + let target = src + .get_events(&bucket, None, None, None) + .unwrap() + .into_iter() + .find(|e| e.data.get("title").and_then(|v| v.as_str()) == Some("sanitized.mp4")) + .unwrap(); + src.delete_events_by_id(&bucket, vec![target.id.unwrap()]) + .unwrap(); + let mut data = target.data.clone(); + data.insert("title".to_string(), json!("Real Video Title")); + src.insert_events( + &bucket, + &[Event { + id: None, + timestamp: target.timestamp, + duration: target.duration, + data, + }], + ) + .unwrap(); + src.force_commit().unwrap(); + sync_push(&src, &dest); + + let got = titles(&dest, &bucket); + assert_eq!(got.len(), 3); + assert!(got.contains(&(t0, "Real Video Title".to_string()))); + assert!( + got.contains(&(t0, "unrelated sibling".to_string())), + "a same-timestamp event with a different duration must keep its title: {got:?}" + ); + assert!(got.contains(&(t1, "later window".to_string()))); +} + #[test] fn edits_older_than_lookback_are_not_reconciled() { // Peak-memory bound: only the 7 days before the resume cursor are compared. From 8594c85ba2a4f882996ec7e94f0899c79d4a251b Mon Sep 17 00:00:00 2001 From: Bob Date: Mon, 14 Sep 2026 03:38:27 +0000 Subject: [PATCH 3/8] fix(sync): cap edit-reconcile fetch and skip post-cursor rows Keep the lookback query unclipped so dest-latest identity stays intact, skip source events that end after the resume cursor, and cap both fetches at 20k events so a dense 7-day window cannot OOM. Git-Session-Id: 0d71 --- aw-sync/src/sync.rs | 24 ++++++++++++++++++++++-- 1 file changed, 22 insertions(+), 2 deletions(-) diff --git a/aw-sync/src/sync.rs b/aw-sync/src/sync.rs index fc3a4b39..748f273c 100644 --- a/aw-sync/src/sync.rs +++ b/aw-sync/src/sync.rs @@ -317,6 +317,9 @@ const BATCH_SIZE: usize = 5; /// already-synced events (ActivityWatch/aw-android#253). Bounded so a full /// historical bucket is never loaded into memory on Android. const EDIT_RECONCILE_LOOKBACK: Duration = Duration::days(7); +/// Hard cap on the lookback fetch (newest first). Time bound alone is not a +/// memory bound if a bucket is extremely dense. +const EDIT_RECONCILE_EVENT_CAP: u64 = 20_000; /// Whether a bucket holds data synced from another host, rather than data /// collected on this host. @@ -473,11 +476,25 @@ fn reconcile_updated_events( }; let lookback_start = resume - EDIT_RECONCILE_LOOKBACK; + // end=None on purpose: an end bound would clip the dest-latest event's + // duration and break (timestamp, duration) identity. Skip source rows + // that end after `resume` in the loop instead; those belong to the + // incremental copy. Cap both fetches so a dense 7-day window cannot OOM. let source_events = ds_from - .get_events(bucket_from.id.as_str(), Some(lookback_start), None, None) + .get_events( + bucket_from.id.as_str(), + Some(lookback_start), + None, + Some(EDIT_RECONCILE_EVENT_CAP), + ) .unwrap(); let dest_events = ds_to - .get_events(bucket_to.id.as_str(), Some(lookback_start), None, None) + .get_events( + bucket_to.id.as_str(), + Some(lookback_start), + None, + Some(EDIT_RECONCILE_EVENT_CAP), + ) .unwrap(); let mut dest_by_identity: HashMap<(DateTime, i64), Vec> = HashMap::new(); @@ -494,6 +511,9 @@ fn reconcile_updated_events( if src.timestamp < lookback_start { continue; } + if src.timestamp + src.duration > resume { + continue; + } let Some(dsts) = dest_by_identity.get(&event_identity(&src)) else { continue; }; From 4f50795adf621b5e0f1b98800c2bc202c3b7d937 Mon Sep 17 00:00:00 2001 From: Bob Date: Mon, 14 Sep 2026 03:39:44 +0000 Subject: [PATCH 4/8] fix(sync): skip only post-cursor starts in edit reconcile Filtering on event end dropped dest-latest title edits whose duration grew past the old resume cursor. Skip when timestamp >= resume instead. Git-Session-Id: 0d71 --- aw-sync/src/sync.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/aw-sync/src/sync.rs b/aw-sync/src/sync.rs index 748f273c..ee11e7e9 100644 --- a/aw-sync/src/sync.rs +++ b/aw-sync/src/sync.rs @@ -511,7 +511,10 @@ fn reconcile_updated_events( if src.timestamp < lookback_start { continue; } - if src.timestamp + src.duration > resume { + // Skip events that start at/after the dest cursor; the incremental + // copy owns those. Do not use end>resume: the dest-latest event starts + // before resume and must still be title-reconciled. + if src.timestamp >= resume { continue; } let Some(dsts) = dest_by_identity.get(&event_identity(&src)) else { From 8ca47fd370569076308e8e29dfa0b7617d619c72 Mon Sep 17 00:00:00 2001 From: Bob Date: Mon, 14 Sep 2026 03:41:21 +0000 Subject: [PATCH 5/8] fix(sync): insert replacement before deleting the stale dest row A crash between delete and insert would drop the event, and the next resume-cursor pass would not resurrect it. Insert first; a later pass skips insert when matching data is already present. Git-Session-Id: 0d71 --- aw-sync/src/sync.rs | 32 +++++++++++++++++--------------- 1 file changed, 17 insertions(+), 15 deletions(-) diff --git a/aw-sync/src/sync.rs b/aw-sync/src/sync.rs index ee11e7e9..152d8b9a 100644 --- a/aw-sync/src/sync.rs +++ b/aw-sync/src/sync.rs @@ -520,24 +520,26 @@ fn reconcile_updated_events( let Some(dsts) = dest_by_identity.get(&event_identity(&src)) else { continue; }; - let Some(dst) = dsts.iter().find(|dst| dst.data != src.data) else { - continue; - }; - let Some(dst_id) = dst.id else { - warn!( - "Cannot reconcile event at {:?} — dest event has no id", - src.timestamp - ); + let stale: Vec = dsts + .iter() + .filter(|dst| dst.data != src.data) + .filter_map(|dst| dst.id) + .collect(); + if stale.is_empty() { continue; - }; - ds_to - .delete_events_by_id(bucket_to.id.as_str(), vec![dst_id]) - .unwrap(); + } let ts = src.timestamp; - let mut replacement = src; - replacement.id = None; + // Insert before delete so a crash cannot drop the row. A later pass + // sees matching data, skips insert, and still removes remaining stale ids. + if !dsts.iter().any(|dst| dst.data == src.data) { + let mut replacement = src; + replacement.id = None; + ds_to + .insert_events(bucket_to.id.as_str(), vec![replacement]) + .unwrap(); + } ds_to - .insert_events(bucket_to.id.as_str(), vec![replacement]) + .delete_events_by_id(bucket_to.id.as_str(), stale) .unwrap(); info!(" ~ Reconciled edited event at {:?}", ts); } From aef0f78a0997fe410f5f0eede77c6294d1268278 Mon Sep 17 00:00:00 2001 From: Bob Date: Mon, 14 Sep 2026 03:48:02 +0000 Subject: [PATCH 6/8] fix(sync): drop in-window event cap on edit reconcile A 20k newest-first cap skipped older edits still inside the 7-day lookback. The time window is the memory bound. Git-Session-Id: 0d71 --- aw-sync/src/sync.rs | 22 +++++----------------- 1 file changed, 5 insertions(+), 17 deletions(-) diff --git a/aw-sync/src/sync.rs b/aw-sync/src/sync.rs index 152d8b9a..8000f931 100644 --- a/aw-sync/src/sync.rs +++ b/aw-sync/src/sync.rs @@ -317,9 +317,6 @@ const BATCH_SIZE: usize = 5; /// already-synced events (ActivityWatch/aw-android#253). Bounded so a full /// historical bucket is never loaded into memory on Android. const EDIT_RECONCILE_LOOKBACK: Duration = Duration::days(7); -/// Hard cap on the lookback fetch (newest first). Time bound alone is not a -/// memory bound if a bucket is extremely dense. -const EDIT_RECONCILE_EVENT_CAP: u64 = 20_000; /// Whether a bucket holds data synced from another host, rather than data /// collected on this host. @@ -478,23 +475,14 @@ fn reconcile_updated_events( // end=None on purpose: an end bound would clip the dest-latest event's // duration and break (timestamp, duration) identity. Skip source rows - // that end after `resume` in the loop instead; those belong to the - // incremental copy. Cap both fetches so a dense 7-day window cannot OOM. + // that start at/after `resume` in the loop instead; those belong to the + // incremental copy. The 7-day lookback is the memory bound; do not also + // cap by count, which would silently skip older in-window edits. let source_events = ds_from - .get_events( - bucket_from.id.as_str(), - Some(lookback_start), - None, - Some(EDIT_RECONCILE_EVENT_CAP), - ) + .get_events(bucket_from.id.as_str(), Some(lookback_start), None, None) .unwrap(); let dest_events = ds_to - .get_events( - bucket_to.id.as_str(), - Some(lookback_start), - None, - Some(EDIT_RECONCILE_EVENT_CAP), - ) + .get_events(bucket_to.id.as_str(), Some(lookback_start), None, None) .unwrap(); let mut dest_by_identity: HashMap<(DateTime, i64), Vec> = HashMap::new(); From 1379d77aeb4a723d1042a4fe24194116daed158a Mon Sep 17 00:00:00 2001 From: Bob Date: Mon, 14 Sep 2026 04:09:14 +0000 Subject: [PATCH 7/8] fix(sync): bound edit reconcile and keep same-identity siblings end=None loaded the entire post-cursor source backlog when dest was behind. Fetch through resume instead. Match source and dest by identity as a group so a sibling with matching data is not deleted as stale. Git-Session-Id: 175f4625-aa96-5219-a447-7df0d9f1a988 --- aw-sync/src/sync.rs | 88 ++++++++++++++------- aw-sync/tests/historical_edit.rs | 132 +++++++++++++++++++++++++++++++ 2 files changed, 192 insertions(+), 28 deletions(-) diff --git a/aw-sync/src/sync.rs b/aw-sync/src/sync.rs index 8000f931..d482c10a 100644 --- a/aw-sync/src/sync.rs +++ b/aw-sync/src/sync.rs @@ -473,21 +473,34 @@ fn reconcile_updated_events( }; let lookback_start = resume - EDIT_RECONCILE_LOOKBACK; - // end=None on purpose: an end bound would clip the dest-latest event's - // duration and break (timestamp, duration) identity. Skip source rows - // that start at/after `resume` in the loop instead; those belong to the - // incremental copy. The 7-day lookback is the memory bound; do not also - // cap by count, which would silently skip older in-window edits. + // Bound both fetches to the lookback window ending at `resume`. + // end=None would load every newer source event when dest is far behind, + // exhausting Android RAM and bypassing the paginated incremental copy. + // get_events clips to the query range; dest-latest ends at `resume`, so + // that clip is a no-op on its (timestamp, duration) identity. Title edits + // keep duration; duration-only updates stay on the heartbeat path. Do not + // also cap by count — a newest-first cap silently skips older in-window + // edits. let source_events = ds_from - .get_events(bucket_from.id.as_str(), Some(lookback_start), None, None) + .get_events( + bucket_from.id.as_str(), + Some(lookback_start), + Some(resume), + None, + ) .unwrap(); let dest_events = ds_to - .get_events(bucket_to.id.as_str(), Some(lookback_start), None, None) + .get_events( + bucket_to.id.as_str(), + Some(lookback_start), + Some(resume), + None, + ) .unwrap(); let mut dest_by_identity: HashMap<(DateTime, i64), Vec> = HashMap::new(); for event in dest_events { - if event.timestamp >= lookback_start { + if event.timestamp >= lookback_start && event.timestamp < resume { dest_by_identity .entry(event_identity(&event)) .or_default() @@ -495,40 +508,59 @@ fn reconcile_updated_events( } } + let mut src_by_identity: HashMap<(DateTime, i64), Vec> = HashMap::new(); for src in source_events { - if src.timestamp < lookback_start { - continue; - } // Skip events that start at/after the dest cursor; the incremental // copy owns those. Do not use end>resume: the dest-latest event starts // before resume and must still be title-reconciled. - if src.timestamp >= resume { + if src.timestamp < lookback_start || src.timestamp >= resume { continue; } - let Some(dsts) = dest_by_identity.get(&event_identity(&src)) else { - continue; + src_by_identity + .entry(event_identity(&src)) + .or_default() + .push(src); + } + + for (identity, srcs) in src_by_identity { + let mut dsts = match dest_by_identity.remove(&identity) { + Some(dsts) if !dsts.is_empty() => dsts, + _ => continue, }; - let stale: Vec = dsts - .iter() - .filter(|dst| dst.data != src.data) - .filter_map(|dst| dst.id) - .collect(); - if stale.is_empty() { + let mut to_insert = Vec::new(); + for src in srcs { + if let Some(idx) = dsts.iter().position(|dst| dst.data == src.data) { + // Still present on dest — keep it, including same-identity + // siblings a later source row must not treat as stale. + dsts.remove(idx); + } else { + to_insert.push(src); + } + } + if to_insert.is_empty() && dsts.is_empty() { continue; } - let ts = src.timestamp; + let ts = identity.0; // Insert before delete so a crash cannot drop the row. A later pass // sees matching data, skips insert, and still removes remaining stale ids. - if !dsts.iter().any(|dst| dst.data == src.data) { - let mut replacement = src; - replacement.id = None; + if !to_insert.is_empty() { + let replacements: Vec = to_insert + .into_iter() + .map(|mut src| { + src.id = None; + src + }) + .collect(); + ds_to + .insert_events(bucket_to.id.as_str(), replacements) + .unwrap(); + } + let stale: Vec = dsts.into_iter().filter_map(|dst| dst.id).collect(); + if !stale.is_empty() { ds_to - .insert_events(bucket_to.id.as_str(), vec![replacement]) + .delete_events_by_id(bucket_to.id.as_str(), stale) .unwrap(); } - ds_to - .delete_events_by_id(bucket_to.id.as_str(), stale) - .unwrap(); info!(" ~ Reconciled edited event at {:?}", ts); } } diff --git a/aw-sync/tests/historical_edit.rs b/aw-sync/tests/historical_edit.rs index ce3dd6ee..99e3e863 100644 --- a/aw-sync/tests/historical_edit.rs +++ b/aw-sync/tests/historical_edit.rs @@ -215,6 +215,138 @@ fn historical_title_edit_reaches_peer() { ); } +#[test] +fn same_identity_siblings_survive_noop_sync() { + // Datastore does not require (timestamp, duration) to be unique. A + // per-source-row pass against a frozen dest snapshot treated every other + // sibling as stale and deleted them all, including on a no-op second sync. + let (src, dest) = memory_pair(); + let bucket = create_bucket(&src, "aw-watcher-android-identity-sib", "phone"); + let t0 = Utc::now() - Duration::hours(2); + let t1 = t0 + Duration::minutes(40); + src.insert_events( + &bucket, + &[ + event_at(t0, Duration::minutes(20), "alpha"), + event_at(t0, Duration::minutes(20), "bravo"), + event_at(t1, Duration::minutes(3), "later window"), + ], + ) + .unwrap(); + src.force_commit().unwrap(); + sync_push(&src, &dest); + sync_push(&src, &dest); + + let got = titles(&dest, &bucket); + assert_eq!( + got.len(), + 3, + "noop reconcile must not drop same-identity siblings: {got:?}" + ); + assert!(got.contains(&(t0, "alpha".to_string()))); + assert!(got.contains(&(t0, "bravo".to_string()))); + assert!(got.contains(&(t1, "later window".to_string()))); +} + +#[test] +fn same_identity_sibling_edit_keeps_the_other() { + let (src, dest) = memory_pair(); + let bucket = create_bucket(&src, "aw-watcher-android-identity-edit", "phone"); + let t0 = Utc::now() - Duration::hours(2); + let t1 = t0 + Duration::minutes(40); + src.insert_events( + &bucket, + &[ + event_at(t0, Duration::minutes(20), "alpha"), + event_at(t0, Duration::minutes(20), "bravo"), + event_at(t1, Duration::minutes(3), "later window"), + ], + ) + .unwrap(); + src.force_commit().unwrap(); + sync_push(&src, &dest); + + let target = src + .get_events(&bucket, None, None, None) + .unwrap() + .into_iter() + .find(|e| e.data.get("title").and_then(|v| v.as_str()) == Some("alpha")) + .unwrap(); + src.delete_events_by_id(&bucket, vec![target.id.unwrap()]) + .unwrap(); + let mut data = target.data.clone(); + data.insert("title".to_string(), json!("alpha-edited")); + src.insert_events( + &bucket, + &[Event { + id: None, + timestamp: target.timestamp, + duration: target.duration, + data, + }], + ) + .unwrap(); + src.force_commit().unwrap(); + sync_push(&src, &dest); + + let got = titles(&dest, &bucket); + assert_eq!( + got.len(), + 3, + "editing one sibling must not drop the other: {got:?}" + ); + assert!(got.contains(&(t0, "alpha-edited".to_string()))); + assert!( + got.contains(&(t0, "bravo".to_string())), + "same timestamp+duration sibling must keep its title: {got:?}" + ); + assert!(got.contains(&(t1, "later window".to_string()))); +} + +#[test] +fn far_behind_dest_still_copies_post_cursor_backlog() { + // Reconcile must not load the post-cursor source backlog (end=resume). + // Those events belong to the bounded incremental copy. + let (src, dest) = memory_pair(); + let bucket = create_bucket(&src, "aw-watcher-android-far-behind", "phone"); + let t_old = Utc::now() - Duration::days(10); + let t_cursor = t_old + Duration::hours(2); + src.insert_events( + &bucket, + &[ + event_at(t_old, Duration::minutes(20), "sanitized.mp4"), + event_at(t_cursor, Duration::minutes(5), "cursor window"), + ], + ) + .unwrap(); + src.force_commit().unwrap(); + sync_push(&src, &dest); + + edit_title(&src, &bucket, t_old, "Real Video Title"); + let mut later = Vec::new(); + for i in 0..12 { + later.push(event_at( + t_cursor + Duration::days(1) + Duration::minutes(i), + Duration::minutes(1), + &format!("later-{i}"), + )); + } + src.insert_events(&bucket, &later).unwrap(); + src.force_commit().unwrap(); + sync_push(&src, &dest); + + let got = titles(&dest, &bucket); + assert_eq!( + got.len(), + 14, + "lookback edit plus paginated backlog: {got:?}" + ); + assert!(got.contains(&(t_old, "Real Video Title".to_string()))); + assert!(got.contains(&(t_cursor, "cursor window".to_string()))); + assert!(got.contains(&(later[0].timestamp, "later-0".to_string()))); + assert!(got.contains(&(later[11].timestamp, "later-11".to_string()))); +} + #[test] fn same_timestamp_sibling_is_not_clobbered() { let (src, dest) = memory_pair(); From eb7e751c401a3a688b1bf4c406e20088af9322a3 Mon Sep 17 00:00:00 2001 From: Bob Date: Wed, 16 Sep 2026 10:07:53 +0000 Subject: [PATCH 8/8] fix(sync): propagate reconcile datastore errors instead of unwrap reconcile_updated_events now returns Result and uses ? on get_events, insert_events, and delete_events_by_id. A datastore failure costs that bucket's reconcile, not a panic that aborts the pass (or the JNI frame on Android). force_commit in delete_events_by_id matches the other AccessMethod methods. Git-Session-Id: 82e7d9e5-0c8a-58c0-b129-a46f96ef6953 --- aw-sync/src/accessmethod.rs | 2 +- aw-sync/src/sync.rs | 47 +++++++++++++++++-------------------- 2 files changed, 23 insertions(+), 26 deletions(-) diff --git a/aw-sync/src/accessmethod.rs b/aw-sync/src/accessmethod.rs index 7160ce04..7bc72c32 100644 --- a/aw-sync/src/accessmethod.rs +++ b/aw-sync/src/accessmethod.rs @@ -68,7 +68,7 @@ impl AccessMethod for Datastore { } fn delete_events_by_id(&self, bucket_id: &str, event_ids: Vec) -> Result<(), String> { Datastore::delete_events_by_id(self, bucket_id, event_ids).map_err(|e| format!("{e:?}"))?; - self.force_commit().unwrap(); + self.force_commit().map_err(|e| format!("{e:?}"))?; Ok(()) } } diff --git a/aw-sync/src/sync.rs b/aw-sync/src/sync.rs index d482c10a..99a4b7a1 100644 --- a/aw-sync/src/sync.rs +++ b/aw-sync/src/sync.rs @@ -467,9 +467,9 @@ fn reconcile_updated_events( bucket_from: &Bucket, bucket_to: &Bucket, resume_sync_at: Option>, -) { +) -> Result<(), String> { let Some(resume) = resume_sync_at else { - return; + return Ok(()); }; let lookback_start = resume - EDIT_RECONCILE_LOOKBACK; @@ -481,22 +481,22 @@ fn reconcile_updated_events( // keep duration; duration-only updates stay on the heartbeat path. Do not // also cap by count — a newest-first cap silently skips older in-window // edits. - let source_events = ds_from - .get_events( - bucket_from.id.as_str(), - Some(lookback_start), - Some(resume), - None, - ) - .unwrap(); - let dest_events = ds_to - .get_events( - bucket_to.id.as_str(), - Some(lookback_start), - Some(resume), - None, - ) - .unwrap(); + // + // Datastore errors return rather than unwrap: a panic here aborts the + // whole pass (and on Android, the JNI frame). The per-bucket skip in + // ActivityWatch/aw-server-rust#697 then drops this bucket, not the daemon. + let source_events = ds_from.get_events( + bucket_from.id.as_str(), + Some(lookback_start), + Some(resume), + None, + )?; + let dest_events = ds_to.get_events( + bucket_to.id.as_str(), + Some(lookback_start), + Some(resume), + None, + )?; let mut dest_by_identity: HashMap<(DateTime, i64), Vec> = HashMap::new(); for event in dest_events { @@ -551,18 +551,15 @@ fn reconcile_updated_events( src }) .collect(); - ds_to - .insert_events(bucket_to.id.as_str(), replacements) - .unwrap(); + ds_to.insert_events(bucket_to.id.as_str(), replacements)?; } let stale: Vec = dsts.into_iter().filter_map(|dst| dst.id).collect(); if !stale.is_empty() { - ds_to - .delete_events_by_id(bucket_to.id.as_str(), stale) - .unwrap(); + ds_to.delete_events_by_id(bucket_to.id.as_str(), stale)?; } info!(" ~ Reconciled edited event at {:?}", ts); } + Ok(()) } /// Syncs a single bucket from one datastore to another @@ -594,7 +591,7 @@ fn sync_one( info!(" + Starting from beginning"); } - reconcile_updated_events(ds_from, ds_to, &bucket_from, &bucket_to, resume_sync_at); + reconcile_updated_events(ds_from, ds_to, &bucket_from, &bucket_to, resume_sync_at)?; // Fetch events in bounded chunks to avoid OOM on devices with limited RAM (e.g. Android). // get_events returns events in descending order (newest first), so we paginate backwards