fix: eliminate per-request plugin manager rebuilds (#6532)

Two root causes drove the plugin manager (including the external OPA
rules plugin) to re-initialize on nearly every request:

Root cause A — DEFAULT_CACHE_TTL=30 forced a full manager rebuild every
30 seconds under sustained traffic. Each rebuild called manager.initialize()
which re-imports and re-instantiates every configured plugin.

Root cause B — per-tool context_id fragmentation created one cache slot
per (team_id, tool_name) pair, triggering O(N tools) cold-start rebuilds.

Fix A: set DEFAULT_CACHE_TTL=0 so managers live for the worker lifetime.
Add PLUGIN_MANAGER_CACHE_TTL config knob (default 0) so operators can
re-enable timed refresh when needed. Explicit invalidation signals
(Redis pub/sub mode_change, binding_change, team_binding_change) already
handle config propagation; TTL-driven expiry was a blunt workaround.

Fix B: introduce TEAM_SCOPED_TOOL_SENTINEL ('__team__'). When
_build_manager finds no per-tool ToolPluginBinding rows, it normalises
the cache key from '<team_id>::<tool_name>' to '<team_id>::__team__'.
get_manager checks the canonical team-scoped slot before spawning a new
build task. All unbound tools in the same team share one manager instance
instead of each creating their own slot. Tools with explicit bindings
retain their per-tool key and dedicated manager.

Security invariant preserved: the manager holds no per-request state.
Per-request identity and auth context are passed as hook call arguments
on every invocation; manager lifetime does not affect cross-tenant
isolation.

Closes #6532

Signed-off-by: Lang-Akshay <akshay.shinde26@ibm.com>

Author: Lang-Akshay

mcpgateway/config.py

@@ -1392,6 +1392,17 @@ def normalize_x_frame_options(cls, v: Optional[str]) -> Optional[str]:
             "Requires a server restart to take effect."
         ),
     )
+    plugin_manager_cache_ttl: int = Field(
+        default=0,
+        ge=0,
+        description=(
+            "TTL in seconds for per-context plugin manager cache entries. "
+            "0 (default) disables expiry so managers live for the worker lifetime and are refreshed "
+            "only via explicit invalidation signals (Redis pub/sub, binding changes). "
+            "Set to a positive integer only when you need periodic config refresh and "
+            "accept the per-TTL re-initialization cost for plugins such as the external OPA rules plugin."
+        ),
+    )
 
     # Values used to detect unconfigured or insecure deployment states
     SENTINEL_VALUES: ClassVar[list[str]] = ["", "UNCONFIGURED"]

mcpgateway/main.py

@@ -1592,6 +1592,7 @@ def _notification_handler_factory(url: str, gateway_id, *, downstream_session_id
                 hook_policies=HOOK_PAYLOAD_POLICIES,
                 observability=None,  # Will be set later if needed
                 db_factory=SessionLocal,
+                cache_ttl=settings.plugin_manager_cache_ttl,
             )
             logger.info("Plugin manager factory initialized")
         except Exception as init_exc:

mcpgateway/plugins/__init__.py

@@ -307,6 +307,7 @@ def init_plugin_manager_factory(
     hook_policies: dict,
     observability: Optional[ObservabilityProvider] = None,
     db_factory: Optional[Callable] = None,
+    cache_ttl: Optional[int] = None,
 ) -> None:
     """Explicitly initialise the global plugin manager factory."""
     global _plugin_manager_factory
@@ -321,6 +322,7 @@ def init_plugin_manager_factory(
         hook_policies=hook_policies,
         observability=observability,
         db_factory=db_factory,
+        cache_ttl=cache_ttl,
     )
 
 

mcpgateway/plugins/gateway_plugin_manager.py

@@ -39,6 +39,11 @@
 logger = logging.getLogger(__name__)
 
 CONTEXT_ID_SEPARATOR = "::"
+# Sentinel tool-name used as the shared cache slot for team-scoped tools that
+# have no per-tool ToolPluginBinding rows.  All tools in a team without explicit
+# bindings resolve to "<team_id>::__team__" so they share one manager instance
+# instead of each creating their own slot.
+TEAM_SCOPED_TOOL_SENTINEL = "__team__"
 
 _LEGACY_MODE_TO_PLUGIN_MODE: dict[str, tuple[PluginMode, Optional[OnError]]] = {
     "enforce": (PluginMode.SEQUENTIAL, None),
@@ -86,7 +91,7 @@ class TenantPluginManagerFactory:
     Call sites should use :func:`make_context_id` to construct them.
     """
 
-    DEFAULT_CACHE_TTL = 30
+    DEFAULT_CACHE_TTL = 0  # 0 = no expiry; managers live for the worker lifetime
     CONTEXT_ID_SEPARATOR = CONTEXT_ID_SEPARATOR
 
     def __init__(
@@ -127,7 +132,16 @@ def plugin_names(self) -> list[str]:
         return [plugin.name for plugin in self._base_config.plugins]
 
     async def get_manager(self, context_id: Optional[str] = None) -> TenantPluginManager:
-        """Get or create a TenantPluginManager for the given context."""
+        """Get or create a TenantPluginManager for the given context.
+
+        For per-tool context IDs (``"<team_id>::<tool_name>"``), the factory
+        first checks whether a team-scoped canonical slot
+        (``"<team_id>::__team__"``) already exists.  When a tool has no
+        per-tool ``ToolPluginBinding`` rows, ``_build_manager`` normalises the
+        key to the team-scoped slot before caching, so all unbound tools in the
+        same team share one manager instance.  The per-tool key is looked up
+        first so that tools *with* bindings still get their own dedicated slot.
+        """
         context_id = context_id or "__global__"
         expired = None
 
@@ -140,6 +154,17 @@ async def get_manager(self, context_id: Optional[str] = None) -> TenantPluginMan
                 else:
                     return entry.manager
 
+            # Fast path: check the canonical team-scoped slot for per-tool keys.
+            # _build_manager stores unbound tools under "<team_id>::__team__"
+            # so a second unbound tool in the same team hits the cache here.
+            if CONTEXT_ID_SEPARATOR in context_id:
+                team_id, tool_name = context_id.split(CONTEXT_ID_SEPARATOR, 1)
+                if tool_name != TEAM_SCOPED_TOOL_SENTINEL:
+                    canonical = f"{team_id}{CONTEXT_ID_SEPARATOR}{TEAM_SCOPED_TOOL_SENTINEL}"
+                    canonical_entry = self._managers.get(canonical)
+                    if canonical_entry is not None and not canonical_entry.is_expired(self._cache_ttl):
+                        return canonical_entry.manager
+
             if expired is not None:
                 try:
                     await expired.manager.shutdown()
@@ -164,10 +189,33 @@ async def get_manager(self, context_id: Optional[str] = None) -> TenantPluginMan
                     self._inflight.pop(context_id, None)
 
     async def _build_manager(self, context_id: str) -> TenantPluginManager:
-        """Create, initialise, and cache a new manager for *context_id*."""
+        """Create, initialise, and cache a new manager for *context_id*.
+
+        When *context_id* is a per-tool key (``"<team_id>::<tool_name>"``) and
+        ``get_config_from_db`` finds no per-tool bindings, the manager is stored
+        under the canonical team-scoped key ``"<team_id>::__team__"`` instead.
+        Subsequent calls for other unbound tools in the same team hit the
+        team-scoped cache entry in ``get_manager`` without rebuilding.
+        """
         manager = None
         try:
             new_config = await self.get_config_from_db(context_id)
+
+            # Normalize unbound per-tool keys to the team-scoped canonical slot.
+            # A tool with no ToolPluginBinding rows produces the same manager as
+            # any other unbound tool in its team; sharing one slot avoids O(tools)
+            # cold-start rebuilds.
+            cache_key = context_id
+            if new_config is None and CONTEXT_ID_SEPARATOR in context_id:
+                team_id, tool_name = context_id.split(CONTEXT_ID_SEPARATOR, 1)
+                if tool_name != TEAM_SCOPED_TOOL_SENTINEL:
+                    cache_key = f"{team_id}{CONTEXT_ID_SEPARATOR}{TEAM_SCOPED_TOOL_SENTINEL}"
+                    async with self._lock:
+                        existing = self._managers.get(cache_key)
+                        if existing is not None and not existing.is_expired(self._cache_ttl):
+                            logger.debug("_build_manager: reusing team-scoped slot %s for %s", cache_key, context_id)
+                            return existing.manager
+
             config = self._merge_tenant_config(new_config)
             config = await self._apply_redis_mode_overrides(config)
 
@@ -180,15 +228,15 @@ async def _build_manager(self, context_id: str) -> TenantPluginManager:
             await manager.initialize()
 
             async with self._lock:
-                old_entry = self._managers.get(context_id)
-                self._managers[context_id] = _CachedManager(manager=manager, created_at=time.monotonic())
+                old_entry = self._managers.get(cache_key)
+                self._managers[cache_key] = _CachedManager(manager=manager, created_at=time.monotonic())
 
             old = old_entry.manager if old_entry is not None else None
             if old is not None and old is not manager:
                 try:
                     await old.shutdown()
                 except Exception:
-                    logger.warning("Failed to shutdown old manager for context_id=%s", context_id)
+                    logger.warning("Failed to shutdown old manager for context_id=%s", cache_key)
 
             return manager