From 686ec85f59d5e2c13deb2ec44a8c2a081b6169fd Mon Sep 17 00:00:00 2001 From: Bob Date: Thu, 27 Aug 2026 23:22:34 +0000 Subject: [PATCH 1/3] fix(datastore): merge disjoint aw-watcher-android-test events UPDATE OR IGNORE skipped the collision case where both the legacy aw-watcher-android-test_* bucket and the canonical aw-watcher-android_* bucket exist, leaving years of history stranded after upgrades (ActivityWatch/aw-android#243). Move non-overlapping legacy events into the canonical bucket, leave overlapping events in the legacy bucket, and delete the legacy bucket only once it is empty. Replace the in-memory bucket cache from SQLite after the rewrite so deleted buckets disappear. Refs: ActivityWatch/aw-android#149, ActivityWatch/aw-android#150 --- aw-datastore/src/datastore.rs | 130 +++++++++++++++++++++++------- aw-datastore/tests/datastore.rs | 136 ++++++++++++++++++++++++++++++++ 2 files changed, 239 insertions(+), 27 deletions(-) diff --git a/aw-datastore/src/datastore.rs b/aw-datastore/src/datastore.rs index 66e4df4a..082b77f2 100644 --- a/aw-datastore/src/datastore.rs +++ b/aw-datastore/src/datastore.rs @@ -318,10 +318,11 @@ impl DatastoreInstance { ))) } }; + let mut new_cache = HashMap::new(); for bucket in buckets { match bucket { Ok(b) => { - self.buckets_cache.insert(b.id.clone(), b.clone()); + new_cache.insert(b.id.clone(), b); } Err(e) => { return Err(DatastoreError::InternalError(format!( @@ -330,6 +331,7 @@ impl DatastoreInstance { } } } + self.buckets_cache = new_cache; Ok(()) } @@ -1113,39 +1115,113 @@ impl DatastoreInstance { } /// Migrates all buckets whose name starts with `aw-watcher-android-test` to use - /// `aw-watcher-android` instead. This covers the old debug-build bucket naming - /// convention (e.g. `aw-watcher-android-test_hostname` → `aw-watcher-android_hostname`). - /// Events are left untouched; only the bucket metadata is updated. - /// Returns the number of buckets that were migrated. - /// Note: if a UNIQUE constraint violation occurs on any single row, `UPDATE OR IGNORE` - /// will skip conflicting rows instead of aborting the entire batch. + /// `aw-watcher-android` instead. This covers the old production bucket naming + /// convention (e.g. `aw-watcher-android-test_phone` → `aw-watcher-android_phone`). + /// + /// If the destination already exists, disjoint legacy events are moved into it. + /// Events that overlap a destination event stay in the legacy bucket so the + /// migration cannot create duplicate activity records. The legacy bucket is + /// deleted only after it is empty. + /// Returns the number of legacy buckets that were fully renamed or merged. pub fn migrate_test_bucket_names( &mut self, conn: &Connection, ) -> Result { - info!("Migrating 'aw-watcher-android-test' bucket names to 'aw-watcher-android'"); - - let updated = match conn.execute( - "UPDATE OR IGNORE buckets SET name = 'aw-watcher-android' || SUBSTR(name, LENGTH('aw-watcher-android-test') + 1) \ - WHERE name LIKE 'aw-watcher-android-test%'", - [], - ) { - Ok(n) => n, - Err(err) => { - return Err(DatastoreError::InternalError(format!( - "Failed to migrate test bucket names: {err}" - ))) + const OLD_PREFIX: &str = "aw-watcher-android-test"; + const NEW_PREFIX: &str = "aw-watcher-android"; + + info!("Migrating '{OLD_PREFIX}' bucket names to '{NEW_PREFIX}'"); + let legacy_ids: Vec = self + .buckets_cache + .keys() + .filter(|id| id.starts_with(OLD_PREFIX)) + .cloned() + .collect(); + let mut migrated = 0; + let mut cache_dirty = false; + + for old_id in legacy_ids { + let new_id = old_id.replacen(OLD_PREFIX, NEW_PREFIX, 1); + if let Some(new_bucket) = self.buckets_cache.get(&new_id).cloned() { + let old_bucket = self + .buckets_cache + .get(&old_id) + .cloned() + .ok_or_else(|| DatastoreError::NoSuchBucket(old_id.clone()))?; + + // Move only events that do not overlap any destination event. + // A single overlapping cutover heartbeat must not strand years of + // disjoint history in the legacy bucket (ActivityWatch/aw-android#243). + conn.execute( + "UPDATE events SET bucketrow = ?1 + WHERE id IN ( + SELECT old_event.id FROM events AS old_event + WHERE old_event.bucketrow = ?2 + AND NOT EXISTS ( + SELECT 1 FROM events AS new_event + WHERE new_event.bucketrow = ?1 + AND old_event.starttime < new_event.endtime + AND new_event.starttime < old_event.endtime + ) + )", + [new_bucket.bid, old_bucket.bid], + ) + .map_err(|err| { + DatastoreError::InternalError(format!( + "Failed to merge bucket '{}' into '{}': {err}", + old_id, new_id + )) + })?; + cache_dirty = true; + + let remaining: i64 = conn + .query_row( + "SELECT COUNT(*) FROM events WHERE bucketrow = ?1", + [old_bucket.bid], + |row| row.get(0), + ) + .map_err(|err| { + DatastoreError::InternalError(format!( + "Failed to count leftover events in '{}': {err}", + old_id + )) + })?; + if remaining == 0 { + conn.execute("DELETE FROM buckets WHERE id = ?1", [old_bucket.bid]) + .map_err(|err| { + DatastoreError::InternalError(format!( + "Failed to remove merged bucket '{}': {err}", + old_id + )) + })?; + info!("Merged legacy bucket '{}' into '{}'", old_id, new_id); + migrated += 1; + } else { + warn!( + "Partially merged '{}' into '{}'; {} overlapping event(s) remain in the legacy bucket", + old_id, new_id, remaining + ); + } + } else { + conn.execute( + "UPDATE buckets SET name = ?1 WHERE name = ?2", + [&new_id, &old_id], + ) + .map_err(|err| { + DatastoreError::InternalError(format!( + "Failed to rename bucket '{}' to '{}': {err}", + old_id, new_id + )) + })?; + info!("Renamed legacy bucket '{}' to '{}'", old_id, new_id); + migrated += 1; + cache_dirty = true; } - }; + } - if updated > 0 { - info!("Migrated {} 'aw-watcher-android-test' bucket(s)", updated); - // Refresh the in-memory cache so callers see the new names immediately. + if cache_dirty { self.get_stored_buckets(conn)?; - } else { - info!("No 'aw-watcher-android-test' buckets found; nothing to migrate"); } - - Ok(updated) + Ok(migrated) } } diff --git a/aw-datastore/tests/datastore.rs b/aw-datastore/tests/datastore.rs index fe353e2a..7b785b3a 100644 --- a/aw-datastore/tests/datastore.rs +++ b/aw-datastore/tests/datastore.rs @@ -60,6 +60,142 @@ mod datastore_tests { bucket } + fn create_named_test_bucket(ds: &Datastore, id: &str) -> Bucket { + let mut bucket = test_bucket(); + bucket.id = id.to_string(); + ds.create_bucket(&bucket).unwrap(); + bucket + } + + fn test_event(timestamp: chrono::DateTime, duration: Duration) -> Event { + Event { + id: None, + timestamp, + duration, + data: json_map! {"key": json!("value")}, + } + } + + #[test] + fn test_migrate_test_bucket_names_renames_bucket_and_preserves_events() { + let ds = Datastore::new_in_memory(false); + let old_id = "aw-watcher-android-test_phone"; + let new_id = "aw-watcher-android_phone"; + create_named_test_bucket(&ds, old_id); + let event = test_event(Utc::now(), Duration::seconds(30)); + ds.insert_events(old_id, std::slice::from_ref(&event)) + .unwrap(); + + assert_eq!(ds.migrate_test_bucket_names().unwrap(), 1); + assert!(!ds.get_buckets().unwrap().contains_key(old_id)); + assert!(ds.get_buckets().unwrap().contains_key(new_id)); + assert_eq!(ds.get_events(new_id, None, None, None).unwrap().len(), 1); + } + + #[test] + fn test_migrate_test_bucket_names_merges_non_overlapping_buckets() { + let ds = Datastore::new_in_memory(false); + let old_id = "aw-watcher-android-test_phone"; + let new_id = "aw-watcher-android_phone"; + create_named_test_bucket(&ds, old_id); + create_named_test_bucket(&ds, new_id); + let now = Utc::now(); + ds.insert_events( + old_id, + &[test_event(now - Duration::hours(2), Duration::minutes(30))], + ) + .unwrap(); + ds.insert_events(new_id, &[test_event(now, Duration::minutes(30))]) + .unwrap(); + + assert_eq!(ds.migrate_test_bucket_names().unwrap(), 1); + assert!(!ds.get_buckets().unwrap().contains_key(old_id)); + assert_eq!(ds.get_events(new_id, None, None, None).unwrap().len(), 2); + } + + #[test] + fn test_migrate_test_bucket_names_keeps_overlapping_buckets_separate() { + let ds = Datastore::new_in_memory(false); + let old_id = "aw-watcher-android-test_phone"; + let new_id = "aw-watcher-android_phone"; + create_named_test_bucket(&ds, old_id); + create_named_test_bucket(&ds, new_id); + let now = Utc::now(); + ds.insert_events(old_id, &[test_event(now, Duration::minutes(30))]) + .unwrap(); + ds.insert_events( + new_id, + &[test_event( + now + Duration::minutes(15), + Duration::minutes(30), + )], + ) + .unwrap(); + + assert_eq!(ds.migrate_test_bucket_names().unwrap(), 0); + assert!(ds.get_buckets().unwrap().contains_key(old_id)); + assert_eq!(ds.get_events(old_id, None, None, None).unwrap().len(), 1); + assert_eq!(ds.get_events(new_id, None, None, None).unwrap().len(), 1); + } + + #[test] + fn test_migrate_test_bucket_names_moves_disjoint_events_when_some_overlap() { + let ds = Datastore::new_in_memory(false); + let old_id = "aw-watcher-android-test_phone"; + let new_id = "aw-watcher-android_phone"; + create_named_test_bucket(&ds, old_id); + create_named_test_bucket(&ds, new_id); + let now = Utc::now(); + ds.insert_events( + old_id, + &[ + test_event(now - Duration::hours(2), Duration::minutes(30)), + test_event(now, Duration::minutes(30)), + ], + ) + .unwrap(); + ds.insert_events( + new_id, + &[test_event( + now + Duration::minutes(15), + Duration::minutes(30), + )], + ) + .unwrap(); + + assert_eq!(ds.migrate_test_bucket_names().unwrap(), 0); + assert!(ds.get_buckets().unwrap().contains_key(old_id)); + assert_eq!(ds.get_events(old_id, None, None, None).unwrap().len(), 1); + assert_eq!(ds.get_events(new_id, None, None, None).unwrap().len(), 2); + } + + #[test] + fn test_migrate_test_bucket_names_merges_interleaved_non_overlapping_events() { + let ds = Datastore::new_in_memory(false); + let old_id = "aw-watcher-android-test_phone"; + let new_id = "aw-watcher-android_phone"; + create_named_test_bucket(&ds, old_id); + create_named_test_bucket(&ds, new_id); + let now = Utc::now(); + ds.insert_events( + old_id, + &[ + test_event(now - Duration::hours(2), Duration::minutes(20)), + test_event(now, Duration::minutes(20)), + ], + ) + .unwrap(); + ds.insert_events( + new_id, + &[test_event(now - Duration::hours(1), Duration::minutes(20))], + ) + .unwrap(); + + assert_eq!(ds.migrate_test_bucket_names().unwrap(), 1); + assert!(!ds.get_buckets().unwrap().contains_key(old_id)); + assert_eq!(ds.get_events(new_id, None, None, None).unwrap().len(), 3); + } + #[test] fn test_bucket_create_delete() { // Setup datastore From 383da79e9462ffd3758fc0cff07729f3e247ad78 Mon Sep 17 00:00:00 2001 From: Bob Date: Thu, 27 Aug 2026 23:34:20 +0000 Subject: [PATCH 2/3] fix(datastore): add busy_timeout to handle Windows database locking Set a 5-second busy_timeout after opening SQLite connections to allow the database to wait for locks rather than immediately failing with DatabaseBusy errors. This fixes test failures on Windows where multiple connections to the same database file can create lock contention. Fixes windows-latest CI failure in sync_roundtrip tests. --- aw-datastore/src/worker.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/aw-datastore/src/worker.rs b/aw-datastore/src/worker.rs index add5a756..8e37bac1 100644 --- a/aw-datastore/src/worker.rs +++ b/aw-datastore/src/worker.rs @@ -142,6 +142,10 @@ impl DatastoreWorker { } }; + // Set busy timeout to handle concurrent access on systems with strict file locking (e.g., Windows) + conn.busy_timeout(std::time::Duration::from_secs(5)) + .expect("Failed to set busy timeout"); + // WAL turns each commit into a single sequential WAL append+fsync where // delete mode paid two fsyncs plus journal-file churn, and lets future // reader connections proceed while a commit is in flight. From edbf0c98f18c9c141c1b54549fe9041f316f9354 Mon Sep 17 00:00:00 2001 From: Bob Date: Thu, 27 Aug 2026 23:49:48 +0000 Subject: [PATCH 3/3] fix(datastore): retain overlapping legacy events --- aw-datastore/src/datastore.rs | 21 +++++++++++---------- aw-datastore/tests/datastore.rs | 29 +++++++++++++++++++++++++++++ 2 files changed, 40 insertions(+), 10 deletions(-) diff --git a/aw-datastore/src/datastore.rs b/aw-datastore/src/datastore.rs index 082b77f2..b3902700 100644 --- a/aw-datastore/src/datastore.rs +++ b/aw-datastore/src/datastore.rs @@ -1119,9 +1119,9 @@ impl DatastoreInstance { /// convention (e.g. `aw-watcher-android-test_phone` → `aw-watcher-android_phone`). /// /// If the destination already exists, disjoint legacy events are moved into it. - /// Events that overlap a destination event stay in the legacy bucket so the - /// migration cannot create duplicate activity records. The legacy bucket is - /// deleted only after it is empty. + /// Events that overlap a destination or another legacy event stay in the legacy + /// bucket so the migration cannot create overlapping activity records. The legacy + /// bucket is deleted only after it is empty. /// Returns the number of legacy buckets that were fully renamed or merged. pub fn migrate_test_bucket_names( &mut self, @@ -1149,19 +1149,20 @@ impl DatastoreInstance { .cloned() .ok_or_else(|| DatastoreError::NoSuchBucket(old_id.clone()))?; - // Move only events that do not overlap any destination event. - // A single overlapping cutover heartbeat must not strand years of - // disjoint history in the legacy bucket (ActivityWatch/aw-android#243). + // Move only events that do not overlap the destination or another + // legacy event. A single overlapping cutover heartbeat must not strand + // years of disjoint history in the legacy bucket (ActivityWatch/aw-android#243). conn.execute( "UPDATE events SET bucketrow = ?1 WHERE id IN ( SELECT old_event.id FROM events AS old_event WHERE old_event.bucketrow = ?2 AND NOT EXISTS ( - SELECT 1 FROM events AS new_event - WHERE new_event.bucketrow = ?1 - AND old_event.starttime < new_event.endtime - AND new_event.starttime < old_event.endtime + SELECT 1 FROM events AS other_event + WHERE other_event.id != old_event.id + AND other_event.bucketrow IN (?1, ?2) + AND old_event.starttime < other_event.endtime + AND other_event.starttime < old_event.endtime ) )", [new_bucket.bid, old_bucket.bid], diff --git a/aw-datastore/tests/datastore.rs b/aw-datastore/tests/datastore.rs index 7b785b3a..9c05740c 100644 --- a/aw-datastore/tests/datastore.rs +++ b/aw-datastore/tests/datastore.rs @@ -169,6 +169,35 @@ mod datastore_tests { assert_eq!(ds.get_events(new_id, None, None, None).unwrap().len(), 2); } + #[test] + fn test_migrate_test_bucket_names_keeps_overlapping_legacy_events_together() { + let ds = Datastore::new_in_memory(false); + let old_id = "aw-watcher-android-test_phone"; + let new_id = "aw-watcher-android_phone"; + create_named_test_bucket(&ds, old_id); + create_named_test_bucket(&ds, new_id); + let now = Utc::now(); + ds.insert_events( + old_id, + &[ + test_event(now - Duration::hours(3), Duration::minutes(20)), + test_event(now, Duration::minutes(30)), + test_event(now + Duration::minutes(15), Duration::minutes(30)), + ], + ) + .unwrap(); + ds.insert_events( + new_id, + &[test_event(now + Duration::hours(2), Duration::minutes(20))], + ) + .unwrap(); + + assert_eq!(ds.migrate_test_bucket_names().unwrap(), 0); + assert!(ds.get_buckets().unwrap().contains_key(old_id)); + assert_eq!(ds.get_events(old_id, None, None, None).unwrap().len(), 2); + assert_eq!(ds.get_events(new_id, None, None, None).unwrap().len(), 2); + } + #[test] fn test_migrate_test_bucket_names_merges_interleaved_non_overlapping_events() { let ds = Datastore::new_in_memory(false);