Harden Apple Watch snapshot synchronization

crates/storage/tests/mobile_watch.rs

@@ -112,6 +112,12 @@ fn replacement_snapshots_reject_replays_conflicts_and_pairing_changes() -> TestR
         receiver.apply(SecretBytes::new(
             reset_revocation.snapshot().expose().to_vec()
         ))?,
+        WatchSnapshotApply::Stale
+    );
+    assert_eq!(
+        receiver.replace_authoritative(SecretBytes::new(
+            reset_revocation.snapshot().expose().to_vec()
+        ))?,
         WatchSnapshotApply::Revoked
     );
     assert!(
@@ -134,12 +140,17 @@ fn replacement_snapshots_reject_replays_conflicts_and_pairing_changes() -> TestR
         "paired-watch-b",
         vec![entry("otp/carol", "Acme", "carol", b"third-secret")],
     )?;
-    assert_eq!(changed_watch.revision(), 4);
+    assert_eq!(changed_watch.revision(), 1);
     assert_eq!(
         receiver.apply(SecretBytes::new(changed_watch.snapshot().expose().to_vec()))?,
+        WatchSnapshotApply::Stale
+    );
+    assert_eq!(
+        receiver
+            .replace_authoritative(SecretBytes::new(changed_watch.snapshot().expose().to_vec()))?,
         WatchSnapshotApply::PairingChanged
     );
-    assert_eq!(receiver.current().expect("new pairing").revision(), 4);
+    assert_eq!(receiver.current().expect("new pairing").revision(), 1);
     let mut fresh_watch = WatchSnapshotReceiver::default();
     assert_eq!(
         fresh_watch.apply(SecretBytes::new(changed_watch.snapshot().expose().to_vec()))?,
@@ -149,14 +160,21 @@ fn replacement_snapshots_reject_replays_conflicts_and_pairing_changes() -> TestR
     let mut damaged = changed_watch.snapshot().expose().to_vec();
     damaged[20] ^= 0x55;
     assert!(fresh_watch.apply(SecretBytes::new(damaged)).is_err());
-    assert!(fresh_watch.current().is_none());
+    assert_eq!(
+        fresh_watch
+            .current()
+            .expect("last valid snapshot")
+            .entries()[0]
+            .account(),
+        "carol"
+    );
     assert_eq!(
         fresh_watch.apply(SecretBytes::new(first.snapshot().expose().to_vec()))?,
         WatchSnapshotApply::Stale
     );
     assert_eq!(
         fresh_watch.apply(SecretBytes::new(changed_watch.snapshot().expose().to_vec()))?,
-        WatchSnapshotApply::Replaced
+        WatchSnapshotApply::Duplicate
     );
 
     let persisted = fs::read_to_string(journal)?;
@@ -193,6 +211,131 @@ fn journal_keeps_revisions_monotonic_across_sender_reloads() -> TestResult {
     Ok(())
 }
 
+#[test]
+fn journal_upgrade_starts_a_new_generation_at_revision_one() -> TestResult {
+    let directory = tempfile::tempdir()?;
+    let journal = directory.path().join("watch-snapshot.toml");
+    let entries = || vec![entry("otp/alice", "Acme", "alice", b"secret")];
+    WatchSnapshotSender::load(journal.clone()).prepare("paired-watch", entries())?;
+
+    let legacy = fs::read_to_string(&journal)?
+        .lines()
+        .filter(|line| !line.starts_with("generation = "))
+        .collect::<Vec<_>>()
+        .join("\n")
+        .replace("revision = 1", "revision = 42");
+    fs::write(&journal, legacy)?;
+
+    let reset = WatchSnapshotSender::load(journal.clone()).prepare("paired-watch", entries())?;
+    assert_eq!(reset.revision(), 1);
+    let reloaded = WatchSnapshotSender::load(journal).prepare("paired-watch", entries())?;
+    assert_eq!(reloaded.revision(), 1);
+    assert_eq!(reset.snapshot().expose(), reloaded.snapshot().expose());
+    Ok(())
+}
+
+#[test]
+fn duplicate_delivery_after_watch_restart_recovers_a_lost_acknowledgement() -> TestResult {
+    let directory = tempfile::tempdir()?;
+    let mut sender = WatchSnapshotSender::load(directory.path().join("watch-snapshot.toml"));
+    let transfer = sender.prepare(
+        "paired-watch",
+        vec![entry("otp/alice", "Acme", "alice", b"secret")],
+    )?;
+    let persisted_snapshot = transfer.snapshot().expose().to_vec();
+
+    let mut interrupted_runtime = WatchRuntime::default();
+    let applied = interrupted_runtime.apply_snapshot(persisted_snapshot.clone())?;
+    assert_eq!(applied.persistence(), WatchPersistenceAction::Replace);
+    assert_eq!(sender.status().state(), MobileWatchSnapshotState::Pending);
+
+    let mut restarted_runtime = WatchRuntime::default();
+    restarted_runtime.apply_snapshot(persisted_snapshot.clone())?;
+    let redelivered = restarted_runtime.apply_snapshot(persisted_snapshot)?;
+    assert_eq!(redelivered.apply(), WatchSnapshotApply::Duplicate);
+    assert_eq!(redelivered.persistence(), WatchPersistenceAction::Keep);
+    assert_eq!(
+        sender.acknowledge(redelivered.receipt())?.state(),
+        MobileWatchSnapshotState::Current
+    );
+    Ok(())
+}
+
+#[test]
+fn authoritative_context_switches_sender_generation_and_rejects_delayed_packets() -> TestResult {
+    let directory = tempfile::tempdir()?;
+    let mut old_sender = WatchSnapshotSender::load(directory.path().join("old.toml"));
+    old_sender.prepare(
+        "paired-watch",
+        vec![entry("otp/alice", "Acme", "alice", b"old-secret")],
+    )?;
+    for revision in 2..5 {
+        old_sender.prepare(
+            "paired-watch",
+            vec![entry(
+                "otp/alice",
+                "Acme",
+                &format!("alice-{revision}"),
+                b"old-secret",
+            )],
+        )?;
+    }
+    let old = old_sender.prepare(
+        "paired-watch",
+        vec![entry("otp/alice", "Acme", "alice-5", b"old-secret")],
+    )?;
+    assert_eq!(old.revision(), 5);
+
+    let mut new_sender = WatchSnapshotSender::load(directory.path().join("new.toml"));
+    let new = new_sender.prepare(
+        "paired-watch",
+        vec![entry("otp/bob", "Acme", "bob", b"new-secret")],
+    )?;
+    assert_eq!(new.revision(), 1);
+
+    let mut runtime = WatchRuntime::default();
+    runtime.apply_snapshot(old.snapshot().expose().to_vec())?;
+    let queued_new = runtime.apply_snapshot(new.snapshot().expose().to_vec())?;
+    assert_eq!(queued_new.apply(), WatchSnapshotApply::Stale);
+    assert_eq!(
+        runtime.presentation_at(59)?.state(),
+        WatchPresentationState::Ready
+    );
+    assert_eq!(runtime.records_at(59)?[0].account(), "alice-5");
+
+    let replaced = runtime.apply_authoritative_snapshot(new.snapshot().expose().to_vec())?;
+    assert_eq!(replaced.apply(), WatchSnapshotApply::PairingChanged);
+    assert_eq!(replaced.persistence(), WatchPersistenceAction::Replace);
+    assert_eq!(runtime.records_at(59)?[0].account(), "bob");
+    assert_eq!(
+        new_sender.acknowledge(replaced.receipt())?.state(),
+        MobileWatchSnapshotState::Current
+    );
+
+    let delayed_old = runtime.apply_snapshot(old.snapshot().expose().to_vec())?;
+    assert_eq!(delayed_old.apply(), WatchSnapshotApply::Stale);
+    assert_eq!(runtime.records_at(59)?[0].account(), "bob");
+
+    let newer = new_sender.prepare(
+        "paired-watch",
+        vec![entry("otp/carol", "Acme", "carol", b"newer-secret")],
+    )?;
+    runtime.apply_authoritative_snapshot(newer.snapshot().expose().to_vec())?;
+    let delayed_context = runtime.apply_authoritative_snapshot(new.snapshot().expose().to_vec())?;
+    assert_eq!(delayed_context.apply(), WatchSnapshotApply::Stale);
+    assert_eq!(runtime.records_at(59)?[0].account(), "carol");
+
+    assert!(runtime.apply_snapshot(vec![0; 64]).is_err());
+    assert_eq!(runtime.records_at(59)?[0].account(), "carol");
+    assert_eq!(
+        runtime.presentation_at(59)?.state(),
+        WatchPresentationState::Stale
+    );
+    runtime.sync_unavailable();
+    assert_eq!(runtime.records_at(59)?[0].account(), "carol");
+    Ok(())
+}
+
 #[test]
 fn watch_runtime_generates_view_ready_totp_and_clears_secrets_when_locked() -> TestResult {
     let directory = tempfile::tempdir()?;
@@ -266,12 +409,12 @@ fn watch_runtime_generates_view_ready_totp_and_clears_secrets_when_locked() -> T
     runtime.apply_snapshot(snapshot.snapshot().expose().to_vec())?;
     assert_eq!(
         runtime.presentation_at(59)?.state(),
-        WatchPresentationState::Stale
+        WatchPresentationState::Empty
     );
     assert!(runtime.apply_snapshot(vec![0; 64]).is_err());
     assert_eq!(
         runtime.presentation_at(59)?.state(),
-        WatchPresentationState::Error
+        WatchPresentationState::Stale
     );
     Ok(())
 }