cratefield_runtime_native/ports/
rate_limit.rs

1//! `RateLimiter` over Redis: a true sliding window (sorted set + one Lua
2//! script), the honest differences from the Workers Rate Limiting
3//! binding documented on [`RedisRateLimiter`].
4
5use std::sync::atomic::{AtomicU64, Ordering};
6use std::time::Duration;
7
8use async_trait::async_trait;
9use cratefield_core::{Config, Decision, RateLimitError, RateLimiter};
10use redis::aio::ConnectionManager;
11
12/// Default `RATE_LIMIT_MAX`: requests allowed per key per window.
13const DEFAULT_LIMIT: u64 = 100;
14
15/// Default `RATE_LIMIT_PERIOD_SECS`.
16const DEFAULT_PERIOD_SECS: u64 = 60;
17
18/// One atomic sliding-window decision. KEYS[1] is the key; ARGV holds
19/// `now_ms`, `window_ms`, the limit and a caller-supplied uniqueness
20/// suffix for the member id. Returns `{allowed, retry_after_ms}`.
21///
22/// Atomicity matters: check-then-add outside a script would let two
23/// concurrent requests both pass the same full window.
24const SLIDING_WINDOW: &str = r"
25redis.call('ZREMRANGEBYSCORE', KEYS[1], '-inf', tonumber(ARGV[1]) - tonumber(ARGV[2]))
26local count = redis.call('ZCARD', KEYS[1])
27if count < tonumber(ARGV[3]) then
28  redis.call('ZADD', KEYS[1], ARGV[1], ARGV[1] .. ':' .. ARGV[4])
29  redis.call('PEXPIRE', KEYS[1], ARGV[2])
30  return {1, 0}
31end
32local oldest = redis.call('ZRANGE', KEYS[1], 0, 0, 'WITHSCORES')
33local retry = tonumber(ARGV[2])
34if oldest[2] then
35  retry = tonumber(oldest[2]) + tonumber(ARGV[2]) - tonumber(ARGV[1])
36end
37redis.call('PEXPIRE', KEYS[1], ARGV[2])
38return {0, math.max(retry, 0)}
39";
40
41/// `RateLimiter` over Redis: each allowed request ZADDs `now` into the
42/// key's sorted set; a request is allowed while the set holds fewer than
43/// `limit` members within the trailing `window`.
44///
45/// # Differences from the Workers Rate Limiting binding (documented,
46/// not papered over)
47///
48/// | | Workers Rate Limiting binding | `RedisRateLimiter` |
49/// |---|---|---|
50/// | Algorithm | fixed-window counter (limit + period per binding, configured in `wrangler.toml`) | sliding window: a request at the window's edge does not get a fresh full budget immediately |
51/// | `retry_after` | the binding exposes none — the port's `Decision::retry_after` is always `None` on Workers | `Some`: milliseconds until the oldest counted request exits the window, so callers can send an honest `Retry-After` |
52/// | Where the limit lives | the binding declaration (per Worker) | config keys `RATE_LIMIT_MAX` / `RATE_LIMIT_PERIOD_SECS`, one budget shared by every module — the same "one venture, one budget per subject" the shared binding gives |
53/// | Scope | counters are per-Cloudflare-edge | one Redis: exact across every process and replica that shares it, and a single point of failure |
54/// | Key space | the binding's own | `rl:`-prefixed in the database shared with [`RedisKv`](crate::RedisKv) |
55///
56/// `Duration`-granularity windows are kept whole: the script works in
57/// milliseconds.
58pub struct RedisRateLimiter {
59    conn: ConnectionManager,
60    script: redis::Script,
61    limit: u64,
62    window_ms: u64,
63    nonce: AtomicU64,
64}
65
66impl RedisRateLimiter {
67    pub fn new(conn: ConnectionManager, limit: u64, window: Duration) -> Self {
68        Self {
69            conn,
70            script: redis::Script::new(SLIDING_WINDOW),
71            limit,
72            window_ms: u64::try_from(window.as_millis()).unwrap_or(1).max(1),
73            nonce: AtomicU64::new(0),
74        }
75    }
76
77    /// `RATE_LIMIT_MAX` (default `100`) and `RATE_LIMIT_PERIOD_SECS`
78    /// (default `60`) from config.
79    #[must_use]
80    pub fn from_env(conn: ConnectionManager, config: &dyn Config) -> Self {
81        let limit = config
82            .get("RATE_LIMIT_MAX")
83            .and_then(|raw| raw.parse().ok())
84            .unwrap_or(DEFAULT_LIMIT);
85        let period_secs = config
86            .get("RATE_LIMIT_PERIOD_SECS")
87            .and_then(|raw| raw.parse().ok())
88            .unwrap_or(DEFAULT_PERIOD_SECS);
89        Self::new(conn, limit, Duration::from_secs(period_secs))
90    }
91}
92
93fn now_ms() -> u64 {
94    u64::try_from(time::OffsetDateTime::now_utc().unix_timestamp_nanos() / 1_000_000).unwrap_or(0)
95}
96
97#[async_trait]
98impl RateLimiter for RedisRateLimiter {
99    async fn limit(&self, key: &str) -> Result<Decision, RateLimitError> {
100        let mut conn = self.conn.clone();
101        let timestamp_ms = now_ms();
102        // Member ids must be unique per counted request; the nonce is
103        // process-local, the timestamp bounds collisions across
104        // processes to the same millisecond.
105        let member_nonce = self.nonce.fetch_add(1, Ordering::Relaxed);
106        let outcome: Vec<i64> = self
107            .script
108            .key(format!("rl:{key}"))
109            .arg(timestamp_ms)
110            .arg(self.window_ms)
111            .arg(self.limit)
112            .arg(member_nonce)
113            .invoke_async(&mut conn)
114            .await
115            .map_err(|err| RateLimitError::Transport(err.to_string()))?;
116        let (allowed, retry_ms) = (
117            outcome.first().copied().unwrap_or(0),
118            outcome.get(1).copied().unwrap_or(0),
119        );
120        Ok(Decision {
121            ok: allowed == 1,
122            retry_after: if retry_ms > 0 {
123                Some(Duration::from_millis(retry_ms.unsigned_abs()))
124            } else {
125                None
126            },
127            // The script reports the verdict and the retry pause; the
128            // window's membership count is not returned, so no quota.
129            quota: None,
130        })
131    }
132}
cratefield_runtime_native/ports/rate_limit.rs

cratefield_runtime_native/ports/
rate_limit.rs
