👾 Havoc: Rate limit double-spend via time-warp race

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;

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();
+    });
+}

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;

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.
 ///