IBM · lucarlig · Apr 22, 2026 · Apr 20, 2026 · Apr 20, 2026 · Apr 20, 2026
@@ -39,6 +39,11 @@ If any configured dimension is exceeded, the plugin returns a violation with HTT
     # Redis options (required when backend: redis)
     redis_url: "redis://redis:6379/0"
     redis_key_prefix: "rl"
+
+    # Backend failure policy (default: "open" — fail-open)
+    # "closed" — return HTTP 503 BACKEND_UNAVAILABLE violation when the
+    # backend can't be reached (correctness over availability)
+    fail_mode: "open"
 ```
 
 ### Configuration reference
@@ -52,8 +57,13 @@ If any configured dimension is exceeded, the plugin returns a violation with HTT
 | `backend` | string | `"memory"` | `"memory"` or `"redis"` |
 | `redis_url` | string | `null` | Redis connection URL (required when `backend: redis`) |
 | `redis_key_prefix` | string | `"rl"` | Prefix for all Redis keys |
+| `fail_mode` | string | `"open"` | Behaviour when the backend can't be reached: `"open"` allows the request through, `"closed"` blocks with a 503 `BACKEND_UNAVAILABLE` violation |
+
+**Rate string format:** `"<count>/<unit>"` where unit is `s`/`sec`/`second`, `m`/`min`/`minute`, or `h`/`hr`/`hour`. Malformed strings raise `ValueError` at startup. Counts above `1_000_000` are rejected as a sanity ceiling — anything higher is almost certainly a misconfig or a denial-of-service vector against the memory backend.
 
-**Rate string format:** `"<count>/<unit>"` where unit is `s`/`sec`/`second`, `m`/`min`/`minute`, or `h`/`hr`/`hour`. Malformed strings raise `ValueError` at startup.
+**Unknown config keys** (e.g. a typo like `redis_ur`) are logged at `WARN` at engine init alongside the accepted-key list, instead of being silently ignored.
+
+**Invalid `fail_mode` values** (e.g. `"clsoed"`) are logged at `WARN` and fall back to `"open"` so an operator's typo surfaces instead of silently disabling the hardening they asked for.
 
 **Omitting a dimension** (e.g. no `by_tenant`) means that dimension is unlimited — no counter is tracked for it.
 
@@ -107,10 +117,26 @@ Each identity (user, tenant, tool) has a bucket that holds up to `count` tokens.
 - `token_bucket`: atomic Lua script — reads `{tokens, last_refill}` hash, refills proportionally, consumes 1 token, writes back — one round-trip, no race condition
 - All gateway instances share the same counter — the configured limit is the true cluster-wide limit
 - Requires `redis_url` to be set
-- If Redis is unavailable, the plugin fails open — the request is allowed through without rate limiting. This is a deliberate design choice: an infrastructure failure must never block legitimate traffic. Operators should monitor for rate-limiter error logs and treat them as high-priority alerts
+- **Backend failure policy** is governed by `fail_mode`:
+  - `"open"` (default) — the request is allowed through without rate limiting. Availability over correctness; an infrastructure failure must never block legitimate traffic. Operators should monitor for rate-limiter error logs and treat them as high-priority alerts.
+  - `"closed"` — the request is blocked with a `PluginViolation` (code `BACKEND_UNAVAILABLE`, HTTP 503, `Retry-After: 1`). Correctness over availability; pick this when a failed rate-limit check is less acceptable than a brief outage.
 
 **Multi-instance deployment (important):** The `memory` backend is local to a single gateway instance — rate limit counters are not shared across replicas. For multi-instance deployments (e.g., behind nginx or on OpenShift with multiple gateway pods), always use `backend: redis` to ensure rate limits are enforced correctly across all instances.
 
+### Tenant-scoped Redis key layout
+
+When the plugin context carries a `tenant_id`, every dimension key is prefixed with it so counters are isolated per tenant:
+
+```
+rl:{tenant_id}:user:{email}:{window_seconds}
+rl:{tenant_id}:tenant:{tenant_id}:{window_seconds}
+rl:{tenant_id}:tool:{tool_name}:{window_seconds}
+```
+
+When `tenant_id` is absent (single-tenant deployments), the prefix is omitted and keys revert to the pre-tenant-scoping layout (`rl:user:{email}:{window}`), so single-tenant behaviour is unchanged.
+
+**Upgrade note:** the first deploy of the tenant-scoping change causes counters under `rl:user:*` / `rl:tool:*` to be orphaned while new writes land at `rl:{tenant}:user:*`. Counters effectively reset once for all in-flight windows — non-event for typical second/minute windows.
+
 ## Examples
 
 ### Single-instance (default config)
@@ -170,6 +196,15 @@ config:
 
 In `permissive` mode the plugin records violations and emits `X-RateLimit-*` headers but does not block requests. Useful for baselining traffic before switching to `enforce`.
 
+## Lifecycle
+
+The plugin participates in the plugin manager's lifecycle contract:
+
+- `async def initialize(self)` — invoked once when the plugin manager constructs the plugin. Logs one `INFO` record naming the active backend (`memory` / `redis`).
+- `async def shutdown(self)` — invoked when the plugin manager tears the plugin down (runtime disable, re-instantiation after a config change). Releases backend-held resources — specifically, drops the Rust core's cached Redis multiplexed connection and the SCRIPT LOAD SHA cache. In-flight requests already hold their own clones of the connection and remain valid; the cached reference is replaced on the next request.
+
+Without `shutdown`, the cached Redis connection would leak across plugin re-instantiation, producing connection churn on the server.
+
 ## Limitations
 
 | Limitation | Severity | Status |

@@ -3,6 +3,8 @@
 
 from __future__ import annotations
 
+import logging
+
 try:
     from mcpgateway.plugins.framework import Plugin, PromptPrehookResult, ToolPreInvokeResult
 except ModuleNotFoundError:
@@ -45,6 +47,7 @@ class RateLimiterConfig:
         "backend",
         "redis_url",
         "redis_key_prefix",
+        "fail_mode",
     )
 
     def __init__(self, **overrides) -> None:
@@ -54,20 +57,46 @@ def __init__(self, **overrides) -> None:
             setattr(self, field, config.get(field))
 
 
+_logger = logging.getLogger(__name__)
+
+
 class RateLimiterPlugin(Plugin):
     """Gateway-facing Plugin subclass that delegates behavior to Rust."""
 
     def __init__(self, config) -> None:
         super().__init__(config)
         self._core = RateLimiterPluginCore(config.config or {})
 
+    async def initialize(self) -> None:
+        """Lifecycle hook: called once when the plugin manager constructs us."""
+        cfg = self.config.config or {}
+        backend = cfg.get("backend", "memory")
+        _logger.info("rate limiter initialized: backend=%s", backend)
+
+    async def shutdown(self) -> None:
+        """Lifecycle hook: release Rust-held resources (e.g. Redis connection).
+
+        The plugin manager calls this on disable and on re-instantiation.
+        Without it, the cached Redis connection leaks until the plugin
+        instance is garbage-collected.
+        """
+        try:
+            self._core.shutdown()
+        except Exception:
+            _logger.exception("rate limiter shutdown: core.shutdown() raised")
+
     async def prompt_pre_fetch(self, payload, context):
+        # The Rust core handles fail_mode policy internally (open vs closed)
+        # and logs backend errors via log_exception. The except here is a
+        # final safety net for the unlikely case that a non-backend bug in
+        # the core escapes as a Python exception.
         try:
             result = self._core.prompt_pre_fetch(payload, context)
             if hasattr(result, "__await__"):
                 return await result
             return result
         except Exception:
+            _logger.warning("rate limiter prompt_pre_fetch: unexpected core error; allowing request", exc_info=True)
             return PromptPrehookResult()
 
     async def tool_pre_invoke(self, payload, context):
@@ -77,6 +106,7 @@ async def tool_pre_invoke(self, payload, context):
                 return await result
             return result
         except Exception:
+            _logger.warning("rate limiter tool_pre_invoke: unexpected core error; allowing request", exc_info=True)
             return ToolPreInvokeResult()
 
 

@@ -102,8 +102,15 @@ class RateLimiterEngine:
         - `backend`: `"memory"` (default) or `"redis"`
         - `redis_url`: required when `backend = "redis"`
         - `redis_key_prefix`: key namespace prefix (default `"rl"`)
+        - `fail_mode`: `"open"` (default) or `"closed"` — handled by the
+          plugin shim, but accepted here so it doesn't trip the unknown-key
+          warning below.
+
+        Any other key in the dict is logged at WARN so misspellings (e.g.
+        `redis_ur` instead of `redis_url`) surface visibly instead of being
+        silently ignored.
         """
-    def check(self, user: builtins.str, tenant: typing.Optional[builtins.str], tool: builtins.str, now_unix: builtins.int, include_retry_after: builtins.bool) -> tuple[builtins.bool, dict, dict]:
+    def check(self, user: builtins.str, tenant: typing.Optional[builtins.str], tool: builtins.str, now_unix: builtins.int, include_retry_after: builtins.bool, context_prefix: typing.Optional[builtins.str]) -> tuple[builtins.bool, dict, dict]:
         r"""
         High-level check: builds dimension keys internally, evaluates, and
         returns pre-built Python dicts for headers and metadata.
@@ -113,13 +120,18 @@ class RateLimiterEngine:
 
         Returns `(allowed, headers_dict, meta_dict)`.
 
+        When `context_prefix` is provided (e.g. a team/tenant ID), it is
+        prepended to every dimension key so that separate plugin instances
+        for different tenants use isolated Redis counters instead of sharing
+        a single key namespace.
+
         **Note:** The Redis backend arm uses `block_on()` on a dedicated Tokio
         runtime, which would deadlock if called from within a Tokio context.
         The Python wrapper routes Redis to `check_async()` instead; this sync
         path is intended for the memory backend.  The `debug_assert` below
         guards against accidental misuse.
         """
-    def check_async(self, user: builtins.str, tenant: typing.Optional[builtins.str], tool: builtins.str, now_unix: builtins.int, include_retry_after: builtins.bool) -> typing.Any:
+    def check_async(self, user: builtins.str, tenant: typing.Optional[builtins.str], tool: builtins.str, now_unix: builtins.int, include_retry_after: builtins.bool, context_prefix: typing.Optional[builtins.str]) -> typing.Any:
         r"""
         Async variant of `check()` for Redis-backed deployments.
 
@@ -129,6 +141,12 @@ class RateLimiterEngine:
 @typing.final
 class RateLimiterPluginCore:
     def __new__(cls, config: dict) -> RateLimiterPluginCore: ...
+    def shutdown(self) -> None:
+        r"""
+        Release backend-held resources (e.g. the cached Redis multiplexed
+        connection). Called by the Python shim's `shutdown()` when the plugin
+        framework tears the plugin down.
+        """
     def prompt_pre_fetch(self, payload: typing.Any, context: typing.Any) -> typing.Any: ...
     def tool_pre_invoke(self, payload: typing.Any, context: typing.Any) -> typing.Any: ...
 

@@ -31,6 +31,10 @@ pub enum ConfigError {
     InvalidRateString(String),
     #[error("rate count must be > 0, got {0}")]
     ZeroCount(u64),
+    #[error(
+        "rate count {0} exceeds the sanity ceiling of {MAX_RATE_COUNT}; if you really need this, raise the limit deliberately"
+    )]
+    CountAboveCeiling(u64),
     #[error("{field}: {message}")]
     FieldError { field: String, message: String },
     #[error(
@@ -39,6 +43,11 @@ pub enum ConfigError {
     InvalidAlgorithm(String),
 }
 
+/// Upper bound for any single rate-limit count. Above this, configurations
+/// are almost certainly typos or denial-of-service vectors against the
+/// memory backend (which allocates per dimension key).
+pub const MAX_RATE_COUNT: u64 = 1_000_000;
+
 /// Parse a rate string like `"30/m"`, `"100/s"`, `"1000/h"`.
 ///
 /// Accepted units (case-insensitive): `s`, `sec`, `second`, `m`, `min`,
@@ -57,6 +66,9 @@ pub fn parse_rate(s: &str) -> Result<RateLimit, ConfigError> {
     if count == 0 {
         return Err(ConfigError::ZeroCount(count));
     }
+    if count > MAX_RATE_COUNT {
+        return Err(ConfigError::CountAboveCeiling(count));
+    }
 
     let window_secs: u64 = match unit_str.trim().to_ascii_lowercase().as_str() {
         "s" | "sec" | "second" => 1,
@@ -198,6 +210,25 @@ mod tests {
         assert!(parse_rate("0/s").is_err());
     }
 
+    // --- Bounds: reject absurd counts so misconfig can't allocate huge
+    //     windows, overflow internal math, or starve other memory.
+    //     See config_hardening_bounds_rationale in the README.
+
+    #[test]
+    fn parse_rate_rejects_count_above_upper_bound() {
+        assert!(
+            parse_rate("99999999/h").is_err(),
+            "parse_rate must reject counts above the sanity ceiling"
+        );
+    }
+
+    #[test]
+    fn parse_rate_accepts_reasonable_large_count() {
+        // 100000/h is plausibly a real quota — must still parse.
+        let r = parse_rate("100000/h").unwrap();
+        assert_eq!(r.count, 100_000);
+    }
+
     // --- Algorithm::from_str ---
 
     #[test]