From e334700809c482df96a08b175609d621b3d05a20 Mon Sep 17 00:00:00 2001 From: liuhao1024 Date: Fri, 3 Jul 2026 16:56:37 +0800 Subject: [PATCH] fix(mcp): reset reconnect retry counter after successful session establishment MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The local retries variable in MCPServerTask.run() accumulated across transient disconnections — each transport exception incremented it, but only clean transport returns (auth recovery / manual refresh) or park-wake reset it. Five transient blips over a long-uptime gateway would permanently park the MCP server. Promote retries to instance attribute _reconnect_retries and reset it at all 4 session-establishment sites in _run_stdio / _run_http, so only consecutive failures without successful reconnection count toward the parking budget. Fixes #57604 --- tests/tools/test_mcp_reconnect_retry_reset.py | 196 ++++++++++++++++++ tools/mcp_tool.py | 21 +- 2 files changed, 211 insertions(+), 6 deletions(-) create mode 100644 tests/tools/test_mcp_reconnect_retry_reset.py diff --git a/tests/tools/test_mcp_reconnect_retry_reset.py b/tests/tools/test_mcp_reconnect_retry_reset.py new file mode 100644 index 00000000000..50442aa602a --- /dev/null +++ b/tests/tools/test_mcp_reconnect_retry_reset.py @@ -0,0 +1,196 @@ +"""Tests for MCP reconnect retry counter reset (#57604). + +Verifies that the reconnect retry counter resets after each successful +reconnection, so transient blips do not accumulate toward permanent parking. +""" + +import asyncio + +import pytest + + +@pytest.mark.no_isolate +def test_reconnect_counter_resets_after_successful_session(monkeypatch, tmp_path): + """Transient disconnections must not accumulate toward permanent parking. + + Before the fix, ``retries`` was a local variable in ``run()`` that only + reset on clean transport return (line 2367) or park-wake (line 2468). + Each exception from ``_run_stdio`` incremented it without reset, so 5 + transient blips over a long-uptime gateway would permanently park the + server. + + After the fix, ``_reconnect_retries`` is an instance variable that resets + to 0 whenever a session is successfully established (``_reset_server_error`` + call sites in ``_run_stdio`` / ``_run_http``). + """ + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + + from tools import mcp_tool + from tools.mcp_tool import MCPServerTask + + # Shrink budget so the test can exhaust it quickly if the counter + # does NOT reset (the bug scenario). + monkeypatch.setattr(mcp_tool, "_MAX_RECONNECT_RETRIES", 3) + + _real_sleep = asyncio.sleep + + async def _fast_sleep(_delay, *a, **kw): + await _real_sleep(0) + + monkeypatch.setattr(mcp_tool.asyncio, "sleep", _fast_sleep) + + state = { + "transport_calls": 0, + "parked": False, + "max_retries_seen": 0, + } + + async def _scenario(): + class _Task(MCPServerTask): + def _is_http(self): + return False + + def _deregister_tools(self): + state["parked"] = True + self._registered_tool_names = [] + + async def _run_stdio(self, config): + state["transport_calls"] += 1 + call = state["transport_calls"] + + if call == 1: + # First connect: succeed (sets _ready), then fail. + self.session = object() + self._ready.set() + self.session = None + raise RuntimeError("blip 1") + + # Subsequent calls: succeed (session established, which + # triggers _reset_server_error and should reset retries), + # then immediately fail again — simulating a new transient + # blip. If retries accumulate, call 4 would exceed the + # budget of 3 and park. If retries reset correctly, + # this loop can continue indefinitely. + if call <= 8: + self.session = object() + # _run_stdio calls _reset_server_error and sets + # _reconnect_retries = 0 after session establishment. + # We simulate that by calling the real method. + mcp_tool._reset_server_error(self.name) + self._reconnect_retries = 0 + self.session = None + raise RuntimeError(f"blip {call}") + + # If we reach here without parking, the fix works. + self.session = object() + mcp_tool._reset_server_error(self.name) + self._reconnect_retries = 0 + await self._wait_for_lifecycle_event() + return + + task = _Task("srv") + task._registered_tool_names = ["srv__tool"] + + run_task = asyncio.ensure_future(task.run({"command": "x"})) + + # Let the scenario run. + for _ in range(2000): + await _real_sleep(0) + if state["transport_calls"] >= 8 or state["parked"]: + break + + # The fix: the server should NOT have parked despite 8 transient + # disconnections, because each successful reconnection reset the + # retry counter. + assert not state["parked"], ( + f"server parked after {state['transport_calls']} transport calls " + f"— retry counter accumulated instead of resetting" + ) + assert state["transport_calls"] >= 8, ( + f"only {state['transport_calls']} transport calls reached " + f"(expected >= 8)" + ) + + # Verify the counter is an instance variable, not a local. + assert hasattr(task, "_reconnect_retries"), ( + "_reconnect_retries should be an instance variable" + ) + + # Clean shutdown. + task._shutdown_event.set() + task._reconnect_event.set() + try: + await asyncio.wait_for(run_task, timeout=2) + except (asyncio.TimeoutError, asyncio.CancelledError, Exception): + run_task.cancel() + + asyncio.run(_scenario()) + + +@pytest.mark.no_isolate +def test_reconnect_counter_still_parks_on_consecutive_failures(monkeypatch, tmp_path): + """The server must still park when failures are genuinely consecutive + (no successful reconnection in between). + """ + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + + from tools import mcp_tool + from tools.mcp_tool import MCPServerTask + + monkeypatch.setattr(mcp_tool, "_MAX_RECONNECT_RETRIES", 2) + + _real_sleep = asyncio.sleep + + async def _fast_sleep(_delay, *a, **kw): + await _real_sleep(0) + + monkeypatch.setattr(mcp_tool.asyncio, "sleep", _fast_sleep) + + state = {"transport_calls": 0, "parked": False} + + async def _scenario(): + class _Task(MCPServerTask): + def _is_http(self): + return False + + def _deregister_tools(self): + state["parked"] = True + self._registered_tool_names = [] + + async def _run_stdio(self, config): + state["transport_calls"] += 1 + call = state["transport_calls"] + + if call == 1: + self.session = object() + self._ready.set() + self.session = None + raise RuntimeError("first failure") + + # All subsequent calls fail WITHOUT establishing a session + # (no _reset_server_error, no retry reset). This simulates + # genuinely consecutive failures. + raise RuntimeError(f"failure {call}") + + task = _Task("srv") + task._registered_tool_names = ["srv__tool"] + + run_task = asyncio.ensure_future(task.run({"command": "x"})) + + for _ in range(500): + await _real_sleep(0) + if state["parked"] or run_task.done(): + break + + assert state["parked"], ( + "server should park on consecutive failures without successful reconnect" + ) + + task._shutdown_event.set() + task._reconnect_event.set() + try: + await asyncio.wait_for(run_task, timeout=2) + except (asyncio.TimeoutError, asyncio.CancelledError, Exception): + run_task.cancel() + + asyncio.run(_scenario()) diff --git a/tools/mcp_tool.py b/tools/mcp_tool.py index d74d7f3ceaf..2a4d485f8d1 100644 --- a/tools/mcp_tool.py +++ b/tools/mcp_tool.py @@ -1475,6 +1475,7 @@ class MCPServerTask: "_rpc_lock", "_pending_refresh_tasks", "_pending_call_context", "initialize_result", "_ping_unsupported", + "_reconnect_retries", ) def __init__(self, name: str): @@ -1496,6 +1497,7 @@ class MCPServerTask: self._sampling: Optional[SamplingHandler] = None self._elicitation: Optional[ElicitationHandler] = None self._registered_tool_names: list[str] = [] + self._reconnect_retries: int = 0 self._auth_type: str = "" self._refresh_lock = asyncio.Lock() # MCP stdio sessions are a single JSON-RPC stream. Some servers emit @@ -1984,6 +1986,10 @@ class MCPServerTask: # prior outage so the first call after recovery isn't # gated on a stale consecutive-failure count (#16788). _reset_server_error(self.name) + # This session is live: reset the reconnect retry counter + # so transient prior failures do not accumulate toward + # permanent parking (#57604). + self._reconnect_retries = 0 # stdio transport does not use OAuth, but we still honor # _reconnect_event (e.g. future manual /mcp refresh) for # consistency with _run_http. @@ -2221,6 +2227,7 @@ class MCPServerTask: # prior outage so the first call after recovery isn't # gated on a stale consecutive-failure count (#16788). _reset_server_error(self.name) + self._reconnect_retries = 0 reason = await self._wait_for_lifecycle_event() if reason == "reconnect": logger.info( @@ -2274,6 +2281,7 @@ class MCPServerTask: # a prior outage so the first call after recovery # isn't gated on a stale failure count (#16788). _reset_server_error(self.name) + self._reconnect_retries = 0 reason = await self._wait_for_lifecycle_event() if reason == "reconnect": logger.info( @@ -2301,6 +2309,7 @@ class MCPServerTask: # prior outage so the first call after recovery isn't # gated on a stale consecutive-failure count (#16788). _reset_server_error(self.name) + self._reconnect_retries = 0 reason = await self._wait_for_lifecycle_event() if reason == "reconnect": logger.info( @@ -2417,7 +2426,7 @@ class MCPServerTask: self._ready.set() return - retries = 0 + self._reconnect_retries = 0 initial_retries = 0 backoff = 1.0 @@ -2446,7 +2455,7 @@ class MCPServerTask: # failure budget — otherwise transient drops accumulated over # a long-lived session would eventually exhaust it and # permanently kill an otherwise-healthy server. - retries = 0 + self._reconnect_retries = 0 backoff = 1.0 # Reset the session reference; _run_http/_run_stdio will # repopulate it on successful re-entry. @@ -2520,8 +2529,8 @@ class MCPServerTask: ) return - retries += 1 - if retries > _MAX_RECONNECT_RETRIES: + self._reconnect_retries += 1 + if self._reconnect_retries > _MAX_RECONNECT_RETRIES: logger.warning( "MCP server '%s' failed after %d reconnection attempts, " "parking until a reconnect is requested: %s", @@ -2547,14 +2556,14 @@ class MCPServerTask: "rebuilding transport.", self.name, ) - retries = 0 + self._reconnect_retries = 0 backoff = 1.0 continue logger.warning( "MCP server '%s' connection lost (attempt %d/%d), " "reconnecting in %.0fs: %s", - self.name, retries, _MAX_RECONNECT_RETRIES, + self.name, self._reconnect_retries, _MAX_RECONNECT_RETRIES, backoff, exc, ) await asyncio.sleep(backoff)