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/message/commit_batch.rs

@@ -704,6 +704,29 @@ 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.
+    fn dedup_batch_by_message(items: Arc<[InboundMessage]>) -> Arc<[InboundMessage]> {
+        if items.len() < 2 {
+            return items;
+        }
+        let mut seen = std::collections::HashSet::with_capacity(items.len());
+        if items
+            .iter()
+            .all(|item| seen.insert(Self::dispatch_key(&item.info)))
+        {
+            return items;
+        }
+        seen.clear();
+        items
+            .iter()
+            .filter(|item| seen.insert(Self::dispatch_key(&item.info)))
+            .cloned()
+            .collect()
+    }
+
     /// 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 +760,13 @@ 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 = Self::dedup_batch_by_message(items);
         let mut reinsert = ReinsertGuard {
             batcher: &self.inbound_commit_batch,
             items: is_drain.then(|| Arc::clone(&items)),
@@ -870,7 +900,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,6 +913,9 @@ impl Client {
         if is_drain {
             self.flush_offline_receipts();
         }
+        for item in items.iter() {
+            self.mark_message_dispatched(&item.info).await;
+        }
         self.core.event_bus.dispatch(Event::Messages(
             MessageBatch::builder()
                 .messages(items)

src/message/dispatch.rs

@@ -13,6 +13,63 @@ 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()
+    }
+
+    /// 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(

src/message/receive.rs

@@ -1873,6 +1873,26 @@ impl Client {
                 skdm_only: true,
                 ..Default::default()
             })
+        } else if self.message_already_dispatched(info).await {
+            // The sender resent a message we already handed to consumers. Ack
+            // it the way the ratchet-level duplicate is acked, so a registered
+            // durability hook still gets its replay instead of a bare ack; the
+            // key share (if any) was scheduled by the delivery that dispatched.
+            // status is acked by the should_ack gate.
+            self.duplicate_dispatch_suppressed
+                .fetch_add(1, Ordering::Relaxed);
+            wacore::telemetry::recv("duplicate_resend");
+            log::debug!(
+                "[msg:{}] already dispatched for this sender; suppressing the resend's event",
+                info.id
+            );
+            if !info.source.chat.is_status_broadcast() {
+                self.ack_or_replay_to_hook(info).await;
+            }
+            Ok(PlaintextHandleOutcome {
+                dispatched: true,
+                ..Default::default()
+            })
         } else {
             let commit_state = self
                 .dispatch_parsed_message(msg, info, app_state_key_share_job.is_some())