feat(retry): quarantine per-(chat, requester) inbound retry receipts

src/cache_config.rs

@@ -213,6 +213,15 @@ pub struct CacheConfig {
     /// Default: 4096.
     pub resend_rate_limiter_capacity: u64,
 
+    /// Max (chat, requester) pairs tracked by the inbound retry-receipt
+    /// quarantine. Unlike the resend limiter (one entry per chat), the
+    /// quarantine keyspace is O(groups x broken members) — a single storming
+    /// 1012-participant group holds ~470 pairs — and evicting an ACTIVE pair
+    /// refunds its full burst on the next receipt, weakening the budget. Pairs
+    /// are tiny (two short strings + one bucket), so size generously.
+    /// Default: 32768.
+    pub retry_mark_quarantine_capacity: u64,
+
     // --- Sent message DB cleanup ---
     /// TTL in seconds for sent messages in DB before periodic cleanup. Must
     /// outlive retry receipts (which can arrive well after a send) or the retry
@@ -292,6 +301,10 @@ impl std::fmt::Debug for CacheConfig {
                 "resend_rate_limiter_capacity",
                 &self.resend_rate_limiter_capacity,
             )
+            .field(
+                "retry_mark_quarantine_capacity",
+                &self.retry_mark_quarantine_capacity,
+            )
             .field("sent_message_ttl_secs", &self.sent_message_ttl_secs)
             .field("msg_secret_policy", &self.msg_secret_policy)
             .field("msg_secret_retention", &self.msg_secret_retention)
@@ -350,6 +363,7 @@ impl Default for CacheConfig {
             chat_lanes_capacity: 5_000,
             group_distribution_locks_capacity: 512,
             resend_rate_limiter_capacity: 4_096,
+            retry_mark_quarantine_capacity: 32_768,
             sent_message_ttl_secs: 7200,
             // Bounded by default: seed only the still-relevant slice of history
             // and prune by per-add-on-kind event-time horizons, so the store no

src/client.rs

@@ -213,6 +213,7 @@ pub struct MemoryReport {
     pub session_locks: u64,
     pub chat_lanes: u64,
     pub resend_rate_limiter_chats: u64,
+    pub retry_mark_quarantine_pairs: u64,
     // -- Unbounded collections --
     pub response_waiters: usize,
     pub node_waiters: usize,
@@ -287,6 +288,11 @@ impl std::fmt::Display for MemoryReport {
             "  resend_rl_chats:        {}",
             self.resend_rate_limiter_chats
         )?;
+        writeln!(
+            f,
+            "  retry_quarantine_pairs: {}",
+            self.retry_mark_quarantine_pairs
+        )?;
         writeln!(f, "--- Unbounded collections ---")?;
         writeln!(f, "  response_waiters:       {}", self.response_waiters)?;
         writeln!(f, "  node_waiters:           {}", self.node_waiters)?;
@@ -612,6 +618,7 @@ pub struct Client {
     /// to a chat (the anti-abuse signal) so a PN to LID fan-out cannot storm into
     /// AccountLocked. Throttled devices still recover via the fresh-SKDM mark.
     pub(crate) resend_rate_limiter: crate::resend_rate_limiter::ResendRateLimiter,
+    pub(crate) retry_mark_quarantine: crate::resend_rate_limiter::RetryMarkQuarantine,
 
     /// Dispatch-once gate for `UndecryptableMessage`: a server resend of a
     /// failed id re-enters the failure path and would otherwise fire a

src/client/accessors.rs

@@ -83,6 +83,12 @@ impl Client {
         self.resend_rate_limiter.set_rate(burst, refill_per_min);
     }
 
+    /// Retune the per-(chat, requester) retry-receipt quarantine live.
+    /// `burst` 0 disables it. See `RetryMarkQuarantine` for rationale.
+    pub fn set_retry_mark_quarantine(&self, burst: u32, refill_per_day: u32) {
+        self.retry_mark_quarantine.set_rate(burst, refill_per_day);
+    }
+
     /// Cumulative wire I/O and activity counters for this client session.
     ///
     /// Always available, no feature gate: recording costs one relaxed atomic
@@ -93,6 +99,7 @@ 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.retry_receipts_quarantined = self.retry_mark_quarantine.throttled_total();
         snapshot
     }
 
@@ -161,6 +168,7 @@ impl Client {
             session_locks: self.session_locks.entry_count(),
             chat_lanes: self.chat_lanes.entry_count(),
             resend_rate_limiter_chats: self.resend_rate_limiter.entry_count(),
+            retry_mark_quarantine_pairs: self.retry_mark_quarantine.entry_count(),
             response_waiters,
             node_waiters: self.node_waiter_count.load(Ordering::Relaxed),
             pending_retries: pending_retries_count,

src/client/lifecycle.rs

@@ -194,6 +194,12 @@ impl Client {
                 crate::resend_rate_limiter::DEFAULT_RESEND_REFILL_PER_MIN,
             ),
 
+            retry_mark_quarantine: crate::resend_rate_limiter::RetryMarkQuarantine::new(
+                cache_config.retry_mark_quarantine_capacity,
+                crate::resend_rate_limiter::DEFAULT_RETRY_MARK_BURST,
+                crate::resend_rate_limiter::DEFAULT_RETRY_MARK_REFILL_PER_DAY,
+            ),
+
             undecryptable_dispatched: cache_config.undecryptable_dispatched.build_with_ttl(),
 
             offline_sync_metrics: Arc::new(OfflineSyncMetrics {

src/resend_rate_limiter.rs

@@ -137,11 +137,158 @@ impl ResendRateLimiter {
     }
 }
 
+/// Default burst for the per-(chat, requester) retry-receipt quarantine: how
+/// many "fresh SKDM" repair attempts one member gets at full speed. One mark
+/// is enough to repair a healthy member (the next send carries the SKDM), so
+/// anything past a small burst is a member whose session never establishes.
+pub(crate) const DEFAULT_RETRY_MARK_BURST: u32 = 2;
+
+/// Sustained repair attempts per day per (chat, requester) once the burst is
+/// spent. Keeps a genuinely-recovering member repairable (a fresh attempt
+/// every ~12h) while bounding a permanently-broken member to O(1)/day instead
+/// of O(group messages)/day.
+pub(crate) const DEFAULT_RETRY_MARK_REFILL_PER_DAY: u32 = 2;
+
+/// Per-(chat, requester) quarantine for inbound group retry receipts.
+///
+/// In large groups a cohort of members whose pairwise sessions never
+/// establish (dead registrations, exhausted prekeys, re-registered LIDs)
+/// sends a retry receipt for every single group message. Each receipt pays
+/// `markForgetSenderKey` (a DB write plus a sender-key cache invalidation for
+/// the whole group), bundle processing and possibly a resend — all upstream
+/// of the per-chat resend cap, which only bounds the resend itself. Observed
+/// in production at ~58k marks per 3.5 days from 468 members of a single
+/// 1012-participant group. This bounds the whole repair path per member:
+/// past the burst, further receipts from the same (chat, requester) are
+/// dropped before any work happens; the bucket refills so real recovery is
+/// still possible.
+pub(crate) struct RetryMarkQuarantine {
+    /// One bucket per (chat user, requester user). Capacity-only: evicting an
+    /// idle pair only forgives rate (it recreates full), never over-restricts.
+    buckets: Cache<(String, String), Arc<Mutex<TokenBucket>>>,
+    burst: AtomicU32,
+    refill_per_day: AtomicU32,
+    throttled_total: AtomicU64,
+}
+
+impl RetryMarkQuarantine {
+    pub(crate) fn new(capacity: u64, burst: u32, refill_per_day: u32) -> Self {
+        Self {
+            buckets: Cache::builder().max_capacity(capacity.max(1)).build(),
+            burst: AtomicU32::new(burst),
+            refill_per_day: AtomicU32::new(refill_per_day),
+            throttled_total: AtomicU64::new(0),
+        }
+    }
+
+    /// Retune the rate live. A `burst` of 0 disables the quarantine.
+    pub(crate) fn set_rate(&self, burst: u32, refill_per_day: u32) {
+        self.burst.store(burst, Ordering::Relaxed);
+        self.refill_per_day.store(refill_per_day, Ordering::Relaxed);
+    }
+
+    /// Try to consume one repair token for (chat, requester). `true` lets the
+    /// retry receipt through; `false` quarantines it. Device is intentionally
+    /// excluded from the key so all devices of a broken account share one
+    /// budget (WA Web re-targets the whole user when the primary goes cold).
+    pub(crate) async fn try_acquire(&self, chat: &Jid, requester: &Jid) -> bool {
+        let burst = self.burst.load(Ordering::Relaxed);
+        if burst == 0 {
+            return true;
+        }
+        let burst = burst as f64;
+        let refill_per_sec = self.refill_per_day.load(Ordering::Relaxed) as f64 / 86_400.0;
+
+        // The owned key allocates two small strings per receipt (no
+        // Borrow<(&str, &str)> for (String, String)); acceptable here — this
+        // runs once per retry receipt, not per message, and replaces a DB
+        // write + whole-group cache invalidation when it quarantines.
+        let key = (chat.user.to_string(), requester.user.to_string());
+        let bucket = self
+            .buckets
+            .get_with(key, async move {
+                Arc::new(Mutex::new(TokenBucket::new(burst, Instant::now())))
+            })
+            .await;
+
+        let allowed = bucket
+            .lock()
+            .await
+            .try_take(Instant::now(), burst, refill_per_sec);
+        if !allowed {
+            self.throttled_total.fetch_add(1, Ordering::Relaxed);
+        }
+        allowed
+    }
+
+    /// Total receipts quarantined since start (observability).
+    pub(crate) fn throttled_total(&self) -> u64 {
+        self.throttled_total.load(Ordering::Relaxed)
+    }
+
+    /// Number of (chat, requester) pairs holding a live bucket (diagnostics).
+    pub(crate) fn entry_count(&self) -> u64 {
+        self.buckets.entry_count()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
     use std::time::Duration;
 
+    #[tokio::test]
+    async fn quarantine_bounds_per_pair_and_isolates_pairs() {
+        // burst 2, refill 0: two repair attempts then quarantined.
+        let q = RetryMarkQuarantine::new(100, 2, 0);
+        let g: Jid = "123-456@g.us".parse().unwrap();
+        let a: Jid = "111@lid".parse().unwrap();
+        let b: Jid = "222@lid".parse().unwrap();
+        assert!(q.try_acquire(&g, &a).await);
+        assert!(q.try_acquire(&g, &a).await);
+        assert!(!q.try_acquire(&g, &a).await, "third receipt quarantined");
+        // Another requester in the same chat has its own budget.
+        assert!(q.try_acquire(&g, &b).await);
+        assert_eq!(q.throttled_total(), 1);
+        // burst 0 disables.
+        let off = RetryMarkQuarantine::new(100, 0, 0);
+        assert!(off.try_acquire(&g, &a).await);
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn quarantine_concurrent_acquires_for_one_pair_do_not_exceed_burst() {
+        // Mirrors the ResendRateLimiter contention test: hundreds of members
+        // hammering one group is the exact storm this guards, so burst
+        // enforcement must hold under concurrent receipts for the SAME pair.
+        let q = Arc::new(RetryMarkQuarantine::new(100, 5, 0));
+        let g: Jid = "123-456@g.us".parse().unwrap();
+        let r: Jid = "999@lid".parse().unwrap();
+        let allowed = Arc::new(AtomicU64::new(0));
+
+        let mut handles = Vec::new();
+        for _ in 0..40 {
+            let q = q.clone();
+            let g = g.clone();
+            let r = r.clone();
+            let allowed = allowed.clone();
+            handles.push(tokio::spawn(async move {
+                if q.try_acquire(&g, &r).await {
+                    allowed.fetch_add(1, Ordering::Relaxed);
+                }
+            }));
+        }
+        for h in handles {
+            h.await.unwrap();
+        }
+
+        assert_eq!(
+            allowed.load(Ordering::Relaxed),
+            5,
+            "exactly burst repairs pass under contention, no bypass"
+        );
+        assert_eq!(q.throttled_total(), 35);
+    }
+
     fn chat(s: &str) -> Jid {
         s.parse().unwrap()
     }

src/retry.rs

@@ -331,6 +331,37 @@ impl Client {
                 .as_ref()
                 .is_some_and(|our_lid| info.requester.is_same_user_as(our_lid));
 
+        // Quarantine members whose sessions never establish: past a small
+        // burst, each further receipt from the same (chat, requester) pair is
+        // dropped BEFORE all repair work — the group-info fetch, the rotateKey
+        // path (own sender-key deletion + whole-group cache invalidation for
+        // unknown senders) and markForgetSenderKey (a DB write + whole-group
+        // sender-key cache invalidation) all run per receipt otherwise, even
+        // when the per-chat cap later drops the resend. Placed after the
+        // message-cache lookup so receipts for already-expired messages (a
+        // cheap no-op) don't burn the pair's repair budget. The bucket refills
+        // (default 2/day) so genuine recovery still works.
+        //
+        // Own companion devices (`is_peer`) are exempt: a secondary of our own
+        // account that was offline for a while or rotated keys can legitimately
+        // need many repair cycles to rebuild its group session, and dropping
+        // those would block group decryption on that device. The storm this
+        // guards is third-party members, never our own devices.
+        if is_group_or_status
+            && !is_peer
+            && !self
+                .retry_mark_quarantine
+                .try_acquire(&info.chat, &info.requester)
+                .await
+        {
+            debug!(
+                "Quarantining retry receipt from {} in {}: repeated undeliverable SKDM (pair budget spent)",
+                info.requester.observe(),
+                info.chat.observe()
+            );
+            return Ok(());
+        }
+
         // Fetch group info (cache-first, server on miss) — used for SKDM rotation + addressing_mode.
         // Without this, a cold cache would silently default to PN semantics for LID groups.
         let cached_group_info = if info.chat.is_group() {

wacore/src/stats.rs

@@ -71,6 +71,9 @@ pub struct StatsSnapshot {
     /// Outbound resends dropped by the per-chat rate limiter. Surfaces storm
     /// chats.
     pub resends_throttled: u64,
+    /// Inbound group retry receipts dropped by the per-(chat, requester)
+    /// quarantine (client-side counter, filled by the owner).
+    pub retry_receipts_quarantined: u64,
     pub last_data_sent_ms: u64,
     pub last_data_received_ms: u64,
 }
@@ -169,6 +172,7 @@ impl SessionStats {
             reconnects: self.reconnects.load(Ordering::Relaxed),
             reconnect_errors: 0,
             resends_throttled: 0,
+            retry_receipts_quarantined: 0,
             last_data_sent_ms: self.last_data_sent_ms.load(Ordering::Relaxed),
             last_data_received_ms: self.last_data_received_ms.load(Ordering::Relaxed),
         }