fix(recv): dispatch a resent message once

src/cache_config.rs

@@ -179,6 +179,13 @@ pub struct CacheConfig {
     /// the same id does not surface a second notification. Default: 5m TTL,
     /// 1000 entries.
     pub undecryptable_dispatched: CacheEntryConfig,
+    /// Dispatch-once gate for a decrypted message: a sender whose outbox
+    /// retries resends the same id re-encrypted, which the ratchet cannot see
+    /// as a duplicate. Default: 5m TTL, 1000 entries. The TTL covers the
+    /// observed resend window (production logs: median 12s between attempts,
+    /// p90 189s, longest plausible resend 285s) and the capacity ~3.6x the
+    /// busiest 5-minute burst measured (278 messages). Capacity 0 disables it.
+    pub dispatched_messages: 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
@@ -280,6 +287,7 @@ impl std::fmt::Debug for CacheConfig {
             .field("recent_messages", &self.recent_messages)
             .field("message_retry_counts", &self.message_retry_counts)
             .field("undecryptable_dispatched", &self.undecryptable_dispatched)
+            .field("dispatched_messages", &self.dispatched_messages)
             .field("pdo_pending_requests", &self.pdo_pending_requests)
             .field("pdo_requested", &self.pdo_requested)
             .field("sender_key_devices_cache", &self.sender_key_devices_cache)
@@ -341,6 +349,7 @@ impl Default for CacheConfig {
             // 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),
+            dispatched_messages: 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),

src/client.rs

@@ -432,6 +432,8 @@ pub struct MemoryReport {
     pub dm_devices_memo: CollectionStats,
     pub message_retry_counts: u64,
     pub undecryptable_dispatched: u64,
+    /// Entries in the dispatch-once gate for decrypted messages.
+    pub dispatched_messages: u64,
     pub pdo_pending_requests: u64,
     pub pdo_requested: u64,
     /// Queued/running history-sync tasks and their logical compressed-payload
@@ -662,6 +664,7 @@ impl std::fmt::Display for MemoryReport {
             "  undec_dispatched:       {}",
             self.undecryptable_dispatched
         )?;
+        writeln!(f, "  dispatched_messages:    {}", self.dispatched_messages)?;
         writeln!(f, "  pdo_pending_requests:   {}", self.pdo_pending_requests)?;
         writeln!(f, "  pdo_requested:          {}", self.pdo_requested)?;
         writeln!(f, "--- Capacity-only caches ---")?;
@@ -1422,6 +1425,16 @@ pub struct Client {
     /// in `WAWebMessageProcessPlaceholder`.
     pub(crate) undecryptable_dispatched: Cache<wacore::types::message::SenderMessageId, ()>,
 
+    /// Dispatch-once gate for a decrypted message. A sender retrying its own
+    /// outbox resends one id as fresh ciphertext on a new ratchet iteration,
+    /// which decrypts as new traffic, so only message identity can collapse it.
+    pub(crate) dispatched_messages: Cache<wacore::types::message::SenderMessageId, ()>,
+
+    /// Lifetime count of resent messages this gate kept from reaching
+    /// consumers. Client-level, so it survives reconnects: the sender's retry
+    /// window does not end because our socket did.
+    pub(crate) duplicate_dispatch_suppressed: AtomicU64,
+
     pub enable_auto_reconnect: Arc<AtomicBool>,
     /// Set by [`Client::pause`] and cleared by [`Client::resume`]: the run loop
     /// parks instead of connecting for as long as it holds.

src/client/accessors.rs

@@ -290,6 +290,8 @@ impl Client {
         let mut snapshot = self.stats.snapshot();
         snapshot.reconnect_errors = self.auto_reconnect_errors.load(Ordering::Relaxed);
         snapshot.resends_throttled = self.resend_rate_limiter.throttled_total();
+        snapshot.messages_suppressed_duplicate =
+            self.duplicate_dispatch_suppressed.load(Ordering::Relaxed);
         snapshot
     }
 
@@ -425,6 +427,7 @@ impl Client {
             dm_devices_memo,
             message_retry_counts: self.message_retry_counts.entry_count(),
             undecryptable_dispatched: self.undecryptable_dispatched.entry_count(),
+            dispatched_messages: self.dispatched_messages.entry_count(),
             pdo_pending_requests: self.pdo_pending_requests.entry_count(),
             pdo_requested: self.pdo_requested.entry_count(),
             history_sync_tasks,

src/client/lifecycle.rs

@@ -476,6 +476,9 @@ impl Client {
 
             undecryptable_dispatched: cache_config.undecryptable_dispatched.build_with_ttl(),
 
+            dispatched_messages: cache_config.dispatched_messages.build_with_ttl(),
+            duplicate_dispatch_suppressed: AtomicU64::new(0),
+
             offline_sync_metrics: Arc::new(OfflineSyncMetrics {
                 active: AtomicBool::new(false),
                 total_messages: AtomicUsize::new(0),

src/features/message_edit.rs

@@ -347,6 +347,19 @@ impl<'a> SecretEncrypted<'a> {
     }
 }
 
+/// Whether the message carries a secret-encrypted addon envelope at all,
+/// regardless of whether that envelope is well formed.
+///
+/// [`extract_secret_encrypted`] answers "can this be opened", which is the
+/// wrong question for anything deciding that the consumer did not get real
+/// content: it returns `None` both for a plain message and for a tagged
+/// envelope that is malformed, and those two must not be treated alike.
+pub fn carries_secret_encrypted(msg: &wa::Message) -> bool {
+    msg.secret_encrypted_message.as_option().is_some()
+        || msg.enc_reaction_message.as_option().is_some()
+        || msg.enc_comment_message.as_option().is_some()
+}
+
 /// Extract any supported `secret_encrypted_message` envelope (EVENT_EDIT,
 /// MESSAGE_EDIT, POLL_EDIT, POLL_ADD_OPTION) from a received message.
 ///

src/message/commit_batch.rs

@@ -704,6 +704,80 @@ impl Client {
         .await
     }
 
+    /// Collapse deliveries of one message inside a batch, keeping the first.
+    ///
+    /// Returns the input untouched when there is nothing to drop, which is
+    /// every live batch (they hold one message) and almost every drained one.
+    ///
+    /// A repeat is identity *and* content: a resend re-encrypts the same
+    /// message, so the decoded protos are equal. Identity alone is not enough,
+    /// because one stanza id can carry several genuinely different payloads
+    /// that share an `info`. The msmsg loop in `handle_incoming_message`
+    /// dispatches every bot-reply part under one id, and a drain batch can hold
+    /// an unresolved secret envelope beside the retry that resolved to the
+    /// inner message; collapsing on identity alone would drop those and ack
+    /// them, losing content with nothing left to redeliver it. Comparing the
+    /// message keeps them and costs a structural compare only where identities
+    /// actually collide, which is the rare case this whole function exists for.
+    fn dedup_batch_by_message(&self, items: Arc<[InboundMessage]>) -> Arc<[InboundMessage]> {
+        if items.len() < 2 {
+            return items;
+        }
+        /// Distinct payloads compared per identity before the collapse gives
+        /// up on it. A resend repeats one message, so a real duplicate is
+        /// found against the first or second entry; an identity accumulating
+        /// more than this is not the pattern this function exists for, and
+        /// without a bound a participant could reuse one id across a full
+        /// 400-message drain batch and force ~80k structural compares on the
+        /// receive path while it holds the processing permit. Giving up keeps
+        /// everything, which is the safe direction.
+        const MAX_COMPARED_PER_ID: usize = 8;
+        type Seen = std::collections::HashMap<
+            wacore::types::message::SenderMessageId,
+            Vec<(Arc<MessageInfo>, Arc<waproto::whatsapp::Message>)>,
+        >;
+        fn keep(seen: &mut Seen, item: &InboundMessage) -> bool {
+            let kept = seen.entry(Client::dispatch_key(&item.info)).or_default();
+            if kept.len() >= MAX_COMPARED_PER_ID {
+                return true;
+            }
+            if kept.iter().any(|(info, m)| {
+                // Never collapse two dispatches of one stanza. They share the
+                // `MessageInfo` the stanza was parsed into, which the msmsg
+                // loop hands to every bot-reply part, so equal parts are not a
+                // repeat of each other. Best-effort by construction: two
+                // separate stanzas always allocate separate infos, so this
+                // never exempts a genuine resend, while a part whose info was
+                // copied on write falls through to the content compare.
+                !Arc::ptr_eq(info, &item.info)
+                    && (Arc::ptr_eq(m, &item.message) || **m == *item.message)
+            }) {
+                return false;
+            }
+            kept.push((Arc::clone(&item.info), Arc::clone(&item.message)));
+            true
+        }
+        let mut seen = Seen::with_capacity(items.len());
+        if items.iter().all(|item| keep(&mut seen, item)) {
+            return items;
+        }
+        seen.clear();
+        let kept: Arc<[InboundMessage]> = items
+            .iter()
+            .filter(|item| keep(&mut seen, item))
+            .cloned()
+            .collect();
+        // Counted like any other suppression: what this drops never reaches a
+        // consumer, so leaving it out would make the metric disagree with what
+        // the consumer saw for exactly the deliveries it exists to explain.
+        for _ in 0..(items.len() - kept.len()) {
+            self.duplicate_dispatch_suppressed
+                .fetch_add(1, Ordering::Relaxed);
+            wacore::telemetry::recv("duplicate_resend");
+        }
+        kept
+    }
+
     /// Commit one batch: durable buffer → Signal flush → hook → clear buffer →
     /// acks → event. Nothing is acked or observable before it is durable (WA
     /// Web's `createSnapshot` ordering); acks precede the event dispatch so a
@@ -737,6 +811,17 @@ impl Client {
             debug_assert!(commit_ticket.is_none());
             return true;
         }
+        // A drain can accumulate two deliveries of one message before the batch
+        // commits: both read the gate before either claim landed, so the
+        // collapse has to happen here too. Every stanza that arrived is still
+        // acked from `arrived`; only what the hook and the consumer see is
+        // reduced to one copy.
+        let arrived = Arc::clone(&items);
+        let items = if self.dispatch_gate_enabled() {
+            self.dedup_batch_by_message(items)
+        } else {
+            items
+        };
         let mut reinsert = ReinsertGuard {
             batcher: &self.inbound_commit_batch,
             items: is_drain.then(|| Arc::clone(&items)),
@@ -840,15 +925,43 @@ impl Client {
             }
             hook_committed = true;
 
-            let delete_keys: Vec<PendingInboundKey<'_>> = items
-                .iter()
-                .zip(&keys)
-                .map(|(item, (chat, sender))| PendingInboundKey {
-                    chat,
-                    sender,
-                    id: &item.info.id,
-                })
-                .collect();
+            // Cleared for every stanza that arrived, not just the ones kept: a
+            // pending row is keyed on the sender exactly as that delivery spelled
+            // it, while the collapse folds spellings together. Deleting only the
+            // kept spelling would leave a row that a later resend replays as an
+            // already committed message. Deleting a row that is not there is a
+            // no-op, so the superset is free.
+            let arrived_keys: Vec<(String, String)>;
+            let delete_keys: Vec<PendingInboundKey<'_>> = if arrived.len() == items.len() {
+                items
+                    .iter()
+                    .zip(&keys)
+                    .map(|(item, (chat, sender))| PendingInboundKey {
+                        chat,
+                        sender,
+                        id: &item.info.id,
+                    })
+                    .collect()
+            } else {
+                arrived_keys = arrived
+                    .iter()
+                    .map(|m| {
+                        (
+                            m.info.source.chat.to_string(),
+                            m.info.source.sender.to_string(),
+                        )
+                    })
+                    .collect();
+                arrived
+                    .iter()
+                    .zip(&arrived_keys)
+                    .map(|(item, (chat, sender))| PendingInboundKey {
+                        chat,
+                        sender,
+                        id: &item.info.id,
+                    })
+                    .collect()
+            };
             if let Err(e) = backend.delete_pending_inbound_batch(&delete_keys).await {
                 // Leftover rows replay as duplicates; the idempotent hook
                 // re-commits and the replay path clears them.
@@ -870,7 +983,7 @@ impl Client {
         // synchronously, so a handler that panics or blocks must not be able
         // to suppress acks for messages the consumer already owns — the
         // pre-batch at-most-once path acked before dispatching too.
-        for item in items.iter() {
+        for item in arrived.iter() {
             self.ack_received_message(&item.info);
         }
         // WA Web `createSnapshot` sends `sendAggregateOfflineReceipts` per
@@ -883,13 +996,35 @@ impl Client {
         if is_drain {
             self.flush_offline_receipts();
         }
+        let dispatched = Arc::clone(&items);
         self.core.event_bus.dispatch(Event::Messages(
             MessageBatch::builder()
                 .messages(items)
                 .origin(origin)
                 .hook_committed(hook_committed)
                 .build(),
         ));
+        // Claimed after the dispatch, never between the acks and it: this is the
+        // one suspension point in that span, and a teardown cancelling it there
+        // would leave the batch acked with its event never sent, which for a
+        // consumer without a hook is a lost message. Cancelled here instead, the
+        // claim is simply missing and a resend dispatches twice.
+        for item in dispatched.iter() {
+            // An unresolved secret envelope is a placeholder in the same sense
+            // as an UndecryptableMessage: what reached the consumer is not the
+            // content. Presence, not extractability: a malformed envelope is
+            // just as unreadable to a consumer as an unopenable one, and
+            // `extract_secret_encrypted` returns `None` for both a plain
+            // message and a malformed envelope. Claiming it would suppress the resend that arrives once
+            // the parent secret is known, so leave it unclaimed and take the
+            // duplicate instead. The batch collapse keeps such a resend for a
+            // related reason: it compares content, and the envelope is not the
+            // content.
+            if crate::features::message_edit::carries_secret_encrypted(&item.message) {
+                continue;
+            }
+            self.mark_message_dispatched(&item.info).await;
+        }
         true
     }
 }

src/message/dispatch.rs

@@ -13,6 +13,70 @@ fn delivery_receipt_burst_warning(
 }
 
 impl Client {
+    /// Has this message already been dispatched to consumers?
+    ///
+    /// A sender whose network is bad re-runs its own outbox: same message id,
+    /// fresh ciphertext on a new ratchet iteration. Neither ratchet can call
+    /// that a duplicate (`DuplicatedMessage` covers only the byte-identical
+    /// stanza the server replays), so identity is the only thing left that says
+    /// the two deliveries are one message.
+    ///
+    /// Read here and written by [`Self::mark_message_dispatched`] once the
+    /// batch carrying the message becomes observable. Those are two points in
+    /// time, so this cannot be an atomic `get_with` the way the sibling
+    /// `undecryptable_dispatched` gate is: claiming at the check would claim
+    /// for a commit that may still fail. `chat_lanes` serializes incoming
+    /// processing per chat and so closes the window for ordinary traffic, but
+    /// two workers for one chat can coexist after a lane eviction. The race
+    /// they leave is a second dispatch of one message, which is the behaviour
+    /// this gate improves on rather than a regression, and it is the safe
+    /// direction: the alternative loses the message.
+    pub(crate) async fn message_already_dispatched(&self, info: &Arc<MessageInfo>) -> bool {
+        self.dispatched_messages
+            .get(&Self::dispatch_key(info))
+            .await
+            .is_some()
+    }
+
+    /// Whether the dispatch-once gate is on. Capacity 0 is its documented off
+    /// switch, and it has to turn off the batch collapse too, or the switch
+    /// would restore the old behaviour for live traffic only.
+    pub(crate) fn dispatch_gate_enabled(&self) -> bool {
+        self.dispatched_messages.configured_capacity() != Some(0)
+    }
+
+    /// Claim a message id, called where a committed batch is dispatched.
+    ///
+    /// Placing it there rather than at the call site is what makes the offline
+    /// drain claim as well: a deferred batch dispatches later, and a claim
+    /// taken before that could be taken for a batch that never commits, which
+    /// would suppress and ack the redelivery with nothing ever handed to a
+    /// consumer, losing the message instead of duplicating it.
+    pub(crate) async fn mark_message_dispatched(&self, info: &Arc<MessageInfo>) {
+        self.dispatched_messages
+            .insert(Self::dispatch_key(info), ())
+            .await;
+    }
+
+    /// The message's identity: chat, id, and the sender without its device.
+    ///
+    /// Dropping the device matches WA Web, whose `MsgKey` for a group message
+    /// takes `participant: asUserWidOrThrow(author)`, a device-less wid. It
+    /// also has to: the server sends `skmsg` with a bare participant and
+    /// `pkmsg` with a device-qualified one, so a resend bundling a rotated
+    /// SKDM would otherwise be spelled differently from the delivery it
+    /// repeats and slip past the gate.
+    ///
+    /// The PN/LID namespace stays as it arrived, deliberately unresolved, for
+    /// the reasons [`Self::dispatch_undecryptable_event`] states at length.
+    pub(crate) fn dispatch_key(info: &Arc<MessageInfo>) -> wacore::types::message::SenderMessageId {
+        wacore::types::message::SenderMessageId::new(
+            info.source.chat.clone(),
+            info.id.clone(),
+            info.source.sender.to_non_ad(),
+        )
+    }
+
     /// Dispatches a successfully parsed message to the event bus and sends a delivery receipt.
     #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.recv.dispatch", level = "debug", skip_all, fields(chat = %info.source.chat.observe(), sender = %info.source.sender.observe(), msg_id = %info.id)))]
     pub(crate) async fn dispatch_parsed_message(