fix(retry): bound outbound resend rate per group to prevent AccountLocked

src/bot.rs

@@ -492,6 +492,7 @@ pub struct BotBuilder<
     initial_push_name: Option<String>,
     cache_config: CacheConfig,
     wanted_pre_key_count: Option<usize>,
+    resend_rate_limit: Option<(u32, u32)>,
     _marker: PhantomData<(B, T, H, R)>,
 }
 
@@ -512,6 +513,7 @@ impl BotBuilder<MissingBackend, DefaultTransportState, DefaultHttpState, Default
             initial_push_name: None,
             cache_config: CacheConfig::default(),
             wanted_pre_key_count: None,
+            resend_rate_limit: None,
             _marker: PhantomData,
         }
     }
@@ -536,6 +538,7 @@ impl<B, T, H, R> BotBuilder<B, T, H, R> {
             initial_push_name: self.initial_push_name,
             cache_config: self.cache_config,
             wanted_pre_key_count: self.wanted_pre_key_count,
+            resend_rate_limit: self.resend_rate_limit,
             _marker: PhantomData,
         }
     }
@@ -834,6 +837,22 @@ impl<B, T, H, R> BotBuilder<B, T, H, R> {
         self
     }
 
+    /// Tune the per-chat outbound resend rate limiter.
+    ///
+    /// Outbound retry resends to a chat are bounded by a token bucket: `burst`
+    /// is the instantaneous allowance, `refill_per_min` the sustained ceiling
+    /// per chat. This caps the aggregate resend rate WhatsApp's anti-abuse
+    /// penalizes during a PN to LID migration fan-out, while throttled devices
+    /// still recover via the fresh-SKDM mark. A `burst` of 0 disables it.
+    ///
+    /// Defaults are conservative (burst 20, refill 10/min) and apply without
+    /// calling this. Can also be retuned live via
+    /// [`Client::set_resend_rate_limit`](crate::Client::set_resend_rate_limit).
+    pub fn with_resend_rate_limit(mut self, burst: u32, refill_per_min: u32) -> Self {
+        self.resend_rate_limit = Some((burst, refill_per_min));
+        self
+    }
+
     /// Set an initial push name on the device before connecting.
     ///
     /// This is included in the `ClientPayload` during registration, allowing the
@@ -955,6 +974,10 @@ impl BotBuilder<Provided, Provided, Provided, Provided> {
             client.set_wanted_pre_key_count(count);
         }
 
+        if let Some((burst, refill_per_min)) = self.resend_rate_limit {
+            client.set_resend_rate_limit(burst, refill_per_min);
+        }
+
         Ok(Bot {
             client,
             sync_task_receiver: Some(sync_task_receiver),

src/cache_config.rs

@@ -201,6 +201,12 @@ pub struct CacheConfig {
     pub session_locks_capacity: u64,
     /// Per-chat lane capacity (combined lock + queue). Default: 5000.
     pub chat_lanes_capacity: u64,
+    /// Per-chat resend rate-limiter capacity: one token-bucket entry per group
+    /// recently driving retry resends. Keep above the count of concurrently
+    /// storming groups: eviction is FIFO and fail-open (an evicted bucket is
+    /// recreated full), so undersizing only forgives rate, never over-throttles.
+    /// Default: 4096.
+    pub resend_rate_limiter_capacity: u64,
 
     // --- Sent message DB cleanup ---
     /// TTL in seconds for sent messages in DB before periodic cleanup. Must
@@ -273,6 +279,10 @@ impl std::fmt::Debug for CacheConfig {
             .field("session_recreate_history", &self.session_recreate_history)
             .field("session_locks_capacity", &self.session_locks_capacity)
             .field("chat_lanes_capacity", &self.chat_lanes_capacity)
+            .field(
+                "resend_rate_limiter_capacity",
+                &self.resend_rate_limiter_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)
@@ -329,6 +339,7 @@ impl Default for CacheConfig {
             // breaking serialization. Size generously to avoid eviction pressure.
             session_locks_capacity: 10_000,
             chat_lanes_capacity: 5_000,
+            resend_rate_limiter_capacity: 4_096,
             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

@@ -194,6 +194,9 @@ pub struct MemoryDiagnostics {
     // -- Capacity-only caches (no TTL) --
     pub session_locks: u64,
     pub chat_lanes: u64,
+    pub resend_rate_limiter_chats: u64,
+    /// Total outbound resends dropped by the per-chat rate limiter since start.
+    pub resends_throttled_total: u64,
     // -- Unbounded collections --
     pub response_waiters: usize,
     pub node_waiters: usize,
@@ -239,6 +242,16 @@ impl std::fmt::Display for MemoryDiagnostics {
         writeln!(f, "--- Capacity-only caches ---")?;
         writeln!(f, "  session_locks:          {}", self.session_locks)?;
         writeln!(f, "  chat_lanes:             {}", self.chat_lanes)?;
+        writeln!(
+            f,
+            "  resend_rl_chats:        {}",
+            self.resend_rate_limiter_chats
+        )?;
+        writeln!(
+            f,
+            "  resends_throttled:      {}",
+            self.resends_throttled_total
+        )?;
         writeln!(f, "--- Unbounded collections ---")?;
         writeln!(f, "  response_waiters:       {}", self.response_waiters)?;
         writeln!(f, "  node_waiters:           {}", self.node_waiters)?;
@@ -455,6 +468,11 @@ pub struct Client {
     /// loop us through prekey fetches.
     pub(crate) session_recreate_history: Cache<wacore_binary::jid::Jid, wacore::time::Instant>,
 
+    /// Per-chat outbound resend rate limiter: bounds the aggregate resend rate
+    /// 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,
+
     /// Dispatch-once gate for `UndecryptableMessage`: a server resend of a
     /// failed id re-enters the failure path and would otherwise fire a
     /// duplicate event. Mirrors WA Web's DB-level placeholder uniqueness

src/client/accessors.rs

@@ -60,6 +60,26 @@ impl Client {
         self.wanted_pre_key_count.load(Ordering::Relaxed)
     }
 
+    /// Retune the per-chat outbound resend rate limiter live (no reconnect).
+    ///
+    /// Outbound resends to a chat are bounded by a token bucket: `burst` is the
+    /// instantaneous allowance and `refill_per_min` the sustained ceiling per
+    /// chat. This caps the aggregate resend rate that WhatsApp's anti-abuse
+    /// penalizes during a PN to LID migration fan-out, while throttled devices
+    /// still recover via the fresh-SKDM mark. A `burst` of 0 disables the limiter.
+    ///
+    /// Takes effect on each chat's next retry; a lowered `burst` clamps a live
+    /// bucket on its next access.
+    pub fn set_resend_rate_limit(&self, burst: u32, refill_per_min: u32) {
+        self.resend_rate_limiter.set_rate(burst, refill_per_min);
+    }
+
+    /// Total outbound resends dropped by the per-chat rate limiter since start.
+    /// Surfaces storm chats without the `debug-diagnostics` feature.
+    pub fn resends_throttled_total(&self) -> u64 {
+        self.resend_rate_limiter.throttled_total()
+    }
+
     /// Returns a snapshot of all internal collection sizes for memory leak detection.
     ///
     /// Moka caches report approximate counts (pending evictions may not be reflected).
@@ -95,6 +115,8 @@ impl Client {
             pdo_requested: self.pdo_requested.entry_count(),
             session_locks: self.session_locks.entry_count(),
             chat_lanes: self.chat_lanes.entry_count(),
+            resend_rate_limiter_chats: self.resend_rate_limiter.entry_count(),
+            resends_throttled_total: self.resend_rate_limiter.throttled_total(),
             response_waiters: self.response_waiters.lock().await.len(),
             node_waiters: self.node_waiter_count.load(Ordering::Relaxed),
             pending_retries: pending_retries_count,

src/client/lifecycle.rs

@@ -185,6 +185,12 @@ impl Client {
 
             session_recreate_history: cache_config.session_recreate_history.build_with_ttl(),
 
+            resend_rate_limiter: crate::resend_rate_limiter::ResendRateLimiter::new(
+                cache_config.resend_rate_limiter_capacity,
+                crate::resend_rate_limiter::DEFAULT_RESEND_BURST,
+                crate::resend_rate_limiter::DEFAULT_RESEND_REFILL_PER_MIN,
+            ),
+
             undecryptable_dispatched: cache_config.undecryptable_dispatched.build_with_ttl(),
 
             offline_sync_metrics: Arc::new(OfflineSyncMetrics {

src/lib.rs

@@ -27,6 +27,7 @@ pub use waproto;
 
 pub mod cache;
 pub mod portable_cache;
+pub(crate) mod resend_rate_limiter;
 
 pub mod cache_config;
 pub use cache_config::{