fix(pdo): request a placeholder resend at most once per message

src/cache_config.rs

@@ -181,6 +181,14 @@ pub struct CacheConfig {
     pub undecryptable_dispatched: CacheEntryConfig,
     /// PDO pending requests (time_to_live). Default: 30s TTL, 200 entries.
     pub pdo_pending_requests: CacheEntryConfig,
+    /// Messages already covered by a placeholder-resend PDO request
+    /// (time_to_live). WA Web keeps a session-lifetime set
+    /// (`WAWebNonMessageDataRequestPlaceholderMessageResendUtils`) so each
+    /// message triggers at most one request; without it, every redelivery of
+    /// an undecryptable message re-asks the phone (a stuck sender resending
+    /// every ~11s produced ~700 requests in 3h). The TTL stands in for
+    /// "session lifetime" with bounded memory. Default: 24h TTL, 512 entries.
+    pub pdo_requested: CacheEntryConfig,
     /// Sender key device tracking cache (time_to_idle). Default: 1h TTI, 500 entries.
     /// Caches per-group SKDM distribution state to avoid DB reads on every group send.
     pub sender_key_devices_cache: CacheEntryConfig,
@@ -260,6 +268,7 @@ impl std::fmt::Debug for CacheConfig {
             .field("message_retry_counts", &self.message_retry_counts)
             .field("undecryptable_dispatched", &self.undecryptable_dispatched)
             .field("pdo_pending_requests", &self.pdo_pending_requests)
+            .field("pdo_requested", &self.pdo_requested)
             .field("sender_key_devices_cache", &self.sender_key_devices_cache)
             .field("session_recreate_history", &self.session_recreate_history)
             .field("session_locks_capacity", &self.session_locks_capacity)
@@ -312,6 +321,7 @@ impl Default for CacheConfig {
             message_retry_counts: CacheEntryConfig::new(one_hour, 500),
             undecryptable_dispatched: CacheEntryConfig::new(five_min, 1_000),
             pdo_pending_requests: CacheEntryConfig::new(Some(Duration::from_secs(30)), 200),
+            pdo_requested: CacheEntryConfig::new(Some(Duration::from_secs(24 * 3600)), 512),
             sender_key_devices_cache: CacheEntryConfig::new(one_hour, 500),
             session_recreate_history: CacheEntryConfig::new(one_hour, 256),
             // Coordination caches hold live mutexes/senders; capacity eviction

src/client.rs

@@ -190,6 +190,7 @@ pub struct MemoryDiagnostics {
     pub message_retry_counts: u64,
     pub undecryptable_dispatched: u64,
     pub pdo_pending_requests: u64,
+    pub pdo_requested: u64,
     // -- Moka caches (capacity-only, no TTL) --
     pub session_locks: u64,
     pub chat_lanes: u64,
@@ -234,6 +235,7 @@ impl std::fmt::Display for MemoryDiagnostics {
             self.undecryptable_dispatched
         )?;
         writeln!(f, "  pdo_pending_requests:   {}", self.pdo_pending_requests)?;
+        writeln!(f, "  pdo_requested:          {}", self.pdo_requested)?;
         writeln!(f, "--- Moka caches (capacity-only) ---")?;
         writeln!(f, "  session_locks:          {}", self.session_locks)?;
         writeln!(f, "  chat_lanes:             {}", self.chat_lanes)?;
@@ -527,6 +529,14 @@ pub struct Client {
     pub(crate) pdo_pending_requests:
         Cache<wacore::types::message::ChatMessageId, crate::pdo::PendingPdoRequest>,
 
+    /// Messages already covered by a placeholder-resend PDO request. Mirrors
+    /// the session-lifetime set in
+    /// `WAWebNonMessageDataRequestPlaceholderMessageResendUtils`: at most one
+    /// request per message, no matter how many times the server redelivers
+    /// the undecryptable original. Entries are dropped on send failure so a
+    /// transient error does not block the next attempt.
+    pub(crate) pdo_requested: Cache<wacore::types::message::ChatMessageId, ()>,
+
     /// LRU cache for device registry (matches WhatsApp Web's 5000 entry limit).
     /// Maps user ID to DeviceListRecord for fast device existence checks.
     /// Backed by persistent storage.

src/client/accessors.rs

@@ -92,6 +92,7 @@ impl Client {
             message_retry_counts: self.message_retry_counts.entry_count(),
             undecryptable_dispatched: self.undecryptable_dispatched.entry_count(),
             pdo_pending_requests: self.pdo_pending_requests.entry_count(),
+            pdo_requested: self.pdo_requested.entry_count(),
             session_locks: self.session_locks.entry_count(),
             chat_lanes: self.chat_lanes.entry_count(),
             response_waiters: self.response_waiters.lock().await.len(),

src/client/lid_pn.rs

@@ -420,6 +420,10 @@ impl Client {
     /// Callers must NOT hold `session_lock_for(<lid_addr>)` for any device
     /// in [0, 100) — `async_lock::Mutex` is not reentrant. The decrypt path
     /// drops its address lock around the call (`try_pn_to_lid_migration_decrypt`).
+    ///
+    /// Returns whether anything moved into a LID slot. When `false`, decrypt
+    /// state is unchanged, so a failed decrypt retried after this call is
+    /// guaranteed to fail identically and callers can skip the retry.
     #[cfg_attr(
         feature = "tracing",
         tracing::instrument(
@@ -428,7 +432,11 @@ impl Client {
             skip_all
         )
     )]
-    pub(crate) async fn migrate_signal_sessions_on_lid_discovery(&self, pn: &str, lid: &str) {
+    pub(crate) async fn migrate_signal_sessions_on_lid_discovery(
+        &self,
+        pn: &str,
+        lid: &str,
+    ) -> bool {
         use log::{info, warn};
         use wacore::types::jid::JidExt;
 
@@ -443,9 +451,11 @@ impl Client {
             .has_state_for_user(pn, backend.as_ref())
             .await
         {
-            return;
+            return false;
         }
 
+        let mut migrated = false;
+
         for device_id in 0..MIGRATION_DEVICE_RANGE {
             // `&str` → `CompactString` is inline for ≤24-byte user parts
             // (all PN/LID identifiers fit), so no String intermediate.
@@ -479,6 +489,7 @@ impl Client {
             {
                 self.signal_cache.put_session(&lid_proto, session).await;
                 self.signal_cache.delete_session(&pn_proto).await;
+                migrated = true;
                 info!(
                     "Migrated session {} -> {} (PN wins on conflict)",
                     pn_proto, lid_proto
@@ -511,6 +522,7 @@ impl Client {
                             .put_identity(&lid_proto, &identity_data)
                             .await;
                         self.signal_cache.delete_identity(&pn_proto).await;
+                        migrated = true;
                         info!("Migrated identity {} -> {}", pn_proto, lid_proto);
                     }
                     Ok(Some(_)) => {
@@ -532,6 +544,7 @@ impl Client {
         if let Err(e) = self.signal_cache.flush(backend.as_ref()).await {
             warn!("Failed to flush signal cache after migration: {e:?}");
         }
+        migrated
     }
 
     /// Look up the LID↔phone mapping for a JID. Cache-aside: falls back to
@@ -1292,4 +1305,54 @@ mod tests {
              stays serialized on the address lock"
         );
     }
+
+    /// `try_pn_to_lid_migration_decrypt` skips its retry decrypt when the
+    /// migration reports nothing moved: with decrypt state unchanged, the
+    /// retry would fail identically and log a second decrypt error for
+    /// every redelivered copy of an undecryptable message.
+    #[tokio::test]
+    async fn migration_reports_whether_anything_moved() {
+        use wacore::libsignal::protocol::SessionRecord;
+        use wacore::types::jid::JidExt as _;
+
+        let client: Arc<Client> = create_test_client().await;
+        let pn = "5500000001111";
+        let lid = "122222222222222";
+
+        client
+            .add_lid_pn_mapping(lid, pn, LearningSource::PeerPnMessage)
+            .await
+            .unwrap();
+
+        assert!(
+            !client
+                .migrate_signal_sessions_on_lid_discovery(pn, lid)
+                .await,
+            "no PN signal state, so nothing can move"
+        );
+
+        let pn_addr = Jid::pn_device(pn.to_string(), 0).to_protocol_address();
+        client
+            .signal_cache
+            .put_session(
+                &pn_addr,
+                SessionRecord::deserialize(&tagged_session_blob(7)).expect("blob deserializes"),
+            )
+            .await;
+        let backend = client.persistence_manager.backend();
+        client.signal_cache.flush(backend.as_ref()).await.unwrap();
+
+        assert!(
+            client
+                .migrate_signal_sessions_on_lid_discovery(pn, lid)
+                .await,
+            "a PN session moved into the LID slot"
+        );
+        assert!(
+            !client
+                .migrate_signal_sessions_on_lid_discovery(pn, lid)
+                .await,
+            "second call finds the PN side already drained"
+        );
+    }
 }

src/client/lifecycle.rs

@@ -224,6 +224,7 @@ impl Client {
             custom_enc_handlers: std::sync::OnceLock::new(),
             chatstate_handlers: Arc::new(RwLock::new(Vec::new())),
             pdo_pending_requests: cache_config.pdo_pending_requests.build_with_ttl(),
+            pdo_requested: cache_config.pdo_requested.build_with_ttl(),
             device_registry_cache: crate::client::device_topology::DeviceRegistryCache::new(
                 cache_config.device_registry_cache.build_typed_ttl(
                     cache_config.cache_stores.device_registry_cache.clone(),

src/message/receive.rs

@@ -416,8 +416,11 @@ impl Client {
                             info.source.sender.observe()
                         );
                     } else {
-                        log::log!(
-                            decrypt_fail_log_level(decrypt_fail_mode),
+                        // WA Web skips the skmsg silently after a retryable
+                        // pkmsg failure (canDecryptNext in
+                        // WAWebMsgProcessingDecryptionHandler); the pkmsg
+                        // failure itself is already logged and retried.
+                        log::debug!(
                             "Skipping skmsg decryption for message {} from {} because pkmsg failed to decrypt.",
                             info.id,
                             info.source.sender.observe()
@@ -1446,12 +1449,25 @@ impl Client {
         // Release the address lock so the migration loop can acquire it for
         // the matching device without re-entering.
         *session_guard = None;
-        self.migrate_signal_sessions_on_lid_discovery(&pn, &sender_jid.user)
+        let migrated = self
+            .migrate_signal_sessions_on_lid_discovery(&pn, &sender_jid.user)
             .await;
         // Re-acquire for the retry decrypt and hand the guard back to the
         // caller for subsequent payloads in the batch.
         *session_guard = Some(session_mutex.lock_arc().await);
 
+        // Nothing moved namespaces, so the retry would hit the exact same
+        // state, fail identically, and log a second decrypt failure for
+        // every redelivered copy of an undecryptable message.
+        if !migrated {
+            log::debug!(
+                "[msg:{}] No PN state to migrate for {}; skipping migration retry decrypt",
+                info.id,
+                info.source.sender.observe()
+            );
+            return MigrationDecryptOutcome::default();
+        }
+
         match message_decrypt(
             parsed_message,
             signal_address,

src/message/retry.rs

@@ -185,7 +185,9 @@ impl Client {
     /// This asks our primary phone to share the already-decrypted message content.
     /// PDO is NOT spawned on subsequent retries to avoid duplicate requests.
     ///
-    /// When max retries is reached, an immediate PDO request is sent as a last resort.
+    /// When max retries is reached, a PDO request is attempted as a last resort;
+    /// the `pdo_requested` memo makes it a no-op if one already went out for
+    /// this message, so capped redeliveries cannot re-ask the phone.
     ///
     /// # Arguments
     /// * `info` - The message info for the failed message
@@ -220,8 +222,12 @@ impl Client {
             .await;
 
         let Some(retry_count) = self.increment_retry_count(&cache_key, reason).await else {
-            log::info!(
-                "Max retries ({}) reached for message {} from {} [{:?}]. Sending immediate PDO request.",
+            // Every further redelivery of a capped message lands here, so
+            // keep it at debug; the high-retry warn already fired on the way
+            // to the cap, and the PDO is a once-per-message no-op after the
+            // first request.
+            log::debug!(
+                "Max retries ({}) reached for message {} from {} [{:?}]. Requesting PDO fallback.",
                 MAX_DECRYPT_RETRIES,
                 info.id,
                 info.source.sender.observe(),