diff --git a/src/cache_config.rs b/src/cache_config.rs index 1fa9b5f5a..dbab3ebe5 100644 --- a/src/cache_config.rs +++ b/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 diff --git a/src/client.rs b/src/client.rs index bbd3461be..6b8b53f1d 100644 --- a/src/client.rs +++ b/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 diff --git a/src/client/accessors.rs b/src/client/accessors.rs index 6a200f01b..9a8c53593 100644 --- a/src/client/accessors.rs +++ b/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, diff --git a/src/client/lifecycle.rs b/src/client/lifecycle.rs index d2c34c4e7..4a2f11a18 100644 --- a/src/client/lifecycle.rs +++ b/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 { diff --git a/src/resend_rate_limiter.rs b/src/resend_rate_limiter.rs index c2c28a5e9..f239902ec 100644 --- a/src/resend_rate_limiter.rs +++ b/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>>, + 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() } diff --git a/src/retry.rs b/src/retry.rs index b2e5b5769..cfff54e97 100644 --- a/src/retry.rs +++ b/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() { diff --git a/wacore/src/stats.rs b/wacore/src/stats.rs index d7a5ef7ab..dd750662f 100644 --- a/wacore/src/stats.rs +++ b/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), }