fix(offline): transport-ack stanzas with only unrecognized enc types

src/message.rs

@@ -504,6 +504,8 @@ impl Client {
         let mut group_payloads = Vec::with_capacity(all_enc_nodes.len());
         let mut max_sender_retry_count: u8 = 0;
         let mut has_hide_fail = false;
+        let mut had_unknown_enc = false;
+        let mut had_custom_handler = false;
 
         for enc_node in &all_enc_nodes {
             // Parse sender retry count (WA Web: e.maybeAttrInt("count") ?? 0)
@@ -528,6 +530,7 @@ impl Client {
                 Some(t) => t,
                 None => {
                     log::warn!("Enc node missing 'type' attribute, skipping");
+                    had_unknown_enc = true;
                     continue;
                 }
             };
@@ -559,6 +562,7 @@ impl Client {
                         }
                     }))
                     .detach();
+                had_custom_handler = true;
                 continue;
             }
 
@@ -569,6 +573,7 @@ impl Client {
                 Some(p) => p,
                 None => {
                     log::warn!("Enc node has no content or unknown type: {enc_type}");
+                    had_unknown_enc = true;
                     continue;
                 }
             };
@@ -597,6 +602,25 @@ impl Client {
             );
         }
 
+        // Stanzas whose enc nodes are all unrecognized (e.g. msmsg from the Meta
+        // AI bot) would otherwise leave the offline queue without any ack,
+        // causing the server to replay them until <stream:error>. Custom
+        // handlers ack on their own; status is acked by the should_ack gate.
+        if session_payloads.is_empty()
+            && group_payloads.is_empty()
+            && had_unknown_enc
+            && !had_custom_handler
+        {
+            log::info!(
+                "[msg:{}] All enc payloads unrecognized; transport-acking to drop from offline queue",
+                info.id
+            );
+            if !info.source.chat.is_status_broadcast() {
+                self.spawn_message_ack(&info);
+            }
+            return None;
+        }
+
         Some(ClassifiedMessage {
             info,
             sender_encryption_jid,
@@ -7256,6 +7280,166 @@ mod tests {
         assert_eq!(to, "5511777776666@s.whatsapp.net");
     }
 
+    /// Stanzas with only unrecognized enc types (e.g. msmsg from the Meta AI
+    /// bot) used to silently fall through with no ack, so the server replayed
+    /// them on every reconnect until <stream:error>. Classify must now emit a
+    /// transport ack to drop the stanza from the offline queue.
+    #[tokio::test]
+    async fn unknown_only_enc_is_transport_acked() {
+        let (client, transport) = capturing_client("msmsg_ack").await;
+        let node = NodeBuilder::new("message")
+            .attr("from", "5511777776666@s.whatsapp.net")
+            .attr("id", "MSMSG1")
+            .attr("type", "text")
+            .children([NodeBuilder::new("enc")
+                .attr("type", "msmsg")
+                .bytes(vec![0u8; 8])
+                .build()])
+            .build();
+        let owned = node_to_arc(node);
+        let classified = client.classify_incoming_message(&owned).await;
+        assert!(
+            classified.is_none(),
+            "unknown-only enc must short-circuit before the process phase"
+        );
+
+        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("unknown-only enc must emit a transport ack");
+        assert_eq!(to, "5511777776666@s.whatsapp.net");
+    }
+
+    /// status@broadcast is already acked by the `should_ack` gate; the fallback
+    /// path must skip it to avoid a duplicate transport ack.
+    #[tokio::test]
+    async fn unknown_only_enc_on_status_skips_fallback_ack() {
+        let (client, transport) = capturing_client("msmsg_status_skip").await;
+        let node = NodeBuilder::new("message")
+            .attr("from", "status@broadcast")
+            .attr("id", "MSMSG_STATUS")
+            .attr("type", "text")
+            .attr("participant", "5511777776666@s.whatsapp.net")
+            .children([NodeBuilder::new("enc")
+                .attr("type", "msmsg")
+                .bytes(vec![0u8; 8])
+                .build()])
+            .build();
+        let owned = node_to_arc(node);
+        let classified = client.classify_incoming_message(&owned).await;
+        assert!(classified.is_none());
+
+        // Give any rogue spawned task time to land on the wire.
+        for _ in 0..16 {
+            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
+            if find_message_ack(&transport.sent()).is_some() {
+                break;
+            }
+        }
+        assert!(
+            find_message_ack(&transport.sent()).is_none(),
+            "status@broadcast must not get a fallback transport ack from classify"
+        );
+    }
+
+    /// A recognized enc alongside an unknown one must still flow through the
+    /// normal process path; the fallback must not short-circuit and prevent
+    /// decryption of the recognized payload.
+    #[tokio::test]
+    async fn mixed_recognized_and_unknown_enc_still_classifies() {
+        let (client, _transport) = capturing_client("msmsg_mixed").await;
+        let node = NodeBuilder::new("message")
+            .attr("from", "5511777776666@s.whatsapp.net")
+            .attr("id", "MIXED1")
+            .attr("type", "text")
+            .children([
+                NodeBuilder::new("enc")
+                    .attr("type", "pkmsg")
+                    .bytes(vec![0u8; 8])
+                    .build(),
+                NodeBuilder::new("enc")
+                    .attr("type", "msmsg")
+                    .bytes(vec![0u8; 8])
+                    .build(),
+            ])
+            .build();
+        let owned = node_to_arc(node);
+        let classified = client
+            .classify_incoming_message(&owned)
+            .await
+            .expect("mixed enc must produce a ClassifiedMessage");
+        assert_eq!(classified.session_payloads.len(), 1);
+        assert!(classified.group_payloads.is_empty());
+    }
+
+    /// When a custom handler claims the unknown enc type, the fallback must NOT
+    /// fire (the handler is responsible for its own ack; firing both could send
+    /// two acks for the same stanza).
+    #[tokio::test]
+    async fn custom_handler_only_skips_fallback_ack() {
+        use crate::types::enc_handler::EncHandler;
+        use async_lock::Mutex as AsyncMutex;
+
+        #[derive(Default)]
+        struct NoopHandler {
+            calls: Arc<AsyncMutex<usize>>,
+        }
+        #[async_trait::async_trait]
+        impl EncHandler for NoopHandler {
+            async fn handle(
+                &self,
+                _client: Arc<Client>,
+                _enc_node: &wacore_binary::Node,
+                _info: &crate::types::message::MessageInfo,
+            ) -> anyhow::Result<()> {
+                *self.calls.lock().await += 1;
+                Ok(())
+            }
+        }
+
+        let (client, transport) = capturing_client("msmsg_custom").await;
+        let calls = Arc::new(AsyncMutex::new(0usize));
+        let handler = Arc::new(NoopHandler {
+            calls: Arc::clone(&calls),
+        });
+        client
+            .custom_enc_handlers
+            .write()
+            .await
+            .insert("msmsg".to_string(), handler as Arc<dyn EncHandler>);
+
+        let node = NodeBuilder::new("message")
+            .attr("from", "5511777776666@s.whatsapp.net")
+            .attr("id", "CUSTOM1")
+            .attr("type", "text")
+            .children([NodeBuilder::new("enc")
+                .attr("type", "msmsg")
+                .bytes(vec![0u8; 8])
+                .build()])
+            .build();
+        let owned = node_to_arc(node);
+        let classified = client.classify_incoming_message(&owned).await;
+        assert!(
+            classified.is_some(),
+            "custom-handled enc must not be short-circuited by the fallback guard"
+        );
+
+        // Let the detached handler + any rogue spawned task run.
+        for _ in 0..16 {
+            tokio::time::sleep(std::time::Duration::from_millis(25)).await;
+        }
+        assert!(
+            find_message_ack(&transport.sent()).is_none(),
+            "custom-handled enc must not get a fallback transport ack from classify"
+        );
+        assert_eq!(*calls.lock().await, 1, "custom handler must be invoked");
+    }
+
     /// 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