refactor(sqlite-storage)!: replace bincode with prost for persisted blobs

src/appstate_sync.rs

@@ -865,4 +865,118 @@ mod tests {
         );
         assert!(matches!(pl.error, Some(CollectionSyncError::Retry { .. })));
     }
+
+    // Companion to snapshot_resync_drops_stale_mutation_macs: a collection whose
+    // version blob reset to 0 (e.g. an old bincode row that no longer decodes) keeps
+    // its pre-reset mutation MACs on disk. When the v0 resync arrives as a genesis
+    // patch (v1) WITHOUT a snapshot, those stale MACs must be wiped before the patch
+    // runs, or its ltHash anchors to index->value entries that aren't part of the
+    // fresh baseline.
+    #[tokio::test]
+    async fn genesis_patch_on_reset_collection_drops_stale_mutation_macs() {
+        let backend = Arc::new(MockBackend::default());
+        let processor =
+            AppStateProcessor::new(backend.clone(), Arc::new(crate::runtime_impl::TokioRuntime));
+        let name = WAPatchName::Regular;
+
+        // Reset collection: version 0 / empty hash, but stale MACs still present.
+        backend
+            .set_version(name.as_str(), HashState::default())
+            .await
+            .unwrap();
+        let stale_index_mac = vec![0xAB; 32];
+        backend
+            .put_mutation_macs(
+                name.as_str(),
+                7,
+                &[AppStateMutationMAC {
+                    index_mac: stale_index_mac.clone(),
+                    value_mac: vec![0xCD; 32],
+                }],
+            )
+            .await
+            .unwrap();
+
+        // A genesis patch (v1) served without a snapshot.
+        let patch_list = PatchList {
+            name,
+            has_more_patches: false,
+            patches: vec![wa::SyncdPatch {
+                version: Some(wa::SyncdVersion { version: Some(1) }),
+                ..Default::default()
+            }],
+            snapshot: None,
+            snapshot_ref: None,
+            error: None,
+        };
+
+        processor
+            .process_patch_list(patch_list, false)
+            .await
+            .expect("genesis patch onto a reset collection should process");
+
+        assert_eq!(
+            backend
+                .get_mutation_mac(name.as_str(), &stale_index_mac)
+                .await
+                .unwrap(),
+            None,
+            "a genesis-patch resync onto a reset collection must clear the stale pre-reset MACs"
+        );
+    }
+
+    // The SNAPSHOT's key_id lives INSIDE its external blob, so get_missing_key_ids on
+    // the un-inlined list can't see it. missing_key_ids_after_inline must download and
+    // inline the blob first, so an absent snapshot key is requested up front instead of
+    // aborting mid-process with KeyNotFound (the regression a paired companion hit when
+    // its snapshot key was absent after the bincode->prost reset).
+    #[tokio::test]
+    async fn missing_key_ids_after_inline_sees_external_snapshot_key() {
+        let backend = Arc::new(MockBackend::default());
+        let processor =
+            AppStateProcessor::new(backend.clone(), Arc::new(crate::runtime_impl::TokioRuntime));
+
+        let snapshot_key_id = b"snapshot-key-xyz".to_vec();
+        let snapshot_bytes = wa::SyncdSnapshot {
+            key_id: Some(wa::KeyId {
+                id: Some(snapshot_key_id.clone()),
+            }),
+            ..Default::default()
+        }
+        .encode_to_vec();
+        let direct_path = "/snapshot/blob".to_string();
+
+        let mut pl = PatchList {
+            name: WAPatchName::Regular,
+            has_more_patches: false,
+            patches: vec![],
+            snapshot: None,
+            snapshot_ref: Some(wa::ExternalBlobReference {
+                direct_path: Some(direct_path),
+                ..Default::default()
+            }),
+            error: None,
+        };
+
+        let download = |_ext: &wa::ExternalBlobReference| -> anyhow::Result<Vec<u8>> {
+            Ok(snapshot_bytes.clone())
+        };
+
+        // Before inlining, the external snapshot's key is invisible.
+        assert!(
+            processor.get_missing_key_ids(&pl).await.unwrap().is_empty(),
+            "the snapshot key is inside the un-downloaded blob, so it can't be seen yet"
+        );
+
+        // After inlining, the absent snapshot key is reported so it gets requested.
+        let missing = processor
+            .missing_key_ids_after_inline(&mut pl, &download)
+            .await
+            .unwrap();
+        assert_eq!(
+            missing,
+            vec![snapshot_key_id],
+            "the snapshot's key must be requestable after inlining the blob"
+        );
+    }
 }

src/client/app_state.rs

@@ -173,6 +173,10 @@ impl Client {
         use wacore::appstate::patch_decode::CollectionSyncError;
         const MAX_ITERATIONS: usize = 5;
         let mut iteration = 0;
+        // Set once we've requested missing decode keys this cycle, so the immediate
+        // refetch doesn't re-request; reset after a page decodes so a later iteration
+        // referencing a different rotated key can repair again.
+        let mut requested_keys = false;
 
         while !pending.is_empty() && iteration < MAX_ITERATIONS {
             iteration += 1;
@@ -224,7 +228,10 @@ impl Client {
 
             // Parse the response once here for pre-download; the same parsed
             // lists are handed to the processor below (no second parse).
-            let patch_lists = wacore::appstate::patch_decode::parse_patch_lists_ref(resp.get())?;
+            let mut patch_lists =
+                wacore::appstate::patch_decode::parse_patch_lists_ref(resp.get())?;
+
+            let proc = self.get_app_state_processor().await;
             {
                 for pl in &patch_lists {
                     // Download external snapshot
@@ -282,11 +289,34 @@ impl Client {
                 }
             };
 
-            // Process the already-parsed collections (no re-parse of the response).
-            let proc = self.get_app_state_processor().await;
+            // Request any missing decode keys before processing. Inline each list's
+            // external blobs first so the SNAPSHOT's key_id (which lives inside the blob,
+            // not the patch metadata) is visible -- otherwise process_patch_lists aborts
+            // with KeyNotFound on the snapshot key before the post-process request runs.
+            // Request once, wait briefly for the re-share, then refetch. Only wait when a
+            // fresh request actually went out (the dedup may suppress it).
+            if !requested_keys {
+                let mut missing_all: Vec<Vec<u8>> = Vec::new();
+                for pl in &mut patch_lists {
+                    if let Ok(m) = proc.missing_key_ids_after_inline(pl, &download).await {
+                        missing_all.extend(m);
+                    }
+                }
+                if !missing_all.is_empty() {
+                    requested_keys = true;
+                    if self.request_keys_and_wait(missing_all).await {
+                        continue;
+                    }
+                }
+            }
+
+            // Process the already-parsed (and inlined) collections.
             let results = proc
                 .process_patch_lists(patch_lists, &download, true)
                 .await?;
+            // A page decoded, so its keys were present; let a later iteration repair a
+            // different rotated key.
+            requested_keys = false;
 
             let mut needs_refetch = Vec::new();
 
@@ -396,6 +426,8 @@ impl Client {
         // has_more_patches=true without advancing the version (WA Web uses 500).
         const MAX_PAGINATION_ITERATIONS: u32 = 500;
         let mut iteration = 0u32;
+        // Request the batch's missing keys at most once per task (see below).
+        let mut requested_missing_keys = false;
 
         while has_more {
             if self.is_shutting_down() {
@@ -440,18 +472,19 @@ impl Client {
 
             let _decode_start = wacore::time::Instant::now();
 
-            // Pre-download all external blobs (snapshot and patch mutations)
-            // We use directPath as the key to identify each blob
+            // Parse the response once here; the same parsed list is handed to the
+            // processor below (no second parse).
+            let mut pl = wacore::appstate::patch_decode::parse_patch_list_ref(resp.get())?;
+            debug!(target: "Client/AppState", "Parsed patch list for {:?}: has_snapshot_ref={} has_more_patches={} patches_count={}",
+                name, pl.snapshot_ref.is_some(), pl.has_more_patches, pl.patches.len());
+
+            let proc = self.get_app_state_processor().await;
+
+            // Pre-download all external blobs (snapshot and patch mutations); keyed by
+            // directPath.
             let mut pre_downloaded: std::collections::HashMap<String, Vec<u8>> =
                 std::collections::HashMap::new();
-
-            // Parse the response once here for pre-download; the same parsed list
-            // is handed to the processor below (no second parse).
-            let pl = wacore::appstate::patch_decode::parse_patch_list_ref(resp.get())?;
             {
-                debug!(target: "Client/AppState", "Parsed patch list for {:?}: has_snapshot_ref={} has_more_patches={} patches_count={}",
-                    name, pl.snapshot_ref.is_some(), pl.has_more_patches, pl.patches.len());
-
                 // Download external snapshot if present
                 if let Some(ext) = &pl.snapshot_ref
                     && let Some(path) = &ext.direct_path
@@ -505,9 +538,26 @@ impl Client {
                 }
             };
 
-            let proc = self.get_app_state_processor().await;
+            // Request any missing decode keys before processing. Inline the blobs first
+            // so the SNAPSHOT's key_id (inside its external blob, not the patch metadata)
+            // is visible -- otherwise process aborts with KeyNotFound on the snapshot key
+            // before the post-process request runs. Request once, wait briefly for the
+            // re-share, then refetch (only when a fresh request actually went out).
+            if !requested_missing_keys
+                && let Ok(missing) = proc.missing_key_ids_after_inline(&mut pl, &download).await
+                && !missing.is_empty()
+            {
+                requested_missing_keys = true;
+                if self.request_keys_and_wait(missing).await {
+                    continue;
+                }
+            }
+
             let (mutations, new_state, list) =
                 proc.process_parsed_patch_list(pl, &download, true).await?;
+            // This page decoded, so its keys were available; allow the next page
+            // (which may reference a different rotated key) to repair again.
+            requested_missing_keys = false;
             let decode_elapsed = _decode_start.elapsed();
             if decode_elapsed.as_millis() > 500 {
                 debug!(target: "Client/AppState", "Patch decode for {:?} took {:?}", name, decode_elapsed);
@@ -541,11 +591,33 @@ impl Client {
         Ok(())
     }
 
+    /// Shared missing-key repair step for both sync paths: request the given keys and,
+    /// only if a fresh request actually went out (the per-key dedup didn't suppress it),
+    /// wait briefly for the primary to re-share. Returns true iff the caller should
+    /// refetch (a request was sent and we waited); false means nothing was requested
+    /// (empty or deduped), so the caller proceeds without stalling.
+    async fn request_keys_and_wait(&self, missing: Vec<Vec<u8>>) -> bool {
+        if missing.is_empty() {
+            return false;
+        }
+        let count = missing.len();
+        let listener = self.initial_keys_synced_notifier.listen();
+        if self.request_missing_keys_with_dedup(missing).await {
+            debug!(target: "Client/AppState", "Requested {count} missing app-state key(s) before processing; waiting up to 10s before refetch");
+            let _ = rt_timeout(&*self.runtime, Duration::from_secs(10), listener).await;
+            return true;
+        }
+        false
+    }
+
     /// Request missing app-state keys with dedup stamps.
     /// On send failure, removes stamps so keys can be retried next sync.
-    async fn request_missing_keys_with_dedup(&self, missing: Vec<Vec<u8>>) {
+    /// Returns true iff a fresh key request was actually sent (some ids passed the
+    /// per-key dedup and the send succeeded), so the caller knows whether a re-share
+    /// is plausibly inbound and worth waiting for.
+    async fn request_missing_keys_with_dedup(&self, missing: Vec<Vec<u8>>) -> bool {
         if missing.is_empty() {
-            return;
+            return false;
         }
         let mut to_request: Vec<Vec<u8>> = Vec::with_capacity(missing.len());
         let mut guard = self.app_state_key_requests.lock().await;
@@ -563,15 +635,18 @@ impl Client {
         }
         guard.retain(|_, t| t.elapsed() < std::time::Duration::from_secs(24 * 3600));
         drop(guard);
-        if !to_request.is_empty()
-            && let Err(e) = self.request_app_state_keys(&to_request).await
-        {
+        if to_request.is_empty() {
+            return false;
+        }
+        if let Err(e) = self.request_app_state_keys(&to_request).await {
             warn!("Failed to send app state key request: {e}");
             let mut guard = self.app_state_key_requests.lock().await;
             for key_id in &to_request {
                 guard.remove(&hex::encode(key_id));
             }
+            return false;
         }
+        true
     }
 
     #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.appstate.request_keys", level = "debug", skip_all, fields(count = raw_key_ids.len()), err(Debug)))]
@@ -580,8 +655,13 @@ impl Client {
             return Ok(());
         }
         let device_snapshot = self.persistence_manager.get_device_snapshot();
+        // Address the request to the PRIMARY (device 0), not our own device JID: this is
+        // a peer message and `device_snapshot.pn` carries OUR device number, so sending
+        // it as-is encrypts to ourselves (no self-session exists) and fails with
+        // "session ... not found". The primary is the app-state key source and we hold a
+        // session with it from pairing. Mirrors whatsmeow's `ownID.ToNonAD()`.
         let own_jid = match device_snapshot.pn.clone() {
-            Some(j) => j,
+            Some(j) => j.to_non_ad(),
             None => {
                 return Err(anyhow::anyhow!(
                     "no own JID available for app-state key request"