diff --git a/src/cache_config.rs b/src/cache_config.rs index 4ef291a9e..4dcc5b5f0 100644 --- a/src/cache_config.rs +++ b/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 diff --git a/src/client.rs b/src/client.rs index e3ed38e0a..250df94d7 100644 --- a/src/client.rs +++ b/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, + /// 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, + /// 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. diff --git a/src/client/accessors.rs b/src/client/accessors.rs index 9ffb85ec1..10fcca1b6 100644 --- a/src/client/accessors.rs +++ b/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(), diff --git a/src/client/lid_pn.rs b/src/client/lid_pn.rs index d56d8aae1..b40825898 100644 --- a/src/client/lid_pn.rs +++ b/src/client/lid_pn.rs @@ -420,6 +420,10 @@ impl Client { /// Callers must NOT hold `session_lock_for()` 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 = 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" + ); + } } diff --git a/src/client/lifecycle.rs b/src/client/lifecycle.rs index 02aa598d1..6ca5b1614 100644 --- a/src/client/lifecycle.rs +++ b/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(), diff --git a/src/message/receive.rs b/src/message/receive.rs index d2d365bf5..fdec96fb5 100644 --- a/src/message/receive.rs +++ b/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, diff --git a/src/message/retry.rs b/src/message/retry.rs index e1413210c..d793aae86 100644 --- a/src/message/retry.rs +++ b/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(), diff --git a/src/pdo.rs b/src/pdo.rs index f1c445a74..ef0020efd 100644 --- a/src/pdo.rs +++ b/src/pdo.rs @@ -98,6 +98,32 @@ impl Client { }; let cache_key = ChatMessageId::new(cache_chat, info.id.clone()); + // One request per message, like WA Web's session-lifetime set in + // WAWebNonMessageDataRequestPlaceholderMessageResendUtils. The + // pending cache below only covers in-flight requests; once the phone + // answers (even without content) it empties, and a sender that keeps + // redelivering the same undecryptable message would otherwise trigger + // a fresh request per copy. Claimed via the single-flight `get_with` + // (same arm as `dispatch_undecryptable_event`): decrypt-failure tasks + // are detached per copy, so a get-then-insert would let two + // concurrent copies both pass the gate, and only the claim winner may + // release the slot on send failure below. + let claimed = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let claimed_clone = claimed.clone(); + self.pdo_requested + .get_with(cache_key.clone(), async move { + claimed_clone.store(true, std::sync::atomic::Ordering::Release); + }) + .await; + if !claimed.load(std::sync::atomic::Ordering::Acquire) { + debug!( + "PDO request already sent for message {} from {}; not re-requesting", + info.id, + info.source.sender.observe() + ); + return Ok(()); + } + if self.pdo_pending_requests.get(&cache_key).await.is_some() { debug!( "PDO request already pending for message {} from {}", @@ -157,16 +183,20 @@ impl Client { peer_target.observe() ); + // A failed send must not consume the once-per-message slot, or a + // transient error would permanently block recovery for this message. if let Err(e) = self .ensure_e2e_sessions(std::slice::from_ref(&peer_target)) .await { self.pdo_pending_requests.remove(&cache_key).await; + self.pdo_requested.remove(&cache_key).await; return Err(e); } if let Err(e) = self.send_peer_message(peer_target, &msg).await { self.pdo_pending_requests.remove(&cache_key).await; + self.pdo_requested.remove(&cache_key).await; warn!( "Failed to send PDO request for message {}: {:?}", info.id, e @@ -356,7 +386,9 @@ impl Client { }; let Some(message) = web_msg_info.message else { - warn!("PDO response WebMessageInfo missing message content"); + // Expected when the phone could not decrypt the message either; + // WA Web only counts this outcome in telemetry, with no warning. + info!("PDO response WebMessageInfo missing message content"); return; }; @@ -684,4 +716,155 @@ mod tests { assert_eq!(info.source.chat.to_string(), peer_lid); assert!(info.source.is_from_me); } + + // Once-per-message memo tests: WA Web sends at most one placeholder + // resend request per message per session + // (WAWebNonMessageDataRequestPlaceholderMessageResendUtils); these pin + // the same contract onto `pdo_requested`. + + fn make_group_message_info( + chat: &str, + sender: &str, + id: &str, + ) -> std::sync::Arc { + use wacore::types::message::{MessageInfo, MessageSource}; + std::sync::Arc::new(MessageInfo { + id: id.to_owned(), + source: MessageSource { + chat: chat.parse().expect("chat jid"), + sender: sender.parse().expect("sender jid"), + is_group: true, + ..Default::default() + }, + timestamp: wacore::time::now_utc(), + ..Default::default() + }) + } + + async fn set_own_pn(client: &std::sync::Arc) { + client + .persistence_manager + .process_command(crate::store::commands::DeviceCommand::SetId(Some( + "5511777776666:2@s.whatsapp.net".parse().expect("own jid"), + ))) + .await; + } + + /// A message that already went through one placeholder resend must not + /// trigger another request, no matter how many times the server + /// redelivers the undecryptable original. + #[tokio::test] + async fn pdo_request_skipped_when_already_requested() { + use wacore::types::message::ChatMessageId; + + let client = setup_reconstruct_client().await; + set_own_pn(&client).await; + + let info = make_group_message_info( + "120363000000000001@g.us", + "203040904720543@lid", + "PDO_ONCE_1", + ); + let key = ChatMessageId::new(info.source.chat.clone(), info.id.clone()); + client.pdo_requested.insert(key.clone(), ()).await; + + let res = client.send_pdo_placeholder_resend_request(&info).await; + + assert!(res.is_ok(), "gated path reports success: {res:?}"); + assert!( + client.pdo_pending_requests.get(&key).await.is_none(), + "gated request must not register a pending entry" + ); + } + + /// A transient send failure must release the once-per-message slot, or + /// one bad send would permanently block recovery for that message. + #[tokio::test] + async fn pdo_request_failure_releases_once_per_message_slot() { + use wacore::types::message::ChatMessageId; + + let client = setup_reconstruct_client().await; + set_own_pn(&client).await; + // A live client has finished offline sync long before any PDO; skip + // the offline-delivery wait so the send failure surfaces immediately. + client + .offline_sync_completed + .store(true, std::sync::atomic::Ordering::Relaxed); + + let info = make_group_message_info( + "120363000000000001@g.us", + "203040904720543@lid", + "PDO_ONCE_2", + ); + let key = ChatMessageId::new(info.source.chat.clone(), info.id.clone()); + + let res = tokio::time::timeout( + std::time::Duration::from_secs(15), + client.send_pdo_placeholder_resend_request(&info), + ) + .await + .expect("send attempt must resolve fast without a live transport"); + + assert!(res.is_err(), "no live transport, the send must fail"); + assert!( + client.pdo_requested.get(&key).await.is_none(), + "failed send must release the once-per-message slot" + ); + assert!( + client.pdo_pending_requests.get(&key).await.is_none(), + "failed send must clear the pending entry" + ); + } + + /// A phone response without content consumes the pending slot but keeps + /// the memo: the phone has nothing to share for this message, so + /// re-asking on the next redelivery cannot produce content either. + #[tokio::test] + async fn pdo_missing_content_response_clears_pending_but_keeps_memo() { + use prost::Message as _; + use wacore::types::message::ChatMessageId; + + let client = setup_reconstruct_client().await; + let chat = "5511999998888@s.whatsapp.net"; + let msg_id = "PDO_ONCE_3"; + let key = ChatMessageId::new(chat.parse().expect("chat jid"), msg_id.to_owned()); + + client.pdo_requested.insert(key.clone(), ()).await; + client + .pdo_pending_requests + .insert( + key.clone(), + super::PendingPdoRequest { + message_info: make_group_message_info(chat, chat, msg_id), + requested_at: wacore::time::Instant::now(), + }, + ) + .await; + + let web_msg = waproto::whatsapp::WebMessageInfo { + key: waproto::whatsapp::MessageKey { + remote_jid: Some(chat.to_owned()), + from_me: Some(false), + id: Some(msg_id.to_owned()), + participant: None, + }, + ..Default::default() + }; + let response = waproto::whatsapp::message::peer_data_operation_request_response_message::peer_data_operation_result::PlaceholderMessageResendResponse { + web_message_info_bytes: Some(web_msg.encode_to_vec()), + }; + + client + .handle_placeholder_resend_response(&response, "req-1") + .await; + + assert!( + client.pdo_pending_requests.get(&key).await.is_none(), + "response consumes the pending slot" + ); + assert!( + client.pdo_requested.get(&key).await.is_some(), + "memo must survive a content-less response" + ); + } }