Dataplane integration fixes (combined view)

.env.example

@@ -347,6 +347,8 @@ OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317
 # Send db config to redis to be consumed by experimental Dataplane
 DATAPLANE_PUBLISHER = false
 
+DATAPLANE_PUBLISHER_INTERVAL_SECONDS = 60
+
 
 
 # =============================================================================

.secrets.baseline

@@ -3,7 +3,7 @@
     "files": "(?x)( package-lock\\.json$ |Cargo\\.lock$ |uv\\.lock$ |go\\.sum$ |mcpgateway/sri_hashes\\.json$ )|^.secrets.baseline$",
     "lines": null
   },
-  "generated_at": "2026-07-08T12:40:34Z",
+  "generated_at": "2026-07-09T15:51:34Z",
   "plugins_used": [
     {
       "name": "AWSKeyDetector"
@@ -116,95 +116,95 @@
         "hashed_secret": "08cd923367890009657eab812753379bdb321eeb",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 708,
+        "line_number": 710,
         "type": "Basic Auth Credentials",
         "verified_result": null
       },
       {
         "hashed_secret": "14f8aa3e560a47851908ab0f04ec856dbc512d93",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 936,
+        "line_number": 938,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "fa9beb99e4029ad5a6615399e7bbae21356086b3",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 1234,
+        "line_number": 1236,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "7b4455a56fbf1d198e45e04c437488514645a82c",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 1260,
+        "line_number": 1262,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "d08f88df745fa7950b104e4a707a31cfce7b5841",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 1268,
+        "line_number": 1270,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "ac371b6dcce28a86c90d12bc57d946a800eebf17",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 1347,
+        "line_number": 1349,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "0b6ec68df700dec4dcd64babd0eda1edccddace1",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 1352,
+        "line_number": 1354,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "4ad6f0082ee224001beb3ca5c3e81c8ceea5ed86",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 1357,
+        "line_number": 1359,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "cb32747fcfb55eaa194c8cd8e4ba7d49ada08a94",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 1363,
+        "line_number": 1365,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "6c178d51b13520496dbc767ed3d9d7aa5803ac72",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 1375,
+        "line_number": 1377,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "ca45060a53fd8a255d1a83ee8d2f025283ccc66e",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 1393,
+        "line_number": 1395,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "910fbf00f58e9bcb095ea26a75cc1d9a3355e671",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 1454,
+        "line_number": 1456,
         "type": "Secret Keyword",
         "verified_result": null
       }

mcpgateway/config.py

@@ -2783,6 +2783,12 @@ def validate_ratelimiter_redis_url(cls, v: Optional[str]) -> Optional[str]:
 
     dataplane_publisher: bool = Field(default=False, description="Send data from CF to Rust experimental dataplane")
 
+    dataplane_publisher_interval_seconds: int = Field(
+        default=60,
+        ge=1,
+        description="Seconds between dataplane publisher snapshots to Redis; UserConfig keys expire 10 seconds after the next expected snapshot",
+    )
+
     # Well-Known URI Configuration
     # ===================================
 

mcpgateway/services/dataplane_publisher.py

@@ -22,6 +22,7 @@
 from sqlalchemy import select
 
 # First-Party
+from mcpgateway.config import settings
 from mcpgateway.db import (
     EmailTeamMember,
     EmailUser,
@@ -42,12 +43,14 @@
 logger = logging.getLogger(__name__)
 
 USER_CONFIG_KEY = "UserConfig"
-REDIS_PUBLISHER_TIME = 60  # Publish interval in seconds
-# Keys are not deleted explicitly; stale configs expire via Redis TTL.
-PUBLISHER_TTL = 70
+REDIS_PUBLISHER_TIME = settings.dataplane_publisher_interval_seconds
+# Keys are not deleted explicitly; stale configs expire via Redis TTL. The
+# TTL must outlive a full missed publish cycle: under load the loop's timer
+# can fire late, and with only a few seconds of margin every UserConfig key
+# expires at once, taking dataplane config down for all subjects until the
+# next snapshot.
+PUBLISHER_TTL = REDIS_PUBLISHER_TIME * 2 + 10
 
-# Worker ID for multi-worker coordination (same pattern as session_affinity.py)
-WORKER_ID = f"{socket.gethostname()}:{os.getpid()}"
 PUBLISHER_LOCK_KEY = "mcpgw:dataplane_publisher:lock"
 
 
@@ -86,6 +89,11 @@ def __init__(self) -> None:
         """Initialize the publisher state and dependent services."""
         self.task: asyncio.Task[None] | None = None
         self._shutdown_event = asyncio.Event()
+        # Computed per instance so each gunicorn worker gets its own id;
+        # a module-level constant is evaluated in the master before fork,
+        # which gives every worker the same id and defeats the lock's
+        # compare-and-swap ownership check (same pattern as session_affinity.py).
+        self.worker_id = f"{socket.gethostname()}:{os.getpid()}"
 
     async def start(self) -> None:
         """Start the background publisher task."""
@@ -137,7 +145,7 @@ async def publish_to_redis(self) -> None:
                 lock_ttl = REDIS_PUBLISHER_TIME + 30  # Lock expires 30s after publish interval
                 acquired = await redis.set(
                     PUBLISHER_LOCK_KEY,
-                    WORKER_ID,
+                    self.worker_id,
                     nx=True,  # Only set if not exists
                     ex=lock_ttl,  # Auto-expire if worker crashes
                 )
@@ -149,7 +157,7 @@ async def publish_to_redis(self) -> None:
                     continue
 
                 # We hold the lock - publish data
-                logger.info("Worker %s publishing dataplane payload...", WORKER_ID)
+                logger.info("Worker %s publishing dataplane payload...", self.worker_id)
                 payload = await self.fetch_payload()
 
                 if payload is None:
@@ -181,7 +189,7 @@ async def publish_to_redis(self) -> None:
                     return 0
                     """
                     try:
-                        await redis.eval(release_script, 1, PUBLISHER_LOCK_KEY, WORKER_ID)
+                        await redis.eval(release_script, 1, PUBLISHER_LOCK_KEY, self.worker_id)
                     except Exception as e:
                         logger.warning("Failed to release lock: %s", e)
 
@@ -211,23 +219,34 @@ def create_payload(
 
             prompt_map = {prompt["id"]: prompt["name"] for prompt in prompts}
 
+            # The dataplane proxies streamable-HTTP upstreams only; other
+            # transports (e.g. SSE, deprecated and removed in the 2026-07-28
+            # MCP protocol update) are excluded so the published config never
+            # advertises a backend the dataplane cannot serve. Backends are
+            # keyed by the gateway slug, not the id: the dataplane uses the
+            # key as the namespace prefix for federated tool names, so slug
+            # keys make it advertise the same names the control plane does
+            # (e.g. "fast-time-echo" instead of "<gateway-uuid>-echo").
             gateway_base = {
-                gateway["id"]: {
+                (gateway.get("slug") or gateway["id"]): {
                     "name": gateway["name"],
                     "url": gateway["url"],
                     "transport": gateway["transport"],
                     "passthrough_headers": gateway["passthrough_headers"] or [],
                 }
                 for gateway in gateways
+                if (gateway["transport"] or "").upper() == "STREAMABLEHTTP"
             }
+            gateway_key_by_id = {gateway["id"]: (gateway.get("slug") or gateway["id"]) for gateway in gateways}
 
             virtual_hosts: dict[str, VirtualHostConfig] = {}
 
             for server in servers:
                 backends: dict[str, BackendConfig] = {}
 
                 for gateway_id, backend_items in server["backend_items"].items():
-                    gateway_config = gateway_base.get(gateway_id)
+                    gateway_key = gateway_key_by_id.get(gateway_id)
+                    gateway_config = gateway_base.get(gateway_key) if gateway_key else None
                     if gateway_config is None:
                         continue
 
@@ -236,13 +255,20 @@ def create_payload(
                     if not backend_items["tools"] and not allowed_resource_names and not allowed_prompt_names:
                         continue
 
-                    backends[gateway_id] = {
+                    backends[gateway_key] = {
                         **gateway_config,
                         "allowed_tool_names": backend_items["tools"],
                         "allowed_resource_names": allowed_resource_names,
                         "allowed_prompt_names": allowed_prompt_names,
                     }
 
+                if not backends:
+                    # No publishable backends: leave the virtual host out so
+                    # the dataplane returns 404 for it and deployments can
+                    # route the request back to the control plane, instead of
+                    # serving an empty tool list that looks like success.
+                    continue
+
                 virtual_hosts[server["id"]] = {"backends": backends}
 
             result[user_email] = {"virtual_hosts": virtual_hosts}
@@ -273,6 +299,7 @@ async def get_data_from_db(self) -> dict[str, Any] | None:
                 gateway_rows = db.execute(
                     select(
                         DbGateway.id,
+                        DbGateway.slug,
                         DbGateway.name,
                         DbGateway.url,
                         DbGateway.transport,
@@ -338,6 +365,7 @@ def _build_user_data(
             "gateways": [
                 {
                     "id": gateway.id,
+                    "slug": gateway.slug,
                     "name": gateway.name,
                     "url": gateway.url,
                     "transport": gateway.transport,