From 749d83add41b4f514555d447eb70aac1c320b25a Mon Sep 17 00:00:00 2001 From: Bob Date: Thu, 17 Sep 2026 14:56:17 +0000 Subject: [PATCH 1/3] fix(aw-sync): scan pre-#697 buckets by origin to avoid re-import after Android hostname migration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes ActivityWatch/aw-server-rust#707. When both the raw-ID and sanitized-ID lookups miss in `get_or_create_sync_bucket`, fall back to scanning the destination for a `-synced-from-` bucket whose base ID matches and whose `$aw.sync.origin` sanitizes to the same value. This recovers the case where a desktop imported an Android peer before #697 landed (`…-synced-from-POCO F8 Ultra` with `$aw.sync.origin = "POCO F8 Ultra"`), then the phone ran ActivityWatch/aw-android#273 and its staging hostname became `poco_f8_ultra` (sanitized, no `$aw.sync.origin` on first-hand buckets). The two direct lookups both produced `…-synced-from-poco_f8_ultra`, missed, and created a new bucket — causing a full re-import (every event twice in /timeline). The fallback scan finds the legacy bucket by matching sanitized origins and reuses it. When two distinct pre-#697 buckets share the same sanitized origin (ambiguous), the function returns an error rather than silently merging distinct histories (#697 :368). Two tests added: - `test_pre697_origin_scan_resumes_legacy_bucket`: desktop with `…-synced-from-POCO F8 Ultra` + `$aw.sync.origin` receives a post-migration pull and resumes without creating a fork. - `test_pre697_origin_scan_refuses_ambiguous_candidates`: two legacy buckets with different raw origins that sanitize identically trigger an error. Git-Session-Id: 3846 --- aw-sync/src/sync.rs | 63 ++++++++++++++++++++ aw-sync/tests/sync.rs | 132 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 195 insertions(+) diff --git a/aw-sync/src/sync.rs b/aw-sync/src/sync.rs index 69d811f7..aef36647 100644 --- a/aw-sync/src/sync.rs +++ b/aw-sync/src/sync.rs @@ -547,6 +547,69 @@ fn get_or_create_sync_bucket( Err(e) => return Err(format!("Failed to get bucket '{sanitized_id}': {e:?}")), } } + + // Pre-#697 origin-based fallback (ActivityWatch/aw-server-rust#707): + // Both exact lookups missed. A desktop that imported the peer before + // ActivityWatch/aw-server-rust#697 landed holds the raw hostname from that + // day (e.g. `…-synced-from-POCO F8 Ultra`) with `$aw.sync.origin` set to + // that same raw value. After ActivityWatch/aw-android#273 migrates the + // phone's staging hostname to `poco_f8_ultra`, first-hand buckets carry no + // `$aw.sync.origin`, so both lookups above miss and the whole history gets + // re-imported. Scan the destination's -synced-from- buckets for one whose + // base ID matches and whose `$aw.sync.origin` sanitizes to the same target. + if !is_push { + let target_base = bucket_from + .id + .split("-synced-from-") + .next() + .unwrap_or(bucket_from.id.as_str()); + let target_sanitized = sanitize_hostname( + sync_origin + .as_deref() + .unwrap_or(bucket_from.hostname.as_str()), + ); + let all_dest = ds_to + .get_buckets() + .map_err(|e| format!("Failed to list dest buckets for origin scan: {e:?}"))?; + let mut candidates: Vec = all_dest + .into_values() + .filter(|b| { + let b_base = b.id.split("-synced-from-").next().unwrap_or(&b.id); + if b_base != target_base { + return false; + } + // $aw.sync.origin must be present and sanitize to the same value. + b.data + .get("$aw.sync.origin") + .and_then(|v| v.as_str()) + .map(|s| sanitize_hostname(s) == target_sanitized) + .unwrap_or(false) + }) + .collect(); + match candidates.len() { + 0 => {} // fall through to create a new bucket + 1 => { + let found = candidates.remove(0); + info!( + " ↩ Reusing pre-#697 bucket '{}' for '{}'", + found.id, bucket_from.id + ); + return Ok(found); + } + n => { + // Two distinct pre-#697 buckets share the same sanitized origin — + // ambiguous. Refuse rather than silently merging distinct histories + // (ActivityWatch/aw-server-rust#697 :368). + let ids: Vec<&str> = candidates.iter().map(|b| b.id.as_str()).collect(); + return Err(format!( + "Cannot resolve destination for '{}': {n} pre-#697 buckets share \ + sanitized origin '{}': {ids:?}; deduplicate manually", + bucket_from.id, target_sanitized + )); + } + } + } + let (final_id, final_hostname) = (sanitized_id, sanitized_hostname); let mut bucket_new = bucket_from.clone(); diff --git a/aw-sync/tests/sync.rs b/aw-sync/tests/sync.rs index d4e02669..3c91d1e3 100644 --- a/aw-sync/tests/sync.rs +++ b/aw-sync/tests/sync.rs @@ -261,6 +261,138 @@ mod sync_tests { ); } + /// A desktop that imported the peer **before** ActivityWatch/aw-server-rust#697 + /// landed holds `…-synced-from-POCO F8 Ultra` (raw, with `$aw.sync.origin` set + /// to the raw value by #697's import stamp). After ActivityWatch/aw-android#273 + /// migrates the phone's hostname to `poco_f8_ultra`, first-hand buckets carry no + /// `$aw.sync.origin`, so the two direct lookups miss. The pre-#697 fallback scan + /// must find the legacy bucket and resume from it rather than creating a new one + /// that triggers a full re-import. + #[test] + fn test_pre697_origin_scan_resumes_legacy_bucket() { + let state = init_teststate(); + + // Post-migration phone bucket: sanitized hostname, no $aw.sync.origin. + let src_bucket: Bucket = serde_json::from_value(serde_json::json!({ + "id": "aw-watcher-android", + "type": "currentwindow", + "hostname": "poco_f8_ultra", + "client": "aw-android" + })) + .unwrap(); + state.ds_src.create_bucket(&src_bucket).unwrap(); + + // Pre-#697 destination bucket: raw ID + $aw.sync.origin stamped by #697. + let legacy_id = "aw-watcher-android-synced-from-POCO F8 Ultra"; + let legacy_bucket: Bucket = serde_json::from_value(serde_json::json!({ + "id": legacy_id, + "type": "currentwindow", + "hostname": "POCO F8 Ultra", + "client": "aw-android", + "data": {"$aw.sync.origin": "POCO F8 Ultra"} + })) + .unwrap(); + state.ds_dest.create_bucket(&legacy_bucket).unwrap(); + + // Insert one event into the source so the sync pass has something to copy. + let ts = chrono::Utc::now(); + let ev: Event = serde_json::from_value(serde_json::json!({ + "timestamp": ts.to_rfc3339(), + "duration": 1, + "data": {"app": "test"} + })) + .unwrap(); + state + .ds_src + .insert_events("aw-watcher-android", &[ev]) + .unwrap(); + state.ds_src.force_commit().unwrap(); + + aw_sync::sync_datastores( + &state.ds_src, + &state.ds_dest, + false, // pull + None, + &SyncSpec::default(), + ) + .unwrap(); + + let dest_buckets = state.ds_dest.get_buckets().unwrap(); + + // The legacy bucket must be reused, not replaced. + assert!( + dest_buckets.contains_key(legacy_id), + "legacy bucket must be preserved" + ); + // No new sanitized fork must appear. + let forked_id = "aw-watcher-android-synced-from-poco_f8_ultra"; + assert!( + !dest_buckets.contains_key(forked_id), + "a sanitized fork must not be created; got: {:?}", + dest_buckets.keys().collect::>() + ); + // Events were imported into the legacy bucket, not lost. + let event_count = state + .ds_dest + .get_event_count(legacy_id, None, None) + .unwrap(); + assert!(event_count > 0, "legacy bucket must have received events"); + } + + /// Two distinct pre-#697 buckets for the same base ID whose `$aw.sync.origin` + /// values sanitize to the same target must trigger an error rather than a + /// silent merge (ActivityWatch/aw-server-rust#697 :368). + #[test] + fn test_pre697_origin_scan_refuses_ambiguous_candidates() { + let state = init_teststate(); + + // Post-migration phone bucket. + let src_bucket: Bucket = serde_json::from_value(serde_json::json!({ + "id": "aw-watcher-android", + "type": "currentwindow", + "hostname": "poco_f8_ultra", + "client": "aw-android" + })) + .unwrap(); + state.ds_src.create_bucket(&src_bucket).unwrap(); + + // Two legacy destination buckets whose origins both sanitize to "poco_f8_ultra". + for (legacy_id, raw_origin) in [ + ( + "aw-watcher-android-synced-from-POCO F8 Ultra", + "POCO F8 Ultra", + ), + ( + "aw-watcher-android-synced-from-Poco F8 Ultra", + "Poco F8 Ultra", + ), + ] { + let b: Bucket = serde_json::from_value(serde_json::json!({ + "id": legacy_id, + "type": "currentwindow", + "hostname": raw_origin, + "client": "aw-android", + "data": {"$aw.sync.origin": raw_origin} + })) + .unwrap(); + state.ds_dest.create_bucket(&b).unwrap(); + } + + let result = aw_sync::sync_datastores( + &state.ds_src, + &state.ds_dest, + false, // pull + None, + &SyncSpec::default(), + ); + // The ambiguous-candidates path must fail rather than silently merge. + let err = result.expect_err("ambiguous pre-#697 buckets must return Err"); + assert!( + err.contains("pre-#697 buckets share"), + "error should explain the ambiguity, got: {err}" + ); + } + /// Case-only hostnames (`PIXEL8`) have no whitespace, so a whitespace-only /// guard would leave the destination as `…-synced-from-PIXEL8`. Android's /// later hostname migration produces `pixel8` and forks the history. From 1d3cc39ad16fe0f1cb18dc8d11f8a735677eb959 Mon Sep 17 00:00:00 2001 From: Bob Date: Thu, 17 Sep 2026 17:26:55 +0000 Subject: [PATCH 2/3] test(aw-sync): prove resume cursor + ambiguous-skip doesn't abort peer sync MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two tightenings per ActivityWatch/aw-server-rust#707 review: 1. test_pre697_origin_scan_resumes_legacy_bucket: seed the legacy bucket with one event at T0, put T0-1h and T0+1h in the source. After sync, assert exactly 2 events (existing + new). 3 would mean the cursor was ignored and the full history was re-imported. 2. test_pre697_origin_scan_refuses_ambiguous_candidates: add a second, healthy source bucket (aw-watcher-window). sync_datastores must return Ok overall — the ambiguous android bucket is skipped (warn+continue), not a fatal abort. Assert the healthy bucket received events; assert neither ambiguous legacy bucket was written to. Git-Session-Id: eacd --- aw-sync/tests/sync.rs | 109 ++++++++++++++++++++++++++++++++++++------ 1 file changed, 94 insertions(+), 15 deletions(-) diff --git a/aw-sync/tests/sync.rs b/aw-sync/tests/sync.rs index 3c91d1e3..879403ab 100644 --- a/aw-sync/tests/sync.rs +++ b/aw-sync/tests/sync.rs @@ -294,17 +294,39 @@ mod sync_tests { .unwrap(); state.ds_dest.create_bucket(&legacy_bucket).unwrap(); - // Insert one event into the source so the sync pass has something to copy. - let ts = chrono::Utc::now(); - let ev: Event = serde_json::from_value(serde_json::json!({ - "timestamp": ts.to_rfc3339(), + // Seed the legacy destination bucket with one event at T0 — simulating + // previously-imported history. This establishes the resume cursor. + let t0 = Utc::now(); + let existing_event: Event = serde_json::from_value(serde_json::json!({ + "timestamp": t0.to_rfc3339(), "duration": 1, - "data": {"app": "test"} + "data": {"app": "existing"} + })) + .unwrap(); + state + .ds_dest + .insert_events(legacy_id, &[existing_event]) + .unwrap(); + state.ds_dest.force_commit().unwrap(); + + // Two source events: one before T0 (already covered) and one after T0 (new). + // The sync must read the cursor from the reused bucket and import only the + // post-T0 event — not re-import everything from scratch. + let before_t0: Event = serde_json::from_value(serde_json::json!({ + "timestamp": (t0 - Duration::hours(1)).to_rfc3339(), + "duration": 1, + "data": {"app": "old"} + })) + .unwrap(); + let after_t0: Event = serde_json::from_value(serde_json::json!({ + "timestamp": (t0 + Duration::hours(1)).to_rfc3339(), + "duration": 1, + "data": {"app": "new"} })) .unwrap(); state .ds_src - .insert_events("aw-watcher-android", &[ev]) + .insert_events("aw-watcher-android", &[before_t0, after_t0]) .unwrap(); state.ds_src.force_commit().unwrap(); @@ -331,12 +353,17 @@ mod sync_tests { "a sanitized fork must not be created; got: {:?}", dest_buckets.keys().collect::>() ); - // Events were imported into the legacy bucket, not lost. + // Exactly 2 events: the pre-existing one at T0 plus the new T0+1h event. + // A count of 3 would mean the cursor was NOT read (full re-import from scratch). let event_count = state .ds_dest .get_event_count(legacy_id, None, None) .unwrap(); - assert!(event_count > 0, "legacy bucket must have received events"); + assert_eq!( + event_count, 2, + "legacy bucket must have exactly 2 events (existing + new); \ + 3 would mean the cursor was ignored and history was re-imported" + ); } /// Two distinct pre-#697 buckets for the same base ID whose `$aw.sync.origin` @@ -346,7 +373,8 @@ mod sync_tests { fn test_pre697_origin_scan_refuses_ambiguous_candidates() { let state = init_teststate(); - // Post-migration phone bucket. + // Ambiguous source bucket: two pre-#697 destination buckets share the same + // sanitized origin — sync_one must skip this bucket with a warning, not abort. let src_bucket: Bucket = serde_json::from_value(serde_json::json!({ "id": "aw-watcher-android", "type": "currentwindow", @@ -356,6 +384,29 @@ mod sync_tests { .unwrap(); state.ds_src.create_bucket(&src_bucket).unwrap(); + // A second, healthy source bucket that must sync successfully even while the + // ambiguous bucket is being skipped — one unresolvable bucket must not abort + // the whole peer sync. + let healthy_id = "aw-watcher-window"; + let healthy_bucket: Bucket = serde_json::from_value(serde_json::json!({ + "id": healthy_id, + "type": "currentwindow", + "hostname": "poco_f8_ultra", + "client": "aw-qt" + })) + .unwrap(); + state.ds_src.create_bucket(&healthy_bucket).unwrap(); + + let ts = Utc::now(); + let ev: Event = serde_json::from_value(serde_json::json!({ + "timestamp": ts.to_rfc3339(), + "duration": 1, + "data": {"app": "test"} + })) + .unwrap(); + state.ds_src.insert_events(healthy_id, &[ev]).unwrap(); + state.ds_src.force_commit().unwrap(); + // Two legacy destination buckets whose origins both sanitize to "poco_f8_ultra". for (legacy_id, raw_origin) in [ ( @@ -378,19 +429,47 @@ mod sync_tests { state.ds_dest.create_bucket(&b).unwrap(); } - let result = aw_sync::sync_datastores( + // The ambiguous android bucket is skipped (warn+continue); the healthy window + // bucket syncs normally. sync_datastores must return Ok overall. + aw_sync::sync_datastores( &state.ds_src, &state.ds_dest, false, // pull None, &SyncSpec::default(), - ); - // The ambiguous-candidates path must fail rather than silently merge. - let err = result.expect_err("ambiguous pre-#697 buckets must return Err"); + ) + .unwrap(); + + let dest_buckets = state.ds_dest.get_buckets().unwrap(); + + // The healthy bucket was synced — it has a destination and received events. + let healthy_dest_id = "aw-watcher-window-synced-from-poco_f8_ultra"; assert!( - err.contains("pre-#697 buckets share"), - "error should explain the ambiguity, got: {err}" + dest_buckets.contains_key(healthy_dest_id), + "healthy bucket must be synced even when another bucket is ambiguous; got: {:?}", + dest_buckets.keys().collect::>() ); + let healthy_count = state + .ds_dest + .get_event_count(healthy_dest_id, None, None) + .unwrap(); + assert!(healthy_count > 0, "healthy bucket must have events synced"); + + // The two ambiguous legacy buckets were not written to — the conflict was + // skipped, not merged. + for legacy_id in [ + "aw-watcher-android-synced-from-POCO F8 Ultra", + "aw-watcher-android-synced-from-Poco F8 Ultra", + ] { + let count = state + .ds_dest + .get_event_count(legacy_id, None, None) + .unwrap(); + assert_eq!( + count, 0, + "ambiguous legacy bucket '{legacy_id}' must not have received events" + ); + } } /// Case-only hostnames (`PIXEL8`) have no whitespace, so a whitespace-only From a90bed070fb38414839064a63795af6e090a4797 Mon Sep 17 00:00:00 2001 From: Bob Date: Thu, 17 Sep 2026 18:26:37 +0000 Subject: [PATCH 3/3] test(aw-sync): pin ambiguity refusal with a non-empty ambiguous source bucket MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The refusal test asserted both legacy candidates received zero events, but the ambiguous source bucket held no events — so a regression that silently reused one candidate would also have copied nothing and the assertions would still pass. Insert an event into the ambiguous source bucket (plus a premise guard that it really has one) so only an actual skip keeps both legacy buckets at 0. Git-Session-Id: 9059e8cf-aa8c-5d86-bc5d-0167df2a66b0 --- aw-sync/tests/sync.rs | 31 ++++++++++++++++++++++++++++++- 1 file changed, 30 insertions(+), 1 deletion(-) diff --git a/aw-sync/tests/sync.rs b/aw-sync/tests/sync.rs index 879403ab..14b46f3d 100644 --- a/aw-sync/tests/sync.rs +++ b/aw-sync/tests/sync.rs @@ -405,8 +405,35 @@ mod sync_tests { })) .unwrap(); state.ds_src.insert_events(healthy_id, &[ev]).unwrap(); + + // The ambiguous bucket must ALSO carry an event. Otherwise a regression that + // silently reused one of the two legacy candidates would copy nothing, both + // legacy buckets would still read 0 events, and the refusal assertions below + // would pass vacuously. With a real event present, only an actual skip keeps + // them at 0. + let ev_ambiguous: Event = serde_json::from_value(serde_json::json!({ + "timestamp": ts.to_rfc3339(), + "duration": 1, + "data": {"app": "ambiguous"} + })) + .unwrap(); + state + .ds_src + .insert_events(&src_bucket.id, &[ev_ambiguous]) + .unwrap(); state.ds_src.force_commit().unwrap(); + // Premise guard: the refusal assertions below are only meaningful if the + // ambiguous source bucket actually has something to copy. + assert_eq!( + state + .ds_src + .get_event_count(&src_bucket.id, None, None) + .unwrap(), + 1, + "premise: the ambiguous source bucket must carry one event" + ); + // Two legacy destination buckets whose origins both sanitize to "poco_f8_ultra". for (legacy_id, raw_origin) in [ ( @@ -456,7 +483,9 @@ mod sync_tests { assert!(healthy_count > 0, "healthy bucket must have events synced"); // The two ambiguous legacy buckets were not written to — the conflict was - // skipped, not merged. + // skipped, not merged. The source bucket holds an event (premise guard + // above), so a silent pick-one-candidate regression would make one of these + // counts 1 and fail here. for legacy_id in [ "aw-watcher-android-synced-from-POCO F8 Ultra", "aw-watcher-android-synced-from-Poco F8 Ultra",