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

src/client/app_state.rs

@@ -396,6 +396,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() {
@@ -506,8 +508,29 @@ impl Client {
             };
 
             let proc = self.get_app_state_processor().await;
+
+            // Keys needed to decode this batch may be absent (e.g. an old bincode
+            // row that no longer decodes). The processor fails on a missing key
+            // before the post-process request path below runs, so request them up
+            // front (once), wait briefly for the primary to re-share, then re-fetch.
+            if !requested_missing_keys
+                && let Ok(missing) = proc.get_missing_key_ids(&pl).await
+                && !missing.is_empty()
+            {
+                requested_missing_keys = true;
+                let count = missing.len();
+                let listener = self.initial_keys_synced_notifier.listen();
+                self.request_missing_keys_with_dedup(missing).await;
+                debug!(target: "Client/AppState", "Requested {count} missing app-state key(s) for {:?}; waiting up to 10s before retrying", name);
+                let _ = rt_timeout(&*self.runtime, Duration::from_secs(10), listener).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);

src/message/special.rs

@@ -113,13 +113,12 @@ impl Client {
             );
         }
 
-        // Notify any waiters (initial full sync) that at least one key share was processed.
-        if stored_count > 0
-            && !self
-                .initial_app_state_keys_received
-                .swap(true, std::sync::atomic::Ordering::Relaxed)
-        {
-            // First time setting; notify any waiters
+        // Mark that keys have arrived (idempotent) and wake every waiter. Notifying
+        // on each share, not just the first, lets the app-state retry loop repair
+        // multiple missing keys one share at a time.
+        if stored_count > 0 {
+            self.initial_app_state_keys_received
+                .store(true, std::sync::atomic::Ordering::Relaxed);
             self.initial_keys_synced_notifier.notify(usize::MAX);
         }
     }

storages/sqlite-storage/Cargo.toml

@@ -18,7 +18,6 @@ bundled-sqlite = ["libsqlite3-sys/bundled"]
 
 [dependencies]
 async-trait = { workspace = true }
-bincode = { version = "2.0.1", features = ["serde"] }
 bytes = { workspace = true }
 diesel = { version = "2.3.10", default-features = false, features = [
     "sqlite",
@@ -30,6 +29,7 @@ diesel_migrations = { version = "2.3.2", default-features = false, features = [
 ] }
 libsqlite3-sys = { version = "0.37", default-features = false, optional = true }
 log = { workspace = true }
+prost = { workspace = true }
 serde_json = { workspace = true, features = ["std"] }
 tokio = { workspace = true, features = ["sync", "rt", "time", "macros"] }
 wacore = { workspace = true }

storages/sqlite-storage/src/lib.rs

@@ -5,5 +5,6 @@
 
 mod schema;
 mod sqlite_store;
+mod wire;
 
 pub use sqlite_store::SqliteStore;

storages/sqlite-storage/src/sqlite_store.rs

@@ -353,12 +353,7 @@ impl SqliteStore {
         let server_cert_chain: Option<Arc<[u8]>> = device_data
             .server_cert_chain
             .as_ref()
-            .map(|chain| {
-                bincode::serde::encode_to_vec(chain, bincode::config::standard())
-                    .map(Arc::from)
-                    .map_err(|e| StoreError::Serialization(Box::new(e)))
-            })
-            .transpose()?;
+            .map(|chain| Arc::from(crate::wire::encode_server_cert_chain(chain)));
         let login_counter = device_data.login_counter;
         let new_lid: Arc<str> = Arc::from(
             device_data
@@ -629,11 +624,8 @@ impl SqliteStore {
                         // change between versions) must NOT block startup —
                         // log it and degrade to None so the next connect
                         // simply pays one XX handshake to repopulate.
-                        match bincode::serde::decode_from_slice(
-                            bytes,
-                            bincode::config::standard(),
-                        ) {
-                            Ok((chain, _)) => Some(chain),
+                        match crate::wire::decode_server_cert_chain(bytes) {
+                            Ok(chain) => Some(chain),
                             Err(e) => {
                                 log::warn!(
                                     "device {} server_cert_chain blob ({} bytes) failed to decode: {e}; \
@@ -992,9 +984,20 @@ impl SqliteStore {
             .map_err(|e| StoreError::Database(Box::new(e)))??;
 
         if let Some(data) = res {
-            let (key, _) = bincode::serde::decode_from_slice(&data, bincode::config::standard())
-                .map_err(|e| StoreError::Serialization(Box::new(e)))?;
-            Ok(Some(key))
+            // An undecodable blob (an old bincode row or genuine corruption) is
+            // treated as absent: the app-state sync path then re-requests the key,
+            // the primary re-shares it, and the next set overwrites it as protobuf.
+            match crate::wire::decode_app_state_sync_key(&data) {
+                Ok(key) => Ok(Some(key)),
+                Err(e) => {
+                    warn!(
+                        "app_state_sync_key blob ({} bytes) failed to decode: {e}; \
+                         treating as absent, key will be re-requested",
+                        data.len()
+                    );
+                    Ok(None)
+                }
+            }
         } else {
             Ok(None)
         }
@@ -1008,8 +1011,7 @@ impl SqliteStore {
     ) -> Result<()> {
         let pool = self.pool.clone();
         let key_id = key_id.to_vec();
-        let data = bincode::serde::encode_to_vec(&key, bincode::config::standard())
-            .map_err(|e| StoreError::Serialization(Box::new(e)))?;
+        let data = crate::wire::encode_app_state_sync_key(&key);
         tokio::task::spawn_blocking(move || -> Result<()> {
             let mut conn = pool
                 .get()
@@ -1081,9 +1083,19 @@ impl SqliteStore {
             .map_err(|e| StoreError::Database(Box::new(e)))??;
 
         if let Some(data) = res {
-            let (state, _) = bincode::serde::decode_from_slice(&data, bincode::config::standard())
-                .map_err(|e| StoreError::Serialization(Box::new(e)))?;
-            Ok(state)
+            // An undecodable blob (an old bincode row or corruption) resets the
+            // collection to default, which simply re-syncs it from version 0.
+            match crate::wire::decode_hash_state(&data) {
+                Ok(state) => Ok(state),
+                Err(e) => {
+                    warn!(
+                        "app_state_version blob ({} bytes) failed to decode: {e}; \
+                         resetting to default, collection will re-sync from 0",
+                        data.len()
+                    );
+                    Ok(HashState::default())
+                }
+            }
         } else {
             Ok(HashState::default())
         }
@@ -1096,8 +1108,7 @@ impl SqliteStore {
         device_id: i32,
     ) -> Result<()> {
         let name = name.to_string();
-        let data = bincode::serde::encode_to_vec(&state, bincode::config::standard())
-            .map_err(|e| StoreError::Serialization(Box::new(e)))?;
+        let data = crate::wire::encode_hash_state(&state);
         self.with_retry("set_app_state_version", || {
             let name = name.clone();
             let data = data.clone();
@@ -3853,7 +3864,7 @@ mod tests {
     /// Round-trips a `CachedServerCertChain` through the SQLite schema:
     /// save → close store → reopen on the same db_name → load. Exercises
     /// the `2026-04-26-000000_add_server_cert_chain` migration plus the
-    /// bincode encode/decode path in `save_device_data_for_device` /
+    /// protobuf encode/decode path in `save_device_data_for_device` /
     /// `load_device_data_for_device` (the part that the in-memory backend
     /// integration tests don't reach).
     #[tokio::test]
@@ -3910,7 +3921,7 @@ mod tests {
 
         // Second store on the SAME shared-cache db: this exercises the
         // exact path a fresh-process load would take — schema migration
-        // already applied, BLOB column present, and the bincode-encoded
+        // already applied, BLOB column present, and the protobuf-encoded
         // chain decoded by the load path.
         let store = SqliteStore::new_for_device(&db_name, device_id)
             .await
@@ -3946,6 +3957,92 @@ mod tests {
         );
     }
 
+    // The migration strategy is self-healing: a row that no longer decodes (an old
+    // bincode blob or genuine corruption) must read back as ABSENT, never an error,
+    // so the sync path re-requests the key / re-syncs the collection from scratch.
+    #[tokio::test]
+    async fn undecodable_blobs_self_heal_to_absent() {
+        use diesel::sql_query;
+        use wacore::appstate::hash::HashState;
+        use wacore::store::traits::AppStateSyncKey;
+
+        let store = create_test_store().await;
+        let device_id = store.device_id;
+
+        // A valid sync key reads back normally.
+        let key_id = b"key-id-1".to_vec();
+        store
+            .set_app_state_sync_key_for_device(
+                &key_id,
+                AppStateSyncKey {
+                    key_data: vec![7u8; 32],
+                    fingerprint: vec![1, 2, 3],
+                    timestamp: 42,
+                },
+                device_id,
+            )
+            .await
+            .expect("set key");
+        assert!(
+            store
+                .get_app_state_sync_key_for_device(&key_id, device_id)
+                .await
+                .expect("get key")
+                .is_some()
+        );
+
+        // Garbage in the blob column -> treated as absent (so it gets re-requested),
+        // not a decode error bubbling up.
+        store
+            .with_retry("corrupt_key", || {
+                Box::new(|conn| {
+                    sql_query("UPDATE app_state_keys SET key_data = X'00ff00ff'")
+                        .execute(conn)
+                        .map(|_| ())
+                })
+            })
+            .await
+            .expect("corrupt key blob");
+        assert!(
+            store
+                .get_app_state_sync_key_for_device(&key_id, device_id)
+                .await
+                .expect("get corrupted key must not error")
+                .is_none(),
+            "an undecodable sync-key blob must read back as absent"
+        );
+
+        // A garbage app-state version blob resets to default (version 0) so the
+        // collection re-syncs from scratch instead of erroring.
+        let name = "critical_block";
+        let state = HashState {
+            version: 9,
+            ..HashState::default()
+        };
+        store
+            .set_app_state_version_for_device(name, state, device_id)
+            .await
+            .expect("set version");
+        store
+            .with_retry("corrupt_version", || {
+                Box::new(|conn| {
+                    sql_query("UPDATE app_state_versions SET state_data = X'00ff00ff'")
+                        .execute(conn)
+                        .map(|_| ())
+                })
+            })
+            .await
+            .expect("corrupt version blob");
+        let healed = store
+            .get_app_state_version_for_device(name, device_id)
+            .await
+            .expect("get corrupted version must not error");
+        assert_eq!(
+            healed.version, 0,
+            "an undecodable version blob must reset to default"
+        );
+    }
+
     #[tokio::test]
     async fn group_metadata_round_trip_sqlite() {
         use wacore::store::traits::ProtocolStore;