feat(ratelimit): cluster-level rate limiting via shared Redis

.github/workflows/ci.yml

@@ -79,6 +79,10 @@ jobs:
  # every Config::load_from_path test as `redis_url` and breaks
  # the entire `aisix-core::config::tests` module.
  CACHE_TEST_REDIS_URL: redis://127.0.0.1:6379
+ # Picked up by crates/aisix-ratelimit/tests/redis_integration.rs
+ # (shared cluster-level counters, #798). Same skip-if-unset /
+ # no-AISIX_-prefix rules as CACHE_TEST_REDIS_URL above.
+ RATELIMIT_TEST_REDIS_URL: redis://127.0.0.1:6379
  # Picked up by crates/aisix-admin/tests/etcd_integration.rs.
  # Same skip-if-unset pattern as the Redis case above.
  ADMIN_TEST_ETCD_URL: http://127.0.0.1:2379

config.example.yaml

@@ -91,6 +91,24 @@ cache:
  # url: "redis://127.0.0.1:6379"
  # mode: "single" # single | cluster | sentinel
 
+# Rate-limit counter backend (api7/AISIX-Cloud#798).
+#
+# `memory` (default) keeps counters in this process, so a cluster of N
+# replicas enforces N× every configured limit. `redis` shares the
+# counters across every replica via one Redis, so the whole cluster
+# enforces ONE global window — set this on multi-replica deployments.
+# May point at the same Redis as `cache` (keys are namespaced
+# `aisix:rl:`). On a Redis outage the limiter fails open to per-replica
+# in-memory counting (logged) so traffic keeps flowing.
+ratelimit:
+ backend: "memory" # memory | redis
+ # redis:
+ # url: "redis://127.0.0.1:6379"
+ # mode: "single" # single | cluster | sentinel
+ # Seconds before an unreleased concurrency slot (crashed replica /
+ # hung upstream) is reclaimed. Redis backend only.
+ # concurrency_ttl_secs: 300
+
 # Models, API keys, provider keys, guardrails, cache policies, and
 # observability exporters are NOT defined in this file. They are stored
 # in etcd and managed via the Admin API (see docs/api-admin.md). This

config.managed.yaml

@@ -22,6 +22,12 @@
 # `AISIX_CACHE__BACKEND`, etc. — every config field is reachable via
 # `AISIX_<UPPER>__<UPPER>` (see crates/aisix-core/src/config.rs).
 #
+# For a multi-replica deployment, enable cluster-level rate limiting so
+# the cluster enforces one global window instead of N× per replica
+# (api7/AISIX-Cloud#798):
+# AISIX_RATELIMIT__BACKEND=redis
+# AISIX_RATELIMIT__REDIS__URL=redis://<host>:6379
+#
 # Subsequent boots re-use the mTLS bundle written under
 # `managed.mtls_dir`.
 

crates/aisix-core/src/config.rs

@@ -43,6 +43,12 @@ pub struct Config {
  pub observability: ObservabilityConfig,
  #[serde(default)]
  pub cache: CacheConfig,
+ /// Rate-limit counter backend. Defaults to per-process memory
+ /// (historical behaviour). Set `backend: redis` with a `redis` block
+ /// to share counters across every DP replica so a cluster enforces
+ /// one global window instead of one-per-replica (api7/AISIX-Cloud#798).
+ #[serde(default)]
+ pub ratelimit: RateLimitConfig,
  /// Optional managed-mode configuration. When `managed.enabled = true`
  /// the admin API and Playground endpoints are **not** bound — the DP
  /// is a pure etcd reader driven by the aisix.cloud control plane.
@@ -559,6 +565,43 @@ impl RedisCacheConfig {
  }
 }
 
+/// Rate-limit counter backend (api7/AISIX-Cloud#798).
+///
+/// `Memory` is the default: per-process fixed-window counters, so an
+/// N-replica cluster enforces N× the configured limit. `Redis` shares
+/// the counters across replicas via a single Redis so the whole cluster
+/// enforces one global window. The `redis` block is required iff
+/// `backend = redis` (validated at boot). Reuses [`RedisCacheConfig`]
+/// for the connection shape; may point at the same Redis as `cache`
+/// (keys are namespaced `aisix:rl:`).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields, default)]
+pub struct RateLimitConfig {
+ pub backend: RateLimitBackend,
+ pub redis: Option<RedisCacheConfig>,
+ /// Seconds after which an unreleased concurrency slot is reclaimed
+ /// (crashed replica / hung upstream). Generous enough for a long
+ /// streaming response. Redis backend only.
+ pub concurrency_ttl_secs: u64,
+}
+
+impl Default for RateLimitConfig {
+ fn default() -> Self {
+ Self {
+ backend: RateLimitBackend::Memory,
+ redis: None,
+ concurrency_ttl_secs: 300,
+ }
+ }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum RateLimitBackend {
+ Memory,
+ Redis,
+}
+
 impl Config {
  /// Load + merge + validate.
  ///
@@ -658,6 +701,20 @@ impl Config {
  "observability.metrics.prometheus.addr invalid socket address: {metrics_addr}"
  )));
  }
+ if self.ratelimit.backend == RateLimitBackend::Redis {
+ if self.ratelimit.redis.is_none() {
+ return Err(BootstrapError::Config(
+ "ratelimit.backend = redis requires a ratelimit.redis block".into(),
+ ));
+ }
+ // A zero concurrency TTL would prune a slot in the same second
+ // it was taken, silently disabling concurrency limiting.
+ if self.ratelimit.concurrency_ttl_secs == 0 {
+ return Err(BootstrapError::Config(
+ "ratelimit.concurrency_ttl_secs must be > 0 for the redis backend".into(),
+ ));
+ }
+ }
  Ok(())
  }
 }
@@ -784,6 +841,93 @@ admin:
  assert!(err.to_string().contains("admin.admin_keys"));
  }
 
+ #[test]
+ fn ratelimit_defaults_to_memory_backend() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+"#,
+ );
+ let cfg = Config::load_from_path(Some(f.path())).unwrap();
+ assert_eq!(cfg.ratelimit.backend, RateLimitBackend::Memory);
+ assert!(cfg.ratelimit.redis.is_none());
+ assert_eq!(cfg.ratelimit.concurrency_ttl_secs, 300);
+ }
+
+ #[test]
+ fn ratelimit_redis_backend_requires_redis_block() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+ratelimit:
+ backend: "redis"
+"#,
+ );
+ let err = Config::load_from_path(Some(f.path())).unwrap_err();
+ assert!(err.to_string().contains("ratelimit.redis"));
+ }
+
+ #[test]
+ fn rejects_zero_concurrency_ttl_for_redis_backend() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+ratelimit:
+ backend: "redis"
+ redis:
+ url: "redis://127.0.0.1:6379"
+ concurrency_ttl_secs: 0
+"#,
+ );
+ let err = Config::load_from_path(Some(f.path())).unwrap_err();
+ assert!(err.to_string().contains("concurrency_ttl_secs"));
+ }
+
+ #[test]
+ fn loads_ratelimit_redis_config() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+ratelimit:
+ backend: "redis"
+ redis:
+ url: "redis://127.0.0.1:6379"
+ concurrency_ttl_secs: 120
+"#,
+ );
+ let cfg = Config::load_from_path(Some(f.path())).unwrap();
+ assert_eq!(cfg.ratelimit.backend, RateLimitBackend::Redis);
+ assert_eq!(
+ cfg.ratelimit.redis.as_ref().unwrap().url,
+ "redis://127.0.0.1:6379"
+ );
+ assert_eq!(cfg.ratelimit.concurrency_ttl_secs, 120);
+ }
+
  #[test]
  fn rejects_invalid_bind_addr() {
  let f = write_yaml(

crates/aisix-core/src/lib.rs

@@ -24,7 +24,8 @@ pub mod snapshot;
 
 pub use config::{
  AdminConfig, CacheBackend, CacheConfig, Config, EtcdConfig, EtcdTlsConfig, ManagedConfig,
- ObservabilityConfig, ProxyConfig, RealIpConfig, TlsConfig,
+ ObservabilityConfig, ProxyConfig, RateLimitBackend, RateLimitConfig, RealIpConfig,
+ RedisCacheConfig, TlsConfig,
 };
 pub use error::{
  AdminError, AdminErrorEnvelope, BootstrapError, ProxyError, ProxyErrorEnvelope, RateLimitScope,

crates/aisix-proxy/src/chat.rs

@@ -245,7 +245,7 @@ pub async fn chat_completions(
  // current window state. We peek *after* the commit so
  // remaining-requests reflects the post-dispatch tally.
  let rl_limits = auth.key().rate_limit.clone().unwrap_or_default();
- if let Some(rl_status) = state.limiter.peek(&api_key_id, &rl_limits) {
+ if let Some(rl_status) = state.limiter.peek(&api_key_id, &rl_limits).await {
  crate::render::inject_ratelimit_headers(&mut success.response, &rl_status);
  state.metrics.set_rate_limit_remaining(
  &api_key_id,
@@ -843,8 +843,9 @@ async fn dispatch(
  &virtual_entry.id,
  &virtual_entry.value,
  );
- let reservation =
- crate::quota::enforce_rate_limit(state, auth, Some(&model_rl)).map_err(&with_model)?;
+ let reservation = crate::quota::enforce_rate_limit(state, auth, Some(&model_rl))
+ .await
+ .map_err(&with_model)?;
 
  let now = created_ts();
 
@@ -1073,7 +1074,7 @@ async fn dispatch(
  // permit was released here, letting a key capped at N run far more
  // than N simultaneous streams (#450).
  let post_stream_keys = reservation.keys();
- let stream_concurrency_hold = reservation.into_stream_hold(Arc::clone(&state.limiter));
+ let stream_concurrency_hold = reservation.into_stream_hold();
  // Capture everything the stream-completion callback needs so
  // it can fire `emit_usage_event` once the terminal SSE chunk
  // has yielded its `usage` block. Telemetry emission has to
@@ -1379,7 +1380,7 @@ async fn dispatch(
  if let (Some(cache), Some(key)) = (policy_cache.as_ref(), cache_key.as_ref()) {
  match cache.get(key).await {
  Ok(Some(cached)) => {
- reservation.commit_tokens(0);
+ reservation.commit_tokens(0).await;
  // #448: a cache hit is client-visible output just like a
  // fresh upstream response, so it must run output guardrails
  // before being returned — not bypass them.
@@ -1690,7 +1691,7 @@ async fn dispatch(
  let provider_request_id = upstream.id.clone();
  let provider_model_version = upstream.model.clone();
  let finish_reason = finish_reason_label(&upstream.finish_reason);
- reservation.commit_tokens(total);
+ reservation.commit_tokens(total).await;
 
  // cp-api recomputes cost server-side from its pricing catalog when
  // ingesting telemetry; the DP just records 0.0 on the wire.
@@ -1860,14 +1861,14 @@ async fn dispatch(
 /// response. `reservation` is the SINGLE entry-level reservation taken
 /// in `dispatch` — ensemble does not add per-sub-call reservations.
 #[allow(clippy::too_many_arguments)]
-async fn dispatch_ensemble<'a>(
- state: &'a ProxyState,
+async fn dispatch_ensemble(
+ state: &ProxyState,
  snapshot: &aisix_core::AisixSnapshot,
  virtual_entry: &aisix_core::ResourceEntry<aisix_core::Model>,
  req: &ChatFormat,
  request_id: &str,
  created_ts: i64,
- reservation: aisix_ratelimit::MultiReservation<'a, aisix_ratelimit::SystemClock>,
+ reservation: aisix_ratelimit::MultiReservation,
  resolved_chain: &Arc<dyn aisix_guardrails::Guardrail>,
  applied_guardrails: &[AppliedGuardrail],
  mut bypass_reason: Option<String>,
@@ -2006,7 +2007,7 @@ async fn dispatch_ensemble<'a>(
  ),
  };
  let survivor_total: u64 = panel.iter().map(|p| u64::from(p.usage.total_tokens)).sum();
- reservation.commit_tokens(survivor_total);
+ reservation.commit_tokens(survivor_total).await;
  for (index, member) in panel.iter().enumerate() {
  emit_panel_member(
  member, index, /* blocked */ false, /* bypass */ "",
@@ -2030,7 +2031,7 @@ async fn dispatch_ensemble<'a>(
  .sum();
  let judge_usage = outcome.response.usage.clone();
  let total_tokens = panel_total + u64::from(judge_usage.total_tokens);
- reservation.commit_tokens(total_tokens);
+ reservation.commit_tokens(total_tokens).await;
 
  // Emit one usage event per sub-call (each panel member + the judge),
  // all sharing `request_id`. `attempt_kind` is `"panel"` / `"judge"`;

crates/aisix-proxy/src/embeddings.rs

@@ -321,7 +321,9 @@ async fn dispatch(
  // and finalise RPM. Embeddings do report prompt_tokens via
  // EmbeddingResponse.usage; thread it through so TPM works
  // here even though other handlers commit 0.
- reservation.commit_tokens(embed_resp.usage.total_tokens as u64);
+ reservation
+ .commit_tokens(embed_resp.usage.total_tokens as u64)
+ .await;
  let provider_label = provider.to_ascii_lowercase();
  // Capture the prompt_tokens count BEFORE moving the
  // embed_resp into the JSON response — the handler needs
@@ -342,7 +344,7 @@ async fn dispatch(
  // (`upstream_called: false` → handler skips emit per the
  // chat.rs convention that we only attribute usage on a
  // real upstream completion).
- reservation.commit_tokens(0);
+ reservation.commit_tokens(0).await;
  let env = ErrorEnvelope::new(msg, "not_implemented");
  Ok(EmbedDispatchSuccess {
  response: (StatusCode::NOT_IMPLEMENTED, Json(env)).into_response(),
@@ -357,7 +359,7 @@ async fn dispatch(
  })
  }
  Err(e) => {
- reservation.commit_tokens(0);
+ reservation.commit_tokens(0).await;
  Err(ProxyError::Bridge(e))
  }
  }

.github/workflows/ci.yml

@@ -79,6 +79,10 @@ jobs:
  # every Config::load_from_path test as `redis_url` and breaks
  # the entire `aisix-core::config::tests` module.
  CACHE_TEST_REDIS_URL: redis://127.0.0.1:6379
+ # Picked up by crates/aisix-ratelimit/tests/redis_integration.rs
+ # (shared cluster-level counters, #798). Same skip-if-unset /
+ # no-AISIX_-prefix rules as CACHE_TEST_REDIS_URL above.
+ RATELIMIT_TEST_REDIS_URL: redis://127.0.0.1:6379
  # Picked up by crates/aisix-admin/tests/etcd_integration.rs.
  # Same skip-if-unset pattern as the Redis case above.
  ADMIN_TEST_ETCD_URL: http://127.0.0.1:2379

config.example.yaml

@@ -91,6 +91,24 @@ cache:
  # url: "redis://127.0.0.1:6379"
  # mode: "single" # single | cluster | sentinel
 
+# Rate-limit counter backend (api7/AISIX-Cloud#798).
+#
+# `memory` (default) keeps counters in this process, so a cluster of N
+# replicas enforces N× every configured limit. `redis` shares the
+# counters across every replica via one Redis, so the whole cluster
+# enforces ONE global window — set this on multi-replica deployments.
+# May point at the same Redis as `cache` (keys are namespaced
+# `aisix:rl:`). On a Redis outage the limiter fails open to per-replica
+# in-memory counting (logged) so traffic keeps flowing.
+ratelimit:
+ backend: "memory" # memory | redis
+ # redis:
+ # url: "redis://127.0.0.1:6379"
+ # mode: "single" # single | cluster | sentinel
+ # Seconds before an unreleased concurrency slot (crashed replica /
+ # hung upstream) is reclaimed. Redis backend only.
+ # concurrency_ttl_secs: 300
+
 # Models, API keys, provider keys, guardrails, cache policies, and
 # observability exporters are NOT defined in this file. They are stored
 # in etcd and managed via the Admin API (see docs/api-admin.md). This

config.managed.yaml

@@ -22,6 +22,12 @@
 # `AISIX_CACHE__BACKEND`, etc. — every config field is reachable via
 # `AISIX_<UPPER>__<UPPER>` (see crates/aisix-core/src/config.rs).
 #
+# For a multi-replica deployment, enable cluster-level rate limiting so
+# the cluster enforces one global window instead of N× per replica
+# (api7/AISIX-Cloud#798):
+# AISIX_RATELIMIT__BACKEND=redis
+# AISIX_RATELIMIT__REDIS__URL=redis://<host>:6379
+#
 # Subsequent boots re-use the mTLS bundle written under
 # `managed.mtls_dir`.
 

crates/aisix-core/src/config.rs

@@ -43,6 +43,12 @@ pub struct Config {
  pub observability: ObservabilityConfig,
  #[serde(default)]
  pub cache: CacheConfig,
+ /// Rate-limit counter backend. Defaults to per-process memory
+ /// (historical behaviour). Set `backend: redis` with a `redis` block
+ /// to share counters across every DP replica so a cluster enforces
+ /// one global window instead of one-per-replica (api7/AISIX-Cloud#798).
+ #[serde(default)]
+ pub ratelimit: RateLimitConfig,
  /// Optional managed-mode configuration. When `managed.enabled = true`
  /// the admin API and Playground endpoints are **not** bound — the DP
  /// is a pure etcd reader driven by the aisix.cloud control plane.
@@ -559,6 +565,43 @@ impl RedisCacheConfig {
  }
 }
 
+/// Rate-limit counter backend (api7/AISIX-Cloud#798).
+///
+/// `Memory` is the default: per-process fixed-window counters, so an
+/// N-replica cluster enforces N× the configured limit. `Redis` shares
+/// the counters across replicas via a single Redis so the whole cluster
+/// enforces one global window. The `redis` block is required iff
+/// `backend = redis` (validated at boot). Reuses [`RedisCacheConfig`]
+/// for the connection shape; may point at the same Redis as `cache`
+/// (keys are namespaced `aisix:rl:`).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields, default)]
+pub struct RateLimitConfig {
+ pub backend: RateLimitBackend,
+ pub redis: Option<RedisCacheConfig>,
+ /// Seconds after which an unreleased concurrency slot is reclaimed
+ /// (crashed replica / hung upstream). Generous enough for a long
+ /// streaming response. Redis backend only.
+ pub concurrency_ttl_secs: u64,
+}
+
+impl Default for RateLimitConfig {
+ fn default() -> Self {
+ Self {
+ backend: RateLimitBackend::Memory,
+ redis: None,
+ concurrency_ttl_secs: 300,
+ }
+ }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum RateLimitBackend {
+ Memory,
+ Redis,
+}
+
 impl Config {
  /// Load + merge + validate.
  ///
@@ -658,6 +701,20 @@ impl Config {
  "observability.metrics.prometheus.addr invalid socket address: {metrics_addr}"
  )));
  }
+ if self.ratelimit.backend == RateLimitBackend::Redis {
+ if self.ratelimit.redis.is_none() {
+ return Err(BootstrapError::Config(
+ "ratelimit.backend = redis requires a ratelimit.redis block".into(),
+ ));
+ }
+ // A zero concurrency TTL would prune a slot in the same second
+ // it was taken, silently disabling concurrency limiting.
+ if self.ratelimit.concurrency_ttl_secs == 0 {
+ return Err(BootstrapError::Config(
+ "ratelimit.concurrency_ttl_secs must be > 0 for the redis backend".into(),
+ ));
+ }
+ }
  Ok(())
  }
 }
@@ -784,6 +841,93 @@ admin:
  assert!(err.to_string().contains("admin.admin_keys"));
  }
 
+ #[test]
+ fn ratelimit_defaults_to_memory_backend() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+"#,
+ );
+ let cfg = Config::load_from_path(Some(f.path())).unwrap();
+ assert_eq!(cfg.ratelimit.backend, RateLimitBackend::Memory);
+ assert!(cfg.ratelimit.redis.is_none());
+ assert_eq!(cfg.ratelimit.concurrency_ttl_secs, 300);
+ }
+
+ #[test]
+ fn ratelimit_redis_backend_requires_redis_block() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+ratelimit:
+ backend: "redis"
+"#,
+ );
+ let err = Config::load_from_path(Some(f.path())).unwrap_err();
+ assert!(err.to_string().contains("ratelimit.redis"));
+ }
+
+ #[test]
+ fn rejects_zero_concurrency_ttl_for_redis_backend() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+ratelimit:
+ backend: "redis"
+ redis:
+ url: "redis://127.0.0.1:6379"
+ concurrency_ttl_secs: 0
+"#,
+ );
+ let err = Config::load_from_path(Some(f.path())).unwrap_err();
+ assert!(err.to_string().contains("concurrency_ttl_secs"));
+ }
+
+ #[test]
+ fn loads_ratelimit_redis_config() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+ratelimit:
+ backend: "redis"
+ redis:
+ url: "redis://127.0.0.1:6379"
+ concurrency_ttl_secs: 120
+"#,
+ );
+ let cfg = Config::load_from_path(Some(f.path())).unwrap();
+ assert_eq!(cfg.ratelimit.backend, RateLimitBackend::Redis);
+ assert_eq!(
+ cfg.ratelimit.redis.as_ref().unwrap().url,
+ "redis://127.0.0.1:6379"
+ );
+ assert_eq!(cfg.ratelimit.concurrency_ttl_secs, 120);
+ }
+
  #[test]
  fn rejects_invalid_bind_addr() {
  let f = write_yaml(

crates/aisix-core/src/lib.rs

@@ -24,7 +24,8 @@ pub mod snapshot;
 
 pub use config::{
  AdminConfig, CacheBackend, CacheConfig, Config, EtcdConfig, EtcdTlsConfig, ManagedConfig,
- ObservabilityConfig, ProxyConfig, RealIpConfig, TlsConfig,
+ ObservabilityConfig, ProxyConfig, RateLimitBackend, RateLimitConfig, RealIpConfig,
+ RedisCacheConfig, TlsConfig,
 };
 pub use error::{
  AdminError, AdminErrorEnvelope, BootstrapError, ProxyError, ProxyErrorEnvelope, RateLimitScope,

crates/aisix-proxy/src/chat.rs

@@ -245,7 +245,7 @@ pub async fn chat_completions(
  // current window state. We peek *after* the commit so
  // remaining-requests reflects the post-dispatch tally.
  let rl_limits = auth.key().rate_limit.clone().unwrap_or_default();
- if let Some(rl_status) = state.limiter.peek(&api_key_id, &rl_limits) {
+ if let Some(rl_status) = state.limiter.peek(&api_key_id, &rl_limits).await {
  crate::render::inject_ratelimit_headers(&mut success.response, &rl_status);
  state.metrics.set_rate_limit_remaining(
  &api_key_id,
@@ -843,8 +843,9 @@ async fn dispatch(
  &virtual_entry.id,
  &virtual_entry.value,
  );
- let reservation =
- crate::quota::enforce_rate_limit(state, auth, Some(&model_rl)).map_err(&with_model)?;
+ let reservation = crate::quota::enforce_rate_limit(state, auth, Some(&model_rl))
+ .await
+ .map_err(&with_model)?;
 
  let now = created_ts();
 
@@ -1073,7 +1074,7 @@ async fn dispatch(
  // permit was released here, letting a key capped at N run far more
  // than N simultaneous streams (#450).
  let post_stream_keys = reservation.keys();
- let stream_concurrency_hold = reservation.into_stream_hold(Arc::clone(&state.limiter));
+ let stream_concurrency_hold = reservation.into_stream_hold();
  // Capture everything the stream-completion callback needs so
  // it can fire `emit_usage_event` once the terminal SSE chunk
  // has yielded its `usage` block. Telemetry emission has to
@@ -1379,7 +1380,7 @@ async fn dispatch(
  if let (Some(cache), Some(key)) = (policy_cache.as_ref(), cache_key.as_ref()) {
  match cache.get(key).await {
  Ok(Some(cached)) => {
- reservation.commit_tokens(0);
+ reservation.commit_tokens(0).await;
  // #448: a cache hit is client-visible output just like a
  // fresh upstream response, so it must run output guardrails
  // before being returned — not bypass them.
@@ -1690,7 +1691,7 @@ async fn dispatch(
  let provider_request_id = upstream.id.clone();
  let provider_model_version = upstream.model.clone();
  let finish_reason = finish_reason_label(&upstream.finish_reason);
- reservation.commit_tokens(total);
+ reservation.commit_tokens(total).await;
 
  // cp-api recomputes cost server-side from its pricing catalog when
  // ingesting telemetry; the DP just records 0.0 on the wire.
@@ -1860,14 +1861,14 @@ async fn dispatch(
 /// response. `reservation` is the SINGLE entry-level reservation taken
 /// in `dispatch` — ensemble does not add per-sub-call reservations.
 #[allow(clippy::too_many_arguments)]
-async fn dispatch_ensemble<'a>(
- state: &'a ProxyState,
+async fn dispatch_ensemble(
+ state: &ProxyState,
  snapshot: &aisix_core::AisixSnapshot,
  virtual_entry: &aisix_core::ResourceEntry<aisix_core::Model>,
  req: &ChatFormat,
  request_id: &str,
  created_ts: i64,
- reservation: aisix_ratelimit::MultiReservation<'a, aisix_ratelimit::SystemClock>,
+ reservation: aisix_ratelimit::MultiReservation,
  resolved_chain: &Arc<dyn aisix_guardrails::Guardrail>,
  applied_guardrails: &[AppliedGuardrail],
  mut bypass_reason: Option<String>,
@@ -2006,7 +2007,7 @@ async fn dispatch_ensemble<'a>(
  ),
  };
  let survivor_total: u64 = panel.iter().map(|p| u64::from(p.usage.total_tokens)).sum();
- reservation.commit_tokens(survivor_total);
+ reservation.commit_tokens(survivor_total).await;
  for (index, member) in panel.iter().enumerate() {
  emit_panel_member(
  member, index, /* blocked */ false, /* bypass */ "",
@@ -2030,7 +2031,7 @@ async fn dispatch_ensemble<'a>(
  .sum();
  let judge_usage = outcome.response.usage.clone();
  let total_tokens = panel_total + u64::from(judge_usage.total_tokens);
- reservation.commit_tokens(total_tokens);
+ reservation.commit_tokens(total_tokens).await;
 
  // Emit one usage event per sub-call (each panel member + the judge),
  // all sharing `request_id`. `attempt_kind` is `"panel"` / `"judge"`;

crates/aisix-proxy/src/embeddings.rs

@@ -321,7 +321,9 @@ async fn dispatch(
  // and finalise RPM. Embeddings do report prompt_tokens via
  // EmbeddingResponse.usage; thread it through so TPM works
  // here even though other handlers commit 0.
- reservation.commit_tokens(embed_resp.usage.total_tokens as u64);
+ reservation
+ .commit_tokens(embed_resp.usage.total_tokens as u64)
+ .await;
  let provider_label = provider.to_ascii_lowercase();
  // Capture the prompt_tokens count BEFORE moving the
  // embed_resp into the JSON response — the handler needs
@@ -342,7 +344,7 @@ async fn dispatch(
  // (`upstream_called: false` → handler skips emit per the
  // chat.rs convention that we only attribute usage on a
  // real upstream completion).
- reservation.commit_tokens(0);
+ reservation.commit_tokens(0).await;
  let env = ErrorEnvelope::new(msg, "not_implemented");
  Ok(EmbedDispatchSuccess {
  response: (StatusCode::NOT_IMPLEMENTED, Json(env)).into_response(),
@@ -357,7 +359,7 @@ async fn dispatch(
  })
  }
  Err(e) => {
- reservation.commit_tokens(0);
+ reservation.commit_tokens(0).await;
  Err(ProxyError::Bridge(e))
  }
  }

.github/workflows/ci.yml

@@ -79,6 +79,10 @@ jobs:
  # every Config::load_from_path test as `redis_url` and breaks
  # the entire `aisix-core::config::tests` module.
  CACHE_TEST_REDIS_URL: redis://127.0.0.1:6379
+ # Picked up by crates/aisix-ratelimit/tests/redis_integration.rs
+ # (shared cluster-level counters, #798). Same skip-if-unset /
+ # no-AISIX_-prefix rules as CACHE_TEST_REDIS_URL above.
+ RATELIMIT_TEST_REDIS_URL: redis://127.0.0.1:6379
  # Picked up by crates/aisix-admin/tests/etcd_integration.rs.
  # Same skip-if-unset pattern as the Redis case above.
  ADMIN_TEST_ETCD_URL: http://127.0.0.1:2379

config.example.yaml

@@ -91,6 +91,24 @@ cache:
  # url: "redis://127.0.0.1:6379"
  # mode: "single" # single | cluster | sentinel
 
+# Rate-limit counter backend (api7/AISIX-Cloud#798).
+#
+# `memory` (default) keeps counters in this process, so a cluster of N
+# replicas enforces N× every configured limit. `redis` shares the
+# counters across every replica via one Redis, so the whole cluster
+# enforces ONE global window — set this on multi-replica deployments.
+# May point at the same Redis as `cache` (keys are namespaced
+# `aisix:rl:`). On a Redis outage the limiter fails open to per-replica
+# in-memory counting (logged) so traffic keeps flowing.
+ratelimit:
+ backend: "memory" # memory | redis
+ # redis:
+ # url: "redis://127.0.0.1:6379"
+ # mode: "single" # single | cluster | sentinel
+ # Seconds before an unreleased concurrency slot (crashed replica /
+ # hung upstream) is reclaimed. Redis backend only.
+ # concurrency_ttl_secs: 300
+
 # Models, API keys, provider keys, guardrails, cache policies, and
 # observability exporters are NOT defined in this file. They are stored
 # in etcd and managed via the Admin API (see docs/api-admin.md). This

config.managed.yaml

@@ -22,6 +22,12 @@
 # `AISIX_CACHE__BACKEND`, etc. — every config field is reachable via
 # `AISIX_<UPPER>__<UPPER>` (see crates/aisix-core/src/config.rs).
 #
+# For a multi-replica deployment, enable cluster-level rate limiting so
+# the cluster enforces one global window instead of N× per replica
+# (api7/AISIX-Cloud#798):
+# AISIX_RATELIMIT__BACKEND=redis
+# AISIX_RATELIMIT__REDIS__URL=redis://<host>:6379
+#
 # Subsequent boots re-use the mTLS bundle written under
 # `managed.mtls_dir`.
 

crates/aisix-core/src/config.rs

@@ -43,6 +43,12 @@ pub struct Config {
  pub observability: ObservabilityConfig,
  #[serde(default)]
  pub cache: CacheConfig,
+ /// Rate-limit counter backend. Defaults to per-process memory
+ /// (historical behaviour). Set `backend: redis` with a `redis` block
+ /// to share counters across every DP replica so a cluster enforces
+ /// one global window instead of one-per-replica (api7/AISIX-Cloud#798).
+ #[serde(default)]
+ pub ratelimit: RateLimitConfig,
  /// Optional managed-mode configuration. When `managed.enabled = true`
  /// the admin API and Playground endpoints are **not** bound — the DP
  /// is a pure etcd reader driven by the aisix.cloud control plane.
@@ -559,6 +565,43 @@ impl RedisCacheConfig {
  }
 }
 
+/// Rate-limit counter backend (api7/AISIX-Cloud#798).
+///
+/// `Memory` is the default: per-process fixed-window counters, so an
+/// N-replica cluster enforces N× the configured limit. `Redis` shares
+/// the counters across replicas via a single Redis so the whole cluster
+/// enforces one global window. The `redis` block is required iff
+/// `backend = redis` (validated at boot). Reuses [`RedisCacheConfig`]
+/// for the connection shape; may point at the same Redis as `cache`
+/// (keys are namespaced `aisix:rl:`).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields, default)]
+pub struct RateLimitConfig {
+ pub backend: RateLimitBackend,
+ pub redis: Option<RedisCacheConfig>,
+ /// Seconds after which an unreleased concurrency slot is reclaimed
+ /// (crashed replica / hung upstream). Generous enough for a long
+ /// streaming response. Redis backend only.
+ pub concurrency_ttl_secs: u64,
+}
+
+impl Default for RateLimitConfig {
+ fn default() -> Self {
+ Self {
+ backend: RateLimitBackend::Memory,
+ redis: None,
+ concurrency_ttl_secs: 300,
+ }
+ }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum RateLimitBackend {
+ Memory,
+ Redis,
+}
+
 impl Config {
  /// Load + merge + validate.
  ///
@@ -658,6 +701,20 @@ impl Config {
  "observability.metrics.prometheus.addr invalid socket address: {metrics_addr}"
  )));
  }
+ if self.ratelimit.backend == RateLimitBackend::Redis {
+ if self.ratelimit.redis.is_none() {
+ return Err(BootstrapError::Config(
+ "ratelimit.backend = redis requires a ratelimit.redis block".into(),
+ ));
+ }
+ // A zero concurrency TTL would prune a slot in the same second
+ // it was taken, silently disabling concurrency limiting.
+ if self.ratelimit.concurrency_ttl_secs == 0 {
+ return Err(BootstrapError::Config(
+ "ratelimit.concurrency_ttl_secs must be > 0 for the redis backend".into(),
+ ));
+ }
+ }
  Ok(())
  }
 }
@@ -784,6 +841,93 @@ admin:
  assert!(err.to_string().contains("admin.admin_keys"));
  }
 
+ #[test]
+ fn ratelimit_defaults_to_memory_backend() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+"#,
+ );
+ let cfg = Config::load_from_path(Some(f.path())).unwrap();
+ assert_eq!(cfg.ratelimit.backend, RateLimitBackend::Memory);
+ assert!(cfg.ratelimit.redis.is_none());
+ assert_eq!(cfg.ratelimit.concurrency_ttl_secs, 300);
+ }
+
+ #[test]
+ fn ratelimit_redis_backend_requires_redis_block() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+ratelimit:
+ backend: "redis"
+"#,
+ );
+ let err = Config::load_from_path(Some(f.path())).unwrap_err();
+ assert!(err.to_string().contains("ratelimit.redis"));
+ }
+
+ #[test]
+ fn rejects_zero_concurrency_ttl_for_redis_backend() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+ratelimit:
+ backend: "redis"
+ redis:
+ url: "redis://127.0.0.1:6379"
+ concurrency_ttl_secs: 0
+"#,
+ );
+ let err = Config::load_from_path(Some(f.path())).unwrap_err();
+ assert!(err.to_string().contains("concurrency_ttl_secs"));
+ }
+
+ #[test]
+ fn loads_ratelimit_redis_config() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+ratelimit:
+ backend: "redis"
+ redis:
+ url: "redis://127.0.0.1:6379"
+ concurrency_ttl_secs: 120
+"#,
+ );
+ let cfg = Config::load_from_path(Some(f.path())).unwrap();
+ assert_eq!(cfg.ratelimit.backend, RateLimitBackend::Redis);
+ assert_eq!(
+ cfg.ratelimit.redis.as_ref().unwrap().url,
+ "redis://127.0.0.1:6379"
+ );
+ assert_eq!(cfg.ratelimit.concurrency_ttl_secs, 120);
+ }
+
  #[test]
  fn rejects_invalid_bind_addr() {
  let f = write_yaml(

crates/aisix-core/src/lib.rs

@@ -24,7 +24,8 @@ pub mod snapshot;
 
 pub use config::{
  AdminConfig, CacheBackend, CacheConfig, Config, EtcdConfig, EtcdTlsConfig, ManagedConfig,
- ObservabilityConfig, ProxyConfig, RealIpConfig, TlsConfig,
+ ObservabilityConfig, ProxyConfig, RateLimitBackend, RateLimitConfig, RealIpConfig,
+ RedisCacheConfig, TlsConfig,
 };
 pub use error::{
  AdminError, AdminErrorEnvelope, BootstrapError, ProxyError, ProxyErrorEnvelope, RateLimitScope,

crates/aisix-proxy/src/chat.rs

@@ -245,7 +245,7 @@ pub async fn chat_completions(
  // current window state. We peek *after* the commit so
  // remaining-requests reflects the post-dispatch tally.
  let rl_limits = auth.key().rate_limit.clone().unwrap_or_default();
- if let Some(rl_status) = state.limiter.peek(&api_key_id, &rl_limits) {
+ if let Some(rl_status) = state.limiter.peek(&api_key_id, &rl_limits).await {
  crate::render::inject_ratelimit_headers(&mut success.response, &rl_status);
  state.metrics.set_rate_limit_remaining(
  &api_key_id,
@@ -843,8 +843,9 @@ async fn dispatch(
  &virtual_entry.id,
  &virtual_entry.value,
  );
- let reservation =
- crate::quota::enforce_rate_limit(state, auth, Some(&model_rl)).map_err(&with_model)?;
+ let reservation = crate::quota::enforce_rate_limit(state, auth, Some(&model_rl))
+ .await
+ .map_err(&with_model)?;
 
  let now = created_ts();
 
@@ -1073,7 +1074,7 @@ async fn dispatch(
  // permit was released here, letting a key capped at N run far more
  // than N simultaneous streams (#450).
  let post_stream_keys = reservation.keys();
- let stream_concurrency_hold = reservation.into_stream_hold(Arc::clone(&state.limiter));
+ let stream_concurrency_hold = reservation.into_stream_hold();
  // Capture everything the stream-completion callback needs so
  // it can fire `emit_usage_event` once the terminal SSE chunk
  // has yielded its `usage` block. Telemetry emission has to
@@ -1379,7 +1380,7 @@ async fn dispatch(
  if let (Some(cache), Some(key)) = (policy_cache.as_ref(), cache_key.as_ref()) {
  match cache.get(key).await {
  Ok(Some(cached)) => {
- reservation.commit_tokens(0);
+ reservation.commit_tokens(0).await;
  // #448: a cache hit is client-visible output just like a
  // fresh upstream response, so it must run output guardrails
  // before being returned — not bypass them.
@@ -1690,7 +1691,7 @@ async fn dispatch(
  let provider_request_id = upstream.id.clone();
  let provider_model_version = upstream.model.clone();
  let finish_reason = finish_reason_label(&upstream.finish_reason);
- reservation.commit_tokens(total);
+ reservation.commit_tokens(total).await;
 
  // cp-api recomputes cost server-side from its pricing catalog when
  // ingesting telemetry; the DP just records 0.0 on the wire.
@@ -1860,14 +1861,14 @@ async fn dispatch(
 /// response. `reservation` is the SINGLE entry-level reservation taken
 /// in `dispatch` — ensemble does not add per-sub-call reservations.
 #[allow(clippy::too_many_arguments)]
-async fn dispatch_ensemble<'a>(
- state: &'a ProxyState,
+async fn dispatch_ensemble(
+ state: &ProxyState,
  snapshot: &aisix_core::AisixSnapshot,
  virtual_entry: &aisix_core::ResourceEntry<aisix_core::Model>,
  req: &ChatFormat,
  request_id: &str,
  created_ts: i64,
- reservation: aisix_ratelimit::MultiReservation<'a, aisix_ratelimit::SystemClock>,
+ reservation: aisix_ratelimit::MultiReservation,
  resolved_chain: &Arc<dyn aisix_guardrails::Guardrail>,
  applied_guardrails: &[AppliedGuardrail],
  mut bypass_reason: Option<String>,
@@ -2006,7 +2007,7 @@ async fn dispatch_ensemble<'a>(
  ),
  };
  let survivor_total: u64 = panel.iter().map(|p| u64::from(p.usage.total_tokens)).sum();
- reservation.commit_tokens(survivor_total);
+ reservation.commit_tokens(survivor_total).await;
  for (index, member) in panel.iter().enumerate() {
  emit_panel_member(
  member, index, /* blocked */ false, /* bypass */ "",
@@ -2030,7 +2031,7 @@ async fn dispatch_ensemble<'a>(
  .sum();
  let judge_usage = outcome.response.usage.clone();
  let total_tokens = panel_total + u64::from(judge_usage.total_tokens);
- reservation.commit_tokens(total_tokens);
+ reservation.commit_tokens(total_tokens).await;
 
  // Emit one usage event per sub-call (each panel member + the judge),
  // all sharing `request_id`. `attempt_kind` is `"panel"` / `"judge"`;

crates/aisix-proxy/src/embeddings.rs

@@ -321,7 +321,9 @@ async fn dispatch(
  // and finalise RPM. Embeddings do report prompt_tokens via
  // EmbeddingResponse.usage; thread it through so TPM works
  // here even though other handlers commit 0.
- reservation.commit_tokens(embed_resp.usage.total_tokens as u64);
+ reservation
+ .commit_tokens(embed_resp.usage.total_tokens as u64)
+ .await;
  let provider_label = provider.to_ascii_lowercase();
  // Capture the prompt_tokens count BEFORE moving the
  // embed_resp into the JSON response — the handler needs
@@ -342,7 +344,7 @@ async fn dispatch(
  // (`upstream_called: false` → handler skips emit per the
  // chat.rs convention that we only attribute usage on a
  // real upstream completion).
- reservation.commit_tokens(0);
+ reservation.commit_tokens(0).await;
  let env = ErrorEnvelope::new(msg, "not_implemented");
  Ok(EmbedDispatchSuccess {
  response: (StatusCode::NOT_IMPLEMENTED, Json(env)).into_response(),
@@ -357,7 +359,7 @@ async fn dispatch(
  })
  }
  Err(e) => {
- reservation.commit_tokens(0);
+ reservation.commit_tokens(0).await;
  Err(ProxyError::Bridge(e))
  }
  }

.github/workflows/ci.yml

@@ -79,6 +79,10 @@ jobs:
  # every Config::load_from_path test as `redis_url` and breaks
  # the entire `aisix-core::config::tests` module.
  CACHE_TEST_REDIS_URL: redis://127.0.0.1:6379
+ # Picked up by crates/aisix-ratelimit/tests/redis_integration.rs
+ # (shared cluster-level counters, #798). Same skip-if-unset /
+ # no-AISIX_-prefix rules as CACHE_TEST_REDIS_URL above.
+ RATELIMIT_TEST_REDIS_URL: redis://127.0.0.1:6379
  # Picked up by crates/aisix-admin/tests/etcd_integration.rs.
  # Same skip-if-unset pattern as the Redis case above.
  ADMIN_TEST_ETCD_URL: http://127.0.0.1:2379

config.example.yaml

@@ -91,6 +91,24 @@ cache:
  # url: "redis://127.0.0.1:6379"
  # mode: "single" # single | cluster | sentinel
 
+# Rate-limit counter backend (api7/AISIX-Cloud#798).
+#
+# `memory` (default) keeps counters in this process, so a cluster of N
+# replicas enforces N× every configured limit. `redis` shares the
+# counters across every replica via one Redis, so the whole cluster
+# enforces ONE global window — set this on multi-replica deployments.
+# May point at the same Redis as `cache` (keys are namespaced
+# `aisix:rl:`). On a Redis outage the limiter fails open to per-replica
+# in-memory counting (logged) so traffic keeps flowing.
+ratelimit:
+ backend: "memory" # memory | redis
+ # redis:
+ # url: "redis://127.0.0.1:6379"
+ # mode: "single" # single | cluster | sentinel
+ # Seconds before an unreleased concurrency slot (crashed replica /
+ # hung upstream) is reclaimed. Redis backend only.
+ # concurrency_ttl_secs: 300
+
 # Models, API keys, provider keys, guardrails, cache policies, and
 # observability exporters are NOT defined in this file. They are stored
 # in etcd and managed via the Admin API (see docs/api-admin.md). This

config.managed.yaml

@@ -22,6 +22,12 @@
 # `AISIX_CACHE__BACKEND`, etc. — every config field is reachable via
 # `AISIX_<UPPER>__<UPPER>` (see crates/aisix-core/src/config.rs).
 #
+# For a multi-replica deployment, enable cluster-level rate limiting so
+# the cluster enforces one global window instead of N× per replica
+# (api7/AISIX-Cloud#798):
+# AISIX_RATELIMIT__BACKEND=redis
+# AISIX_RATELIMIT__REDIS__URL=redis://<host>:6379
+#
 # Subsequent boots re-use the mTLS bundle written under
 # `managed.mtls_dir`.
 

crates/aisix-core/src/config.rs

@@ -43,6 +43,12 @@ pub struct Config {
  pub observability: ObservabilityConfig,
  #[serde(default)]
  pub cache: CacheConfig,
+ /// Rate-limit counter backend. Defaults to per-process memory
+ /// (historical behaviour). Set `backend: redis` with a `redis` block
+ /// to share counters across every DP replica so a cluster enforces
+ /// one global window instead of one-per-replica (api7/AISIX-Cloud#798).
+ #[serde(default)]
+ pub ratelimit: RateLimitConfig,
  /// Optional managed-mode configuration. When `managed.enabled = true`
  /// the admin API and Playground endpoints are **not** bound — the DP
  /// is a pure etcd reader driven by the aisix.cloud control plane.
@@ -559,6 +565,43 @@ impl RedisCacheConfig {
  }
 }
 
+/// Rate-limit counter backend (api7/AISIX-Cloud#798).
+///
+/// `Memory` is the default: per-process fixed-window counters, so an
+/// N-replica cluster enforces N× the configured limit. `Redis` shares
+/// the counters across replicas via a single Redis so the whole cluster
+/// enforces one global window. The `redis` block is required iff
+/// `backend = redis` (validated at boot). Reuses [`RedisCacheConfig`]
+/// for the connection shape; may point at the same Redis as `cache`
+/// (keys are namespaced `aisix:rl:`).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields, default)]
+pub struct RateLimitConfig {
+ pub backend: RateLimitBackend,
+ pub redis: Option<RedisCacheConfig>,
+ /// Seconds after which an unreleased concurrency slot is reclaimed
+ /// (crashed replica / hung upstream). Generous enough for a long
+ /// streaming response. Redis backend only.
+ pub concurrency_ttl_secs: u64,
+}
+
+impl Default for RateLimitConfig {
+ fn default() -> Self {
+ Self {
+ backend: RateLimitBackend::Memory,
+ redis: None,
+ concurrency_ttl_secs: 300,
+ }
+ }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum RateLimitBackend {
+ Memory,
+ Redis,
+}
+
 impl Config {
  /// Load + merge + validate.
  ///
@@ -658,6 +701,20 @@ impl Config {
  "observability.metrics.prometheus.addr invalid socket address: {metrics_addr}"
  )));
  }
+ if self.ratelimit.backend == RateLimitBackend::Redis {
+ if self.ratelimit.redis.is_none() {
+ return Err(BootstrapError::Config(
+ "ratelimit.backend = redis requires a ratelimit.redis block".into(),
+ ));
+ }
+ // A zero concurrency TTL would prune a slot in the same second
+ // it was taken, silently disabling concurrency limiting.
+ if self.ratelimit.concurrency_ttl_secs == 0 {
+ return Err(BootstrapError::Config(
+ "ratelimit.concurrency_ttl_secs must be > 0 for the redis backend".into(),
+ ));
+ }
+ }
  Ok(())
  }
 }
@@ -784,6 +841,93 @@ admin:
  assert!(err.to_string().contains("admin.admin_keys"));
  }
 
+ #[test]
+ fn ratelimit_defaults_to_memory_backend() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+"#,
+ );
+ let cfg = Config::load_from_path(Some(f.path())).unwrap();
+ assert_eq!(cfg.ratelimit.backend, RateLimitBackend::Memory);
+ assert!(cfg.ratelimit.redis.is_none());
+ assert_eq!(cfg.ratelimit.concurrency_ttl_secs, 300);
+ }
+
+ #[test]
+ fn ratelimit_redis_backend_requires_redis_block() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+ratelimit:
+ backend: "redis"
+"#,
+ );
+ let err = Config::load_from_path(Some(f.path())).unwrap_err();
+ assert!(err.to_string().contains("ratelimit.redis"));
+ }
+
+ #[test]
+ fn rejects_zero_concurrency_ttl_for_redis_backend() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+ratelimit:
+ backend: "redis"
+ redis:
+ url: "redis://127.0.0.1:6379"
+ concurrency_ttl_secs: 0
+"#,
+ );
+ let err = Config::load_from_path(Some(f.path())).unwrap_err();
+ assert!(err.to_string().contains("concurrency_ttl_secs"));
+ }
+
+ #[test]
+ fn loads_ratelimit_redis_config() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+ratelimit:
+ backend: "redis"
+ redis:
+ url: "redis://127.0.0.1:6379"
+ concurrency_ttl_secs: 120
+"#,
+ );
+ let cfg = Config::load_from_path(Some(f.path())).unwrap();
+ assert_eq!(cfg.ratelimit.backend, RateLimitBackend::Redis);
+ assert_eq!(
+ cfg.ratelimit.redis.as_ref().unwrap().url,
+ "redis://127.0.0.1:6379"
+ );
+ assert_eq!(cfg.ratelimit.concurrency_ttl_secs, 120);
+ }
+
  #[test]
  fn rejects_invalid_bind_addr() {
  let f = write_yaml(

crates/aisix-core/src/lib.rs

@@ -24,7 +24,8 @@ pub mod snapshot;
 
 pub use config::{
  AdminConfig, CacheBackend, CacheConfig, Config, EtcdConfig, EtcdTlsConfig, ManagedConfig,
- ObservabilityConfig, ProxyConfig, RealIpConfig, TlsConfig,
+ ObservabilityConfig, ProxyConfig, RateLimitBackend, RateLimitConfig, RealIpConfig,
+ RedisCacheConfig, TlsConfig,
 };
 pub use error::{
  AdminError, AdminErrorEnvelope, BootstrapError, ProxyError, ProxyErrorEnvelope, RateLimitScope,

crates/aisix-proxy/src/chat.rs

@@ -245,7 +245,7 @@ pub async fn chat_completions(
  // current window state. We peek *after* the commit so
  // remaining-requests reflects the post-dispatch tally.
  let rl_limits = auth.key().rate_limit.clone().unwrap_or_default();
- if let Some(rl_status) = state.limiter.peek(&api_key_id, &rl_limits) {
+ if let Some(rl_status) = state.limiter.peek(&api_key_id, &rl_limits).await {
  crate::render::inject_ratelimit_headers(&mut success.response, &rl_status);
  state.metrics.set_rate_limit_remaining(
  &api_key_id,
@@ -843,8 +843,9 @@ async fn dispatch(
  &virtual_entry.id,
  &virtual_entry.value,
  );
- let reservation =
- crate::quota::enforce_rate_limit(state, auth, Some(&model_rl)).map_err(&with_model)?;
+ let reservation = crate::quota::enforce_rate_limit(state, auth, Some(&model_rl))
+ .await
+ .map_err(&with_model)?;
 
  let now = created_ts();
 
@@ -1073,7 +1074,7 @@ async fn dispatch(
  // permit was released here, letting a key capped at N run far more
  // than N simultaneous streams (#450).
  let post_stream_keys = reservation.keys();
- let stream_concurrency_hold = reservation.into_stream_hold(Arc::clone(&state.limiter));
+ let stream_concurrency_hold = reservation.into_stream_hold();
  // Capture everything the stream-completion callback needs so
  // it can fire `emit_usage_event` once the terminal SSE chunk
  // has yielded its `usage` block. Telemetry emission has to
@@ -1379,7 +1380,7 @@ async fn dispatch(
  if let (Some(cache), Some(key)) = (policy_cache.as_ref(), cache_key.as_ref()) {
  match cache.get(key).await {
  Ok(Some(cached)) => {
- reservation.commit_tokens(0);
+ reservation.commit_tokens(0).await;
  // #448: a cache hit is client-visible output just like a
  // fresh upstream response, so it must run output guardrails
  // before being returned — not bypass them.
@@ -1690,7 +1691,7 @@ async fn dispatch(
  let provider_request_id = upstream.id.clone();
  let provider_model_version = upstream.model.clone();
  let finish_reason = finish_reason_label(&upstream.finish_reason);
- reservation.commit_tokens(total);
+ reservation.commit_tokens(total).await;
 
  // cp-api recomputes cost server-side from its pricing catalog when
  // ingesting telemetry; the DP just records 0.0 on the wire.
@@ -1860,14 +1861,14 @@ async fn dispatch(
 /// response. `reservation` is the SINGLE entry-level reservation taken
 /// in `dispatch` — ensemble does not add per-sub-call reservations.
 #[allow(clippy::too_many_arguments)]
-async fn dispatch_ensemble<'a>(
- state: &'a ProxyState,
+async fn dispatch_ensemble(
+ state: &ProxyState,
  snapshot: &aisix_core::AisixSnapshot,
  virtual_entry: &aisix_core::ResourceEntry<aisix_core::Model>,
  req: &ChatFormat,
  request_id: &str,
  created_ts: i64,
- reservation: aisix_ratelimit::MultiReservation<'a, aisix_ratelimit::SystemClock>,
+ reservation: aisix_ratelimit::MultiReservation,
  resolved_chain: &Arc<dyn aisix_guardrails::Guardrail>,
  applied_guardrails: &[AppliedGuardrail],
  mut bypass_reason: Option<String>,
@@ -2006,7 +2007,7 @@ async fn dispatch_ensemble<'a>(
  ),
  };
  let survivor_total: u64 = panel.iter().map(|p| u64::from(p.usage.total_tokens)).sum();
- reservation.commit_tokens(survivor_total);
+ reservation.commit_tokens(survivor_total).await;
  for (index, member) in panel.iter().enumerate() {
  emit_panel_member(
  member, index, /* blocked */ false, /* bypass */ "",
@@ -2030,7 +2031,7 @@ async fn dispatch_ensemble<'a>(
  .sum();
  let judge_usage = outcome.response.usage.clone();
  let total_tokens = panel_total + u64::from(judge_usage.total_tokens);
- reservation.commit_tokens(total_tokens);
+ reservation.commit_tokens(total_tokens).await;
 
  // Emit one usage event per sub-call (each panel member + the judge),
  // all sharing `request_id`. `attempt_kind` is `"panel"` / `"judge"`;

crates/aisix-proxy/src/embeddings.rs

@@ -321,7 +321,9 @@ async fn dispatch(
  // and finalise RPM. Embeddings do report prompt_tokens via
  // EmbeddingResponse.usage; thread it through so TPM works
  // here even though other handlers commit 0.
- reservation.commit_tokens(embed_resp.usage.total_tokens as u64);
+ reservation
+ .commit_tokens(embed_resp.usage.total_tokens as u64)
+ .await;
  let provider_label = provider.to_ascii_lowercase();
  // Capture the prompt_tokens count BEFORE moving the
  // embed_resp into the JSON response — the handler needs
@@ -342,7 +344,7 @@ async fn dispatch(
  // (`upstream_called: false` → handler skips emit per the
  // chat.rs convention that we only attribute usage on a
  // real upstream completion).
- reservation.commit_tokens(0);
+ reservation.commit_tokens(0).await;
  let env = ErrorEnvelope::new(msg, "not_implemented");
  Ok(EmbedDispatchSuccess {
  response: (StatusCode::NOT_IMPLEMENTED, Json(env)).into_response(),
@@ -357,7 +359,7 @@ async fn dispatch(
  })
  }
  Err(e) => {
- reservation.commit_tokens(0);
+ reservation.commit_tokens(0).await;
  Err(ProxyError::Bridge(e))
  }
  }

.github/workflows/ci.yml

@@ -79,6 +79,10 @@ jobs:
  # every Config::load_from_path test as `redis_url` and breaks
  # the entire `aisix-core::config::tests` module.
  CACHE_TEST_REDIS_URL: redis://127.0.0.1:6379
+ # Picked up by crates/aisix-ratelimit/tests/redis_integration.rs
+ # (shared cluster-level counters, #798). Same skip-if-unset /
+ # no-AISIX_-prefix rules as CACHE_TEST_REDIS_URL above.
+ RATELIMIT_TEST_REDIS_URL: redis://127.0.0.1:6379
  # Picked up by crates/aisix-admin/tests/etcd_integration.rs.
  # Same skip-if-unset pattern as the Redis case above.
  ADMIN_TEST_ETCD_URL: http://127.0.0.1:2379

config.example.yaml

@@ -91,6 +91,24 @@ cache:
  # url: "redis://127.0.0.1:6379"
  # mode: "single" # single | cluster | sentinel
 
+# Rate-limit counter backend (api7/AISIX-Cloud#798).
+#
+# `memory` (default) keeps counters in this process, so a cluster of N
+# replicas enforces N× every configured limit. `redis` shares the
+# counters across every replica via one Redis, so the whole cluster
+# enforces ONE global window — set this on multi-replica deployments.
+# May point at the same Redis as `cache` (keys are namespaced
+# `aisix:rl:`). On a Redis outage the limiter fails open to per-replica
+# in-memory counting (logged) so traffic keeps flowing.
+ratelimit:
+ backend: "memory" # memory | redis
+ # redis:
+ # url: "redis://127.0.0.1:6379"
+ # mode: "single" # single | cluster | sentinel
+ # Seconds before an unreleased concurrency slot (crashed replica /
+ # hung upstream) is reclaimed. Redis backend only.
+ # concurrency_ttl_secs: 300
+
 # Models, API keys, provider keys, guardrails, cache policies, and
 # observability exporters are NOT defined in this file. They are stored
 # in etcd and managed via the Admin API (see docs/api-admin.md). This

config.managed.yaml

@@ -22,6 +22,12 @@
 # `AISIX_CACHE__BACKEND`, etc. — every config field is reachable via
 # `AISIX_<UPPER>__<UPPER>` (see crates/aisix-core/src/config.rs).
 #
+# For a multi-replica deployment, enable cluster-level rate limiting so
+# the cluster enforces one global window instead of N× per replica
+# (api7/AISIX-Cloud#798):
+# AISIX_RATELIMIT__BACKEND=redis
+# AISIX_RATELIMIT__REDIS__URL=redis://<host>:6379
+#
 # Subsequent boots re-use the mTLS bundle written under
 # `managed.mtls_dir`.
 

crates/aisix-core/src/config.rs

@@ -43,6 +43,12 @@ pub struct Config {
  pub observability: ObservabilityConfig,
  #[serde(default)]
  pub cache: CacheConfig,
+ /// Rate-limit counter backend. Defaults to per-process memory
+ /// (historical behaviour). Set `backend: redis` with a `redis` block
+ /// to share counters across every DP replica so a cluster enforces
+ /// one global window instead of one-per-replica (api7/AISIX-Cloud#798).
+ #[serde(default)]
+ pub ratelimit: RateLimitConfig,
  /// Optional managed-mode configuration. When `managed.enabled = true`
  /// the admin API and Playground endpoints are **not** bound — the DP
  /// is a pure etcd reader driven by the aisix.cloud control plane.
@@ -559,6 +565,43 @@ impl RedisCacheConfig {
  }
 }
 
+/// Rate-limit counter backend (api7/AISIX-Cloud#798).
+///
+/// `Memory` is the default: per-process fixed-window counters, so an
+/// N-replica cluster enforces N× the configured limit. `Redis` shares
+/// the counters across replicas via a single Redis so the whole cluster
+/// enforces one global window. The `redis` block is required iff
+/// `backend = redis` (validated at boot). Reuses [`RedisCacheConfig`]
+/// for the connection shape; may point at the same Redis as `cache`
+/// (keys are namespaced `aisix:rl:`).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields, default)]
+pub struct RateLimitConfig {
+ pub backend: RateLimitBackend,
+ pub redis: Option<RedisCacheConfig>,
+ /// Seconds after which an unreleased concurrency slot is reclaimed
+ /// (crashed replica / hung upstream). Generous enough for a long
+ /// streaming response. Redis backend only.
+ pub concurrency_ttl_secs: u64,
+}
+
+impl Default for RateLimitConfig {
+ fn default() -> Self {
+ Self {
+ backend: RateLimitBackend::Memory,
+ redis: None,
+ concurrency_ttl_secs: 300,
+ }
+ }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum RateLimitBackend {
+ Memory,
+ Redis,
+}
+
 impl Config {
  /// Load + merge + validate.
  ///
@@ -658,6 +701,20 @@ impl Config {
  "observability.metrics.prometheus.addr invalid socket address: {metrics_addr}"
  )));
  }
+ if self.ratelimit.backend == RateLimitBackend::Redis {
+ if self.ratelimit.redis.is_none() {
+ return Err(BootstrapError::Config(
+ "ratelimit.backend = redis requires a ratelimit.redis block".into(),
+ ));
+ }
+ // A zero concurrency TTL would prune a slot in the same second
+ // it was taken, silently disabling concurrency limiting.
+ if self.ratelimit.concurrency_ttl_secs == 0 {
+ return Err(BootstrapError::Config(
+ "ratelimit.concurrency_ttl_secs must be > 0 for the redis backend".into(),
+ ));
+ }
+ }
  Ok(())
  }
 }
@@ -784,6 +841,93 @@ admin:
  assert!(err.to_string().contains("admin.admin_keys"));
  }
 
+ #[test]
+ fn ratelimit_defaults_to_memory_backend() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+"#,
+ );
+ let cfg = Config::load_from_path(Some(f.path())).unwrap();
+ assert_eq!(cfg.ratelimit.backend, RateLimitBackend::Memory);
+ assert!(cfg.ratelimit.redis.is_none());
+ assert_eq!(cfg.ratelimit.concurrency_ttl_secs, 300);
+ }
+
+ #[test]
+ fn ratelimit_redis_backend_requires_redis_block() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+ratelimit:
+ backend: "redis"
+"#,
+ );
+ let err = Config::load_from_path(Some(f.path())).unwrap_err();
+ assert!(err.to_string().contains("ratelimit.redis"));
+ }
+
+ #[test]
+ fn rejects_zero_concurrency_ttl_for_redis_backend() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+ratelimit:
+ backend: "redis"
+ redis:
+ url: "redis://127.0.0.1:6379"
+ concurrency_ttl_secs: 0
+"#,
+ );
+ let err = Config::load_from_path(Some(f.path())).unwrap_err();
+ assert!(err.to_string().contains("concurrency_ttl_secs"));
+ }
+
+ #[test]
+ fn loads_ratelimit_redis_config() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+ratelimit:
+ backend: "redis"
+ redis:
+ url: "redis://127.0.0.1:6379"
+ concurrency_ttl_secs: 120
+"#,
+ );
+ let cfg = Config::load_from_path(Some(f.path())).unwrap();
+ assert_eq!(cfg.ratelimit.backend, RateLimitBackend::Redis);
+ assert_eq!(
+ cfg.ratelimit.redis.as_ref().unwrap().url,
+ "redis://127.0.0.1:6379"
+ );
+ assert_eq!(cfg.ratelimit.concurrency_ttl_secs, 120);
+ }
+
  #[test]
  fn rejects_invalid_bind_addr() {
  let f = write_yaml(

crates/aisix-core/src/lib.rs

@@ -24,7 +24,8 @@ pub mod snapshot;
 
 pub use config::{
  AdminConfig, CacheBackend, CacheConfig, Config, EtcdConfig, EtcdTlsConfig, ManagedConfig,
- ObservabilityConfig, ProxyConfig, RealIpConfig, TlsConfig,
+ ObservabilityConfig, ProxyConfig, RateLimitBackend, RateLimitConfig, RealIpConfig,
+ RedisCacheConfig, TlsConfig,
 };
 pub use error::{
  AdminError, AdminErrorEnvelope, BootstrapError, ProxyError, ProxyErrorEnvelope, RateLimitScope,

crates/aisix-proxy/src/chat.rs

@@ -245,7 +245,7 @@ pub async fn chat_completions(
  // current window state. We peek *after* the commit so
  // remaining-requests reflects the post-dispatch tally.
  let rl_limits = auth.key().rate_limit.clone().unwrap_or_default();
- if let Some(rl_status) = state.limiter.peek(&api_key_id, &rl_limits) {
+ if let Some(rl_status) = state.limiter.peek(&api_key_id, &rl_limits).await {
  crate::render::inject_ratelimit_headers(&mut success.response, &rl_status);
  state.metrics.set_rate_limit_remaining(
  &api_key_id,
@@ -843,8 +843,9 @@ async fn dispatch(
  &virtual_entry.id,
  &virtual_entry.value,
  );
- let reservation =
- crate::quota::enforce_rate_limit(state, auth, Some(&model_rl)).map_err(&with_model)?;
+ let reservation = crate::quota::enforce_rate_limit(state, auth, Some(&model_rl))
+ .await
+ .map_err(&with_model)?;
 
  let now = created_ts();
 
@@ -1073,7 +1074,7 @@ async fn dispatch(
  // permit was released here, letting a key capped at N run far more
  // than N simultaneous streams (#450).
  let post_stream_keys = reservation.keys();
- let stream_concurrency_hold = reservation.into_stream_hold(Arc::clone(&state.limiter));
+ let stream_concurrency_hold = reservation.into_stream_hold();
  // Capture everything the stream-completion callback needs so
  // it can fire `emit_usage_event` once the terminal SSE chunk
  // has yielded its `usage` block. Telemetry emission has to
@@ -1379,7 +1380,7 @@ async fn dispatch(
  if let (Some(cache), Some(key)) = (policy_cache.as_ref(), cache_key.as_ref()) {
  match cache.get(key).await {
  Ok(Some(cached)) => {
- reservation.commit_tokens(0);
+ reservation.commit_tokens(0).await;
  // #448: a cache hit is client-visible output just like a
  // fresh upstream response, so it must run output guardrails
  // before being returned — not bypass them.
@@ -1690,7 +1691,7 @@ async fn dispatch(
  let provider_request_id = upstream.id.clone();
  let provider_model_version = upstream.model.clone();
  let finish_reason = finish_reason_label(&upstream.finish_reason);
- reservation.commit_tokens(total);
+ reservation.commit_tokens(total).await;
 
  // cp-api recomputes cost server-side from its pricing catalog when
  // ingesting telemetry; the DP just records 0.0 on the wire.
@@ -1860,14 +1861,14 @@ async fn dispatch(
 /// response. `reservation` is the SINGLE entry-level reservation taken
 /// in `dispatch` — ensemble does not add per-sub-call reservations.
 #[allow(clippy::too_many_arguments)]
-async fn dispatch_ensemble<'a>(
- state: &'a ProxyState,
+async fn dispatch_ensemble(
+ state: &ProxyState,
  snapshot: &aisix_core::AisixSnapshot,
  virtual_entry: &aisix_core::ResourceEntry<aisix_core::Model>,
  req: &ChatFormat,
  request_id: &str,
  created_ts: i64,
- reservation: aisix_ratelimit::MultiReservation<'a, aisix_ratelimit::SystemClock>,
+ reservation: aisix_ratelimit::MultiReservation,
  resolved_chain: &Arc<dyn aisix_guardrails::Guardrail>,
  applied_guardrails: &[AppliedGuardrail],
  mut bypass_reason: Option<String>,
@@ -2006,7 +2007,7 @@ async fn dispatch_ensemble<'a>(
  ),
  };
  let survivor_total: u64 = panel.iter().map(|p| u64::from(p.usage.total_tokens)).sum();
- reservation.commit_tokens(survivor_total);
+ reservation.commit_tokens(survivor_total).await;
  for (index, member) in panel.iter().enumerate() {
  emit_panel_member(
  member, index, /* blocked */ false, /* bypass */ "",
@@ -2030,7 +2031,7 @@ async fn dispatch_ensemble<'a>(
  .sum();
  let judge_usage = outcome.response.usage.clone();
  let total_tokens = panel_total + u64::from(judge_usage.total_tokens);
- reservation.commit_tokens(total_tokens);
+ reservation.commit_tokens(total_tokens).await;
 
  // Emit one usage event per sub-call (each panel member + the judge),
  // all sharing `request_id`. `attempt_kind` is `"panel"` / `"judge"`;

crates/aisix-proxy/src/embeddings.rs

@@ -321,7 +321,9 @@ async fn dispatch(
  // and finalise RPM. Embeddings do report prompt_tokens via
  // EmbeddingResponse.usage; thread it through so TPM works
  // here even though other handlers commit 0.
- reservation.commit_tokens(embed_resp.usage.total_tokens as u64);
+ reservation
+ .commit_tokens(embed_resp.usage.total_tokens as u64)
+ .await;
  let provider_label = provider.to_ascii_lowercase();
  // Capture the prompt_tokens count BEFORE moving the
  // embed_resp into the JSON response — the handler needs
@@ -342,7 +344,7 @@ async fn dispatch(
  // (`upstream_called: false` → handler skips emit per the
  // chat.rs convention that we only attribute usage on a
  // real upstream completion).
- reservation.commit_tokens(0);
+ reservation.commit_tokens(0).await;
  let env = ErrorEnvelope::new(msg, "not_implemented");
  Ok(EmbedDispatchSuccess {
  response: (StatusCode::NOT_IMPLEMENTED, Json(env)).into_response(),
@@ -357,7 +359,7 @@ async fn dispatch(
  })
  }
  Err(e) => {
- reservation.commit_tokens(0);
+ reservation.commit_tokens(0).await;
  Err(ProxyError::Bridge(e))
  }
  }

.github/workflows/ci.yml

@@ -79,6 +79,10 @@ jobs:
  # every Config::load_from_path test as `redis_url` and breaks
  # the entire `aisix-core::config::tests` module.
  CACHE_TEST_REDIS_URL: redis://127.0.0.1:6379
+ # Picked up by crates/aisix-ratelimit/tests/redis_integration.rs
+ # (shared cluster-level counters, #798). Same skip-if-unset /
+ # no-AISIX_-prefix rules as CACHE_TEST_REDIS_URL above.
+ RATELIMIT_TEST_REDIS_URL: redis://127.0.0.1:6379
  # Picked up by crates/aisix-admin/tests/etcd_integration.rs.
  # Same skip-if-unset pattern as the Redis case above.
  ADMIN_TEST_ETCD_URL: http://127.0.0.1:2379

config.example.yaml

@@ -91,6 +91,24 @@ cache:
  # url: "redis://127.0.0.1:6379"
  # mode: "single" # single | cluster | sentinel
 
+# Rate-limit counter backend (api7/AISIX-Cloud#798).
+#
+# `memory` (default) keeps counters in this process, so a cluster of N
+# replicas enforces N× every configured limit. `redis` shares the
+# counters across every replica via one Redis, so the whole cluster
+# enforces ONE global window — set this on multi-replica deployments.
+# May point at the same Redis as `cache` (keys are namespaced
+# `aisix:rl:`). On a Redis outage the limiter fails open to per-replica
+# in-memory counting (logged) so traffic keeps flowing.
+ratelimit:
+ backend: "memory" # memory | redis
+ # redis:
+ # url: "redis://127.0.0.1:6379"
+ # mode: "single" # single | cluster | sentinel
+ # Seconds before an unreleased concurrency slot (crashed replica /
+ # hung upstream) is reclaimed. Redis backend only.
+ # concurrency_ttl_secs: 300
+
 # Models, API keys, provider keys, guardrails, cache policies, and
 # observability exporters are NOT defined in this file. They are stored
 # in etcd and managed via the Admin API (see docs/api-admin.md). This

config.managed.yaml

@@ -22,6 +22,12 @@
 # `AISIX_CACHE__BACKEND`, etc. — every config field is reachable via
 # `AISIX_<UPPER>__<UPPER>` (see crates/aisix-core/src/config.rs).
 #
+# For a multi-replica deployment, enable cluster-level rate limiting so
+# the cluster enforces one global window instead of N× per replica
+# (api7/AISIX-Cloud#798):
+# AISIX_RATELIMIT__BACKEND=redis
+# AISIX_RATELIMIT__REDIS__URL=redis://<host>:6379
+#
 # Subsequent boots re-use the mTLS bundle written under
 # `managed.mtls_dir`.
 

crates/aisix-core/src/config.rs

@@ -43,6 +43,12 @@ pub struct Config {
  pub observability: ObservabilityConfig,
  #[serde(default)]
  pub cache: CacheConfig,
+ /// Rate-limit counter backend. Defaults to per-process memory
+ /// (historical behaviour). Set `backend: redis` with a `redis` block
+ /// to share counters across every DP replica so a cluster enforces
+ /// one global window instead of one-per-replica (api7/AISIX-Cloud#798).
+ #[serde(default)]
+ pub ratelimit: RateLimitConfig,
  /// Optional managed-mode configuration. When `managed.enabled = true`
  /// the admin API and Playground endpoints are **not** bound — the DP
  /// is a pure etcd reader driven by the aisix.cloud control plane.
@@ -559,6 +565,43 @@ impl RedisCacheConfig {
  }
 }
 
+/// Rate-limit counter backend (api7/AISIX-Cloud#798).
+///
+/// `Memory` is the default: per-process fixed-window counters, so an
+/// N-replica cluster enforces N× the configured limit. `Redis` shares
+/// the counters across replicas via a single Redis so the whole cluster
+/// enforces one global window. The `redis` block is required iff
+/// `backend = redis` (validated at boot). Reuses [`RedisCacheConfig`]
+/// for the connection shape; may point at the same Redis as `cache`
+/// (keys are namespaced `aisix:rl:`).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields, default)]
+pub struct RateLimitConfig {
+ pub backend: RateLimitBackend,
+ pub redis: Option<RedisCacheConfig>,
+ /// Seconds after which an unreleased concurrency slot is reclaimed
+ /// (crashed replica / hung upstream). Generous enough for a long
+ /// streaming response. Redis backend only.
+ pub concurrency_ttl_secs: u64,
+}
+
+impl Default for RateLimitConfig {
+ fn default() -> Self {
+ Self {
+ backend: RateLimitBackend::Memory,
+ redis: None,
+ concurrency_ttl_secs: 300,
+ }
+ }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum RateLimitBackend {
+ Memory,
+ Redis,
+}
+
 impl Config {
  /// Load + merge + validate.
  ///
@@ -658,6 +701,20 @@ impl Config {
  "observability.metrics.prometheus.addr invalid socket address: {metrics_addr}"
  )));
  }
+ if self.ratelimit.backend == RateLimitBackend::Redis {
+ if self.ratelimit.redis.is_none() {
+ return Err(BootstrapError::Config(
+ "ratelimit.backend = redis requires a ratelimit.redis block".into(),
+ ));
+ }
+ // A zero concurrency TTL would prune a slot in the same second
+ // it was taken, silently disabling concurrency limiting.
+ if self.ratelimit.concurrency_ttl_secs == 0 {
+ return Err(BootstrapError::Config(
+ "ratelimit.concurrency_ttl_secs must be > 0 for the redis backend".into(),
+ ));
+ }
+ }
  Ok(())
  }
 }
@@ -784,6 +841,93 @@ admin:
  assert!(err.to_string().contains("admin.admin_keys"));
  }
 
+ #[test]
+ fn ratelimit_defaults_to_memory_backend() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+"#,
+ );
+ let cfg = Config::load_from_path(Some(f.path())).unwrap();
+ assert_eq!(cfg.ratelimit.backend, RateLimitBackend::Memory);
+ assert!(cfg.ratelimit.redis.is_none());
+ assert_eq!(cfg.ratelimit.concurrency_ttl_secs, 300);
+ }
+
+ #[test]
+ fn ratelimit_redis_backend_requires_redis_block() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+ratelimit:
+ backend: "redis"
+"#,
+ );
+ let err = Config::load_from_path(Some(f.path())).unwrap_err();
+ assert!(err.to_string().contains("ratelimit.redis"));
+ }
+
+ #[test]
+ fn rejects_zero_concurrency_ttl_for_redis_backend() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+ratelimit:
+ backend: "redis"
+ redis:
+ url: "redis://127.0.0.1:6379"
+ concurrency_ttl_secs: 0
+"#,
+ );
+ let err = Config::load_from_path(Some(f.path())).unwrap_err();
+ assert!(err.to_string().contains("concurrency_ttl_secs"));
+ }
+
+ #[test]
+ fn loads_ratelimit_redis_config() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+ratelimit:
+ backend: "redis"
+ redis:
+ url: "redis://127.0.0.1:6379"
+ concurrency_ttl_secs: 120
+"#,
+ );
+ let cfg = Config::load_from_path(Some(f.path())).unwrap();
+ assert_eq!(cfg.ratelimit.backend, RateLimitBackend::Redis);
+ assert_eq!(
+ cfg.ratelimit.redis.as_ref().unwrap().url,
+ "redis://127.0.0.1:6379"
+ );
+ assert_eq!(cfg.ratelimit.concurrency_ttl_secs, 120);
+ }
+
  #[test]
  fn rejects_invalid_bind_addr() {
  let f = write_yaml(

crates/aisix-core/src/lib.rs

@@ -24,7 +24,8 @@ pub mod snapshot;
 
 pub use config::{
  AdminConfig, CacheBackend, CacheConfig, Config, EtcdConfig, EtcdTlsConfig, ManagedConfig,
- ObservabilityConfig, ProxyConfig, RealIpConfig, TlsConfig,
+ ObservabilityConfig, ProxyConfig, RateLimitBackend, RateLimitConfig, RealIpConfig,
+ RedisCacheConfig, TlsConfig,
 };
 pub use error::{
  AdminError, AdminErrorEnvelope, BootstrapError, ProxyError, ProxyErrorEnvelope, RateLimitScope,

crates/aisix-proxy/src/chat.rs

@@ -245,7 +245,7 @@ pub async fn chat_completions(
  // current window state. We peek *after* the commit so
  // remaining-requests reflects the post-dispatch tally.
  let rl_limits = auth.key().rate_limit.clone().unwrap_or_default();
- if let Some(rl_status) = state.limiter.peek(&api_key_id, &rl_limits) {
+ if let Some(rl_status) = state.limiter.peek(&api_key_id, &rl_limits).await {
  crate::render::inject_ratelimit_headers(&mut success.response, &rl_status);
  state.metrics.set_rate_limit_remaining(
  &api_key_id,
@@ -843,8 +843,9 @@ async fn dispatch(
  &virtual_entry.id,
  &virtual_entry.value,
  );
- let reservation =
- crate::quota::enforce_rate_limit(state, auth, Some(&model_rl)).map_err(&with_model)?;
+ let reservation = crate::quota::enforce_rate_limit(state, auth, Some(&model_rl))
+ .await
+ .map_err(&with_model)?;
 
  let now = created_ts();
 
@@ -1073,7 +1074,7 @@ async fn dispatch(
  // permit was released here, letting a key capped at N run far more
  // than N simultaneous streams (#450).
  let post_stream_keys = reservation.keys();
- let stream_concurrency_hold = reservation.into_stream_hold(Arc::clone(&state.limiter));
+ let stream_concurrency_hold = reservation.into_stream_hold();
  // Capture everything the stream-completion callback needs so
  // it can fire `emit_usage_event` once the terminal SSE chunk
  // has yielded its `usage` block. Telemetry emission has to
@@ -1379,7 +1380,7 @@ async fn dispatch(
  if let (Some(cache), Some(key)) = (policy_cache.as_ref(), cache_key.as_ref()) {
  match cache.get(key).await {
  Ok(Some(cached)) => {
- reservation.commit_tokens(0);
+ reservation.commit_tokens(0).await;
  // #448: a cache hit is client-visible output just like a
  // fresh upstream response, so it must run output guardrails
  // before being returned — not bypass them.
@@ -1690,7 +1691,7 @@ async fn dispatch(
  let provider_request_id = upstream.id.clone();
  let provider_model_version = upstream.model.clone();
  let finish_reason = finish_reason_label(&upstream.finish_reason);
- reservation.commit_tokens(total);
+ reservation.commit_tokens(total).await;
 
  // cp-api recomputes cost server-side from its pricing catalog when
  // ingesting telemetry; the DP just records 0.0 on the wire.
@@ -1860,14 +1861,14 @@ async fn dispatch(
 /// response. `reservation` is the SINGLE entry-level reservation taken
 /// in `dispatch` — ensemble does not add per-sub-call reservations.
 #[allow(clippy::too_many_arguments)]
-async fn dispatch_ensemble<'a>(
- state: &'a ProxyState,
+async fn dispatch_ensemble(
+ state: &ProxyState,
  snapshot: &aisix_core::AisixSnapshot,
  virtual_entry: &aisix_core::ResourceEntry<aisix_core::Model>,
  req: &ChatFormat,
  request_id: &str,
  created_ts: i64,
- reservation: aisix_ratelimit::MultiReservation<'a, aisix_ratelimit::SystemClock>,
+ reservation: aisix_ratelimit::MultiReservation,
  resolved_chain: &Arc<dyn aisix_guardrails::Guardrail>,
  applied_guardrails: &[AppliedGuardrail],
  mut bypass_reason: Option<String>,
@@ -2006,7 +2007,7 @@ async fn dispatch_ensemble<'a>(
  ),
  };
  let survivor_total: u64 = panel.iter().map(|p| u64::from(p.usage.total_tokens)).sum();
- reservation.commit_tokens(survivor_total);
+ reservation.commit_tokens(survivor_total).await;
  for (index, member) in panel.iter().enumerate() {
  emit_panel_member(
  member, index, /* blocked */ false, /* bypass */ "",
@@ -2030,7 +2031,7 @@ async fn dispatch_ensemble<'a>(
  .sum();
  let judge_usage = outcome.response.usage.clone();
  let total_tokens = panel_total + u64::from(judge_usage.total_tokens);
- reservation.commit_tokens(total_tokens);
+ reservation.commit_tokens(total_tokens).await;
 
  // Emit one usage event per sub-call (each panel member + the judge),
  // all sharing `request_id`. `attempt_kind` is `"panel"` / `"judge"`;

crates/aisix-proxy/src/embeddings.rs

@@ -321,7 +321,9 @@ async fn dispatch(
  // and finalise RPM. Embeddings do report prompt_tokens via
  // EmbeddingResponse.usage; thread it through so TPM works
  // here even though other handlers commit 0.
- reservation.commit_tokens(embed_resp.usage.total_tokens as u64);
+ reservation
+ .commit_tokens(embed_resp.usage.total_tokens as u64)
+ .await;
  let provider_label = provider.to_ascii_lowercase();
  // Capture the prompt_tokens count BEFORE moving the
  // embed_resp into the JSON response — the handler needs
@@ -342,7 +344,7 @@ async fn dispatch(
  // (`upstream_called: false` → handler skips emit per the
  // chat.rs convention that we only attribute usage on a
  // real upstream completion).
- reservation.commit_tokens(0);
+ reservation.commit_tokens(0).await;
  let env = ErrorEnvelope::new(msg, "not_implemented");
  Ok(EmbedDispatchSuccess {
  response: (StatusCode::NOT_IMPLEMENTED, Json(env)).into_response(),
@@ -357,7 +359,7 @@ async fn dispatch(
  })
  }
  Err(e) => {
- reservation.commit_tokens(0);
+ reservation.commit_tokens(0).await;
  Err(ProxyError::Bridge(e))
  }
  }

.github/workflows/ci.yml

@@ -79,6 +79,10 @@ jobs:
  # every Config::load_from_path test as `redis_url` and breaks
  # the entire `aisix-core::config::tests` module.
  CACHE_TEST_REDIS_URL: redis://127.0.0.1:6379
+ # Picked up by crates/aisix-ratelimit/tests/redis_integration.rs
+ # (shared cluster-level counters, #798). Same skip-if-unset /
+ # no-AISIX_-prefix rules as CACHE_TEST_REDIS_URL above.
+ RATELIMIT_TEST_REDIS_URL: redis://127.0.0.1:6379
  # Picked up by crates/aisix-admin/tests/etcd_integration.rs.
  # Same skip-if-unset pattern as the Redis case above.
  ADMIN_TEST_ETCD_URL: http://127.0.0.1:2379

config.example.yaml

@@ -91,6 +91,24 @@ cache:
  # url: "redis://127.0.0.1:6379"
  # mode: "single" # single | cluster | sentinel
 
+# Rate-limit counter backend (api7/AISIX-Cloud#798).
+#
+# `memory` (default) keeps counters in this process, so a cluster of N
+# replicas enforces N× every configured limit. `redis` shares the
+# counters across every replica via one Redis, so the whole cluster
+# enforces ONE global window — set this on multi-replica deployments.
+# May point at the same Redis as `cache` (keys are namespaced
+# `aisix:rl:`). On a Redis outage the limiter fails open to per-replica
+# in-memory counting (logged) so traffic keeps flowing.
+ratelimit:
+ backend: "memory" # memory | redis
+ # redis:
+ # url: "redis://127.0.0.1:6379"
+ # mode: "single" # single | cluster | sentinel
+ # Seconds before an unreleased concurrency slot (crashed replica /
+ # hung upstream) is reclaimed. Redis backend only.
+ # concurrency_ttl_secs: 300
+
 # Models, API keys, provider keys, guardrails, cache policies, and
 # observability exporters are NOT defined in this file. They are stored
 # in etcd and managed via the Admin API (see docs/api-admin.md). This

config.managed.yaml

@@ -22,6 +22,12 @@
 # `AISIX_CACHE__BACKEND`, etc. — every config field is reachable via
 # `AISIX_<UPPER>__<UPPER>` (see crates/aisix-core/src/config.rs).
 #
+# For a multi-replica deployment, enable cluster-level rate limiting so
+# the cluster enforces one global window instead of N× per replica
+# (api7/AISIX-Cloud#798):
+# AISIX_RATELIMIT__BACKEND=redis
+# AISIX_RATELIMIT__REDIS__URL=redis://<host>:6379
+#
 # Subsequent boots re-use the mTLS bundle written under
 # `managed.mtls_dir`.
 

crates/aisix-core/src/config.rs

@@ -43,6 +43,12 @@ pub struct Config {
  pub observability: ObservabilityConfig,
  #[serde(default)]
  pub cache: CacheConfig,
+ /// Rate-limit counter backend. Defaults to per-process memory
+ /// (historical behaviour). Set `backend: redis` with a `redis` block
+ /// to share counters across every DP replica so a cluster enforces
+ /// one global window instead of one-per-replica (api7/AISIX-Cloud#798).
+ #[serde(default)]
+ pub ratelimit: RateLimitConfig,
  /// Optional managed-mode configuration. When `managed.enabled = true`
  /// the admin API and Playground endpoints are **not** bound — the DP
  /// is a pure etcd reader driven by the aisix.cloud control plane.
@@ -559,6 +565,43 @@ impl RedisCacheConfig {
  }
 }
 
+/// Rate-limit counter backend (api7/AISIX-Cloud#798).
+///
+/// `Memory` is the default: per-process fixed-window counters, so an
+/// N-replica cluster enforces N× the configured limit. `Redis` shares
+/// the counters across replicas via a single Redis so the whole cluster
+/// enforces one global window. The `redis` block is required iff
+/// `backend = redis` (validated at boot). Reuses [`RedisCacheConfig`]
+/// for the connection shape; may point at the same Redis as `cache`
+/// (keys are namespaced `aisix:rl:`).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields, default)]
+pub struct RateLimitConfig {
+ pub backend: RateLimitBackend,
+ pub redis: Option<RedisCacheConfig>,
+ /// Seconds after which an unreleased concurrency slot is reclaimed
+ /// (crashed replica / hung upstream). Generous enough for a long
+ /// streaming response. Redis backend only.
+ pub concurrency_ttl_secs: u64,
+}
+
+impl Default for RateLimitConfig {
+ fn default() -> Self {
+ Self {
+ backend: RateLimitBackend::Memory,
+ redis: None,
+ concurrency_ttl_secs: 300,
+ }
+ }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum RateLimitBackend {
+ Memory,
+ Redis,
+}
+
 impl Config {
  /// Load + merge + validate.
  ///
@@ -658,6 +701,20 @@ impl Config {
  "observability.metrics.prometheus.addr invalid socket address: {metrics_addr}"
  )));
  }
+ if self.ratelimit.backend == RateLimitBackend::Redis {
+ if self.ratelimit.redis.is_none() {
+ return Err(BootstrapError::Config(
+ "ratelimit.backend = redis requires a ratelimit.redis block".into(),
+ ));
+ }
+ // A zero concurrency TTL would prune a slot in the same second
+ // it was taken, silently disabling concurrency limiting.
+ if self.ratelimit.concurrency_ttl_secs == 0 {
+ return Err(BootstrapError::Config(
+ "ratelimit.concurrency_ttl_secs must be > 0 for the redis backend".into(),
+ ));
+ }
+ }
  Ok(())
  }
 }
@@ -784,6 +841,93 @@ admin:
  assert!(err.to_string().contains("admin.admin_keys"));
  }
 
+ #[test]
+ fn ratelimit_defaults_to_memory_backend() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+"#,
+ );
+ let cfg = Config::load_from_path(Some(f.path())).unwrap();
+ assert_eq!(cfg.ratelimit.backend, RateLimitBackend::Memory);
+ assert!(cfg.ratelimit.redis.is_none());
+ assert_eq!(cfg.ratelimit.concurrency_ttl_secs, 300);
+ }
+
+ #[test]
+ fn ratelimit_redis_backend_requires_redis_block() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+ratelimit:
+ backend: "redis"
+"#,
+ );
+ let err = Config::load_from_path(Some(f.path())).unwrap_err();
+ assert!(err.to_string().contains("ratelimit.redis"));
+ }
+
+ #[test]
+ fn rejects_zero_concurrency_ttl_for_redis_backend() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+ratelimit:
+ backend: "redis"
+ redis:
+ url: "redis://127.0.0.1:6379"
+ concurrency_ttl_secs: 0
+"#,
+ );
+ let err = Config::load_from_path(Some(f.path())).unwrap_err();
+ assert!(err.to_string().contains("concurrency_ttl_secs"));
+ }
+
+ #[test]
+ fn loads_ratelimit_redis_config() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+ratelimit:
+ backend: "redis"
+ redis:
+ url: "redis://127.0.0.1:6379"
+ concurrency_ttl_secs: 120
+"#,
+ );
+ let cfg = Config::load_from_path(Some(f.path())).unwrap();
+ assert_eq!(cfg.ratelimit.backend, RateLimitBackend::Redis);
+ assert_eq!(
+ cfg.ratelimit.redis.as_ref().unwrap().url,
+ "redis://127.0.0.1:6379"
+ );
+ assert_eq!(cfg.ratelimit.concurrency_ttl_secs, 120);
+ }
+
  #[test]
  fn rejects_invalid_bind_addr() {
  let f = write_yaml(

crates/aisix-core/src/lib.rs

@@ -24,7 +24,8 @@ pub mod snapshot;
 
 pub use config::{
  AdminConfig, CacheBackend, CacheConfig, Config, EtcdConfig, EtcdTlsConfig, ManagedConfig,
- ObservabilityConfig, ProxyConfig, RealIpConfig, TlsConfig,
+ ObservabilityConfig, ProxyConfig, RateLimitBackend, RateLimitConfig, RealIpConfig,
+ RedisCacheConfig, TlsConfig,
 };
 pub use error::{
  AdminError, AdminErrorEnvelope, BootstrapError, ProxyError, ProxyErrorEnvelope, RateLimitScope,

crates/aisix-proxy/src/chat.rs

@@ -245,7 +245,7 @@ pub async fn chat_completions(
  // current window state. We peek *after* the commit so
  // remaining-requests reflects the post-dispatch tally.
  let rl_limits = auth.key().rate_limit.clone().unwrap_or_default();
- if let Some(rl_status) = state.limiter.peek(&api_key_id, &rl_limits) {
+ if let Some(rl_status) = state.limiter.peek(&api_key_id, &rl_limits).await {
  crate::render::inject_ratelimit_headers(&mut success.response, &rl_status);
  state.metrics.set_rate_limit_remaining(
  &api_key_id,
@@ -843,8 +843,9 @@ async fn dispatch(
  &virtual_entry.id,
  &virtual_entry.value,
  );
- let reservation =
- crate::quota::enforce_rate_limit(state, auth, Some(&model_rl)).map_err(&with_model)?;
+ let reservation = crate::quota::enforce_rate_limit(state, auth, Some(&model_rl))
+ .await
+ .map_err(&with_model)?;
 
  let now = created_ts();
 
@@ -1073,7 +1074,7 @@ async fn dispatch(
  // permit was released here, letting a key capped at N run far more
  // than N simultaneous streams (#450).
  let post_stream_keys = reservation.keys();
- let stream_concurrency_hold = reservation.into_stream_hold(Arc::clone(&state.limiter));
+ let stream_concurrency_hold = reservation.into_stream_hold();
  // Capture everything the stream-completion callback needs so
  // it can fire `emit_usage_event` once the terminal SSE chunk
  // has yielded its `usage` block. Telemetry emission has to
@@ -1379,7 +1380,7 @@ async fn dispatch(
  if let (Some(cache), Some(key)) = (policy_cache.as_ref(), cache_key.as_ref()) {
  match cache.get(key).await {
  Ok(Some(cached)) => {
- reservation.commit_tokens(0);
+ reservation.commit_tokens(0).await;
  // #448: a cache hit is client-visible output just like a
  // fresh upstream response, so it must run output guardrails
  // before being returned — not bypass them.
@@ -1690,7 +1691,7 @@ async fn dispatch(
  let provider_request_id = upstream.id.clone();
  let provider_model_version = upstream.model.clone();
  let finish_reason = finish_reason_label(&upstream.finish_reason);
- reservation.commit_tokens(total);
+ reservation.commit_tokens(total).await;
 
  // cp-api recomputes cost server-side from its pricing catalog when
  // ingesting telemetry; the DP just records 0.0 on the wire.
@@ -1860,14 +1861,14 @@ async fn dispatch(
 /// response. `reservation` is the SINGLE entry-level reservation taken
 /// in `dispatch` — ensemble does not add per-sub-call reservations.
 #[allow(clippy::too_many_arguments)]
-async fn dispatch_ensemble<'a>(
- state: &'a ProxyState,
+async fn dispatch_ensemble(
+ state: &ProxyState,
  snapshot: &aisix_core::AisixSnapshot,
  virtual_entry: &aisix_core::ResourceEntry<aisix_core::Model>,
  req: &ChatFormat,
  request_id: &str,
  created_ts: i64,
- reservation: aisix_ratelimit::MultiReservation<'a, aisix_ratelimit::SystemClock>,
+ reservation: aisix_ratelimit::MultiReservation,
  resolved_chain: &Arc<dyn aisix_guardrails::Guardrail>,
  applied_guardrails: &[AppliedGuardrail],
  mut bypass_reason: Option<String>,
@@ -2006,7 +2007,7 @@ async fn dispatch_ensemble<'a>(
  ),
  };
  let survivor_total: u64 = panel.iter().map(|p| u64::from(p.usage.total_tokens)).sum();
- reservation.commit_tokens(survivor_total);
+ reservation.commit_tokens(survivor_total).await;
  for (index, member) in panel.iter().enumerate() {
  emit_panel_member(
  member, index, /* blocked */ false, /* bypass */ "",
@@ -2030,7 +2031,7 @@ async fn dispatch_ensemble<'a>(
  .sum();
  let judge_usage = outcome.response.usage.clone();
  let total_tokens = panel_total + u64::from(judge_usage.total_tokens);
- reservation.commit_tokens(total_tokens);
+ reservation.commit_tokens(total_tokens).await;
 
  // Emit one usage event per sub-call (each panel member + the judge),
  // all sharing `request_id`. `attempt_kind` is `"panel"` / `"judge"`;

crates/aisix-proxy/src/embeddings.rs

@@ -321,7 +321,9 @@ async fn dispatch(
  // and finalise RPM. Embeddings do report prompt_tokens via
  // EmbeddingResponse.usage; thread it through so TPM works
  // here even though other handlers commit 0.
- reservation.commit_tokens(embed_resp.usage.total_tokens as u64);
+ reservation
+ .commit_tokens(embed_resp.usage.total_tokens as u64)
+ .await;
  let provider_label = provider.to_ascii_lowercase();
  // Capture the prompt_tokens count BEFORE moving the
  // embed_resp into the JSON response — the handler needs
@@ -342,7 +344,7 @@ async fn dispatch(
  // (`upstream_called: false` → handler skips emit per the
  // chat.rs convention that we only attribute usage on a
  // real upstream completion).
- reservation.commit_tokens(0);
+ reservation.commit_tokens(0).await;
  let env = ErrorEnvelope::new(msg, "not_implemented");
  Ok(EmbedDispatchSuccess {
  response: (StatusCode::NOT_IMPLEMENTED, Json(env)).into_response(),
@@ -357,7 +359,7 @@ async fn dispatch(
  })
  }
  Err(e) => {
- reservation.commit_tokens(0);
+ reservation.commit_tokens(0).await;
  Err(ProxyError::Bridge(e))
  }
  }

.github/workflows/ci.yml

@@ -79,6 +79,10 @@ jobs:
  # every Config::load_from_path test as `redis_url` and breaks
  # the entire `aisix-core::config::tests` module.
  CACHE_TEST_REDIS_URL: redis://127.0.0.1:6379
+ # Picked up by crates/aisix-ratelimit/tests/redis_integration.rs
+ # (shared cluster-level counters, #798). Same skip-if-unset /
+ # no-AISIX_-prefix rules as CACHE_TEST_REDIS_URL above.
+ RATELIMIT_TEST_REDIS_URL: redis://127.0.0.1:6379
  # Picked up by crates/aisix-admin/tests/etcd_integration.rs.
  # Same skip-if-unset pattern as the Redis case above.
  ADMIN_TEST_ETCD_URL: http://127.0.0.1:2379

config.example.yaml

@@ -91,6 +91,24 @@ cache:
  # url: "redis://127.0.0.1:6379"
  # mode: "single" # single | cluster | sentinel
 
+# Rate-limit counter backend (api7/AISIX-Cloud#798).
+#
+# `memory` (default) keeps counters in this process, so a cluster of N
+# replicas enforces N× every configured limit. `redis` shares the
+# counters across every replica via one Redis, so the whole cluster
+# enforces ONE global window — set this on multi-replica deployments.
+# May point at the same Redis as `cache` (keys are namespaced
+# `aisix:rl:`). On a Redis outage the limiter fails open to per-replica
+# in-memory counting (logged) so traffic keeps flowing.
+ratelimit:
+ backend: "memory" # memory | redis
+ # redis:
+ # url: "redis://127.0.0.1:6379"
+ # mode: "single" # single | cluster | sentinel
+ # Seconds before an unreleased concurrency slot (crashed replica /
+ # hung upstream) is reclaimed. Redis backend only.
+ # concurrency_ttl_secs: 300
+
 # Models, API keys, provider keys, guardrails, cache policies, and
 # observability exporters are NOT defined in this file. They are stored
 # in etcd and managed via the Admin API (see docs/api-admin.md). This

config.managed.yaml

@@ -22,6 +22,12 @@
 # `AISIX_CACHE__BACKEND`, etc. — every config field is reachable via
 # `AISIX_<UPPER>__<UPPER>` (see crates/aisix-core/src/config.rs).
 #
+# For a multi-replica deployment, enable cluster-level rate limiting so
+# the cluster enforces one global window instead of N× per replica
+# (api7/AISIX-Cloud#798):
+# AISIX_RATELIMIT__BACKEND=redis
+# AISIX_RATELIMIT__REDIS__URL=redis://<host>:6379
+#
 # Subsequent boots re-use the mTLS bundle written under
 # `managed.mtls_dir`.
 

crates/aisix-core/src/config.rs

@@ -43,6 +43,12 @@ pub struct Config {
  pub observability: ObservabilityConfig,
  #[serde(default)]
  pub cache: CacheConfig,
+ /// Rate-limit counter backend. Defaults to per-process memory
+ /// (historical behaviour). Set `backend: redis` with a `redis` block
+ /// to share counters across every DP replica so a cluster enforces
+ /// one global window instead of one-per-replica (api7/AISIX-Cloud#798).
+ #[serde(default)]
+ pub ratelimit: RateLimitConfig,
  /// Optional managed-mode configuration. When `managed.enabled = true`
  /// the admin API and Playground endpoints are **not** bound — the DP
  /// is a pure etcd reader driven by the aisix.cloud control plane.
@@ -559,6 +565,43 @@ impl RedisCacheConfig {
  }
 }
 
+/// Rate-limit counter backend (api7/AISIX-Cloud#798).
+///
+/// `Memory` is the default: per-process fixed-window counters, so an
+/// N-replica cluster enforces N× the configured limit. `Redis` shares
+/// the counters across replicas via a single Redis so the whole cluster
+/// enforces one global window. The `redis` block is required iff
+/// `backend = redis` (validated at boot). Reuses [`RedisCacheConfig`]
+/// for the connection shape; may point at the same Redis as `cache`
+/// (keys are namespaced `aisix:rl:`).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields, default)]
+pub struct RateLimitConfig {
+ pub backend: RateLimitBackend,
+ pub redis: Option<RedisCacheConfig>,
+ /// Seconds after which an unreleased concurrency slot is reclaimed
+ /// (crashed replica / hung upstream). Generous enough for a long
+ /// streaming response. Redis backend only.
+ pub concurrency_ttl_secs: u64,
+}
+
+impl Default for RateLimitConfig {
+ fn default() -> Self {
+ Self {
+ backend: RateLimitBackend::Memory,
+ redis: None,
+ concurrency_ttl_secs: 300,
+ }
+ }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum RateLimitBackend {
+ Memory,
+ Redis,
+}
+
 impl Config {
  /// Load + merge + validate.
  ///
@@ -658,6 +701,20 @@ impl Config {
  "observability.metrics.prometheus.addr invalid socket address: {metrics_addr}"
  )));
  }
+ if self.ratelimit.backend == RateLimitBackend::Redis {
+ if self.ratelimit.redis.is_none() {
+ return Err(BootstrapError::Config(
+ "ratelimit.backend = redis requires a ratelimit.redis block".into(),
+ ));
+ }
+ // A zero concurrency TTL would prune a slot in the same second
+ // it was taken, silently disabling concurrency limiting.
+ if self.ratelimit.concurrency_ttl_secs == 0 {
+ return Err(BootstrapError::Config(
+ "ratelimit.concurrency_ttl_secs must be > 0 for the redis backend".into(),
+ ));
+ }
+ }
  Ok(())
  }
 }
@@ -784,6 +841,93 @@ admin:
  assert!(err.to_string().contains("admin.admin_keys"));
  }
 
+ #[test]
+ fn ratelimit_defaults_to_memory_backend() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+"#,
+ );
+ let cfg = Config::load_from_path(Some(f.path())).unwrap();
+ assert_eq!(cfg.ratelimit.backend, RateLimitBackend::Memory);
+ assert!(cfg.ratelimit.redis.is_none());
+ assert_eq!(cfg.ratelimit.concurrency_ttl_secs, 300);
+ }
+
+ #[test]
+ fn ratelimit_redis_backend_requires_redis_block() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+ratelimit:
+ backend: "redis"
+"#,
+ );
+ let err = Config::load_from_path(Some(f.path())).unwrap_err();
+ assert!(err.to_string().contains("ratelimit.redis"));
+ }
+
+ #[test]
+ fn rejects_zero_concurrency_ttl_for_redis_backend() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+ratelimit:
+ backend: "redis"
+ redis:
+ url: "redis://127.0.0.1:6379"
+ concurrency_ttl_secs: 0
+"#,
+ );
+ let err = Config::load_from_path(Some(f.path())).unwrap_err();
+ assert!(err.to_string().contains("concurrency_ttl_secs"));
+ }
+
+ #[test]
+ fn loads_ratelimit_redis_config() {
+ let f = write_yaml(
+ r#"
+etcd:
+ endpoints: ["http://localhost:2379"]
+proxy:
+ addr: "0.0.0.0:3000"
+admin:
+ addr: "127.0.0.1:3001"
+ admin_keys: ["k1"]
+ratelimit:
+ backend: "redis"
+ redis:
+ url: "redis://127.0.0.1:6379"
+ concurrency_ttl_secs: 120
+"#,
+ );
+ let cfg = Config::load_from_path(Some(f.path())).unwrap();
+ assert_eq!(cfg.ratelimit.backend, RateLimitBackend::Redis);
+ assert_eq!(
+ cfg.ratelimit.redis.as_ref().unwrap().url,
+ "redis://127.0.0.1:6379"
+ );
+ assert_eq!(cfg.ratelimit.concurrency_ttl_secs, 120);
+ }
+
  #[test]
  fn rejects_invalid_bind_addr() {
  let f = write_yaml(

crates/aisix-core/src/lib.rs

@@ -24,7 +24,8 @@ pub mod snapshot;
 
 pub use config::{
  AdminConfig, CacheBackend, CacheConfig, Config, EtcdConfig, EtcdTlsConfig, ManagedConfig,
- ObservabilityConfig, ProxyConfig, RealIpConfig, TlsConfig,
+ ObservabilityConfig, ProxyConfig, RateLimitBackend, RateLimitConfig, RealIpConfig,
+ RedisCacheConfig, TlsConfig,
 };
 pub use error::{
  AdminError, AdminErrorEnvelope, BootstrapError, ProxyError, ProxyErrorEnvelope, RateLimitScope,

crates/aisix-proxy/src/chat.rs

@@ -245,7 +245,7 @@ pub async fn chat_completions(
  // current window state. We peek *after* the commit so
  // remaining-requests reflects the post-dispatch tally.
  let rl_limits = auth.key().rate_limit.clone().unwrap_or_default();
- if let Some(rl_status) = state.limiter.peek(&api_key_id, &rl_limits) {
+ if let Some(rl_status) = state.limiter.peek(&api_key_id, &rl_limits).await {
  crate::render::inject_ratelimit_headers(&mut success.response, &rl_status);
  state.metrics.set_rate_limit_remaining(
  &api_key_id,
@@ -843,8 +843,9 @@ async fn dispatch(
  &virtual_entry.id,
  &virtual_entry.value,
  );
- let reservation =
- crate::quota::enforce_rate_limit(state, auth, Some(&model_rl)).map_err(&with_model)?;
+ let reservation = crate::quota::enforce_rate_limit(state, auth, Some(&model_rl))
+ .await
+ .map_err(&with_model)?;
 
  let now = created_ts();
 
@@ -1073,7 +1074,7 @@ async fn dispatch(
  // permit was released here, letting a key capped at N run far more
  // than N simultaneous streams (#450).
  let post_stream_keys = reservation.keys();
- let stream_concurrency_hold = reservation.into_stream_hold(Arc::clone(&state.limiter));
+ let stream_concurrency_hold = reservation.into_stream_hold();
  // Capture everything the stream-completion callback needs so
  // it can fire `emit_usage_event` once the terminal SSE chunk
  // has yielded its `usage` block. Telemetry emission has to
@@ -1379,7 +1380,7 @@ async fn dispatch(
  if let (Some(cache), Some(key)) = (policy_cache.as_ref(), cache_key.as_ref()) {
  match cache.get(key).await {
  Ok(Some(cached)) => {
- reservation.commit_tokens(0);
+ reservation.commit_tokens(0).await;
  // #448: a cache hit is client-visible output just like a
  // fresh upstream response, so it must run output guardrails
  // before being returned — not bypass them.
@@ -1690,7 +1691,7 @@ async fn dispatch(
  let provider_request_id = upstream.id.clone();
  let provider_model_version = upstream.model.clone();
  let finish_reason = finish_reason_label(&upstream.finish_reason);
- reservation.commit_tokens(total);
+ reservation.commit_tokens(total).await;
 
  // cp-api recomputes cost server-side from its pricing catalog when
  // ingesting telemetry; the DP just records 0.0 on the wire.
@@ -1860,14 +1861,14 @@ async fn dispatch(
 /// response. `reservation` is the SINGLE entry-level reservation taken
 /// in `dispatch` — ensemble does not add per-sub-call reservations.
 #[allow(clippy::too_many_arguments)]
-async fn dispatch_ensemble<'a>(
- state: &'a ProxyState,
+async fn dispatch_ensemble(
+ state: &ProxyState,
  snapshot: &aisix_core::AisixSnapshot,
  virtual_entry: &aisix_core::ResourceEntry<aisix_core::Model>,
  req: &ChatFormat,
  request_id: &str,
  created_ts: i64,
- reservation: aisix_ratelimit::MultiReservation<'a, aisix_ratelimit::SystemClock>,
+ reservation: aisix_ratelimit::MultiReservation,
  resolved_chain: &Arc<dyn aisix_guardrails::Guardrail>,
  applied_guardrails: &[AppliedGuardrail],
  mut bypass_reason: Option<String>,
@@ -2006,7 +2007,7 @@ async fn dispatch_ensemble<'a>(
  ),
  };
  let survivor_total: u64 = panel.iter().map(|p| u64::from(p.usage.total_tokens)).sum();
- reservation.commit_tokens(survivor_total);
+ reservation.commit_tokens(survivor_total).await;
  for (index, member) in panel.iter().enumerate() {
  emit_panel_member(
  member, index, /* blocked */ false, /* bypass */ "",
@@ -2030,7 +2031,7 @@ async fn dispatch_ensemble<'a>(
  .sum();
  let judge_usage = outcome.response.usage.clone();
  let total_tokens = panel_total + u64::from(judge_usage.total_tokens);
- reservation.commit_tokens(total_tokens);
+ reservation.commit_tokens(total_tokens).await;
 
  // Emit one usage event per sub-call (each panel member + the judge),
  // all sharing `request_id`. `attempt_kind` is `"panel"` / `"judge"`;

crates/aisix-proxy/src/embeddings.rs

@@ -321,7 +321,9 @@ async fn dispatch(
  // and finalise RPM. Embeddings do report prompt_tokens via
  // EmbeddingResponse.usage; thread it through so TPM works
  // here even though other handlers commit 0.
- reservation.commit_tokens(embed_resp.usage.total_tokens as u64);
+ reservation
+ .commit_tokens(embed_resp.usage.total_tokens as u64)
+ .await;
  let provider_label = provider.to_ascii_lowercase();
  // Capture the prompt_tokens count BEFORE moving the
  // embed_resp into the JSON response — the handler needs
@@ -342,7 +344,7 @@ async fn dispatch(
  // (`upstream_called: false` → handler skips emit per the
  // chat.rs convention that we only attribute usage on a
  // real upstream completion).
- reservation.commit_tokens(0);
+ reservation.commit_tokens(0).await;
  let env = ErrorEnvelope::new(msg, "not_implemented");
  Ok(EmbedDispatchSuccess {
  response: (StatusCode::NOT_IMPLEMENTED, Json(env)).into_response(),
@@ -357,7 +359,7 @@ async fn dispatch(
  })
  }
  Err(e) => {
- reservation.commit_tokens(0);
+ reservation.commit_tokens(0).await;
  Err(ProxyError::Bridge(e))
  }
  }