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/dispatch.rs

@@ -13,6 +13,56 @@ 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`] after the
+    /// dispatch is durable; the read-then-write is safe because `chat_lanes`
+    /// serializes incoming processing per chat, and both deliveries of one
+    /// message are in one chat.
+    ///
+    /// Keyed on the wire spelling of the sender, deliberately unresolved, for
+    /// the reasons [`Self::dispatch_undecryptable_event`] states at length.
+    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 once its dispatch is durable.
+    ///
+    /// Only `Durable` claims it. A deferred commit (the offline drain) may
+    /// still fail, and a claim taken on one that does would let the redelivery
+    /// be suppressed and acked with nothing ever handed to a consumer, losing
+    /// the message instead of duplicating it. The cost is that a resend
+    /// arriving mid-drain still dispatches twice; drains carry the server's
+    /// byte-identical replays, which the ratchet rejects on its own.
+    pub(crate) async fn mark_message_dispatched(
+        &self,
+        info: &Arc<MessageInfo>,
+        state: &InboundCommitState,
+    ) {
+        if matches!(state, InboundCommitState::Durable) {
+            self.dispatched_messages
+                .insert(Self::dispatch_key(info), ())
+                .await;
+        }
+    }
+
+    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.clone(),
+        )
+    }
+
     /// 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,10 +1873,31 @@ 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())
                 .await;
+            self.mark_message_dispatched(info, &commit_state).await;
             if let Some((requester, request)) = app_state_key_share_job {
                 match commit_state {
                     InboundCommitState::Durable => {