diff --git a/aw-datastore/src/datastore.rs b/aw-datastore/src/datastore.rs index 66e4df4a..b3902700 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,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 { - 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 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) } } 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. diff --git a/aw-datastore/tests/datastore.rs b/aw-datastore/tests/datastore.rs index fe353e2a..9c05740c 100644 --- a/aw-datastore/tests/datastore.rs +++ b/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, 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