diff --git a/autumn/src/security/rate_limit.rs b/autumn/src/security/rate_limit.rs index 0fa1107d0..f11a28356 100644 --- a/autumn/src/security/rate_limit.rs +++ b/autumn/src/security/rate_limit.rs @@ -189,7 +189,7 @@ impl MemoryStore { .saturating_duration_since(bucket.last_refill) .as_secs_f64(); bucket.tokens = elapsed.mul_add(refill_per_sec, bucket.tokens).min(burst); - bucket.last_refill = now; + bucket.last_refill = std::cmp::max(bucket.last_refill, now); let result = if bucket.tokens >= 1.0 { bucket.tokens -= 1.0; diff --git a/autumn/tests/integration/chaos_channels_snapshot_loom.rs b/autumn/tests/integration/chaos_channels_snapshot_loom.rs new file mode 100644 index 000000000..ccf77a774 --- /dev/null +++ b/autumn/tests/integration/chaos_channels_snapshot_loom.rs @@ -0,0 +1,25 @@ +#![cfg(feature = "ws")] +use autumn_web::channels::Channels; +use loom::thread; + +#[test] +fn channels_concurrent_snapshot_and_publish() { + loom::model(|| { + let channels = Channels::new(32); + let tx1 = channels.sender("test_gc"); + let tx2 = tx1.clone(); + + let c1 = channels.clone(); + let t1 = thread::spawn(move || { + let snap = c1.snapshot(); + let _ = snap.get("test_gc").map(|s| s.subscriber_count); + }); + + let t2 = thread::spawn(move || { + tx2.send("hello").ok(); + }); + + let _ = t1.join().unwrap(); + t2.join().unwrap(); + }); +} diff --git a/autumn/tests/integration/mod.rs b/autumn/tests/integration/mod.rs index c3c5967a7..cf0be33bb 100644 --- a/autumn/tests/integration/mod.rs +++ b/autumn/tests/integration/mod.rs @@ -25,6 +25,7 @@ mod chaos_channels_concurrent_loom; mod chaos_channels_loom; #[cfg(feature = "ws")] mod chaos_channels_proptest; +mod chaos_channels_snapshot_loom; #[cfg(feature = "ws")] mod chaos_channels_subscribe_loom; mod chaos_job_client_loom; diff --git a/autumn/tests/loom_models.rs b/autumn/tests/loom_models.rs index 8273f63ef..60838927b 100644 --- a/autumn/tests/loom_models.rs +++ b/autumn/tests/loom_models.rs @@ -261,6 +261,83 @@ fn rate_limit_bucket_single_winner() { }); } +/// Models the `MemoryStore::decide` rate limit bucket time-warp race condition. +/// +/// In `MemoryStore::decide`, `now = Instant::now()` is called OUTSIDE the bucket +/// lock. Thread A can capture a very early `now` and stall. Thread B captures a later +/// `now`, runs `decide`, consumes tokens, and updates `last_refill` to its later `now`. +/// Thread A then wakes up, locks the bucket, and applies its EARLY `now`. +/// +/// `bucket.tokens = elapsed.mul_add(refill_per_sec, bucket.tokens).min(burst);` +/// (elapsed is 0 because `saturating_duration_since` on a future `last_refill` is 0) +/// +/// Then thread A does `bucket.last_refill = now` (the EARLY `now`). +/// This clobbers thread B's later `last_refill` with an older timestamp. +/// The next request will observe an artificially long `elapsed` duration because the +/// `last_refill` was time-warped backwards, granting full tokens immediately and breaking +/// the rate limit envelope. +#[test] +fn rate_limit_time_warp() { + let show_bug = std::env::var_os("RATE_LIMIT_LOOM_SHOW_BUG").is_some(); + loom::model(move || { + // Models `Instant` using an integer + let bucket_tokens = Arc::new(Mutex::new(0.0)); + let bucket_last_refill = Arc::new(Mutex::new(0)); + + let t1 = { + let tokens = bucket_tokens.clone(); + let last_refill = bucket_last_refill.clone(); + thread::spawn(move || { + let now = 10; + let mut guard_t = tokens.lock().unwrap(); + let mut guard_l = last_refill.lock().unwrap(); + + // simulate elapsed logic + let elapsed = if now > *guard_l { now - *guard_l } else { 0 }; + *guard_t = (*guard_t + elapsed as f64).min(5.0); + + if show_bug { + // BUGGY: update last_refill to our captured `now` regardless of what it was + *guard_l = now; + } else { + // FIXED: update last_refill to `now` ONLY IF `now` is >= current last_refill + *guard_l = std::cmp::max(*guard_l, now); + } + }) + }; + + let t2 = { + let tokens = bucket_tokens.clone(); + let last_refill = bucket_last_refill.clone(); + thread::spawn(move || { + let now = 20; // happens later + let mut guard_t = tokens.lock().unwrap(); + let mut guard_l = last_refill.lock().unwrap(); + + let elapsed = if now > *guard_l { now - *guard_l } else { 0 }; + *guard_t = (*guard_t + elapsed as f64).min(5.0); + + if show_bug { + *guard_l = now; + } else { + *guard_l = std::cmp::max(*guard_l, now); + } + }) + }; + + t1.join().unwrap(); + t2.join().unwrap(); + + // The time should not go backwards. Thread 2 happened at T=20, so the max observed time + // must be retained to prevent time warping backwards to T=10 if Thread 1 won the lock last. + assert_eq!( + *bucket_last_refill.lock().unwrap(), + 20, + "last_refill must not travel backwards" + ); + }); +} + /// Models the exactly-once `requests_active` decrement across the /// `MetricsFuture` completion race. ///