From b2b8b6f65ebc384abba9da06cd1de0678c044e44 Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Sun, 12 Jul 2026 17:47:03 -0700 Subject: [PATCH] fix(mcp): schedule node reload after admin write instead of blocking on it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The auto-notify added in the prior commit awaited _notify_nodes_mcp_reload inline in create/update/delete, coupling each admin write's latency — and success — to cluster reachability: on a large cluster with slow/unreachable nodes the write could hang up to ceil(nodes/fan_out_limit)*30s behind the fan-out, and a post-commit fan-out error would 500 a write that already landed. Schedule the fan-out as a BackgroundTask that runs AFTER the 200 instead — the "trigger, not drain" contract already used by _cascade_cancel_to_children — so the write's response is never blocked on, nor failed by, the fan-out. The pre-existing registry-install path is converted the same way for consistency. There is no periodic node->DB reconcile, so a node that misses the reload serves a stale MCP catalog until the next POST /reload. The background _run therefore logs any unreached node (or a systemic fan-out fault) at WARNING — visible at the default INFO level — rather than swallowing it; the per-node status view also surfaces the divergence. A non-2xx reply from a node's reload/action endpoint now counts as a failure (raise_for_status) rather than a reached node, so neither the WARNING nor the operator /reload results miss a 5xx node. Revert the getattr None-guard on _notify_nodes_mcp_reload: it turned the operator-triggered POST /reload into a silent success ({} with 200) when the fan-out infra was absent — a fail-loudly violation — and diverged from the unguarded sibling _notify_nodes_mcp_action. The helper is drain-style again, awaited only by /reload (which must surface fan-out failures); writes go through the best-effort scheduler. Tests: assert the reload is NOT scheduled on a delete/update 404 or a create secret-store 503; that an unreached-node, raising, or non-2xx fan-out is logged at WARNING / recorded as an error; and that operator POST /reload fails loudly (500) without fan-out infra. --- tests/test_mcp_admin_api.py | 127 +++++++++++++++++++++++++++++++++++- turnstone/console/server.py | 106 ++++++++++++++++++++++-------- 2 files changed, 204 insertions(+), 29 deletions(-) diff --git a/tests/test_mcp_admin_api.py b/tests/test_mcp_admin_api.py index 289cefbd..78418ee8 100644 --- a/tests/test_mcp_admin_api.py +++ b/tests/test_mcp_admin_api.py @@ -4,11 +4,13 @@ from __future__ import annotations import base64 import json +import logging import uuid from types import SimpleNamespace from typing import TYPE_CHECKING, Any from unittest.mock import AsyncMock, MagicMock, patch +import httpx import pytest from starlette.applications import Starlette from starlette.middleware import Middleware @@ -1934,6 +1936,28 @@ class TestNotifyNodesMcpReload: assert "error" in result["n1"] assert "refused" in result["n1"]["error"] + @pytest.mark.anyio + async def test_records_error_on_non_2xx(self): + """A node replying non-2xx (e.g. 503) is recorded as an error, not + counted as a reached node — raise_for_status() routes the status into + the error path so a stale node trips the 'did not reach' WARNING, and + the (unused) response body is never consulted.""" + http_req = httpx.Request("POST", "http://n1:8000/x") + resp = MagicMock() + resp.raise_for_status.side_effect = httpx.HTTPStatusError( + "503", request=http_req, response=httpx.Response(503, request=http_req) + ) + client = AsyncMock() + client.post.return_value = resp + req = _fake_request( + {"node_id": "n1", "server_url": "http://n1:8000"}, + proxy_client=client, + ) + result = await _notify_nodes_mcp_reload(req) + assert "n1" in result + assert "error" in result["n1"] + resp.json.assert_not_called() + @pytest.mark.anyio async def test_empty_cluster(self): req = _fake_request() @@ -2024,10 +2048,29 @@ class TestAdminMcpReloadEndpoint: assert data["results"]["n1"] == {"reloaded": 2} assert "error" in data["results"]["n2"] + def test_reload_fails_loud_without_fanout_infra(self, storage: SQLiteBackend) -> None: + """F4 guard: the operator reload drains + reports, so with storage and + admin.mcp permission but no collector/proxy_client on app.state it must + fail loudly (500) — never silently 200 with empty results (which a + re-introduced None-guard would do).""" + app = Starlette( + routes=_ROUTES, + middleware=[Middleware(_InjectAuthMiddleware)], + ) + app.state.auth_storage = storage + # Deliberately omit app.state.collector / proxy_client. + c = TestClient(app, raise_server_exceptions=False) + r = c.post("/v1/api/admin/mcp-servers/reload") + assert r.status_code == 500 + class TestMcpWriteAutoReload: - """create / update / delete auto-notify nodes so a write reaches them (and - active per-user pools re-prime) without a separate /reload.""" + """create / update / delete schedule a node reload (after the 200) so a + write reaches nodes — and active per-user pools re-prime — without a + separate /reload. The fan-out rides only the success response; an error + return schedules nothing. (The error paths tested here return before the + row is written; a post-write secret-apply failure is a separate pre-existing + partial-write path, not exercised here.)""" def test_create_notifies_nodes(self, client: TestClient) -> None: with patch( @@ -2060,6 +2103,86 @@ class TestMcpWriteAutoReload: assert r.status_code == 200 notify.assert_awaited_once() + def test_delete_does_not_notify_on_missing_server(self, client: TestClient) -> None: + """A 404 (server not found) returns before the success response, so no + node reload is scheduled — the fan-out rides only the success path.""" + with patch( + "turnstone.console.server._notify_nodes_mcp_reload", + new_callable=AsyncMock, + return_value={}, + ) as notify: + r = client.delete("/v1/api/admin/mcp-servers/does-not-exist") + assert r.status_code == 404 + notify.assert_not_awaited() + + def test_update_does_not_notify_on_missing_server(self, client: TestClient) -> None: + """A 404 on update likewise schedules no reload.""" + with patch( + "turnstone.console.server._notify_nodes_mcp_reload", + new_callable=AsyncMock, + return_value={}, + ) as notify: + r = client.put("/v1/api/admin/mcp-servers/does-not-exist", json={"enabled": False}) + assert r.status_code == 404 + notify.assert_not_awaited() + + def test_create_does_not_notify_on_secret_store_503( + self, client_no_token_store: TestClient + ) -> None: + """A create that 503s on the OAuth-secret token-store gate returns an + error before any write — so no reload is scheduled.""" + with patch( + "turnstone.console.server._notify_nodes_mcp_reload", + new_callable=AsyncMock, + return_value={}, + ) as notify: + r = client_no_token_store.post( + "/v1/api/admin/mcp-servers", + json={ + "name": "no-notify-503", + "transport": "streamable-http", + "url": "https://mcp.example.com/sse", + "auth_type": "oauth_user", + "oauth_client_id": "cli_abc", + "oauth_client_secret": "secret-value", + }, + ) + assert r.status_code == 503, r.text + notify.assert_not_awaited() + + def test_write_warns_when_reload_reaches_no_node( + self, client: TestClient, caplog: pytest.LogCaptureFixture + ) -> None: + """A background fan-out that leaves nodes unreached is surfaced at + WARNING (not swallowed at debug) — operators need a signal the cluster + catalog may be stale, since there is no periodic node reconcile.""" + with ( + patch( + "turnstone.console.server._notify_nodes_mcp_reload", + new_callable=AsyncMock, + return_value={"n1": {"error": "Connection refused"}}, + ), + caplog.at_level(logging.WARNING), + ): + _create_server(client, name="warn-on-stale") + assert any("did not reach" in r.getMessage() for r in caplog.records) + + def test_write_warns_when_reload_fan_out_raises( + self, client: TestClient, caplog: pytest.LogCaptureFixture + ) -> None: + """A systemic fan-out fault (the whole reload raises) is logged at + WARNING rather than lost, for the same reason.""" + with ( + patch( + "turnstone.console.server._notify_nodes_mcp_reload", + new_callable=AsyncMock, + side_effect=RuntimeError("collector exploded"), + ), + caplog.at_level(logging.WARNING), + ): + _create_server(client, name="warn-on-fault") + assert any("fan-out failed after admin write" in r.getMessage() for r in caplog.records) + # --------------------------------------------------------------------------- # Node reload endpoint: POST /v1/api/_internal/mcp-reload diff --git a/turnstone/console/server.py b/turnstone/console/server.py index 97390089..71f69ca4 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -9726,11 +9726,13 @@ async def admin_registry_install(request: Request) -> JSONResponse: ip, ) - # Auto-reload nodes for one-click UX - await _notify_nodes_mcp_reload(request) - server_row = storage.get_mcp_server(server_id) - return JSONResponse(_mcp_server_to_detail(_mask_mcp_secrets(server_row or {}))) + # Auto-reload nodes for one-click UX — scheduled after the response so the + # install isn't blocked on cluster fan-out (see _schedule_mcp_reload). + return JSONResponse( + _mcp_server_to_detail(_mask_mcp_secrets(server_row or {})), + background=_schedule_mcp_reload(request), + ) # --------------------------------------------------------------------------- @@ -10454,13 +10456,14 @@ async def admin_create_mcp_server(request: Request) -> JSONResponse: ip, ) - # Auto-reload nodes so the new server reaches them (and per-user pools - # re-prime for active sessions) without a separate /reload — mirrors the - # registry-install path. - await _notify_nodes_mcp_reload(request) - server = storage.get_mcp_server(server_id) - return JSONResponse(_mcp_server_to_detail(_mask_mcp_secrets(server or {}))) + # Auto-reload nodes so the new server reaches them (and per-user pools + # re-prime for active sessions) without a separate /reload — scheduled after + # the response so the write isn't blocked on cluster fan-out. + return JSONResponse( + _mcp_server_to_detail(_mask_mcp_secrets(server or {})), + background=_schedule_mcp_reload(request), + ) async def admin_get_mcp_server(request: Request) -> JSONResponse: @@ -10806,12 +10809,14 @@ async def admin_update_mcp_server(request: Request) -> JSONResponse: ip, ) - # Auto-reload nodes so the edit (incl. a pool auth_type flip) reaches them - # and active sessions re-prime, without a separate /reload. - await _notify_nodes_mcp_reload(request) - server = storage.get_mcp_server(server_id) - return JSONResponse(_mcp_server_to_detail(_mask_mcp_secrets(server or {}))) + # Auto-reload nodes so the edit (incl. a pool auth_type flip) reaches them + # and active sessions re-prime — scheduled after the response so the write + # isn't blocked on cluster fan-out. + return JSONResponse( + _mcp_server_to_detail(_mask_mcp_secrets(server or {})), + background=_schedule_mcp_reload(request), + ) async def admin_delete_mcp_server(request: Request) -> JSONResponse: @@ -10863,25 +10868,23 @@ async def admin_delete_mcp_server(request: Request) -> JSONResponse: ip, ) - # Auto-reload nodes so they drop the removed server without a separate - # /reload. - await _notify_nodes_mcp_reload(request) - - return JSONResponse({"status": "ok"}) + # Auto-reload nodes so they drop the removed server — scheduled after the + # response so the delete isn't blocked on cluster fan-out. + return JSONResponse({"status": "ok"}, background=_schedule_mcp_reload(request)) async def _notify_nodes_mcp_reload(request: Request) -> dict[str, Any]: """Tell all nodes to re-read the mcp_servers DB table and reconcile. - Best-effort: if the cluster fan-out infra isn't present (e.g. a minimal - test app), the DB write has already landed and nodes pick the change up on - their next reconcile — so skip rather than 500 the caller. + Drain-style: awaited inline only by the operator-triggered ``POST /reload``, + which reports the per-node results and must fail loudly if the fan-out infra + is absent. Admin *writes* (create/update/delete/registry-install) never await + this — they schedule it best-effort via ``_schedule_mcp_reload`` so a write's + response isn't blocked on (nor failed by) cluster reachability. """ - collector: ClusterCollector | None = getattr(request.app.state, "collector", None) - client: httpx.AsyncClient | None = getattr(request.app.state, "proxy_client", None) - if collector is None or client is None: - return {} + collector: ClusterCollector = request.app.state.collector nodes = collector.get_all_nodes() + client: httpx.AsyncClient = request.app.state.proxy_client headers = _proxy_auth_headers(request) sem = asyncio.Semaphore(_get_fan_out_limit(request)) @@ -10897,6 +10900,9 @@ async def _notify_nodes_mcp_reload(request: Request) -> dict[str, Any]: headers=headers, timeout=30, ) + # A non-2xx reply is a failed reload, not a reached node (httpx + # does not raise on status); surface it as an error so _run warns. + resp.raise_for_status() return node_id, resp.json() except Exception as exc: log.debug("Failed to notify node %s for MCP reload", node_id, exc_info=True) @@ -10906,6 +10912,49 @@ async def _notify_nodes_mcp_reload(request: Request) -> dict[str, Any]: return {nid: data for nid, data in results if data is not None} +def _schedule_mcp_reload(request: Request) -> BackgroundTask: + """Best-effort node-reload fan-out to run AFTER an admin MCP write's 200. + + Returned as a ``BackgroundTask`` — the "trigger, not drain" contract also + used by ``_cascade_cancel_to_children``: the admin write's response is never + blocked on the per-node fan-out (each node POST can hit a 30s timeout under + the fan-out semaphore), so a fan-out failure can't fail a write that already + committed. + + There is no periodic node→DB reconcile — a node that misses this reload + keeps serving a stale MCP catalog until the next ``POST /reload`` (or a node + restart). So ``_run`` does NOT swallow failures: any unreached node (or a + systemic fan-out fault) is logged at WARNING (visible at the default INFO + level), and the per-node status view surfaces the divergence. Only + ``POST /reload`` awaits the fan-out inline, where draining and reporting + per-node results to the operator is the point. + """ + + async def _run() -> None: + try: + results = await _notify_nodes_mcp_reload(request) + except Exception: + log.warning( + "auto mcp-reload fan-out failed after admin write; nodes may serve a " + "stale MCP catalog until the next POST /reload", + exc_info=True, + ) + return + unreached = sorted( + nid for nid, data in results.items() if isinstance(data, dict) and "error" in data + ) + if unreached: + log.warning( + "auto mcp-reload did not reach %d of %d node(s) after admin write " + "(stale MCP catalog until next reload): %s", + len(unreached), + len(results), + ", ".join(unreached), + ) + + return BackgroundTask(_run) + + async def _notify_nodes_mcp_action(request: Request, action: str, name: str) -> dict[str, Any]: """Tell all nodes to perform a per-server MCP action. @@ -10932,6 +10981,9 @@ async def _notify_nodes_mcp_action(request: Request, action: str, name: str) -> headers=headers, timeout=30, ) + # A non-2xx reply is a failed action, not a reached node (httpx + # does not raise on status); surface it as an error to the operator. + resp.raise_for_status() return node_id, resp.json() except Exception as exc: log.debug(