1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
//! Rate limiter enum dispatch — routes calls to the active backend.
//!
//! `RateLimiter` is the public handle used by the rest of the server.
//! It wraps either the in-memory or the Redis backend behind a uniform
//! async API so callers never need to know which backend is active.
#[cfg(feature = "redis-rate-limiting")]
use super::redis::RedisRateLimiter;
use super::{
config::{CheckResult, RateLimitConfig, RateLimitingSecurityConfig},
in_memory::InMemoryRateLimiter,
};
/// Rate limiter that dispatches to either an in-memory or Redis backend.
///
/// Construct via [`RateLimiter::new`] (in-memory, default) or
/// `RateLimiter::new_redis` (distributed Redis, requires the
/// `redis-rate-limiting` Cargo feature).
#[non_exhaustive]
pub enum RateLimiter {
/// Single-node token-bucket limiter backed by `HashMap` with `RwLock`.
InMemory(InMemoryRateLimiter),
/// Distributed token-bucket limiter backed by Redis Lua scripts.
#[cfg(feature = "redis-rate-limiting")]
Redis(RedisRateLimiter),
}
impl RateLimiter {
/// Whether this limiter's budget is shared across replicas (#874).
///
/// The in-memory backend tracks budgets per process, so a deployment of N
/// replicas enforces N times the configured rate.
#[must_use]
pub const fn is_distributed(&self) -> bool {
match self {
Self::InMemory(_) => false,
#[cfg(feature = "redis-rate-limiting")]
Self::Redis(_) => true,
}
}
/// Create an in-memory rate limiter.
#[must_use]
pub fn new(config: RateLimitConfig) -> Self {
Self::InMemory(InMemoryRateLimiter::new(config))
}
/// Create a Redis-backed distributed rate limiter.
///
/// # Errors
///
/// Returns an error if the Redis URL is invalid or the initial connection
/// attempt fails.
#[cfg(feature = "redis-rate-limiting")]
pub async fn new_redis(url: &str, config: RateLimitConfig) -> Result<Self, redis::RedisError> {
let rl = RedisRateLimiter::new(url, config).await?;
Ok(Self::Redis(rl))
}
/// Attach per-path rules from `[security.rate_limiting]` auth endpoint fields.
#[must_use]
pub fn with_path_rules_from_security(self, sec: &RateLimitingSecurityConfig) -> Self {
match self {
Self::InMemory(rl) => Self::InMemory(rl.with_path_rules_from_security(sec)),
#[cfg(feature = "redis-rate-limiting")]
Self::Redis(rl) => Self::Redis(rl.with_path_rules_from_security(sec)),
}
}
/// Return the active rate limit configuration.
#[must_use]
pub const fn config(&self) -> &RateLimitConfig {
match self {
Self::InMemory(rl) => rl.config(),
#[cfg(feature = "redis-rate-limiting")]
Self::Redis(rl) => rl.config(),
}
}
/// Number of per-path rate limit rules registered.
#[must_use]
pub const fn path_rule_count(&self) -> usize {
match self {
Self::InMemory(rl) => rl.path_rule_count(),
#[cfg(feature = "redis-rate-limiting")]
Self::Redis(rl) => rl.path_rule_count(),
}
}
/// Seconds a client should wait before retrying after a per-path rate limit rejection.
///
/// Returns the window duration for the matching path rule (e.g. 60s for an
/// auth/start rule with 5 req/60s), not the IP token-bucket interval.
#[must_use]
pub fn retry_after_for_path(&self, path: &str) -> u32 {
match self {
Self::InMemory(rl) => rl.retry_after_for_path(path),
#[cfg(feature = "redis-rate-limiting")]
Self::Redis(rl) => rl.retry_after_for_path(path),
}
}
/// Check whether a request from `ip` is within the global IP rate limit.
///
/// The `tenant_id` parameter was removed in #1143. It was taken raw from an
/// `X-Tenant-ID` header and folded into the in-memory key, so a caller varying it
/// got a fresh full bucket per request — the limit did not limit. The Redis
/// backend never accepted it, so the two disagreed about what a bucket was.
#[doc(hidden)] // Internal-pub: invoked by rate_limit_middleware; downstream configures rate limiting via TOML, not by calling this directly.
pub async fn check_ip_limit(&self, ip: &str) -> CheckResult {
match self {
Self::InMemory(rl) => rl.check_ip_limit(ip).await,
#[cfg(feature = "redis-rate-limiting")]
Self::Redis(rl) => rl.check_ip_limit(ip).await,
}
}
/// Check whether a request from `user_id` is within the per-user limit.
///
/// `user_id` must be a **verified** identity — see
/// [`InMemoryRateLimiter::check_user_limit`]. The HTTP middleware no longer calls
/// this; gRPC does, after authenticating.
#[doc(hidden)] // Internal-pub: invoked by rate_limit_middleware; downstream configures rate limiting via TOML, not by calling this directly.
pub async fn check_user_limit(&self, user_id: &str) -> CheckResult {
match self {
Self::InMemory(rl) => rl.check_user_limit(user_id).await,
#[cfg(feature = "redis-rate-limiting")]
Self::Redis(rl) => rl.check_user_limit(user_id).await,
}
}
/// Check the per-path rate limit for a request from `ip` to `path`.
///
/// Returns an allowed [`CheckResult`] when no rule governs the path.
/// `CheckResult::retry_after_secs` reflects the actual per-path window, not
/// the global IP rate.
pub async fn check_path_limit(&self, path: &str, ip: &str) -> CheckResult {
match self {
Self::InMemory(rl) => rl.check_path_limit(path, ip).await,
#[cfg(feature = "redis-rate-limiting")]
Self::Redis(rl) => rl.check_path_limit(path, ip).await,
}
}
/// Check the per-tenant rate limit for a request to tenant `key`.
///
/// Each tenant gets its own token bucket with the specified `rps` and `burst`
/// from the tenant's quota configuration.
///
/// # Note
///
/// Redis backend does not support per-tenant limits (falls back to allow-all).
pub async fn check_tenant_limit(&self, tenant_key: &str, rps: u32, burst: u32) -> CheckResult {
match self {
Self::InMemory(rl) => rl.check_tenant_limit(tenant_key, rps, burst).await,
#[cfg(feature = "redis-rate-limiting")]
Self::Redis(_) => CheckResult::allow(f64::from(burst)),
}
}
/// Evict stale in-memory buckets.
///
/// No-op for the Redis backend — Redis handles expiry via `PEXPIRE`.
pub async fn cleanup(&self) {
match self {
Self::InMemory(rl) => rl.cleanup().await,
#[cfg(feature = "redis-rate-limiting")]
Self::Redis(_) => {},
}
}
/// How many live buckets the in-memory backend is holding, across all four maps.
///
/// `None` for the Redis backend, which keeps no local map — deliberately not `0`,
/// so a caller cannot mistake "no map here" for "the map is empty".
///
/// This exists because the sweep's *effect* was otherwise unobservable outside the
/// `rate_limit` module: the maps are `pub(super)`, so only the unit tests next to
/// them could see a bucket appear or disappear, and those call `cleanup()` directly
/// — proving the sweep WORKS, never that it is SCHEDULED. #1080's defect was
/// precisely a `cleanup()` with no caller, so "is it scheduled" is the question that
/// needs an answer from outside (#1173).
#[must_use]
pub fn live_bucket_count(&self) -> Option<usize> {
match self {
Self::InMemory(rl) => Some(rl.live_bucket_count()),
#[cfg(feature = "redis-rate-limiting")]
Self::Redis(_) => None,
}
}
/// Conservative static estimate of how long (in seconds) a client must wait
/// before the IP-level bucket refills one token: `ceil(1 / rps_per_ip)`.
///
/// Used when no backend-computed `retry_after_ms` is available (e.g., the
/// in-memory backend before the precise value is plumbed end-to-end, or as
/// a fallback on Redis errors). Minimum 1 second.
#[must_use]
pub fn retry_after_secs(&self) -> u32 {
let rps = self.config().rps_per_ip;
if rps == 0 {
return 1;
}
#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
// Reason: ceil(1/rps) is always a small positive integer
{
((1.0_f64 / f64::from(rps)).ceil() as u32).max(1)
}
}
}