ewilliams28/turnstone
1
0
Fork 0
mirror of https://github.com/turnstonelabs/turnstone.git synced 2026-08-12 23:12:23 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(mcp): schedule node reload after admin write instead of blocking on it

Browse Source 
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.
This commit is contained in:
Patrick Buckley
2026-07-12 17:47:03 -07:00
parent d6ccc5ed17
commit b2b8b6f65e
2 changed files with 204 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_mcp_admin_api.py
+125 -2
View File
@@ -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
turnstone/console/server.py
+79 -27
View File
@@ -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(