fix(datastore): merge disjoint aw-watcher-android-test events on name collision (#661)

* 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

* 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.

* fix(datastore): retain overlapping legacy events

Author: TimeToBuildBob

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,114 @@ 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 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,
         conn: &Connection,
     ) -> Result<usize, DatastoreError> {
-        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<String> = 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 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 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],
+                )
+                .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)
     }
 }

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.

aw-datastore/tests/datastore.rs

@@ -60,6 +60,171 @@ 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<Utc>, 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_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);
+        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