From 0ac782b41ca4b77539e4fee49e33c6679b1bf67c Mon Sep 17 00:00:00 2001 From: Jeff Gardner <202880+erskingardner@users.noreply.github.com> Date: Sun, 5 Jul 2026 14:32:45 +0100 Subject: [PATCH 1/4] fix: shard rate limiter and add live cache-size gauge MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two capacity-path fixes for the rate limiter (#123, #125). #123 — bounded/sharded capacity eviction. The admission path ran an O(CLEANUP_BATCH_SIZE=1000) eviction scan while holding the limiter's single global write lock, so a key-cardinality flood serialized every rate-limit check behind that scan. Fix on two axes: - Shard the limiter into up to MAX_SHARDS (32) independent RwLock stripes keyed by hash(key) % N, so checks for keys in different stripes never contend on the same lock. Tiny caches (< MIN_ENTRIES_PER_SHARD = 256) stay single-stripe so small-capacity LRU eviction semantics are byte-for-byte identical to before. Per-stripe capacities sum to exactly max_entries. - Use a small admission-path scan bound (ADMISSION_SCAN_LIMIT = 32), distinct from the periodic-cleanup CLEANUP_BATCH_SIZE (1000), so a full stripe costs bounded, small work per check. The #206 reservation identity is preserved: next_hit_id remains a single process-wide AtomicU64 shared across stripes, so reservation ids stay globally unique and a delayed rollback can never remove a newer hit. The #210 admission-eviction reporting is unchanged. cleanup() now sweeps every stripe (up to CLEANUP_BATCH_SIZE each). #125 — live cache-size gauge. check_and_increment now returns a sampled total entry count (RateLimitCheck::sampled_cache_len) on roughly one in GAUGE_SAMPLE_INTERVAL (64) stripe-mutating admissions, summed across stripes to keep the metric's per-cache_type label shape. The caller publishes it so transponder_rate_limit_cache_size tracks growth toward capacity instead of lagging until the 60s cleanup tick, without a metric write per check. Adds tests for single-vs-sharded stripe selection, capacity summing, cross-stripe reservation uniqueness, the bounded admission scan, the sampled gauge, and concurrent cross-stripe admissions. Two large-cache tests that asserted the old 1000-wide global scan were rewritten: one to a single-stripe variant preserving the exact semantic, plus a new sharding-aware sibling. Part of #218. Co-Authored-By: Claude Fable 5 --- src/rate_limiter.rs | 579 +++++++++++++++++++++++++++++++++++++------- 1 file changed, 495 insertions(+), 84 deletions(-) diff --git a/src/rate_limiter.rs b/src/rate_limiter.rs index 7166000..21e0c24 100644 --- a/src/rate_limiter.rs +++ b/src/rate_limiter.rs @@ -4,7 +4,8 @@ //! to prevent spam and replay attacks. use std::collections::VecDeque; -use std::hash::Hash; +use std::collections::hash_map::RandomState; +use std::hash::{BuildHasher, Hash}; use std::num::NonZeroUsize; use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; @@ -41,6 +42,45 @@ pub const DEFAULT_GLOBAL_UNWRAP_LIMIT_PER_HOUR: u32 = 30_000; /// Maximum entries to scan per cleanup cycle. const CLEANUP_BATCH_SIZE: usize = 1000; +/// Maximum entries scanned by the admission-path eviction probe. +/// +/// Distinct from [`CLEANUP_BATCH_SIZE`]: admission runs under the hot per-check +/// write lock, so it must cost bounded, small work even when a stripe is at +/// capacity under a cardinality flood (see #123). The periodic [`cleanup`] +/// path — which runs on a timer, not the hot path — keeps the larger +/// [`CLEANUP_BATCH_SIZE`] sweep. A stale or below-limit victim is almost always +/// found near the LRU tail within this window; if none is, the unknown key is +/// rejected with `ExceededCapacityLimit` exactly as before, just after scanning +/// far fewer entries. +/// +/// [`cleanup`]: RateLimiter::cleanup +const ADMISSION_SCAN_LIMIT: usize = 32; + +/// Number of entries below which the limiter uses a single stripe. +/// +/// Sharding a tiny cache would make per-stripe capacity zero or one, changing +/// the LRU admission-eviction semantics that callers (and tests) rely on for +/// small `max_entries`. Below this threshold the limiter keeps a single stripe +/// so its behavior is byte-for-byte identical to the pre-sharding limiter; +/// above it, entries are striped for lock locality under cardinality pressure. +const MIN_ENTRIES_PER_SHARD: usize = 256; + +/// Maximum number of stripes a sharded limiter uses. +/// +/// Bounds lock/allocation overhead; production caches (100k keys) land here, +/// giving 32-way lock locality so a cardinality flood contends 1/32 of the +/// admission checks per lock instead of serializing them all (see #123). +const MAX_SHARDS: usize = 32; + +/// Sampling divisor for the opportunistic live cache-size gauge. +/// +/// The gauge is refreshed on roughly one in `GAUGE_SAMPLE_INTERVAL` admissions +/// that mutate a stripe, so cache-saturation onset is visible within a few +/// dozen admissions instead of only at the 60s cleanup tick (#125), without +/// paying a metric write on every hot-path check. Must be a power of two so the +/// sampling test is a cheap bit-mask. +const GAUGE_SAMPLE_INTERVAL: u64 = 64; + /// Token identifying one admitted hit for identity-aware rollback. /// /// IDs are unique within a limiter instance, including across entry eviction and @@ -66,6 +106,12 @@ pub enum RateLimitResult { pub struct RateLimitCheck { result: RateLimitResult, admission_evicted: bool, + /// A sampled snapshot of the limiter's total entry count, present only on + /// the roughly one-in-[`GAUGE_SAMPLE_INTERVAL`] admissions selected for an + /// opportunistic live cache-size gauge update (#125). `None` on the calls + /// that were not sampled, so the caller updates the gauge cheaply and + /// rarely rather than on every hot-path check. + sampled_cache_len: Option, } impl RateLimitCheck { @@ -92,6 +138,17 @@ impl RateLimitCheck { pub fn reservation(self) -> Option { self.result.reservation() } + + /// Returns a sampled total-entry count for opportunistic gauge updates. + /// + /// `Some(len)` on the sampled fraction of admissions (see + /// [`GAUGE_SAMPLE_INTERVAL`]); `None` otherwise. Lets the caller keep the + /// `transponder_rate_limit_cache_size` gauge fresh as the cache grows + /// toward capacity without a metric write per check. + #[must_use] + pub fn sampled_cache_len(self) -> Option { + self.sampled_cache_len + } } impl PartialEq for RateLimitCheck { @@ -243,21 +300,64 @@ impl Default for RateLimitConfig { /// per-key precision but does not bypass limits because the global unwrap /// limiter still bounds total admission. pub struct RateLimiter { - entries: RwLock>, + /// Independent LRU stripes, each behind its own `RwLock`. A key is routed + /// to `stripes[hash(key) % stripes.len()]`, so admission checks for keys in + /// different stripes never contend on the same lock — localizing both the + /// bounded admission scan and lock waits under a cardinality flood (#123). + stripes: Vec>>, + /// Stable per-limiter hasher so a key always routes to the same stripe. + hasher: RandomState, max_per_minute: u32, max_per_hour: u32, + /// Process-wide monotonic reservation-id source. Shared across ALL stripes + /// so a [`RateLimitReservation`] is unique within the limiter regardless of + /// which stripe holds the key — preserving the #206 identity guarantee that + /// a delayed rollback cannot remove a newer hit for the same key. next_hit_id: AtomicU64, + /// Counts stripe-mutating admissions to drive the sampled gauge (#125). + gauge_sample_counter: AtomicU64, } impl RateLimiter { /// Creates a new rate limiter with the given configuration. pub fn new(config: RateLimitConfig) -> Self { + let max_entries = config.max_entries.get(); + // Keep tiny caches single-stripe so small-`max_entries` LRU eviction + // semantics are byte-for-byte identical to the pre-sharding limiter; + // only stripe larger caches, and divide capacity evenly across stripes. + let shard_count = (max_entries / MIN_ENTRIES_PER_SHARD) + .clamp(1, MAX_SHARDS) + .max(1); + let base = max_entries / shard_count; + let remainder = max_entries % shard_count; + + let stripes = (0..shard_count) + .map(|i| { + // Distribute the remainder so the summed stripe capacity equals + // the configured `max_entries` exactly. + let cap = base + usize::from(i < remainder); + let cap = NonZeroUsize::new(cap).expect("per-stripe capacity is non-zero"); + RwLock::new(LruCache::new(cap)) + }) + .collect(); + Self { - entries: RwLock::new(LruCache::new(config.max_entries)), + stripes, + hasher: RandomState::new(), max_per_minute: config.max_per_minute, max_per_hour: config.max_per_hour, next_hit_id: AtomicU64::new(0), + gauge_sample_counter: AtomicU64::new(0), + } + } + + /// Routes a key to its stripe by stable hash. + fn stripe_for(&self, key: &K) -> &RwLock> { + if self.stripes.len() == 1 { + return &self.stripes[0]; } + let idx = (self.hasher.hash_one(key) as usize) % self.stripes.len(); + &self.stripes[idx] } fn evict_lru_admission_candidate( @@ -269,7 +369,10 @@ impl RateLimiter { let mut stale_candidate = None; let mut below_limit_candidate = None; - for (key, entry) in entries.iter_mut().rev().take(CLEANUP_BATCH_SIZE) { + // Bounded, small scan: unlike the periodic cleanup sweep this runs on + // the admission hot path under the stripe write lock, so it must be O(K) + // with a small K even when the stripe is full (#123). + for (key, entry) in entries.iter_mut().rev().take(ADMISSION_SCAN_LIMIT) { entry.prune(now); if entry.is_empty() { stale_candidate = Some(key.clone()); @@ -291,6 +394,31 @@ impl RateLimiter { false } + /// Samples the total entry count for the live gauge (#125). + /// + /// Returns `Some(total_len)` roughly once per [`GAUGE_SAMPLE_INTERVAL`] + /// stripe-mutating admissions, and `None` otherwise, so the caller refreshes + /// the cache-size gauge frequently enough to catch saturation onset without + /// a metric write on every check. The total is summed across stripes to keep + /// the metric's existing per-`cache_type` label shape (one value per + /// limiter, not per stripe). + async fn sampled_total_len(&self) -> Option { + let n = self.gauge_sample_counter.fetch_add(1, Ordering::Relaxed); + if !n.is_multiple_of(GAUGE_SAMPLE_INTERVAL) { + return None; + } + Some(self.total_len().await) + } + + /// Sums the entry count across all stripes. + async fn total_len(&self) -> usize { + let mut total = 0; + for stripe in &self.stripes { + total += stripe.read().await.len(); + } + total + } + /// Checks if a request is allowed and increments counters if so. /// /// Returns: @@ -301,61 +429,83 @@ impl RateLimiter { /// victim for the unknown key pub async fn check_and_increment(&self, key: &K) -> RateLimitCheck { let now = Instant::now(); - let mut entries = self.entries.write().await; - let mut admission_evicted = false; - - // Get or create entry (updates access position). At capacity, admit an - // unknown key by evicting an old stale or still-unlimited entry. Entries - // already sitting at their rate limit are protected so cache pressure - // cannot reset a limited key's counters. - let entry = if let Some(entry) = entries.get_mut(key) { - entry - } else { - if entries.len() >= entries.cap().get() { - if Self::evict_lru_admission_candidate( - &mut entries, - now, - self.max_per_minute, - self.max_per_hour, - ) { - admission_evicted = true; - } else { - return RateLimitCheck { - result: RateLimitResult::ExceededCapacityLimit, - admission_evicted: false, - }; + let mutated; + let (result, admission_evicted) = { + let mut entries = self.stripe_for(key).write().await; + let mut admission_evicted = false; + + // Get or create entry (updates access position). At capacity, admit + // an unknown key by evicting an old stale or still-unlimited entry. + // Entries already sitting at their rate limit are protected so cache + // pressure cannot reset a limited key's counters. + let entry = if let Some(entry) = entries.get_mut(key) { + entry + } else { + if entries.len() >= entries.cap().get() { + if Self::evict_lru_admission_candidate( + &mut entries, + now, + self.max_per_minute, + self.max_per_hour, + ) { + admission_evicted = true; + } else { + return RateLimitCheck { + result: RateLimitResult::ExceededCapacityLimit, + admission_evicted: false, + sampled_cache_len: None, + }; + } } + entries.put(key.clone(), RateLimitEntry::new()); + entries.get_mut(key).expect("just inserted") + }; + + entry.prune(now); + + // Check minute limit first (more likely to be hit) + if entry.minute_hits.len() >= self.max_per_minute as usize { + return RateLimitCheck { + result: RateLimitResult::ExceededMinuteLimit, + admission_evicted, + sampled_cache_len: None, + }; } - entries.put(key.clone(), RateLimitEntry::new()); - entries.get_mut(key).expect("just inserted") - }; - entry.prune(now); + // Check hour limit + if entry.hour_hits.len() >= self.max_per_hour as usize { + return RateLimitCheck { + result: RateLimitResult::ExceededHourLimit, + admission_evicted, + sampled_cache_len: None, + }; + } - // Check minute limit first (more likely to be hit) - if entry.minute_hits.len() >= self.max_per_minute as usize { - return RateLimitCheck { - result: RateLimitResult::ExceededMinuteLimit, - admission_evicted, - }; - } + // Increment counters + let hit_id = self.next_hit_id.fetch_add(1, Ordering::Relaxed); + entry.minute_hits.push_back((now, hit_id)); + entry.hour_hits.push_back((now, hit_id)); + mutated = true; - // Check hour limit - if entry.hour_hits.len() >= self.max_per_hour as usize { - return RateLimitCheck { - result: RateLimitResult::ExceededHourLimit, + ( + RateLimitResult::Allowed(RateLimitReservation(hit_id)), admission_evicted, - }; - } + ) + }; - // Increment counters - let hit_id = self.next_hit_id.fetch_add(1, Ordering::Relaxed); - entry.minute_hits.push_back((now, hit_id)); - entry.hour_hits.push_back((now, hit_id)); + // Opportunistically refresh the live size gauge outside the stripe lock + // just taken. Only sampled admissions read the (cross-stripe) length so + // the hot path stays cheap (#125). + let sampled_cache_len = if mutated { + self.sampled_total_len().await + } else { + None + }; RateLimitCheck { - result: RateLimitResult::Allowed(RateLimitReservation(hit_id)), + result, admission_evicted, + sampled_cache_len, } } @@ -369,7 +519,7 @@ impl RateLimiter { /// failed admission leaves no window-start trace for the next real request. pub async fn rollback_increment(&self, key: &K, reservation: RateLimitReservation) { let hit_id = reservation.0; - let mut entries = self.entries.write().await; + let mut entries = self.stripe_for(key).write().await; let should_remove = if let Some(entry) = entries.get_mut(key) { remove_hit(&mut entry.minute_hits, hit_id); remove_hit(&mut entry.hour_hits, hit_id); @@ -387,45 +537,50 @@ impl RateLimiter { /// /// Entries are considered stale if both windows have expired /// (no activity in the last hour). Processes up to `CLEANUP_BATCH_SIZE` - /// entries per call. + /// entries per stripe per call. pub async fn cleanup(&self) -> CleanupStats { - let mut entries = self.entries.write().await; let now = Instant::now(); - let before = entries.len(); let hour = Duration::from_secs(3600); + let mut evicted = 0; + let mut remaining = 0; + + for stripe in &self.stripes { + let mut entries = stripe.write().await; + let before = entries.len(); + + // Collect stale keys (hour window expired) + let stale: Vec = entries + .iter() + .rev() + .take(CLEANUP_BATCH_SIZE) + .filter(|(_, entry)| match entry.hour_hits.back() { + Some((hit, _)) => now.duration_since(*hit) >= hour, + None => true, + }) + .map(|(k, _)| k.clone()) + .collect(); + + for key in &stale { + entries.pop(key); + } - // Collect stale keys (hour window expired) - let stale: Vec = entries - .iter() - .rev() - .take(CLEANUP_BATCH_SIZE) - .filter(|(_, entry)| match entry.hour_hits.back() { - Some((hit, _)) => now.duration_since(*hit) >= hour, - None => true, - }) - .map(|(k, _)| k.clone()) - .collect(); - - for key in &stale { - entries.pop(key); + evicted += stale.len(); + remaining += before - stale.len(); } - CleanupStats { - evicted: stale.len(), - remaining: before - stale.len(), - } + CleanupStats { evicted, remaining } } /// Returns the current number of entries in the rate limiter. #[cfg(test)] pub async fn len(&self) -> usize { - self.entries.read().await.len() + self.total_len().await } /// Checks the current count without incrementing. #[cfg(test)] pub async fn peek_counts(&self, key: &K) -> Option<(u32, u32)> { - let entries = self.entries.read().await; + let entries = self.stripe_for(key).read().await; entries .peek(key) .map(|entry| (entry.minute_hits.len() as u32, entry.hour_hits.len() as u32)) @@ -711,13 +866,19 @@ mod tests { async fn test_cleanup_scans_stale_lru_entries_first() { tokio::time::pause(); + // Single-stripe capacity (below MIN_ENTRIES_PER_SHARD) but larger than + // ADMISSION_SCAN_LIMIT, so the fill never triggers admission eviction + // and cleanup's per-stripe CLEANUP_BATCH_SIZE scan reclaims every stale + // entry while the recently-touched key survives. Sharding-aware variant + // lives in test_cleanup_reclaims_stale_entries_across_stripes. + let capacity = MIN_ENTRIES_PER_SHARD - 1; let limiter: RateLimiter = RateLimiter::new(RateLimitConfig { max_per_minute: 10, max_per_hour: 100, - max_entries: entries(CLEANUP_BATCH_SIZE + 1), + max_entries: entries(capacity), }); - for key in 0..=CLEANUP_BATCH_SIZE as u64 { + for key in 0..capacity as u64 { assert!(limiter.check_and_increment(&key).await.is_allowed()); } @@ -726,10 +887,49 @@ mod tests { assert!(limiter.check_and_increment(&0u64).await.is_allowed()); let stats = limiter.cleanup().await; - assert_eq!(stats.evicted, CLEANUP_BATCH_SIZE); + assert_eq!(stats.evicted, capacity - 1); assert_eq!(stats.remaining, 1); } + #[tokio::test] + async fn test_cleanup_reclaims_stale_entries_across_stripes() { + tokio::time::pause(); + + // A cache large enough to shard: cleanup must reclaim stale entries in + // every stripe (each stripe scans up to CLEANUP_BATCH_SIZE, far more + // than any single stripe holds here) and keep the recently-touched key. + let capacity = MIN_ENTRIES_PER_SHARD * MAX_SHARDS; + let limiter: RateLimiter = RateLimiter::new(RateLimitConfig { + max_per_minute: 10, + max_per_hour: 100, + max_entries: entries(capacity), + }); + + for key in 0..capacity as u64 { + assert!(limiter.check_and_increment(&key).await.is_allowed()); + } + // Uneven hash routing can evict a few below-limit keys during the fill, + // so occupancy may be at or just below the aggregate capacity. + let filled = limiter.len().await; + assert!(filled <= capacity && filled >= capacity * 9 / 10); + + tokio::time::advance(Duration::from_secs(3601)).await; + + // Refresh one key so it is no longer stale; its stripe still holds it. + assert!(limiter.check_and_increment(&0u64).await.is_allowed()); + let refreshed = limiter.peek_counts(&0u64).await; + + let stats = limiter.cleanup().await; + // Every stale entry across every stripe is reclaimed; only the refreshed + // key (if it survived the fill) remains. + let survivors = usize::from(refreshed.is_some()); + assert_eq!(stats.remaining, survivors); + assert_eq!(limiter.len().await, survivors); + if survivors == 1 { + assert_eq!(limiter.peek_counts(&0u64).await, Some((1, 1))); + } + } + #[tokio::test] async fn test_admission_eviction_reports_side_effect() { let limiter: RateLimiter = RateLimiter::new(RateLimitConfig { @@ -817,34 +1017,57 @@ mod tests { } #[tokio::test] - async fn test_admission_scan_prunes_stale_candidate_when_cache_exceeds_batch_size() { + async fn test_admission_scan_prunes_stale_candidate_within_bounded_window() { tokio::time::pause(); + // Single-stripe capacity larger than the bounded admission scan window + // but below MIN_ENTRIES_PER_SHARD. When the whole stripe is stale, the + // admission scan (capped at ADMISSION_SCAN_LIMIT from the LRU tail) still + // finds a stale victim near the tail, prunes it, evicts it, and admits + // the new key — the previously-hit entries stay untouched and total size + // is unchanged. + let capacity = MIN_ENTRIES_PER_SHARD - 1; let limiter: RateLimiter = RateLimiter::new(RateLimitConfig { max_per_minute: 10, max_per_hour: 100, - max_entries: entries(CLEANUP_BATCH_SIZE + 1), + max_entries: entries(capacity), }); - for key in 0..=CLEANUP_BATCH_SIZE as u64 { + for key in 0..capacity as u64 { assert!(limiter.check_and_increment(&key).await.is_allowed()); } - assert_eq!(limiter.len().await, CLEANUP_BATCH_SIZE + 1); + assert_eq!(limiter.len().await, capacity); tokio::time::advance(Duration::from_secs(3601)).await; - let new_key = CLEANUP_BATCH_SIZE as u64 + 1; - assert!(limiter.check_and_increment(&new_key).await.is_allowed()); + let new_key = capacity as u64; + let check = limiter.check_and_increment(&new_key).await; + assert!(check.is_allowed()); + assert!(check.admission_evicted()); - assert_eq!(limiter.len().await, CLEANUP_BATCH_SIZE + 1); + // Total capacity is unchanged: a stale LRU-tail victim (key 0) was + // evicted to admit the new key, and the most-recently-inserted prior key + // is retained. + assert_eq!(limiter.len().await, capacity); assert_eq!(limiter.peek_counts(&0u64).await, None); assert_eq!( - limiter.peek_counts(&(CLEANUP_BATCH_SIZE as u64)).await, + limiter.peek_counts(&(capacity as u64 - 1)).await, Some((1, 1)) ); assert_eq!(limiter.peek_counts(&new_key).await, Some((1, 1))); } + #[test] + fn test_admission_scan_bound_is_small_relative_to_cleanup_batch() { + // The admission-path scan must be far smaller than the periodic-cleanup + // batch so a full stripe costs bounded, small work per check under a + // cardinality flood (#123), rather than reusing the 1000-entry sweep. + const { + assert!(ADMISSION_SCAN_LIMIT < CLEANUP_BATCH_SIZE); + assert!(ADMISSION_SCAN_LIMIT <= 32); + } + } + #[tokio::test] async fn test_rejects_new_keys_at_capacity_when_no_safe_eviction_exists() { let limiter: RateLimiter = RateLimiter::new(RateLimitConfig { @@ -935,4 +1158,192 @@ mod tests { // Key 2 is independent assert!(limiter.check_and_increment(&key2).await.is_allowed()); } + + #[tokio::test] + async fn test_small_cache_uses_single_stripe() { + // Caches below MIN_ENTRIES_PER_SHARD keep a single stripe so their LRU + // admission-eviction semantics are identical to the pre-sharding limiter. + let limiter: RateLimiter = RateLimiter::new(RateLimitConfig { + max_per_minute: 10, + max_per_hour: 100, + max_entries: entries(MIN_ENTRIES_PER_SHARD - 1), + }); + assert_eq!(limiter.stripes.len(), 1); + } + + #[tokio::test] + async fn test_large_cache_is_sharded_and_capacity_sums_to_max_entries() { + // A large cache is striped, and the summed per-stripe capacity equals + // the configured max_entries exactly (remainder distributed), so the + // aggregate key bound is preserved. + let max = MIN_ENTRIES_PER_SHARD * MAX_SHARDS + 7; + let limiter: RateLimiter = RateLimiter::new(RateLimitConfig { + max_per_minute: 10, + max_per_hour: 100, + max_entries: entries(max), + }); + assert_eq!(limiter.stripes.len(), MAX_SHARDS); + + let mut summed = 0usize; + for stripe in &limiter.stripes { + summed += stripe.read().await.cap().get(); + } + assert_eq!(summed, max); + } + + #[tokio::test] + async fn test_sharded_limiter_tracks_distinct_keys_and_limits_repeats() { + // Distinct keys spread across stripes. Sharding routes keys by hash, so + // stripe occupancy is uneven and the aggregate tracked-key count can sit + // slightly below max_entries once some stripe fills and rejects its + // overflow — an inherent, documented property of a sharded LRU. What + // must hold: every admitted key is independently rate-limited on repeat. + let max = MIN_ENTRIES_PER_SHARD * 2; + let limiter: RateLimiter = RateLimiter::new(RateLimitConfig { + max_per_minute: 1, + max_per_hour: 100, + max_entries: entries(max), + }); + + let mut admitted = Vec::new(); + for key in 0..max as u64 { + if limiter.check_and_increment(&key).await.is_allowed() { + admitted.push(key); + } + } + // The vast majority of distinct keys are admitted (uneven hashing costs + // only a small tail to per-stripe capacity rejection). + assert!( + admitted.len() >= max * 9 / 10, + "expected most distinct keys admitted, got {}/{max}", + admitted.len() + ); + // Every admitted key is at its per-minute limit of 1; a repeat is + // rejected by the minute limit, proving per-key accounting survives. + for key in &admitted { + assert_eq!( + limiter.check_and_increment(key).await, + RateLimitResult::ExceededMinuteLimit + ); + } + assert!(limiter.len().await <= max); + } + + #[tokio::test] + async fn test_reservation_ids_unique_across_stripes() { + // The reservation-id source is shared process-wide, so a rollback with a + // reservation from one stripe never removes a hit in another stripe even + // when ids would collide under per-stripe counters (#206 identity). + let max = MIN_ENTRIES_PER_SHARD * 4; + let limiter: RateLimiter = RateLimiter::new(RateLimitConfig { + max_per_minute: 10, + max_per_hour: 100, + max_entries: entries(max), + }); + + let mut reservations = Vec::new(); + for key in 0..max as u64 { + let reservation = limiter + .check_and_increment(&key) + .await + .reservation() + .expect("admit"); + reservations.push(reservation); + } + let unique: std::collections::HashSet = reservations.iter().map(|r| r.0).collect(); + assert_eq!( + unique.len(), + reservations.len(), + "reservation ids must be globally unique across stripes" + ); + } + + #[tokio::test] + async fn test_check_and_increment_samples_live_cache_size() { + // The very first admission is sampled (counter starts at 0), so the + // caller can seed the live gauge immediately; subsequent admissions are + // sampled about once per GAUGE_SAMPLE_INTERVAL. + let limiter: RateLimiter = RateLimiter::new(RateLimitConfig { + max_per_minute: 1_000, + max_per_hour: 100_000, + max_entries: entries(1), + }); + + let first = limiter.check_and_increment(&0u64).await; + assert_eq!( + first.sampled_cache_len(), + Some(1), + "first admission must publish a live size sample" + ); + + // The next GAUGE_SAMPLE_INTERVAL-1 admissions are not sampled. + let mut sampled = 0usize; + for _ in 1..GAUGE_SAMPLE_INTERVAL { + if limiter + .check_and_increment(&0u64) + .await + .sampled_cache_len() + .is_some() + { + sampled += 1; + } + } + assert_eq!(sampled, 0, "only 1-in-interval admissions are sampled"); + + // The interval-th admission after the first is sampled again. + assert!( + limiter + .check_and_increment(&0u64) + .await + .sampled_cache_len() + .is_some() + ); + } + + #[tokio::test] + async fn test_rejected_admission_carries_no_gauge_sample() { + // A capacity/limit rejection must not publish a size sample: it neither + // mutated a stripe nor advanced growth. + let limiter: RateLimiter = RateLimiter::new(RateLimitConfig { + max_per_minute: 1, + max_per_hour: 100, + max_entries: entries(1), + }); + + assert!(limiter.check_and_increment(&0u64).await.is_allowed()); + let rejected = limiter.check_and_increment(&0u64).await; + assert_eq!(rejected, RateLimitResult::ExceededMinuteLimit); + assert_eq!(rejected.sampled_cache_len(), None); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn test_concurrent_admissions_across_stripes_do_not_deadlock() { + use std::sync::Arc; + + // A cardinality flood of distinct keys hammering a sharded limiter + // concurrently must make progress (stripes lock independently) and admit + // every distinct key within aggregate capacity. + let max = MIN_ENTRIES_PER_SHARD * MAX_SHARDS; + let limiter: Arc> = Arc::new(RateLimiter::new(RateLimitConfig { + max_per_minute: 5, + max_per_hour: 1_000, + max_entries: entries(max), + })); + + let mut handles = Vec::new(); + for shard in 0..8u64 { + let limiter = Arc::clone(&limiter); + handles.push(tokio::spawn(async move { + for i in 0..500u64 { + let key = shard * 1_000 + i; + let _ = limiter.check_and_increment(&key).await; + } + })); + } + for handle in handles { + handle.await.expect("task panicked"); + } + // No panic/deadlock; the limiter stays within its aggregate bound. + assert!(limiter.len().await <= max); + } } From 3f771ed00e4623fefd8257304a9b85ce9948acf0 Mon Sep 17 00:00:00 2001 From: Jeff Gardner <202880+erskingardner@users.noreply.github.com> Date: Sun, 5 Jul 2026 14:32:55 +0100 Subject: [PATCH 2/4] fix: expose pre-charge dispatch admission checks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add PushDispatcher::accepts(platform) and PushDispatcher::token_is_encodable(payload) so the event processor can reject undispatchable tokens BEFORE charging the rate limiters (#177). accepts mirrors dispatch()'s platform-client presence filter; the token-encoding check mirrors its FCM UTF-8 filter. The existing filters inside dispatch() are kept as a defensive second layer (and their comment now says so) — in the normal event path they no longer fire because process_inner pre-filters, but dispatch still never enqueues an undeliverable token if a caller skips the pre-filter. Adds unit tests for both helpers. Part of #218. Co-Authored-By: Claude Fable 5 --- src/push/dispatcher.rs | 85 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 85 insertions(+) diff --git a/src/push/dispatcher.rs b/src/push/dispatcher.rs index de581e1..9b09dc4 100644 --- a/src/push/dispatcher.rs +++ b/src/push/dispatcher.rs @@ -524,6 +524,14 @@ impl PushDispatcher { let mut messages = Vec::with_capacity(requested_count); for payload in payloads { + // These platform/encoding filters are now a defensive second layer: + // `process_inner` pre-filters undispatchable tokens (unconfigured + // platform / non-UTF-8 FCM) via `accepts` / `token_is_encodable` + // BEFORE charging the rate limiters, so it can refund and count the + // drop (#177). Keeping the checks here too means `dispatch` never + // enqueues an undeliverable token even if a caller skips the + // pre-filter, but in the normal event path these `continue`s do not + // fire. // Extract token as Zeroizing for automatic cleanup let (platform, token): (Platform, Zeroizing) = match payload.platform { Platform::Apns => { @@ -625,6 +633,39 @@ impl PushDispatcher { Ok(message_count) } + /// Whether the dispatcher has a client wired for `platform`. + /// + /// This is the cheap pre-charge admission gate: `process_inner` calls it + /// *before* spending rate-limit budget so a token targeting an unconfigured + /// platform is dropped without charging (and refunding) the limiters (#177). + /// It mirrors the platform-client presence check inside [`Self::dispatch`] + /// (`apns_client.is_some()` / `fcm_client.is_some()`), keeping the two in + /// lock-step: `dispatch` still filters defensively, but a token this returns + /// `false` for never reaches it. + #[must_use] + pub fn accepts(&self, platform: Platform) -> bool { + match platform { + Platform::Apns => self.apns_client.is_some(), + Platform::Fcm => self.fcm_client.is_some(), + } + } + + /// Whether `payload`'s device token is transport-encodable for its platform. + /// + /// FCM tokens are sent as UTF-8 strings, so a non-UTF-8 device-token blob is + /// undeliverable and [`Self::dispatch`] would silently drop it. Checking here + /// lets `process_inner` treat that token as a terminal drop and refund / + /// count it, rather than spend rate-limit budget on a notification that can + /// never be sent (#177). APNs tokens are hex-encoded from raw bytes, so they + /// are always encodable. + #[must_use] + pub fn token_is_encodable(payload: &TokenPayload) -> bool { + match payload.platform { + Platform::Apns => true, + Platform::Fcm => payload.device_token_string().is_some(), + } + } + /// Check if APNs is configured and ready. #[must_use] pub fn has_apns(&self) -> bool { @@ -999,6 +1040,50 @@ mod tests { assert_eq!(dispatcher.dispatch(vec![]).await.unwrap(), 0); } + #[tokio::test] + async fn test_accepts_reflects_configured_platforms() { + use crate::config::ApnsConfig; + use crate::push::ApnsClient; + + // No clients: accepts nothing. + let none = PushDispatcher::new(None, None); + assert!(!none.accepts(Platform::Apns)); + assert!(!none.accepts(Platform::Fcm)); + + // APNs-only: accepts APNs, rejects FCM (mirrors dispatch()'s filter). + let apns_config = ApnsConfig { + enabled: true, + key_id: "KEY123".to_string(), + team_id: "TEAM456".to_string(), + private_key_path: String::new(), + environment: crate::config::ApnsEnvironment::Sandbox, + bundle_id: "com.example.app".to_string(), + payload_mode: Default::default(), + }; + let apns_only = PushDispatcher::new(Some(ApnsClient::mock(apns_config, true)), None); + assert!(apns_only.accepts(Platform::Apns)); + assert!(!apns_only.accepts(Platform::Fcm)); + } + + #[test] + fn test_token_is_encodable_matches_dispatch_encoding() { + // APNs tokens (hex-encoded) are always encodable, including binary bytes. + assert!(PushDispatcher::token_is_encodable(&TokenPayload { + platform: Platform::Apns, + device_token: vec![0xff, 0x00, 0x01], + })); + // UTF-8 FCM token is encodable. + assert!(PushDispatcher::token_is_encodable(&TokenPayload { + platform: Platform::Fcm, + device_token: b"fcm-token-123".to_vec(), + })); + // Non-UTF-8 FCM token is NOT encodable — dispatch() would drop it. + assert!(!PushDispatcher::token_is_encodable(&TokenPayload { + platform: Platform::Fcm, + device_token: vec![0xff, 0xfe, 0x00, 0x01], + })); + } + #[tokio::test] async fn test_dispatch_without_clients() { let dispatcher = PushDispatcher::new(None, None); From ce625bc5191c73f081b56a90eefd6635376d6b07 Mon Sep 17 00:00:00 2001 From: Jeff Gardner <202880+erskingardner@users.noreply.github.com> Date: Sun, 5 Jul 2026 14:33:25 +0100 Subject: [PATCH 3/4] fix: transactional rate-limit admission lifecycle MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Make per-token rate-limit charging transactional in process_inner so no admission path can strand a spent increment (#170, #177), fold the redundant dedup lock on the success path (#197), and wire the live rate-limit cache-size gauge (#125). AdmissionGuard (RAII-style, explicit-consume). After the encrypted-token limiter charges a token, the charge is held by an AdmissionGuard that every exit path must resolve: - commit() — fully admitted; hands the encrypted+device charges to the dispatch-failure rollback ledger. - refund() — dropped after charging; rolls back every charge held. - keep_charge() — dropped but the charge is intentionally retained. The guard's Drop asserts it was resolved. Because the limiter refund is async (the stripe lock is a tokio RwLock), a blocking Drop-time refund isn't viable, so this is an explicit-consume guard rather than fire-and-forget RAII — the trade the tracker calls out as acceptable. It still centralizes the refund decision per exit path. #170 — device-limiter reject now refunds the encrypted charge. Previously a device reject `continue`d without rolling back the encrypted-token increment already spent for that token, so a legitimate transient redelivery could read as a replay. guard.refund() on the device-reject branch fixes this. The decrypt-failure path keeps its charge via guard.keep_charge() (documented replay defense). #177 — dispatch-level drops refund and are counted. Tokens targeting an unconfigured platform, or (FCM) carrying non-UTF-8 device bytes, were dropped silently inside dispatch() AFTER both limiters were charged, with no metric. process_inner now pre-filters via dispatcher.accepts() / token_is_encodable() right after decrypt (platform is only known then) and before the device charge: it refunds the encrypted charge and records record_push_failed(platform, "unconfigured" | "invalid_encoding"). These drops mark the event terminally dropped (not a retryable rate shed). #197 — single dedup write-lock on the success path. mark_seen refreshed the completion timestamp by re-put'ing the entry under a second lock, redundantly re-writing the dedup_cache_size gauge. SeenEventStore gains mark_terminal, which mutates the existing reservation entry in place (refreshing seen_at, flipping terminal, promoting LRU position) and reports whether the set size changed, so the gauge is written only when it actually moved. Completion-timestamp-refresh and release_reservation semantics are preserved. #125 — the sampled cross-stripe cache length from check_and_increment is published to transponder_rate_limit_cache_size on the admission path. Tests: device-reject refund, unconfigured-platform and non-UTF-8 FCM drop (refund + metric), decrypt-failure charge retention, a multi-thread concurrent device-reject interleaving test proving no encrypted charge is stranded, the live gauge updating before cleanup, and the single dedup-gauge update. The create_processor_with_rate_limits helper now wires a configured APNs mock so its device/encrypted rate-limit tests exercise dispatchable tokens (previously they relied on the #177 bug of charging budget for tokens dropped silently by an unconfigured dispatcher). Fixes #170 Fixes #177 Fixes #197 Part of #218. Co-Authored-By: Claude Fable 5 --- src/nostr/events.rs | 709 ++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 686 insertions(+), 23 deletions(-) diff --git a/src/nostr/events.rs b/src/nostr/events.rs index e847b1f..4f58d22 100644 --- a/src/nostr/events.rs +++ b/src/nostr/events.rs @@ -21,13 +21,14 @@ use tokio::time::Instant; use tracing::{debug, trace, warn}; use crate::crypto::nip59::DEFAULT_MAX_TOKENS_PER_EVENT; -use crate::crypto::{Nip59Handler, TokenDecryptor, TokenPayload}; +use crate::crypto::{Nip59Handler, Platform, TokenDecryptor, TokenPayload}; use crate::error::{Error, Result}; use crate::metrics::{EventOutcome, Metrics, OperationOutcome}; use crate::push::PushDispatcher; use crate::rate_limiter::{ DEFAULT_GLOBAL_UNWRAP_LIMIT_PER_HOUR, DEFAULT_GLOBAL_UNWRAP_LIMIT_PER_MINUTE, DEFAULT_MAX_SIZE, - DEFAULT_RATE_LIMIT_PER_HOUR, DEFAULT_RATE_LIMIT_PER_MINUTE, RateLimitConfig, RateLimiter, + DEFAULT_RATE_LIMIT_PER_HOUR, DEFAULT_RATE_LIMIT_PER_MINUTE, RateLimitConfig, + RateLimitReservation, RateLimiter, }; /// Maximum NIP-59 timestamp randomization window for gift wraps. @@ -139,6 +140,37 @@ impl SeenEventStore { } } + /// Refresh an existing reservation into a terminal entry in place. + /// + /// On the success hot path the ID is already present from `try_reserve`, so + /// this mutates its `SeenEvent` (new `seen_at`, `terminal = true`) instead of + /// re-inserting a fresh value. Returns `true` when the set size changed + /// (i.e. the entry was absent and had to be inserted — e.g. an entry evicted + /// by LRU pressure between reservation and completion), so the caller only + /// pays a `dedup_cache_size` gauge write when the length actually moved. This + /// removes the redundant second gauge update the double-lock success path + /// carried (#197) while preserving the completion-timestamp-refresh and + /// terminal-flip semantics `mark_seen` provided. + fn mark_terminal(&mut self, event_id: EventId, seen_at: Instant) -> bool { + match self { + Self::Bounded(seen) => { + if let Some(existing) = seen.peek_mut(&event_id) { + *existing = SeenEvent::terminal(seen_at); + // Promote to most-recently-used to match the prior `put` + // behavior, which refreshed LRU position on completion. + seen.promote(&event_id); + false + } else { + seen.put(event_id, SeenEvent::terminal(seen_at)); + true + } + } + Self::Retained(seen) => seen + .insert(event_id, SeenEvent::terminal(seen_at)) + .is_none(), + } + } + fn expired_keys(&self, now: Instant, retention: Duration) -> Vec { match self { Self::Bounded(seen) => seen @@ -348,6 +380,127 @@ enum ProcessOutcome { RateLimitedShed, } +/// A per-token admission charge that must be explicitly resolved. +/// +/// Created after the encrypted-token limiter admits a token, and (once the +/// device-token limiter also admits) after that charge too. It makes the +/// charge/refund lifecycle transactional so no admission path can silently +/// strand a spent rate-limit increment (the class of bug behind #170 and #177): +/// +/// - [`commit`](AdmissionGuard::commit) — the token was fully admitted and +/// handed toward the dispatcher; the charges are handed off to the caller's +/// dispatch-failure rollback ledger. Returns the keys and reservations. +/// - [`refund`](AdmissionGuard::refund) — the token is being dropped *after* +/// being charged (device-limiter reject per #170, or an undispatchable +/// platform / non-UTF-8 token discovered post-decrypt per #177). Every charge +/// the guard holds is rolled back so the drop leaves no spent budget. +/// - [`keep_charge`](AdmissionGuard::keep_charge) — the token is dropped but its +/// charge is *intentionally retained* (decrypt failure: an invalid encrypted +/// blob must still spend replay/spam budget, documented in #170). +/// +/// The `Drop` impl asserts the guard was resolved. Because the limiter refund is +/// `async` (the stripe lock is a tokio `RwLock`), a `Drop`-time auto-refund +/// would have to block; the guard is therefore an explicit-consume guard rather +/// than a fire-and-forget RAII one. This keeps the refund decision in exactly +/// one place per exit path while remaining `async`-correct — the trade the +/// tracker calls out as acceptable when a blocking Drop is not. +#[must_use = "an AdmissionGuard must be committed, refunded, or explicitly kept"] +struct AdmissionGuard<'a> { + encrypted_limiter: &'a RateLimiter<[u8; 32]>, + device_limiter: &'a RateLimiter<[u8; 32]>, + encrypted_key: [u8; 32], + encrypted_reservation: RateLimitReservation, + /// Present once the device-token limiter has also admitted the token. + device: Option<([u8; 32], RateLimitReservation)>, + resolved: bool, +} + +impl<'a> AdmissionGuard<'a> { + /// Record the encrypted-token charge just made for a token. + fn new( + encrypted_limiter: &'a RateLimiter<[u8; 32]>, + device_limiter: &'a RateLimiter<[u8; 32]>, + encrypted_key: [u8; 32], + encrypted_reservation: RateLimitReservation, + ) -> Self { + Self { + encrypted_limiter, + device_limiter, + encrypted_key, + encrypted_reservation, + device: None, + resolved: false, + } + } + + /// Record the device-token charge for the same token. + fn add_device_charge(&mut self, device_key: [u8; 32], reservation: RateLimitReservation) { + self.device = Some((device_key, reservation)); + } + + /// Roll back every charge this guard holds (encrypted, and device if any). + /// + /// Used on every drop-after-charge path: a device-limiter reject (#170) and + /// a post-decrypt undispatchable/non-UTF-8 drop (#177). + async fn refund(mut self) { + self.encrypted_limiter + .rollback_increment(&self.encrypted_key, self.encrypted_reservation) + .await; + if let Some((device_key, reservation)) = self.device { + self.device_limiter + .rollback_increment(&device_key, reservation) + .await; + } + self.resolved = true; + } + + /// Intentionally keep the charge(s) — the drop is a deliberate budget spend. + fn keep_charge(mut self) { + self.resolved = true; + } + + /// Hand off the fully-admitted charges to the caller's rollback ledger. + fn commit(mut self) -> AdmittedCharges { + self.resolved = true; + let (device_key, device_reservation) = self + .device + .expect("commit requires a device-token charge to have been recorded"); + AdmittedCharges { + encrypted_key: self.encrypted_key, + encrypted_reservation: self.encrypted_reservation, + device_key, + device_reservation, + } + } +} + +impl Drop for AdmissionGuard<'_> { + fn drop(&mut self) { + debug_assert!( + self.resolved, + "AdmissionGuard dropped without commit/refund/keep_charge — a rate-limit charge would be stranded" + ); + } +} + +/// Fully-admitted rate-limit charges for one token, recorded so a later +/// dispatch-admission failure can roll back exactly the increments it spent. +struct AdmittedCharges { + encrypted_key: [u8; 32], + encrypted_reservation: RateLimitReservation, + device_key: [u8; 32], + device_reservation: RateLimitReservation, +} + +/// Metrics label for a push platform (matches the dispatcher's `platform` +/// label values so drop counters aggregate with send counters). +fn platform_metric_label(platform: Platform) -> &'static str { + match platform { + Platform::Apns => "apns", + Platform::Fcm => "fcm", + } +} + /// Event processor for handling incoming gift-wrapped notifications. pub struct EventProcessor { nip59_handler: Nip59Handler, @@ -830,13 +983,15 @@ impl EventProcessor { // Rate limiting happens BEFORE decryption intentionally: // Prevents wasting CPU on tokens we'll immediately rate-limit anyway. let mut payloads = Vec::with_capacity(token_bytes.len()); - let mut admitted_rate_limit_keys = Vec::with_capacity(token_bytes.len()); + let mut admitted_rate_limit_keys: Vec = + Vec::with_capacity(token_bytes.len()); // Track whether a zero-admission outcome is purely a transient per-token // rate-limit shed. `rate_limited_any` records that at least one token was // dropped by a limiter; `terminally_dropped_any` records that at least one - // token was dropped for a permanent reason (undecryptable per MIP-05), so - // the event genuinely carried invalid tokens rather than momentarily - // over-budget ones. + // token was dropped for a permanent reason (undecryptable per MIP-05, or + // targeting an unconfigured platform / carrying a non-encodable token per + // #177), so the event genuinely carried undispatchable tokens rather than + // momentarily over-budget ones. let mut rate_limited_any = false; let mut terminally_dropped_any = false; for bytes in token_bytes { @@ -851,6 +1006,7 @@ impl EventProcessor { { m.record_rate_limit_admission_eviction("encrypted_token"); } + self.publish_rate_limit_gauge("encrypted_token", encrypted_result.sampled_cache_len()); if !encrypted_result.is_allowed() { rate_limited_any = true; trace!( @@ -865,6 +1021,16 @@ impl EventProcessor { let encrypted_reservation = encrypted_result .reservation() .expect("allowed encrypted-token admission must carry a reservation"); + // The encrypted-token limiter has now been charged. From here every + // exit path must resolve this guard: refund on a device reject or an + // undispatchable-token drop, keep on decrypt failure, commit on full + // admission. This is the fix for the #170 stranded encrypted charge. + let mut guard = AdmissionGuard::new( + &self.encrypted_token_limiter, + &self.device_token_limiter, + encrypted_key, + encrypted_reservation, + ); // Decrypt the token let decrypt_started_at = StageTimer::start(); @@ -880,11 +1046,12 @@ impl EventProcessor { p } Err(e) => { - // Silently ignore invalid tokens per MIP-05 spec + // Silently ignore invalid tokens per MIP-05 spec. // Keep the encrypted-token rate-limit increment: invalid // encrypted blobs should still spend replay/spam budget. // A decrypt failure is permanent, so a zero-admission event // that hit one is terminal, not a retryable rate shed. + guard.keep_charge(); terminally_dropped_any = true; trace!(error = %e, "Failed to decrypt token (ignoring)"); if let Some(ref m) = self.metrics { @@ -898,6 +1065,34 @@ impl EventProcessor { } }; + // Pre-charge dispatch filter (#177): the platform/token is only known + // after decrypt, but it is known BEFORE the device-token limiter is + // charged. If the dispatcher could not send this token — its platform + // is unconfigured, or (FCM) the token bytes are not UTF-8 — drop it + // now, refunding the encrypted charge (no device charge yet) and + // recording a real drop metric, instead of letting `dispatch()` drop + // it silently after both limiters were charged. + if !self.push_dispatcher.accepts(payload.platform) { + guard.refund().await; + terminally_dropped_any = true; + let platform = platform_metric_label(payload.platform); + trace!(platform, "Dropping token: platform not configured"); + if let Some(ref m) = self.metrics { + m.record_push_failed(platform, "unconfigured"); + } + continue; + } + if !PushDispatcher::token_is_encodable(&payload) { + guard.refund().await; + terminally_dropped_any = true; + let platform = platform_metric_label(payload.platform); + trace!(platform, "Dropping token: device token not encodable"); + if let Some(ref m) = self.metrics { + m.record_push_failed(platform, "invalid_encoding"); + } + continue; + } + // Rate limit check 2: Is this device token within rate limits? let device_key = Self::hash_device_token_key(&payload); let device_result = self @@ -909,7 +1104,14 @@ impl EventProcessor { { m.record_rate_limit_admission_eviction("device_token"); } + self.publish_rate_limit_gauge("device_token", device_result.sampled_cache_len()); if !device_result.is_allowed() { + // The device limiter rejected this token. Refund the encrypted + // charge already spent for it (#170): without this, the encrypted + // blob's replay/spam budget was consumed for a token that was + // never admitted, so a legitimate transient redelivery could read + // as a replay. + guard.refund().await; rate_limited_any = true; trace!( reason = device_result.limit_reason(), @@ -923,16 +1125,12 @@ impl EventProcessor { let device_reservation = device_result .reservation() .expect("allowed device-token admission must carry a reservation"); + guard.add_device_charge(device_key, device_reservation); payloads.push(payload); // Keep this paired with `payloads`: dispatch admission failure // rolls back exactly the rate-limit increments for admitted work. - admitted_rate_limit_keys.push(( - encrypted_key, - encrypted_reservation, - device_key, - device_reservation, - )); + admitted_rate_limit_keys.push(guard.commit()); } if payloads.is_empty() { @@ -963,14 +1161,12 @@ impl EventProcessor { Ok(ProcessOutcome::Admitted) } Err(e) => { - for (encrypted_key, encrypted_reservation, device_key, device_reservation) in - &admitted_rate_limit_keys - { + for charges in &admitted_rate_limit_keys { self.encrypted_token_limiter - .rollback_increment(encrypted_key, *encrypted_reservation) + .rollback_increment(&charges.encrypted_key, charges.encrypted_reservation) .await; self.device_token_limiter - .rollback_increment(device_key, *device_reservation) + .rollback_increment(&charges.device_key, charges.device_reservation) .await; } @@ -986,6 +1182,20 @@ impl EventProcessor { } } + /// Publish a sampled live rate-limit cache size to the gauge (#125). + /// + /// `check_and_increment` returns `Some(len)` on the sampled fraction of + /// admissions and `None` otherwise, so this refreshes the + /// `transponder_rate_limit_cache_size` gauge as the cache grows toward + /// capacity — making saturation onset visible without waiting for the 60s + /// cleanup tick — while keeping the metric's existing per-`cache_type` + /// label shape and paying a gauge write only on sampled calls. + fn publish_rate_limit_gauge(&self, cache_type: &str, sampled_len: Option) { + if let (Some(len), Some(m)) = (sampled_len, self.metrics.as_ref()) { + m.set_rate_limit_cache_size(cache_type, len); + } + } + /// Hash arbitrary bytes to a fixed-size key for rate limiting. fn hash_bytes(data: &[u8]) -> [u8; 32] { let mut hasher = Sha256::new(); @@ -1067,10 +1277,14 @@ impl EventProcessor { let now = Instant::now(); { let mut seen = self.seen_events.write().await; - seen.put(event_id, SeenEvent::terminal(now)); - - // Update cache size metric - if let Some(ref m) = self.metrics { + // Refresh the existing reservation in place. On the success hot path + // the ID is already present from `try_reserve`, so the size does not + // change and the redundant second `dedup_cache_size` gauge write the + // old double-lock path carried is skipped (#197). The gauge is only + // touched on the rare path where the entry was evicted between + // reservation and completion and had to be re-inserted. + let size_changed = seen.mark_terminal(event_id, now); + if size_changed && let Some(ref m) = self.metrics { m.set_dedup_cache_size(seen.len()); } } @@ -2516,7 +2730,23 @@ mod tests { secp256k1::SecretKey::from_slice(&server_keys.secret_key().to_secret_bytes()) .expect("valid secret key"); let token_decryptor = TokenDecryptor::new(&mut secp_secret_key); - let push_dispatcher = Arc::new(PushDispatcher::new(None, None)); + // Wire a configured APNs mock so APNs tokens are genuinely dispatchable. + // With no client the #177 pre-charge filter would (correctly) drop and + // refund every APNs token as "unconfigured", so the device/encrypted + // rate-limit paths these helpers exercise would never actually charge. + let apns_config = ApnsConfig { + enabled: true, + key_id: "KEY123".to_string(), + team_id: "TEAM456".to_string(), + private_key_path: String::new(), + environment: crate::config::ApnsEnvironment::Sandbox, + bundle_id: "com.example.app".to_string(), + payload_mode: Default::default(), + }; + let push_dispatcher = Arc::new(PushDispatcher::new( + Some(ApnsClient::mock(apns_config, true)), + None, + )); EventProcessor::with_full_config( nip59_handler, token_decryptor, @@ -2731,6 +2961,439 @@ mod tests { assert!(result4.unwrap(), "Different encrypted token should work"); } + #[tokio::test] + async fn test_device_reject_refunds_encrypted_charge() { + // #170: when the device-token limiter rejects a token, the encrypted + // charge already spent for that same token must be rolled back, so a + // legitimate transient redelivery does not read as a replay. + use crate::test_vectors::{ + GiftWrapBuilder, NotificationContentBuilder, TestToken, TokenEncryptor, + }; + use base64::prelude::*; + + let server_keys = Keys::generate(); + let sender_keys = Keys::generate(); + let device_token = "aabbccdd11223344aabbccdd11223344aabbccdd11223344aabbccdd11223344"; + + // Device limiter allows 1/min; encrypted allows many. The device token + // is the same across events (same device), encrypted blobs differ. + let processor = create_processor_with_rate_limits( + &server_keys, + TokenRateLimitConfig { + max_cache_size: NonZeroUsize::new(100).unwrap(), + max_tokens_per_event: NonZeroUsize::new(DEFAULT_MAX_TOKENS_PER_EVENT).unwrap(), + encrypted_token_per_minute: 100, + encrypted_token_per_hour: 1000, + device_token_per_minute: 1, + device_token_per_hour: 100, + global_unwrap_per_minute: 1000, + global_unwrap_per_hour: 10000, + }, + ); + + // First event: admitted, consumes the single device slot. + let event1 = + scenarios::single_apns_notification(&server_keys, &sender_keys, device_token).await; + assert!(processor.process(&event1).await.unwrap()); + + // Second event: a fresh encrypted blob for the SAME device. The device + // limiter is now at its 1/min cap and rejects the token. + let encryptor = TokenEncryptor::from_keys(&server_keys); + let encrypted = encryptor.encrypt(&TestToken::apns(device_token)); + let encrypted_key = EventProcessor::hash_bytes(&encrypted); + let content = NotificationContentBuilder::new(&server_keys) + .with_raw_token(BASE64_STANDARD.encode(&encrypted)) + .build(); + let event2 = GiftWrapBuilder::new(server_keys.clone(), sender_keys.clone()) + .build(&content) + .await; + + assert!(!processor.process(&event2).await.unwrap()); + + // The encrypted charge for event2's blob was refunded on the device + // reject: its key holds no residual hit (the entry is removed once both + // counters reach zero). + assert_eq!( + processor + .encrypted_token_limiter + .peek_counts(&encrypted_key) + .await, + None, + "device reject must refund the encrypted-token charge (#170)" + ); + } + + #[tokio::test] + async fn test_unconfigured_platform_token_refunds_and_counts_drop() { + // #177: a decrypted token whose platform has no configured client must + // be dropped BEFORE it can strand rate-limit budget, and the drop must + // be visible via a real metric. + use crate::push::FcmClient; + use crate::test_vectors::{ + GiftWrapBuilder, NotificationContentBuilder, TestToken, TokenEncryptor, + }; + use base64::prelude::*; + + let server_keys = Keys::generate(); + let sender_keys = Keys::generate(); + + // FCM-only dispatcher, but the event carries an APNs token → unconfigured. + let nip59_handler = Nip59Handler::new(server_keys.clone()); + let mut secp_secret_key = + secp256k1::SecretKey::from_slice(&server_keys.secret_key().to_secret_bytes()) + .expect("valid secret key"); + let token_decryptor = TokenDecryptor::new(&mut secp_secret_key); + let fcm_config = crate::config::FcmConfig { + enabled: true, + service_account_path: String::new(), + project_id: "test-project".to_string(), + }; + let metrics = Metrics::new().expect("metrics"); + let push_dispatcher = Arc::new(PushDispatcher::with_metrics( + None, + Some(FcmClient::mock(fcm_config, true)), + Some(metrics.clone()), + )); + let processor = EventProcessor::with_full_config( + nip59_handler, + token_decryptor, + push_dispatcher, + DEFAULT_MAX_DEDUP_CACHE_SIZE, + TokenRateLimitConfig::default(), + Some(metrics.clone()), + ); + + let encryptor = TokenEncryptor::from_keys(&server_keys); + let encrypted = encryptor.encrypt(&TestToken::apns( + "aabbccdd11223344aabbccdd11223344aabbccdd11223344aabbccdd11223344", + )); + let encrypted_key = EventProcessor::hash_bytes(&encrypted); + let content = NotificationContentBuilder::new(&server_keys) + .with_raw_token(BASE64_STANDARD.encode(&encrypted)) + .build(); + let event = GiftWrapBuilder::new(server_keys.clone(), sender_keys) + .build(&content) + .await; + + // The APNs token is dropped as unconfigured; the event carried nothing + // dispatchable (terminal, not a retryable rate shed). + assert!(processor.process(&event).await.unwrap()); + + // The encrypted charge was refunded — no stranded budget (#177). + assert_eq!( + processor + .encrypted_token_limiter + .peek_counts(&encrypted_key) + .await, + None, + "unconfigured-platform drop must refund the encrypted charge" + ); + // The device limiter was never charged (drop happens before it). + assert_eq!(processor.device_token_limiter.len().await, 0); + // The drop is recorded by a real metric. + assert_eq!( + counter_value( + &metrics, + "transponder_push_failed_total", + &[("platform", "apns"), ("reason", "unconfigured")], + ), + 1.0, + "unconfigured-platform drop must be counted (#177)" + ); + } + + #[tokio::test] + async fn test_non_utf8_fcm_token_refunds_and_counts_drop() { + // #177: an FCM token whose device bytes are not UTF-8 is undeliverable. + // It must be dropped before charging the device limiter, its encrypted + // charge refunded, and the drop counted as invalid_encoding. + use crate::crypto::token::{ENCRYPTED_TOKEN_SIZE, PLATFORM_FCM}; + use crate::push::FcmClient; + use crate::test_vectors::{ + GiftWrapBuilder, NotificationContentBuilder, TestToken, TokenEncryptor, + }; + use base64::prelude::*; + + let server_keys = Keys::generate(); + let sender_keys = Keys::generate(); + + let fcm_config = crate::config::FcmConfig { + enabled: true, + service_account_path: String::new(), + project_id: "test-project".to_string(), + }; + let metrics = Metrics::new().expect("metrics"); + let nip59_handler = Nip59Handler::new(server_keys.clone()); + let mut secp_secret_key = + secp256k1::SecretKey::from_slice(&server_keys.secret_key().to_secret_bytes()) + .expect("valid secret key"); + let token_decryptor = TokenDecryptor::new(&mut secp_secret_key); + let push_dispatcher = Arc::new(PushDispatcher::with_metrics( + None, + Some(FcmClient::mock(fcm_config, true)), + Some(metrics.clone()), + )); + let processor = EventProcessor::with_full_config( + nip59_handler, + token_decryptor, + push_dispatcher, + DEFAULT_MAX_DEDUP_CACHE_SIZE, + TokenRateLimitConfig::default(), + Some(metrics.clone()), + ); + + // Build an FCM token whose device bytes are not valid UTF-8. + let encryptor = TokenEncryptor::from_keys(&server_keys); + let non_utf8_fcm = TestToken { + platform: PLATFORM_FCM, + device_token: vec![0xff, 0xfe, 0x00, 0x01], + }; + let encrypted = encryptor.encrypt(&non_utf8_fcm); + assert_eq!(encrypted.len(), ENCRYPTED_TOKEN_SIZE); + let encrypted_key = EventProcessor::hash_bytes(&encrypted); + let content = NotificationContentBuilder::new(&server_keys) + .with_raw_token(BASE64_STANDARD.encode(&encrypted)) + .build(); + let event = GiftWrapBuilder::new(server_keys.clone(), sender_keys) + .build(&content) + .await; + + assert!(processor.process(&event).await.unwrap()); + + assert_eq!( + processor + .encrypted_token_limiter + .peek_counts(&encrypted_key) + .await, + None, + "non-UTF-8 FCM drop must refund the encrypted charge" + ); + assert_eq!(processor.device_token_limiter.len().await, 0); + assert_eq!( + counter_value( + &metrics, + "transponder_push_failed_total", + &[("platform", "fcm"), ("reason", "invalid_encoding")], + ), + 1.0, + "non-UTF-8 FCM drop must be counted as invalid_encoding (#177)" + ); + } + + #[tokio::test] + async fn test_decrypt_failure_keeps_encrypted_charge() { + // #170 note: the decrypt-failure path intentionally KEEPS the encrypted + // charge (invalid blobs still spend replay/spam budget). Guard against a + // regression from the refund refactor. + use crate::test_vectors::{GiftWrapBuilder, NotificationContentBuilder}; + use base64::prelude::*; + + let server_keys = Keys::generate(); + let sender_keys = Keys::generate(); + + // A correctly-sized but undecryptable blob (random bytes). + let junk = vec![0x11u8; crate::crypto::token::ENCRYPTED_TOKEN_SIZE]; + let encrypted_key = EventProcessor::hash_bytes(&junk); + let content = NotificationContentBuilder::new(&server_keys) + .with_raw_token(BASE64_STANDARD.encode(&junk)) + .build(); + let event = GiftWrapBuilder::new(server_keys.clone(), sender_keys) + .build(&content) + .await; + + let processor = create_processor_with_rate_limits( + &server_keys, + TokenRateLimitConfig { + max_cache_size: NonZeroUsize::new(100).unwrap(), + max_tokens_per_event: NonZeroUsize::new(DEFAULT_MAX_TOKENS_PER_EVENT).unwrap(), + encrypted_token_per_minute: 100, + encrypted_token_per_hour: 1000, + device_token_per_minute: 100, + device_token_per_hour: 1000, + global_unwrap_per_minute: 1000, + global_unwrap_per_hour: 10000, + }, + ); + + assert!(processor.process(&event).await.unwrap()); + + // The encrypted charge is retained: the blob spent its replay budget. + assert_eq!( + processor + .encrypted_token_limiter + .peek_counts(&encrypted_key) + .await, + Some((1, 1)), + "decrypt-failure path must keep the encrypted charge (#170 note)" + ); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn test_concurrent_device_rejects_do_not_strand_encrypted_charges() { + // Concurrency regression for the #170 guard: many distinct-blob events + // for the SAME device are processed in parallel. The device limiter + // admits only a few; the rest are device-rejected and must each refund + // their (distinct) encrypted charge. Afterwards the encrypted limiter + // must hold no more residual charges than the device limiter admitted — + // no interleaving may strand an encrypted increment. + use crate::test_vectors::{ + GiftWrapBuilder, NotificationContentBuilder, TestToken, TokenEncryptor, + }; + use base64::prelude::*; + + let server_keys = Keys::generate(); + let sender_keys = Keys::generate(); + let device_token = "aabbccdd11223344aabbccdd11223344aabbccdd11223344aabbccdd11223344"; + + const DEVICE_LIMIT: u32 = 3; + let processor = Arc::new(create_processor_with_rate_limits( + &server_keys, + TokenRateLimitConfig { + max_cache_size: NonZeroUsize::new(1000).unwrap(), + max_tokens_per_event: NonZeroUsize::new(DEFAULT_MAX_TOKENS_PER_EVENT).unwrap(), + encrypted_token_per_minute: 1000, + encrypted_token_per_hour: 10000, + device_token_per_minute: DEVICE_LIMIT, + device_token_per_hour: 1000, + global_unwrap_per_minute: 100000, + global_unwrap_per_hour: 1000000, + }, + )); + + // Build many events, each with a distinct encrypted blob for the same + // device token, and track each blob's encrypted-limiter key. + let encryptor = TokenEncryptor::from_keys(&server_keys); + const EVENTS: usize = 40; + let mut events = Vec::with_capacity(EVENTS); + let mut encrypted_keys = Vec::with_capacity(EVENTS); + for _ in 0..EVENTS { + let encrypted = encryptor.encrypt(&TestToken::apns(device_token)); + encrypted_keys.push(EventProcessor::hash_bytes(&encrypted)); + let content = NotificationContentBuilder::new(&server_keys) + .with_raw_token(BASE64_STANDARD.encode(&encrypted)) + .build(); + events.push( + GiftWrapBuilder::new(server_keys.clone(), sender_keys.clone()) + .build(&content) + .await, + ); + } + + let mut handles = Vec::with_capacity(EVENTS); + for event in events { + let processor = Arc::clone(&processor); + handles.push(tokio::spawn(async move { processor.process(&event).await })); + } + for handle in handles { + let _ = handle.await.expect("task panicked"); + } + + // Count encrypted charges that remain across every distinct blob. + let mut residual_encrypted = 0usize; + for key in &encrypted_keys { + if processor + .encrypted_token_limiter + .peek_counts(key) + .await + .is_some() + { + residual_encrypted += 1; + } + } + + // The device limiter admitted at most DEVICE_LIMIT tokens; every other + // blob's encrypted charge was refunded on its device reject. Anything + // above DEVICE_LIMIT would be a stranded charge. + assert!( + residual_encrypted <= DEVICE_LIMIT as usize, + "stranded encrypted charges: {residual_encrypted} > device limit {DEVICE_LIMIT}" + ); + // All admitted tokens share one device key (same device token), so the + // device limiter holds a single entry whose hit count equals the number + // of admitted tokens. Residual encrypted charges must equal exactly that + // — one retained encrypted charge per admitted (device-charged) token, + // none stranded. + let device_payload = TokenPayload { + platform: Platform::Apns, + device_token: hex::decode(device_token).expect("valid hex"), + }; + let device_key = EventProcessor::hash_device_token_key(&device_payload); + let device_hits = processor + .device_token_limiter + .peek_counts(&device_key) + .await + .map(|(minute, _)| minute as usize) + .unwrap_or(0); + assert_eq!( + residual_encrypted, device_hits, + "residual encrypted charges must match device-admitted hit count exactly" + ); + } + + #[tokio::test] + async fn test_live_rate_limit_cache_size_gauge_updates_before_cleanup() { + // #125: the rate-limit cache-size gauge must reflect growth on the + // admission path, not only at the 60s cleanup tick. The first admission + // is sampled, so the gauge is non-zero after processing without any + // cleanup call. + let server_keys = Keys::generate(); + let sender_keys = Keys::generate(); + let device_token = "aabbccdd11223344aabbccdd11223344aabbccdd11223344aabbccdd11223344"; + + let event = + scenarios::single_apns_notification(&server_keys, &sender_keys, device_token).await; + let (processor, metrics) = create_processor_with_metrics(&server_keys); + + assert!(processor.process(&event).await.unwrap()); + + // No cleanup() has run; the gauge was updated opportunistically on the + // sampled admission. + assert_eq!( + metric_gauge_value( + &metrics, + "transponder_rate_limit_cache_size", + &[("type", "encrypted_token")], + ), + 1.0, + "encrypted-token cache-size gauge must update on admission (#125)" + ); + assert_eq!( + metric_gauge_value( + &metrics, + "transponder_rate_limit_cache_size", + &[("type", "device_token")], + ), + 1.0, + "device-token cache-size gauge must update on admission (#125)" + ); + } + + #[tokio::test] + async fn test_successful_path_updates_dedup_gauge_once() { + // #197: the successful path folds the completion refresh so it does not + // redundantly re-write the dedup cache-size gauge. After processing one + // fresh event the gauge reads exactly 1 (reservation), and mark_seen's + // in-place terminal refresh does not change it. + let server_keys = Keys::generate(); + let sender_keys = Keys::generate(); + let device_token = "aabbccdd11223344aabbccdd11223344aabbccdd11223344aabbccdd11223344"; + + let event = + scenarios::single_apns_notification(&server_keys, &sender_keys, device_token).await; + let (processor, metrics) = create_processor_with_metrics(&server_keys); + + assert!(processor.process(&event).await.unwrap()); + + assert_eq!( + gauge_value(&metrics, "transponder_dedup_cache_size"), + 1.0, + "dedup gauge reflects the single retained terminal entry" + ); + assert!(processor.is_duplicate(&event.id).await); + // The completion refresh kept the entry terminal (a replay dedups). + assert!(!processor.process(&event).await.unwrap()); + } + #[tokio::test] async fn test_rate_limit_window_reset() { use crate::test_vectors::{ From efaf5291ea2c2f325fe3e6fb30e9665be97de251 Mon Sep 17 00:00:00 2001 From: Jeff Gardner <202880+erskingardner@users.noreply.github.com> Date: Sun, 5 Jul 2026 14:41:58 +0100 Subject: [PATCH 4/4] test: assert pre-filter drop leaves delivery-health untouched MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reconciliation regression for the merge with the push-provider rework (#233), which added DeliveryHealth streak state feeding /ready. A token dropped by the #177 pre-filter in process_inner never reaches the send path, so it is not a delivery attempt and must not touch a provider's DeliveryHealth streak. This test seeds a real FCM hard-failure streak just below the flagging threshold, processes an unconfigured-APNs drop, and asserts APNs stays delivering (no spurious streak increment) and FCM's seeded streak survives (one more failure flips it), proving the pre-filter neither increments nor resets the streak — no double-count, no missed update. Part of #218. Co-Authored-By: Claude Fable 5 --- src/nostr/events.rs | 100 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 100 insertions(+) diff --git a/src/nostr/events.rs b/src/nostr/events.rs index 4f58d22..9558b38 100644 --- a/src/nostr/events.rs +++ b/src/nostr/events.rs @@ -3102,6 +3102,106 @@ mod tests { ); } + #[tokio::test] + async fn test_prefiltered_drop_does_not_touch_delivery_health() { + // Reconciliation invariant with the push-provider rework (#233): a token + // dropped by the #177 pre-filter in process_inner never reaches the send + // path, so it is NOT a delivery attempt and must not touch the provider's + // DeliveryHealth streak (which gates /ready). No streak increment (would + // wrongly flag a healthy provider) and no reset (would wrongly clear a + // real failure streak). + use crate::push::FcmClient; + use crate::test_vectors::{ + GiftWrapBuilder, NotificationContentBuilder, TestToken, TokenEncryptor, + }; + use base64::prelude::*; + + let server_keys = Keys::generate(); + let sender_keys = Keys::generate(); + + // FCM-only dispatcher; the event carries an APNs token → pre-filtered as + // unconfigured before it can reach send_push. + let nip59_handler = Nip59Handler::new(server_keys.clone()); + let mut secp_secret_key = + secp256k1::SecretKey::from_slice(&server_keys.secret_key().to_secret_bytes()) + .expect("valid secret key"); + let token_decryptor = TokenDecryptor::new(&mut secp_secret_key); + let fcm_config = crate::config::FcmConfig { + enabled: true, + service_account_path: String::new(), + project_id: "test-project".to_string(), + }; + let metrics = Metrics::new().expect("metrics"); + let push_dispatcher = Arc::new(PushDispatcher::with_metrics( + None, + Some(FcmClient::mock(fcm_config, true)), + Some(metrics.clone()), + )); + // Retain a handle to inspect DeliveryHealth after processing. + let dispatcher_handle = Arc::clone(&push_dispatcher); + let processor = EventProcessor::with_full_config( + nip59_handler, + token_decryptor, + push_dispatcher, + DEFAULT_MAX_DEDUP_CACHE_SIZE, + TokenRateLimitConfig::default(), + Some(metrics.clone()), + ); + + // Seed a real hard-failure streak on FCM just below the flagging + // threshold (a genuine prior outage), so we can prove a pre-filtered APNs + // drop neither resets FCM's streak nor touches APNs's. + use crate::push::dispatcher::DELIVERY_FAILURE_STREAK_THRESHOLD; + for _ in 0..DELIVERY_FAILURE_STREAK_THRESHOLD - 1 { + dispatcher_handle + .delivery_health() + .record_hard_failure(Platform::Fcm); + } + + let encryptor = TokenEncryptor::from_keys(&server_keys); + let encrypted = encryptor.encrypt(&TestToken::apns( + "aabbccdd11223344aabbccdd11223344aabbccdd11223344aabbccdd11223344", + )); + let content = NotificationContentBuilder::new(&server_keys) + .with_raw_token(BASE64_STANDARD.encode(&encrypted)) + .build(); + let event = GiftWrapBuilder::new(server_keys.clone(), sender_keys) + .build(&content) + .await; + + assert!(processor.process(&event).await.unwrap()); + + // APNs was never a delivery attempt: its streak is untouched (still 0 → + // delivering), so the pre-filter did not spuriously flag it. + assert!( + dispatcher_handle.is_apns_delivering(), + "a pre-filtered APNs drop must not increment APNs delivery-health streak" + ); + // FCM's genuine prior streak was NOT reset by the unrelated APNs drop. + assert_eq!( + counter_value( + &metrics, + "transponder_push_failed_total", + &[("platform", "apns"), ("reason", "unconfigured")], + ), + 1.0 + ); + assert!( + dispatcher_handle.is_fcm_delivering(), + "streak just below threshold is still delivering" + ); + // One more FCM hard failure reaches the threshold — which only holds if + // the seeded streak survived, proving the APNs pre-filter drop did not + // silently reset FCM's real hard-failure streak. + dispatcher_handle + .delivery_health() + .record_hard_failure(Platform::Fcm); + assert!( + !dispatcher_handle.is_fcm_delivering(), + "pre-filter drop must not have reset FCM's real hard-failure streak" + ); + } + #[tokio::test] async fn test_non_utf8_fcm_token_refunds_and_counts_drop() { // #177: an FCM token whose device bytes are not UTF-8 is undeliverable.