From 73085eb6809547f2ce936c2e0b73bfcd3ef36a11 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= Date: Wed, 27 May 2026 20:31:30 -0300 Subject: [PATCH] fix(offline): drain offline queue by acking every received message Duplicate, undecryptable, unavailable and own-account fan-out messages were acknowledged inconsistently, so the server replayed them on every reconnect until it force-closed the stream. Ensure every received message is acked: - Ack duplicates and own-account fan-out (own non-peer gets a transport ack with the correct to=from, since its delivery receipt is suppressed). - Send a transport ack on decrypt failure and on ; the retry receipt alone does not clear the offline queue. - Nack generic group decrypt errors, matching the session catch-all. - Send the retry receipt, PDO request and transport ack in one ordered, flushed task so a disconnect can't clear a stanza before its resend request (retry/PDO) goes out. - Send delivery receipts as type="inactive" unless marked online, matching whatsmeow's background-companion behavior. - Raise message_retry_counts TTL to 1h so the retry cap survives spaced redeliveries. - Lower decrypt-fail="hide" failures to DEBUG. - Recognize stream:error with an ack child instead of logging it as unknown. --- src/cache_config.rs | 7 +- src/client.rs | 310 +++++++++++++++++- src/features/presence.rs | 10 +- src/message.rs | 672 ++++++++++++++++++++++++++++++++++----- src/pdo.rs | 99 ++---- src/receipt.rs | 58 +++- 6 files changed, 996 insertions(+), 160 deletions(-) diff --git a/src/cache_config.rs b/src/cache_config.rs index c454a272c..9cbe5b6fd 100644 --- a/src/cache_config.rs +++ b/src/cache_config.rs @@ -170,7 +170,8 @@ pub struct CacheConfig { /// Default: capacity 0 (disabled — DB-only, matching WA Web). /// Set capacity > 0 to enable a fast in-memory cache in front of the DB. pub recent_messages: CacheEntryConfig, - /// Message retry counts (time_to_live). Default: 5m TTL, 500 entries. + /// Message retry counts (time_to_live). Default: 1h TTL, 500 entries. + /// Long enough that the MAX_DECRYPT_RETRIES cap survives spaced redeliveries. pub message_retry_counts: CacheEntryConfig, /// Dedup key for `UndecryptableMessage` dispatch so a server resend of /// the same id does not surface a second notification. Default: 5m TTL, @@ -247,7 +248,9 @@ impl Default for CacheConfig { device_registry_cache: CacheEntryConfig::new(one_hour, 1_000), lid_pn_cache: CacheEntryConfig::new(None, u64::MAX), recent_messages: CacheEntryConfig::new(five_min, 0), - message_retry_counts: CacheEntryConfig::new(five_min, 500), + // 1h so the MAX_DECRYPT_RETRIES cap survives spaced redeliveries; a + // 5m TTL expired between reconnects so the count never reached the cap. + 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), sender_key_devices_cache: CacheEntryConfig::new(one_hour, 500), diff --git a/src/client.rs b/src/client.rs index 168ad04f2..a8488fde9 100644 --- a/src/client.rs +++ b/src/client.rs @@ -312,6 +312,10 @@ pub struct Client { /// TOCTOU races where `try_lock()` fails due to contention, not disconnection. is_connected: Arc, + /// whatsmeow's `sendActiveReceipts`: 0 = inactive (default), 1 = active + /// (presence available), 2 = forced. When 0, delivery receipts use `type="inactive"`. + send_active_receipts: AtomicU32, + /// Per-process counter of consecutive Noise IK handshake failures, scoped /// to the lifetime of this `Client`. Mirrors `K` in WA Web's /// `WAWebOpenChatSocket` (`ChatSocket.js`): on the first failure within a @@ -771,6 +775,7 @@ impl Client { is_connecting: Arc::new(AtomicBool::new(false)), is_running: Arc::new(AtomicBool::new(false)), is_connected: Arc::new(AtomicBool::new(false)), + send_active_receipts: AtomicU32::new(0), ik_handshake_failures: Arc::new(AtomicU32::new(0)), shutdown_notifier: wacore::runtime::ShutdownNotifier::new(), connection_shutdown: std::sync::Mutex::new(wacore::runtime::ShutdownNotifier::new()), @@ -1415,6 +1420,11 @@ impl Client { // is_connected==true with a cleared socket. send_node() independently // checks the socket, but this ordering avoids a confusing state window. self.is_connected.store(false, Ordering::Release); + // Presence doesn't survive reconnects: demote presence-driven active + // receipts (1 -> 0), leaving a forced value (2) untouched. + let _ = + self.send_active_receipts + .compare_exchange(1, 0, Ordering::AcqRel, Ordering::Acquire); // Drop per-chat lanes so workers exit via channel close. self.chat_lanes.invalidate_all(); // Clear pending retries so stale keys from detached scopeguard @@ -1945,6 +1955,38 @@ impl Client { self.send_raw_bytes(buf).await } + /// Send a transport ack so the server stops replaying a stanza from the + /// offline queue. Awaitable so callers can order it after a retry receipt + /// in a single flushed task. + pub(crate) async fn send_transport_ack(&self, info: &crate::types::message::MessageInfo) { + let source = message_ack_source_node(info); + let own_pn = self.get_pn().await; + match encode_ack_bytes(&source.as_node_ref(), own_pn.as_ref()) { + Ok(Some(buf)) => { + if let Err(e) = self.send_raw_bytes(buf).await + && !e.is_transport_unavailable() + { + log::warn!("Failed to send transport ack for undecryptable message: {e:?}"); + } + } + Ok(None) => {} + Err(e) => log::warn!("Failed to encode transport ack: {e}"), + } + } + + /// Spawn [`Self::send_transport_ack`], tracked via `outbound_flush` so + /// `disconnect()` flushes it (issue #571), same as delivery receipts. + pub(crate) fn spawn_message_ack( + self: &Arc, + info: &Arc, + ) { + let client = Arc::clone(self); + let info = Arc::clone(info); + self.outbound_flush.spawn(&*self.runtime, async move { + client.send_transport_ack(&info).await; + }); + } + pub async fn set_passive(&self, passive: bool) -> Result<(), crate::request::IqError> { use wacore::iq::passive::PassiveModeSpec; self.execute(PassiveModeSpec::new(passive)).await @@ -3338,7 +3380,24 @@ impl Client { // but here is_fully_ready() gates prekey uploads and we want them to keep // working while the socket is still alive. Severity is warn!, not error!, // because the connection is intentionally preserved. - warn!("Unknown stream error: {}", DisplayableNodeRef(node)); + // WA Web (StreamError.js) knows (type "ack"); + // name it instead of "Unknown". Root cause is usually an un-acked + // offline stanza; the server's drives the reconnect. + if let Some(ack) = node.get_optional_child("ack") { + let id = ack + .get_attr("id") + .map(|v| v.as_str().to_string()) + .unwrap_or_default(); + let class = ack + .get_attr("class") + .map(|v| v.as_str().to_string()) + .unwrap_or_default(); + warn!( + "Stream error: server rejected ack (class={class:?}, id={id}); reconnect will follow on stream end" + ); + } else { + warn!("Unknown stream error: {}", DisplayableNodeRef(node)); + } self.core.event_bus.dispatch(Event::StreamError( crate::types::events::StreamError { code: code.to_string(), @@ -3456,6 +3515,32 @@ impl Client { self.is_connected.load(Ordering::Acquire) } + /// Whether delivery receipts should be sent active (rendered as ticks) vs + /// `type="inactive"`. Mirrors whatsmeow's `sendActiveReceipts != 0`. + pub(crate) fn receipts_are_active(&self) -> bool { + self.send_active_receipts.load(Ordering::Acquire) != 0 + } + + /// Force active delivery receipts even when offline (whatsmeow's + /// `SetForceActiveDeliveryReceipts`); off restores the default. + pub fn set_force_active_delivery_receipts(&self, active: bool) { + self.send_active_receipts + .store(if active { 2 } else { 0 }, Ordering::Release); + } + + /// CAS so a forced value (2) is preserved (whatsmeow's `CompareAndSwap`). + pub(crate) fn mark_receipts_active_on_presence(&self) { + let _ = + self.send_active_receipts + .compare_exchange(0, 1, Ordering::AcqRel, Ordering::Acquire); + } + + pub(crate) fn mark_receipts_inactive_on_presence(&self) { + let _ = + self.send_active_receipts + .compare_exchange(1, 0, Ordering::AcqRel, Ordering::Acquire); + } + pub fn is_logged_in(&self) -> bool { self.is_logged_in.load(Ordering::Relaxed) } @@ -4035,6 +4120,29 @@ fn encode_ack_bytes( Ok(Some(buf)) } +/// Minimal `` stanza carrying the attrs `encode_ack_bytes` needs, +/// reconstructed after the node tree has been dropped. The original `from` +/// is the group for group/broadcast stanzas and the sender otherwise (sender +/// keeps the device qualifier; `chat` is device-stripped for DMs). Mirrors +/// whatsmeow's `sendAck` (`to`=from, copy recipient/participant). +fn message_ack_source_node(info: &crate::types::message::MessageInfo) -> Node { + let from = if info.source.is_group { + &info.source.chat + } else { + &info.source.sender + }; + let mut builder = NodeBuilder::new("message") + .attr("id", &info.id) + .attr("from", from); + if let Some(recipient) = &info.source.recipient { + builder = builder.attr("recipient", recipient); + } + if info.source.is_group { + builder = builder.attr("participant", &info.source.sender); + } + builder.build() +} + /// Build an ack Node (used in tests for structure verification). #[cfg(test)] fn build_ack_node(node: &wacore_binary::NodeRef<'_>, own_device_pn: Option<&Jid>) -> Option { @@ -5369,6 +5477,46 @@ mod tests { client } + #[tokio::test] + async fn delivery_receipt_activity_state_machine() { + let client = create_offline_sync_test_client().await; + assert!( + !client.receipts_are_active(), + "default is inactive (background companion)" + ); + client.mark_receipts_active_on_presence(); + assert!(client.receipts_are_active(), "presence available -> active"); + client.mark_receipts_inactive_on_presence(); + assert!( + !client.receipts_are_active(), + "presence unavailable -> inactive" + ); + client.set_force_active_delivery_receipts(true); + assert!(client.receipts_are_active(), "forced active"); + client.mark_receipts_inactive_on_presence(); + assert!( + client.receipts_are_active(), + "forced (2) survives a presence-unavailable CAS(1,0)" + ); + client.set_force_active_delivery_receipts(false); + assert!(!client.receipts_are_active()); + + // Teardown resets presence-driven active (so it doesn't leak across + // reconnects) but preserves a forced value. + client.mark_receipts_active_on_presence(); + client.cleanup_connection_state().await; + assert!( + !client.receipts_are_active(), + "teardown resets presence-driven active" + ); + client.set_force_active_delivery_receipts(true); + client.cleanup_connection_state().await; + assert!( + client.receipts_are_active(), + "teardown preserves forced active" + ); + } + #[tokio::test] async fn test_ib_thread_metadata_does_not_end_sync() { let client = create_offline_sync_test_client().await; @@ -6045,6 +6193,166 @@ mod tests { ); } + /// Own-account fan-out ack must address back to the original `from` (own + /// LID) echoing `recipient`, not to the chat. Guards against regressing to + /// the chat-addressed `build_nack_node` style. + #[test] + fn test_message_ack_source_node_own_device_addressing() { + use crate::types::message::{MessageInfo, MessageSource}; + // Own-account branch: sender == `from` (device-qualified), chat is the + // device-stripped recipient. `to` must come from sender, not chat. + let info = MessageInfo { + id: "AC055553E56A2C12DE592DAD6353C477".to_string(), + source: MessageSource { + sender: "236395184570386@lid".parse().expect("sender"), + chat: "156535032389744@lid".parse().expect("chat"), + recipient: Some("156535032389744@lid".parse().expect("recipient")), + is_group: false, + ..Default::default() + }, + ..Default::default() + }; + let own_device_pn: Jid = "559984726662:95@s.whatsapp.net" + .parse() + .expect("own device PN JID should parse"); + + let source = message_ack_source_node(&info); + let built = build_ack_node(&source.as_node_ref(), Some(&own_device_pn)) + .expect("message ack should be buildable"); + + assert!(built.attrs.get("class").is_some_and(|v| v == "message")); + assert!( + built + .attrs + .get("to") + .is_some_and(|v| v == "236395184570386@lid"), + "ack `to` must be the original `from` (own LID), not the chat" + ); + assert!( + built + .attrs + .get("recipient") + .is_some_and(|v| v == "156535032389744@lid"), + "ack must echo `recipient` so the server can route/clear it" + ); + assert!( + !built.attrs.contains_key("type"), + "message-class acks never carry a `type`" + ); + } + + /// Common incoming DM from another user: `to` is the device-qualified + /// sender, with no `recipient`/`participant` synthesised. + #[test] + fn test_message_ack_source_node_incoming_dm_addressing() { + use crate::types::message::{MessageInfo, MessageSource}; + let info = MessageInfo { + id: "MSGID".to_string(), + source: MessageSource { + sender: "5511999998888:3@s.whatsapp.net".parse().expect("sender"), + chat: "5511999998888@s.whatsapp.net".parse().expect("chat"), + is_group: false, + ..Default::default() + }, + ..Default::default() + }; + let own_device_pn: Jid = "559984726662:95@s.whatsapp.net" + .parse() + .expect("own device PN JID should parse"); + + let source = message_ack_source_node(&info); + let built = build_ack_node(&source.as_node_ref(), Some(&own_device_pn)) + .expect("dm ack should be buildable"); + + assert!( + built + .attrs + .get("to") + .is_some_and(|v| v == "5511999998888:3@s.whatsapp.net"), + "ack `to` must be the device-qualified sender (the original `from`)" + ); + assert!(!built.attrs.contains_key("recipient")); + assert!(!built.attrs.contains_key("participant")); + } + + /// status@broadcast (is_group=true in the parser) addresses the ack to the + /// status chat, with the sender as participant, not to the sender. + #[test] + fn test_message_ack_source_node_status_addressing() { + use crate::types::message::{MessageInfo, MessageSource}; + let info = MessageInfo { + id: "STATUSMSG".to_string(), + source: MessageSource { + chat: "status@broadcast".parse().expect("status chat"), + sender: "181531758878822@lid".parse().expect("participant"), + is_group: true, + ..Default::default() + }, + ..Default::default() + }; + let own_device_pn: Jid = "559984726662:95@s.whatsapp.net" + .parse() + .expect("own device PN JID should parse"); + + let source = message_ack_source_node(&info); + let built = build_ack_node(&source.as_node_ref(), Some(&own_device_pn)) + .expect("status ack should be buildable"); + + assert!( + built + .attrs + .get("to") + .is_some_and(|v| v == "status@broadcast"), + "status ack `to` must be the status chat, not the sender" + ); + assert!( + built + .attrs + .get("participant") + .is_some_and(|v| v == "181531758878822@lid"), + "status ack must preserve the sending participant" + ); + } + + /// Group failure ack: `to` is the group, `participant` is preserved. + #[test] + fn test_message_ack_source_node_group_addressing() { + use crate::types::message::{MessageInfo, MessageSource}; + // Group branch: chat == group `from`, sender == participant. + let info = MessageInfo { + id: "GROUPMSGID".to_string(), + source: MessageSource { + chat: "120363011111111111@g.us".parse().expect("group"), + sender: "181531758878822@lid".parse().expect("participant"), + is_group: true, + ..Default::default() + }, + ..Default::default() + }; + let own_device_pn: Jid = "559984726662:95@s.whatsapp.net" + .parse() + .expect("own device PN JID should parse"); + + let source = message_ack_source_node(&info); + let built = build_ack_node(&source.as_node_ref(), Some(&own_device_pn)) + .expect("group message ack should be buildable"); + + assert!( + built + .attrs + .get("to") + .is_some_and(|v| v == "120363011111111111@g.us"), + "group ack `to` must be the group JID" + ); + assert!( + built + .attrs + .get("participant") + .is_some_and(|v| v == "181531758878822@lid"), + "group ack must preserve the sending `participant`" + ); + } + /// Smoke test: server ping with xmlns but no id attribute is handled. #[tokio::test] async fn test_handle_iq_ping_without_id() { diff --git a/src/features/presence.rs b/src/features/presence.rs index dd9f7ea77..f0f58c8cd 100644 --- a/src/features/presence.rs +++ b/src/features/presence.rs @@ -82,8 +82,14 @@ impl<'a> Presence<'a> { return Err(PresenceError::PushNameEmpty); } - if status == PresenceStatus::Available { - self.client.send_unified_session().await; + // Track receipt activity like whatsmeow: available -> active receipts, + // unavailable -> back to inactive (a forced value is preserved). + match status { + PresenceStatus::Available => { + self.client.send_unified_session().await; + self.client.mark_receipts_active_on_presence(); + } + PresenceStatus::Unavailable => self.client.mark_receipts_inactive_on_presence(), } let presence_type = status.as_str(); diff --git a/src/message.rs b/src/message.rs index 700f11a4e..3a33a79d3 100644 --- a/src/message.rs +++ b/src/message.rs @@ -73,6 +73,16 @@ pub(crate) struct ClassifiedMessage { /// WhatsApp Web logs metrics when retry count exceeds this value. const HIGH_RETRY_COUNT_THRESHOLD: u8 = 3; +/// `decrypt-fail="hide"` failures are expected (addon/fan-out), so log them at +/// DEBUG to avoid WARN spam. Mode never changes control flow: retry + ack still +/// fire (WA Web retries regardless of `hide`). +fn decrypt_fail_log_level(mode: crate::types::events::DecryptFailMode) -> log::Level { + match mode { + crate::types::events::DecryptFailMode::Hide => log::Level::Debug, + crate::types::events::DecryptFailMode::Show => log::Level::Warn, + } +} + pub(crate) use wacore::protocol::retry::RetryReason; impl Client { @@ -88,18 +98,38 @@ impl Client { msg.get_base_message().get_ephemeral_expiration(); } - // Tracked so `disconnect()` can flush in-flight receipts (issue #571). - let client_clone = self.clone(); - let info_for_receipt = Arc::clone(&info); - self.outbound_flush.spawn(&*self.runtime, async move { - client_clone.send_delivery_receipt(&info_for_receipt).await; - }); + self.ack_received_message(&info); self.core .event_bus .dispatch(Event::Message(Arc::new(msg), info)); } + /// Acknowledge a received message so the server drops it from the offline + /// queue: a delivery receipt when applicable, else a transport ack for + /// own-account fan-out (its receipt is suppressed but the stanza still needs + /// clearing; the ack carries the correct to=from). status is acked by the + /// `should_ack` gate, newsletters/empty ids need nothing here. + fn ack_received_message(self: &Arc, info: &Arc) { + if info.id.is_empty() || info.source.chat.is_newsletter() { + return; + } + if Self::should_send_delivery_receipt(info) { + self.spawn_delivery_receipt(info); + } else if !info.source.chat.is_status_broadcast() { + self.spawn_message_ack(info); + } + } + + /// Spawn a delivery receipt, tracked so `disconnect()` can flush it (issue #571). + fn spawn_delivery_receipt(self: &Arc, info: &Arc) { + let client = self.clone(); + let info = Arc::clone(info); + self.outbound_flush.spawn(&*self.runtime, async move { + client.send_delivery_receipt(&info).await; + }); + } + /// Handles a newsletter plaintext message. /// Newsletters are not E2E encrypted and use the tag directly. async fn handle_newsletter_message( @@ -180,8 +210,17 @@ impl Client { was_fresh } - /// Dispatch an undecryptable event (once per msg id, matching WA Web's - /// DB-level placeholder uniqueness) and spawn a retry receipt. + /// Dispatch an undecryptable event, then send the retry receipt and the + /// transport ack in one ordered, flushed task. + /// + /// The retry asks the sender to re-encrypt; the ack clears the stanza from + /// the server's offline queue (the retry alone does not). Both run in a + /// single `outbound_flush` task so `disconnect()` flushes them together and + /// the retry always goes out before the ack: if only one makes it, it is the + /// retry, so the message is never cleared without a resend request. status is + /// also acked here (flushed) rather than relying on the detached `should_ack` + /// gate, which can be dropped mid-flush on disconnect; the server dedups the + /// resulting duplicate ack. /// /// Returns `true` to be assigned to `dispatched_undecryptable` flag. async fn handle_decrypt_failure( @@ -197,7 +236,16 @@ impl Client { decrypt_fail_mode, ) .await; - self.spawn_retry_receipt(info, reason); + let client = Arc::clone(self); + let info = Arc::clone(info); + self.outbound_flush.spawn(&*self.runtime, async move { + // Only ack once the resend request is actually out; otherwise leave + // the stanza queued so the server redelivers and we retry. + let resend_sent = client.run_retry_receipt(&info, reason).await; + if resend_sent { + client.send_transport_ack(&info).await; + } + }); true } @@ -265,66 +313,79 @@ impl Client { /// # Arguments /// * `info` - The message info for the failed message /// * `reason` - The retry reason code (matches WhatsApp Web's RetryReason enum) + #[cfg(test)] fn spawn_retry_receipt(self: &Arc<Self>, info: &Arc<MessageInfo>, reason: RetryReason) { let client = Arc::clone(self); let info = Arc::clone(info); + self.outbound_flush.spawn(&*self.runtime, async move { + client.run_retry_receipt(&info, reason).await; + }); + } - self.runtime.spawn(Box::pin(async move { - let cache_key = client - .make_retry_cache_key(&info.source.chat, &info.id, &info.source.sender) - .await; + /// Increment the retry count and send the retry receipt (or, at the cap, a + /// last-resort PDO). Awaitable so it can be ordered before the transport ack. + /// + /// Returns whether the caller should send the ack: `false` when we intended + /// to retry but the send failed (so the stanza stays queued for another try), + /// `true` when the resend went out or we deliberately gave up at the cap. + async fn run_retry_receipt( + self: &Arc<Self>, + info: &Arc<MessageInfo>, + reason: RetryReason, + ) -> bool { + let cache_key = self + .make_retry_cache_key(&info.source.chat, &info.id, &info.source.sender) + .await; - // Atomically increment retry count and check if we should continue - let Some(retry_count) = client.increment_retry_count(&cache_key, reason).await else { - // Max retries reached - log::info!( - "Max retries ({}) reached for message {} from {} [{:?}]. Sending immediate PDO request.", - MAX_DECRYPT_RETRIES, - info.id, - info.source.sender, - reason - ); - // Send PDO request immediately (no delay) as last resort - client.spawn_pdo_request_with_options(&info, true); - return; - }; + 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.", + MAX_DECRYPT_RETRIES, + info.id, + info.source.sender, + reason + ); + // Capped: give up and clear the backlog regardless of PDO outcome. + self.run_pdo_request(info).await; + return true; + }; - // Log warning for high retry counts (like WhatsApp Web's MessageHighRetryCount) - if retry_count > HIGH_RETRY_COUNT_THRESHOLD { - log::warn!( - "High retry count ({}) for message {} from {} [{:?}]", + if retry_count > HIGH_RETRY_COUNT_THRESHOLD { + log::warn!( + "High retry count ({}) for message {} from {} [{:?}]", + retry_count, + info.id, + info.source.sender, + reason + ); + } + + let retry_sent = match self.send_retry_receipt(info, retry_count, reason).await { + Ok(()) => { + debug!( + "Sent retry receipt #{} for message {} from {} [{:?}]", + retry_count, info.id, info.source.sender, reason + ); + true + } + Err(e) => { + log::error!( + "Failed to send retry receipt #{} for message {} [{:?}]: {:?}", retry_count, info.id, - info.source.sender, - reason + reason, + e ); + false } + }; - // Send the retry receipt with the actual retry count and reason - match client.send_retry_receipt(&info, retry_count, reason).await { - Ok(()) => { - debug!( - "Sent retry receipt #{} for message {} from {} [{:?}]", - retry_count, info.id, info.source.sender, reason - ); - } - Err(e) => { - log::error!( - "Failed to send retry receipt #{} for message {} [{:?}]: {:?}", - retry_count, - info.id, - reason, - e - ); - } - } - - // Only spawn PDO on the FIRST retry to avoid duplicate requests. - // The PDO cache also provides deduplication, but this reduces unnecessary work. - if retry_count == 1 { - client.spawn_pdo_request(&info); - } - })).detach(); + // First retry only, to avoid duplicate PDO requests. Awaited so it runs + // before the caller's ack; the retry receipt already landed first. + if retry_count == 1 { + self.run_pdo_request(info).await; + } + retry_sent } pub(crate) async fn handle_incoming_message(self: Arc<Self>, node: Arc<OwnedNodeRef>) { @@ -414,8 +475,10 @@ impl Client { info.id, unavailable_type ); - // Ack is handled by the framework; PDO asks the primary phone to relay the message - self.spawn_pdo_request_with_options(&info, true); + // PDO is the only recovery here (no retry receipt), so run it before + // the transport ack in one flush task: the ack must not clear the + // offline queue before the PDO request goes out. status is acked by + // the should_ack gate. Mirrors whatsmeow's request-then-ack. self.dispatch_undecryptable_event( Arc::clone(&info), true, @@ -423,6 +486,17 @@ impl Client { crate::types::events::DecryptFailMode::Show, ) .await; + let client = Arc::clone(self); + let info2 = Arc::clone(&info); + let skip_ack = info.source.chat.is_status_broadcast(); + self.outbound_flush.spawn(&*self.runtime, async move { + // Only ack once the PDO request is out (or skipped as ancient); + // a transient send failure leaves it queued for redelivery. + let pdo_sent = client.run_pdo_request(&info2).await; + if !skip_ack && pdo_sent { + client.send_transport_ack(&info2).await; + } + }); return None; } @@ -679,9 +753,11 @@ impl Client { info.source.sender ); } else { - warn!( + log::log!( + decrypt_fail_log_level(decrypt_fail_mode), "Skipping skmsg decryption for message {} from {} because pkmsg failed to decrypt.", - info.id, info.source.sender + info.id, + info.source.sender ); if !session_dispatched_undecryptable { self.dispatch_undecryptable_event( @@ -708,9 +784,11 @@ impl Client { && !session_payloads.is_empty() { // Edge case: message with only msg/pkmsg that failed to decrypt, no skmsg - warn!( + log::log!( + decrypt_fail_log_level(decrypt_fail_mode), "Message {} from {} failed to decrypt and has no group content. Dispatching UndecryptableMessage event.", - info.id, info.source.sender + info.id, + info.source.sender ); // Dispatch UndecryptableMessage event for messages that failed to decrypt // (This should not cause double-dispatching since process_session_enc_batch @@ -723,6 +801,17 @@ impl Client { ) .await; // Do NOT send delivery receipt - transport ack is sufficient + } else if session_had_duplicates + && !session_decrypted_successfully + && !session_dispatched_undecryptable + && !info.source.chat.is_status_broadcast() + { + // Duplicate (already-processed) with no group content: ack it so the + // server drops it from the offline queue (whatsmeow/WA Web treat + // old-counter like success). status is acked by the should_ack gate + // (a status SKDM pkmsg can reach here), so skip it to avoid a + // redundant receipt. + self.ack_received_message(&info); } // Flush cached Signal state to DB (matches WA Web's flushBufferToDiskIfNotMemOnlyMode) @@ -1111,7 +1200,8 @@ impl Client { } else { (RetryReason::InvalidMessage, "InvalidMessage") }; - log::warn!( + log::log!( + decrypt_fail_log_level(decrypt_fail_mode), "[msg:{}] Decryption failed for {} message from {} due to {label}. \ Sending retry receipt.", info.id, @@ -1260,7 +1350,12 @@ impl Client { iteration, counter ); - // This is expected when messages are redelivered, just continue silently + // Redelivered duplicate: ack it so the server drops it from the + // offline queue. status is already acked by the should_ack gate, + // so skip it to avoid a redundant receipt. + if !info.source.chat.is_status_broadcast() { + self.ack_received_message(info); + } } Err(SignalProtocolError::NoSenderKeyState(msg)) => { if info.is_expired_status() { @@ -1302,13 +1397,27 @@ impl Client { continue; } - log::error!( + log::log!( + decrypt_fail_log_level(decrypt_fail_mode), "Group batch decrypt failed [msg:{}] for group {} sender {}: {:?}", info.id, sender_key_name.group_id(), sender_key_name.sender_id(), e ); + // Always surface the failure to consumers; nack only non-status + // (status is acked by the should_ack gate) so the server drops + // it from the offline queue. + self.dispatch_undecryptable_event( + Arc::clone(info), + false, + crate::types::events::UnavailableType::Unknown, + decrypt_fail_mode, + ) + .await; + if !info.source.chat.is_status_broadcast() { + self.spawn_nack(info, NackReason::UnhandledError, None); + } } } } @@ -6397,7 +6506,7 @@ mod tests { assert_eq!(info.source.chat.server, wacore_binary::Server::Broadcast); - client.spawn_pdo_request_with_options(&info, true); + client.run_pdo_request(&info).await; tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; } @@ -6414,7 +6523,7 @@ mod tests { assert_eq!(info.source.chat.server, wacore_binary::Server::Broadcast); - client.spawn_pdo_request_with_options(&info, true); + client.run_pdo_request(&info).await; tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; } @@ -6430,7 +6539,7 @@ mod tests { assert_ne!(info.source.chat.server, wacore_binary::Server::Broadcast); - client.spawn_pdo_request_with_options(&info, true); + client.run_pdo_request(&info).await; tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; } @@ -6446,7 +6555,7 @@ mod tests { info.source.is_from_me = true; let info = Arc::new(info); - client.spawn_pdo_request_with_options(&info, true); + client.run_pdo_request(&info).await; tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; } @@ -6465,7 +6574,7 @@ mod tests { let cache_key = ChatMessageId::new(info.source.chat.clone(), info.id.clone()); - client.spawn_pdo_request_with_options(&info, true); + client.run_pdo_request(&info).await; tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; assert!( @@ -6491,7 +6600,7 @@ mod tests { let cache_key = ChatMessageId::new(info.source.chat.clone(), info.id.clone()); - client.spawn_pdo_request_with_options(&info, true); + client.run_pdo_request(&info).await; tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; assert!( @@ -6730,6 +6839,423 @@ mod tests { ); } + #[test] + fn test_decrypt_fail_log_level_gated_on_hide() { + use crate::types::events::DecryptFailMode; + assert_eq!( + decrypt_fail_log_level(DecryptFailMode::Hide), + log::Level::Debug + ); + assert_eq!( + decrypt_fail_log_level(DecryptFailMode::Show), + log::Level::Warn + ); + } + + /// Decrypt one captured noise frame (zero-key, counter-based, empty AAD) to + /// its marshalled node bytes; strips the 3-byte frame header. + fn decode_frame(index: usize, frame: &[u8]) -> Option<Vec<u8>> { + use wacore::handshake::NoiseCipher; + if frame.len() <= 3 { + return None; + } + let cipher = NoiseCipher::new(&[0u8; 32]).expect("32-byte key"); + let mut buf = frame[3..].to_vec(); + cipher + .decrypt_in_place_with_counter(index as u32, &mut buf) + .ok()?; + (!buf.is_empty()).then_some(buf) + } + + /// First `<ack class="message">` on the wire as `(to, recipient)`. + fn find_message_ack(frames: &[bytes::Bytes]) -> Option<(String, Option<String>)> { + for (i, frame) in frames.iter().enumerate() { + let Some(buf) = decode_frame(i, frame) else { + continue; + }; + let Ok(node) = wacore_binary::marshal::unmarshal_ref(&buf[1..]) else { + continue; + }; + if node.tag.as_ref() == "ack" + && node + .get_attr("class") + .is_some_and(|v| v.as_str() == "message") + && let Some(to) = node.get_attr("to") + { + let recipient = node.get_attr("recipient").map(|v| v.as_str().to_string()); + return Some((to.as_str().to_string(), recipient)); + } + } + None + } + + /// Count delivery `<receipt>` (anything but type="retry") on the wire for `id`. + fn delivery_receipts_for(frames: &[bytes::Bytes], id: &str) -> usize { + let mut count = 0; + for (i, frame) in frames.iter().enumerate() { + let Some(buf) = decode_frame(i, frame) else { + continue; + }; + let Ok(node) = wacore_binary::marshal::unmarshal_ref(&buf[1..]) else { + continue; + }; + if node.tag.as_ref() == "receipt" + && node.get_attr("id").is_some_and(|v| v.as_str() == id) + && node + .get_attr("type") + .as_ref() + .map(|v| v.as_str()) + .as_deref() + != Some("retry") + { + count += 1; + } + } + count + } + + /// A stanza that fails to decrypt must emit a transport `<ack class="message">` + /// (else the server replays it on every reconnect forever), addressed to the + /// original `from` echoing `recipient`. Uses `Hide` (the production reactions + /// carried `decrypt-fail="hide"`) to also guard that hide does not suppress + /// the ack. BadMac so the retry carries no keys (no device account needed). + #[tokio::test] + async fn decrypt_failure_emits_transport_ack() { + let (client, transport) = capturing_client("decrypt_fail_ack").await; + + let sender: Jid = "236395184570386@lid".parse().expect("sender JID"); + let recipient: Jid = "156535032389744@lid".parse().expect("recipient JID"); + let info = Arc::new(MessageInfo { + id: "AC055553E56A2C12DE592DAD6353C477".to_string(), + source: crate::types::message::MessageSource { + sender: sender.clone(), + chat: recipient.clone(), + recipient: Some(recipient.clone()), + ..Default::default() + }, + ..Default::default() + }); + + client + .handle_decrypt_failure( + &info, + RetryReason::BadMac, + crate::types::events::DecryptFailMode::Hide, + ) + .await; + + // retry + ack are detached spawns; poll the wire until the ack appears. + let mut found = None; + for _ in 0..80 { + if let Some(a) = find_message_ack(&transport.sent()) { + found = Some(a); + break; + } + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + } + let (to, recipient_attr) = found.expect( + "decrypt failure must emit a transport <ack class=message> \ + (else the server redelivers the stanza forever)", + ); + assert_eq!( + to, "236395184570386@lid", + "ack `to` must be the original `from` (own LID), not the chat" + ); + assert_eq!( + recipient_attr.as_deref(), + Some("156535032389744@lid"), + "ack must echo `recipient` for own-account fan-out" + ); + } + + /// If the resend request fails to send, the stanza must NOT be acked, so the + /// server keeps it queued for another try. Here NoSession needs keys, which + /// need a device account this harness lacks, so send_retry_receipt errors. + #[tokio::test] + async fn decrypt_failure_does_not_ack_when_retry_send_fails() { + let (client, transport) = capturing_client("retry_fail_no_ack").await; + let sender: Jid = "5511777776666@s.whatsapp.net".parse().expect("sender"); + let info = Arc::new(MessageInfo { + id: "NOACK1".to_string(), + source: crate::types::message::MessageSource { + sender: sender.clone(), + chat: sender.clone(), + ..Default::default() + }, + ..Default::default() + }); + client + .handle_decrypt_failure( + &info, + RetryReason::NoSession, + crate::types::events::DecryptFailMode::Show, + ) + .await; + tokio::time::sleep(std::time::Duration::from_millis(150)).await; + assert!( + find_message_ack(&transport.sent()).is_none(), + "must not ack when the resend request failed to send" + ); + } + + /// The retry receipt must be sent before the transport ack (one ordered + /// flushed task), so a disconnect mid-flush can never clear the stanza from + /// the offline queue without the sender having received a resend request. + #[tokio::test] + async fn decrypt_failure_sends_retry_before_ack() { + let (client, transport) = capturing_client("retry_before_ack").await; + let sender: Jid = "5511777776666@s.whatsapp.net".parse().expect("sender"); + let info = Arc::new(MessageInfo { + id: "RBA1".to_string(), + source: crate::types::message::MessageSource { + sender: sender.clone(), + chat: sender.clone(), + ..Default::default() + }, + ..Default::default() + }); + // BadMac (not NoSession) so the retry receipt carries no keys and needs + // no device account in this harness. + client + .handle_decrypt_failure( + &info, + RetryReason::BadMac, + crate::types::events::DecryptFailMode::Show, + ) + .await; + + let find = |tag: &str, retry: bool| -> Option<usize> { + let frames = transport.sent(); + for (i, frame) in frames.iter().enumerate() { + let Some(buf) = decode_frame(i, frame) else { + continue; + }; + let Ok(node) = wacore_binary::marshal::unmarshal_ref(&buf[1..]) else { + continue; + }; + let is_retry = node.get_attr("type").is_some_and(|v| v.as_str() == "retry"); + if node.tag.as_ref() == tag && is_retry == retry { + return Some(i); + } + } + None + }; + + let mut retry_idx = None; + let mut ack_idx = None; + for _ in 0..80 { + retry_idx = find("receipt", true); + ack_idx = find("ack", false); + if retry_idx.is_some() && ack_idx.is_some() { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + } + let retry_idx = retry_idx.expect("retry receipt must be sent"); + let ack_idx = ack_idx.expect("transport ack must be sent"); + assert!( + retry_idx < ack_idx, + "retry receipt (frame {retry_idx}) must be sent before the ack (frame {ack_idx})" + ); + } + + /// status@broadcast is already acked by the `should_ack` gate post-dispatch, + /// so the decrypt-failure path must NOT emit a second transport ack + /// (whatsmeow/WA Web send exactly one per message). The retry receipt still + /// goes out. + #[tokio::test] + async fn status_broadcast_decrypt_failure_acks_to_chat() { + let (client, transport) = capturing_client("status_fail_ack").await; + let info = Arc::new(MessageInfo { + id: "STATUSMSGID".to_string(), + source: crate::types::message::MessageSource { + sender: "236395184570386@lid".parse().expect("sender"), + chat: "status@broadcast".parse().expect("status chat"), + is_group: true, + ..Default::default() + }, + ..Default::default() + }); + + client + .handle_decrypt_failure( + &info, + RetryReason::BadMac, + crate::types::events::DecryptFailMode::Show, + ) + .await; + + // status failures are acked from the flushed task (not just the detached + // should_ack gate), so the ack survives a disconnect mid-flush. + let mut found = None; + for _ in 0..80 { + if let Some(a) = find_message_ack(&transport.sent()) { + found = Some(a); + break; + } + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + } + let (to, _) = found.expect("status failure must emit a flushed transport ack"); + assert_eq!( + to, "status@broadcast", + "status ack `to` must be the status chat" + ); + } + + /// Run a single session ciphertext through the full classify->process path. + async fn process_session_ct( + client: &Arc<Client>, + sender: &Jid, + id: &str, + ct: &wacore::libsignal::protocol::CiphertextMessage, + ) { + use wacore::libsignal::protocol::CiphertextMessage; + let (enc_type, bytes) = match ct { + CiphertextMessage::PreKeySignalMessage(m) => ("pkmsg", m.serialized().to_vec()), + CiphertextMessage::SignalMessage(m) => ("msg", m.serialized().to_vec()), + _ => panic!("unexpected ciphertext type"), + }; + let enc = NodeBuilder::new("enc") + .attr("type", enc_type) + .bytes(bytes) + .build(); + let enc_ref = enc.as_node_ref(); + let payload = EncPayload::from_node_ref(&enc_ref).unwrap(); + let info = Arc::new(MessageInfo { + id: id.to_string(), + source: crate::types::message::MessageSource { + sender: sender.clone(), + chat: sender.clone(), + ..Default::default() + }, + ..Default::default() + }); + client + .clone() + .process_classified_message(ClassifiedMessage { + info, + sender_encryption_jid: sender.clone(), + session_payloads: vec![payload], + group_payloads: vec![], + max_sender_retry_count: 0, + decrypt_fail_mode: crate::types::events::DecryptFailMode::Show, + }) + .await; + } + + /// Regression for the offline-backlog disconnect: an already-processed + /// (duplicate) message must get its own delivery receipt, else the server + /// replays it every reconnect until it force-closes the stream. Pre-fix only + /// the first (success) delivery was acked; the duplicate was skipped silently. + #[tokio::test] + async fn duplicate_message_is_acked_with_delivery_receipt() { + let (client, transport) = capturing_client("dup_receipt").await; + let (bundle, bob_jid) = bobs_prekey_bundle(&client).await; + let bob_addr = bob_jid.to_protocol_address(); + let mut alice = AlicePeer::new("5511888887777@s.whatsapp.net").await; + alice.install_bob_session(&bob_addr, &bundle).await; + + // Establish the session, then mark Alice's prekey acked so her next message + // is a plain SignalMessage. Re-submitting it is a clean duplicate. + let establish = alice.encrypt(&bob_addr, b"establish").await; + process_session_ct(&client, &alice.jid, "EST", &establish).await; + if let Some(record) = alice.sessions.0.get_mut(&bob_addr) + && let Some(state) = record.session_state_mut() + { + state.clear_unacknowledged_pre_key_message(); + } + + // A real (padded) Message so the success path also emits its receipt. + let plaintext = wacore::messages::MessageUtils::encode_and_pad(&wa::Message { + conversation: Some("hi".to_string()), + ..Default::default() + }); + let msg = alice.encrypt(&bob_addr, &plaintext).await; + process_session_ct(&client, &alice.jid, "DUP", &msg).await; // success + process_session_ct(&client, &alice.jid, "DUP", &msg).await; // duplicate + + let mut count = 0; + for _ in 0..80 { + count = delivery_receipts_for(&transport.sent(), "DUP"); + if count >= 2 { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + } + assert_eq!( + count, 2, + "duplicate must get its own delivery receipt (pre-fix: only the first send was acked)" + ); + } + + /// Own-account fan-out (is_from_me, non-peer) has its delivery receipt + /// suppressed by `should_send_delivery_receipt`, so it must instead get a + /// transport ack (to = own LID) or the server replays it forever. + #[tokio::test] + async fn own_account_message_acked_via_transport_ack() { + let (client, transport) = capturing_client("own_ack").await; + let own = Arc::new(MessageInfo { + id: "OWN1".to_string(), + source: crate::types::message::MessageSource { + sender: "236395184570386@lid".parse().expect("sender"), + chat: "156535032389744@lid".parse().expect("chat"), + recipient: Some("156535032389744@lid".parse().expect("recipient")), + is_from_me: true, + ..Default::default() + }, + ..Default::default() + }); + client.ack_received_message(&own); + + let mut found = None; + for _ in 0..80 { + if let Some(a) = find_message_ack(&transport.sent()) { + found = Some(a); + break; + } + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + } + let (to, _) = found.expect("own-account message must get a transport ack"); + assert_eq!( + to, "236395184570386@lid", + "ack must be addressed to the own LID" + ); + assert_eq!( + delivery_receipts_for(&transport.sent(), "OWN1"), + 0, + "own non-peer must NOT get a delivery receipt (it's suppressed)" + ); + } + + /// An `<unavailable>` message (no `<enc>`) must be transport-acked so the + /// server stops replaying it (DM/group aren't covered by the should_ack gate). + #[tokio::test] + async fn unavailable_message_is_transport_acked() { + let (client, transport) = capturing_client("unavail_ack").await; + let node = NodeBuilder::new("message") + .attr("from", "5511777776666@s.whatsapp.net") + .attr("id", "UNAVAIL1") + .attr("type", "text") + .children([NodeBuilder::new("unavailable") + .attr("type", "view_once") + .build()]) + .build(); + let owned = node_to_arc(node); + let classified = client.classify_incoming_message(&owned).await; + assert!(classified.is_none(), "unavailable path returns None"); + + let mut found = None; + for _ in 0..80 { + if let Some(a) = find_message_ack(&transport.sent()) { + found = Some(a); + break; + } + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + } + let (to, _) = found.expect("unavailable message must get a transport ack"); + assert_eq!(to, "5511777776666@s.whatsapp.net"); + } + /// Security regression: a self-only `app_state_sync_key_share` protocol /// message must be honoured only when it originates from our own account. /// A spoofed one from a peer must be dropped (otherwise a peer could inject diff --git a/src/pdo.rs b/src/pdo.rs index 5ff4df191..010db5630 100644 --- a/src/pdo.rs +++ b/src/pdo.rs @@ -19,7 +19,6 @@ use crate::types::message::MessageInfo; use log::{debug, info, warn}; use prost::Message; use std::sync::Arc; -use std::time::Duration; use wacore::types::message::{ ChatMessageId, EditAttribute, MessageCategory, MessageSource, MsgMetaInfo, }; @@ -452,83 +451,37 @@ impl Client { }) } - /// Spawns a PDO request for a message that failed to decrypt. - /// This is called alongside the retry receipt to increase chances of recovery. + /// Age-gated PDO send, awaitable so it can run before a transport ack inside + /// one flush task (when PDO is the sole recovery, e.g. `<unavailable>`). + /// `fromMe` is NOT excluded: own-device fan-out that fails to decrypt has PDO + /// as its only recovery (WAWebNonMessageDataRequestPlaceholderMessageResendUtils). /// - /// When `immediate` is true, the PDO request is sent without delay. - /// This is used when we've exhausted retry attempts and need immediate PDO recovery. - pub(crate) fn spawn_pdo_request_with_options( - self: &Arc<Self>, - info: &Arc<MessageInfo>, - immediate: bool, - ) { - // `fromMe` is NOT excluded here: when the user's other devices send a - // message and the fanout copy to this client fails to decrypt, PDO is - // the only recovery path. Matches WAWebNonMessageDataRequestPlaceholderMessageResendUtils. - - // Avoid asking the phone to re-deliver ancient messages during offline - // sync or long reconnect tails. Matches the - // `placeholder_message_resend_maximum_days_limit` AB prop (default 14d) - // enforced by WAWebNonMessageDataRequestPlaceholderMessageResendUtils. - // Compare in seconds to stay bit-for-bit with WA Web's `age_s > i` - // check — `num_days()` truncates and would let 14d1h through. - const PDO_MAX_AGE: chrono::Duration = chrono::Duration::days(14); - let age = wacore::time::now_utc().signed_duration_since(info.timestamp); - if age > PDO_MAX_AGE { + /// Returns `false` only on a transient send failure: the caller must then + /// NOT ack, so the stanza stays in the offline queue for another attempt. + /// Age-skip counts as a deliberate give-up (`true`), so ancient stanzas are + /// still cleared. + pub(crate) async fn run_pdo_request(self: &Arc<Self>, info: &Arc<MessageInfo>) -> bool { + // Skip ancient messages (14d, matching the AB prop), compared in seconds + // like WA Web's `age_s > i`. Uses the wacore time primitive (mockable). + const PDO_MAX_AGE_SECS: i64 = 14 * 24 * 60 * 60; + let age_secs = wacore::time::now_secs() - info.timestamp.timestamp(); + if age_secs > PDO_MAX_AGE_SECS { debug!( - "PDO request skipped for message {} (age {}s exceeds {}s limit)", + "PDO request skipped for message {} (age {age_secs}s exceeds {PDO_MAX_AGE_SECS}s limit)", info.id, - age.num_seconds(), - PDO_MAX_AGE.num_seconds(), ); - return; + return true; + } + match self.send_pdo_placeholder_resend_request(info).await { + Ok(()) => true, + Err(e) => { + warn!( + "Failed to send PDO request for message {} from {}: {:?}", + info.id, info.source.sender, e + ); + false + } } - - let client_clone = Arc::clone(self); - let info_clone = Arc::clone(info); - // Per-connection: on disconnect/reconnect the signal fires and we bail - // before inserting into `pdo_pending_requests`, preventing a 30s TTL - // strand on an entry that can no longer receive its response. - let shutdown = self.connection_shutdown_signal(); - - self.runtime - .spawn(Box::pin(async move { - use futures::FutureExt; - - if !immediate { - // Delay lets the retry receipt land before we pile PDO on top. - futures::select! { - _ = client_clone - .runtime - .sleep(Duration::from_millis(500)) - .fuse() => {} - _ = wacore::runtime::wait_for_shutdown(&shutdown).fuse() => { - return; - } - } - } - - if shutdown.is_fired() { - return; - } - - if let Err(e) = client_clone - .send_pdo_placeholder_resend_request(&info_clone) - .await - { - warn!( - "Failed to send PDO request for message {} from {}: {:?}", - info_clone.id, info_clone.source.sender, e - ); - } - })) - .detach(); - } - - /// Spawns a PDO request for a message that failed to decrypt. - /// This is called alongside the retry receipt to increase chances of recovery. - pub(crate) fn spawn_pdo_request(self: &Arc<Self>, info: &Arc<MessageInfo>) { - self.spawn_pdo_request_with_options(info, false); } } diff --git a/src/receipt.rs b/src/receipt.rs index 22c4a1f5e..81d04364b 100644 --- a/src/receipt.rs +++ b/src/receipt.rs @@ -17,16 +17,23 @@ use wacore_binary::OwnedNodeRef; /// `(t.isGroup() || t.isBroadcast()) && r ? DEVICE_JID(r) : DROP_ATTR`, so /// status broadcasts (isBroadcast = true) also carry the original poster's /// JID. Without it the server can't map the ack back to the status owner. -fn build_delivery_receipt_node(info: &crate::types::message::MessageInfo) -> wacore_binary::Node { +/// `active=false` sends `type="inactive"` (not rendered as ticks), matching +/// whatsmeow's background companion. Peer/status keep their own type/context. +fn build_delivery_receipt_node( + info: &crate::types::message::MessageInfo, + active: bool, +) -> wacore_binary::Node { let mut builder = NodeBuilder::new("receipt") .attr("id", &info.id) .attr("to", &info.source.chat); + let is_status = info.source.chat.is_status_broadcast(); if info.category == MessageCategory::Peer { builder = builder.attr("type", "peer_msg"); + } else if !active && !is_status { + builder = builder.attr("type", "inactive"); } - let is_status = info.source.chat.is_status_broadcast(); if info.source.is_group || is_status { builder = builder.attr("participant", &info.source.sender); } @@ -76,7 +83,7 @@ fn build_nack_node( } impl Client { - fn should_send_delivery_receipt(info: &crate::types::message::MessageInfo) -> bool { + pub(crate) fn should_send_delivery_receipt(info: &crate::types::message::MessageInfo) -> bool { if info.id.is_empty() || info.source.chat.is_newsletter() { return false; } @@ -257,7 +264,7 @@ impl Client { return; } - let receipt_node = build_delivery_receipt_node(info); + let receipt_node = build_delivery_receipt_node(info, self.receipts_are_active()); debug!(target: "Client/Receipt", "Sending {} receipt for message {} to {}", if info.category == MessageCategory::Peer { "peer_msg" } else { "delivery" }, @@ -397,7 +404,7 @@ mod tests { // `Send/DeliveryReceiptJob.js`. Status broadcasts must carry BOTH so // the server can map the ack back to the status owner. let info = info_with("status@broadcast", "12345@s.whatsapp.net", false); - let node = build_delivery_receipt_node(&info); + let node = build_delivery_receipt_node(&info, true); assert_eq!(node.tag, "receipt"); assert_eq!( node.attrs.get("context").map(|v| v.as_str()).as_deref(), @@ -412,12 +419,45 @@ mod tests { #[test] fn delivery_receipt_for_dm_has_no_context_no_participant() { let info = info_with("12345@s.whatsapp.net", "12345@s.whatsapp.net", false); - let node = build_delivery_receipt_node(&info); + let node = build_delivery_receipt_node(&info, true); assert!(node.attrs.get("context").is_none()); assert!(node.attrs.get("participant").is_none()); assert!(node.attrs.get("type").is_none()); } + #[test] + fn delivery_receipt_is_inactive_when_not_active() { + let info = info_with("12345@s.whatsapp.net", "12345@s.whatsapp.net", false); + let inactive = build_delivery_receipt_node(&info, false); + assert_eq!( + inactive.attrs.get("type").map(|v| v.as_str()).as_deref(), + Some("inactive"), + "a passive companion sends inactive delivery receipts" + ); + let active = build_delivery_receipt_node(&info, true); + assert!(active.attrs.get("type").is_none()); + } + + #[test] + fn status_and_peer_receipts_ignore_inactive() { + let status = info_with("status@broadcast", "12345@s.whatsapp.net", false); + let node = build_delivery_receipt_node(&status, false); + // status keeps context, never type=inactive + assert!(node.attrs.get("type").is_none()); + assert_eq!( + node.attrs.get("context").map(|v| v.as_str()).as_deref(), + Some("status") + ); + + let mut peer = info_with("12345@s.whatsapp.net", "12345@s.whatsapp.net", false); + peer.category = MessageCategory::Peer; + let node = build_delivery_receipt_node(&peer, false); + assert_eq!( + node.attrs.get("type").map(|v| v.as_str()).as_deref(), + Some("peer_msg") + ); + } + #[test] fn delivery_receipt_for_group_carries_participant() { let info = info_with( @@ -425,7 +465,7 @@ mod tests { "15551234567@s.whatsapp.net", true, ); - let node = build_delivery_receipt_node(&info); + let node = build_delivery_receipt_node(&info, true); assert_eq!( node.attrs.get("participant").map(|v| v.as_str()).as_deref(), Some("15551234567@s.whatsapp.net") @@ -445,7 +485,7 @@ mod tests { // participant, no context. Matches WA Web's DROP_ATTR gating. let mut info = info_with("12345@s.whatsapp.net", "12345@s.whatsapp.net", false); info.category = MessageCategory::Peer; - let node = build_delivery_receipt_node(&info); + let node = build_delivery_receipt_node(&info, true); assert_eq!( node.attrs.get("type").map(|v| v.as_str()).as_deref(), Some("peer_msg") @@ -461,7 +501,7 @@ mod tests { // status owner from it regardless of the peer_msg type. let mut info = info_with("status@broadcast", "12345@s.whatsapp.net", false); info.category = MessageCategory::Peer; - let node = build_delivery_receipt_node(&info); + let node = build_delivery_receipt_node(&info, true); assert_eq!( node.attrs.get("participant").map(|v| v.as_str()).as_deref(), Some("12345@s.whatsapp.net")