TDD: Add test for race condition in McpClient.start() double initialization
CI / lint (push) Successful in 1m29s
CI / typecheck (push) Successful in 1m19s
CI / quality (push) Successful in 1m5s
CI / build (push) Successful in 47s
CI / security (push) Successful in 1m35s
CI / helm (push) Successful in 45s
CI / push-validation (push) Successful in 26s
CI / benchmark-regression (push) Has been skipped
CI / integration_tests (push) Successful in 4m15s
CI / unit_tests (push) Successful in 9m58s
CI / docker (push) Successful in 1m33s
CI / e2e_tests (push) Failing after 15m15s
CI / coverage (push) Successful in 12m27s
CI / status-check (push) Failing after 3s
CI / benchmark-publish (push) Successful in 1h30m14s
CI / benchmark-publish (pull_request) Has been skipped
CI / benchmark-regression (pull_request) Failing after 47s
CI / coverage (pull_request) Successful in 14m32s
CI / lint (pull_request) Successful in 1m8s
CI / quality (pull_request) Successful in 1m28s
CI / security (pull_request) Successful in 1m49s
CI / build (pull_request) Successful in 28s
CI / typecheck (pull_request) Successful in 2m24s
CI / helm (pull_request) Successful in 46s
CI / push-validation (pull_request) Successful in 22s
CI / unit_tests (pull_request) Successful in 6m26s
CI / integration_tests (pull_request) Successful in 4m40s
CI / e2e_tests (pull_request) Successful in 4m47s
CI / docker (pull_request) Successful in 1m31s
CI / status-check (pull_request) Successful in 4s
CI / lint (push) Successful in 1m29s
CI / typecheck (push) Successful in 1m19s
CI / quality (push) Successful in 1m5s
CI / build (push) Successful in 47s
CI / security (push) Successful in 1m35s
CI / helm (push) Successful in 45s
CI / push-validation (push) Successful in 26s
CI / benchmark-regression (push) Has been skipped
CI / integration_tests (push) Successful in 4m15s
CI / unit_tests (push) Successful in 9m58s
CI / docker (push) Successful in 1m33s
CI / e2e_tests (push) Failing after 15m15s
CI / coverage (push) Successful in 12m27s
CI / status-check (push) Failing after 3s
CI / benchmark-publish (push) Successful in 1h30m14s
CI / benchmark-publish (pull_request) Has been skipped
CI / benchmark-regression (pull_request) Failing after 47s
CI / coverage (pull_request) Successful in 14m32s
CI / lint (pull_request) Successful in 1m8s
CI / quality (pull_request) Successful in 1m28s
CI / security (pull_request) Successful in 1m49s
CI / build (pull_request) Successful in 28s
CI / typecheck (pull_request) Successful in 2m24s
CI / helm (pull_request) Successful in 46s
CI / push-validation (pull_request) Successful in 22s
CI / unit_tests (pull_request) Successful in 6m26s
CI / integration_tests (pull_request) Successful in 4m40s
CI / e2e_tests (pull_request) Successful in 4m47s
CI / docker (pull_request) Successful in 1m31s
CI / status-check (pull_request) Successful in 4s
Adds a Behave TDD issue-capture test for bug #10438: McpClient.start() releases the threading.RLock after setting _state to STARTING but before calling connect() and discover_tools(). Concurrent callers can both pass the _started idempotency check and call discover_tools() multiple times. The test uses @tdd_expected_fail so CI passes while the bug is unfixed. A counting mock transport records connect() and discover_tools() calls. Five threads call start() concurrently through a threading.Barrier to maximise the chance of the race manifesting. ISSUES CLOSED: #10402
This commit was merged in pull request #10915.
This commit is contained in:
@@ -0,0 +1,237 @@
|
||||
"""Step definitions for features/tdd_mcp_client_start_race.feature.
|
||||
|
||||
TDD issue-capture scenario for #10438: McpClient.start() race condition
|
||||
allows concurrent double initialization.
|
||||
|
||||
McpClient.start() releases the threading.RLock after setting _state to
|
||||
STARTING but before calling connect() and discover_tools(). Two concurrent
|
||||
callers can both pass the _started idempotency check (still False) and both
|
||||
proceed to call connect() and discover_tools() simultaneously.
|
||||
|
||||
The test uses @tdd_expected_fail so that CI passes while the bug is unfixed.
|
||||
The underlying assertion ("connect() called exactly once") fails because the
|
||||
race allows multiple threads to trigger initialization. The tag inversion
|
||||
causes the scenario to be reported as passed.
|
||||
|
||||
Race detection uses a threading.Barrier to synchronise threads so they all
|
||||
enter start() at approximately the same instant, maximising the chance of
|
||||
the race manifesting. A counting mock transport records how many times
|
||||
connect() and discover_tools() are called.
|
||||
|
||||
When bug #10438 is fixed (by checking _state == STARTING inside the lock),
|
||||
the assertions will pass and the @tdd_expected_fail tag must be removed.
|
||||
|
||||
See: https://git.cleverthis.com/cleveragents/cleveragents-core/issues/10438
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
from behave import given, then, when
|
||||
from behave.runner import Context
|
||||
|
||||
from cleveragents.mcp.adapter import MCPServerConfig
|
||||
from cleveragents.mcp.client import McpClient, McpClientConfig
|
||||
from features.mocks.mock_mcp_transport import MockMCPTransport
|
||||
|
||||
|
||||
class _CountingMockTransport(MockMCPTransport):
|
||||
"""Mock transport that counts connect() and discover_tools() calls.
|
||||
|
||||
Uses a threading.Lock to make the counters thread-safe so that
|
||||
concurrent calls are accurately recorded.
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
tools = [
|
||||
{
|
||||
"name": "test_tool",
|
||||
"description": "Mock tool for race condition test",
|
||||
"inputSchema": {},
|
||||
}
|
||||
]
|
||||
super().__init__(tools=tools)
|
||||
self._counter_lock = threading.Lock()
|
||||
self.connect_call_count: int = 0
|
||||
self.discover_call_count: int = 0
|
||||
# Add a small sleep inside connect() to widen the race window,
|
||||
# making it more likely that multiple threads enter concurrently.
|
||||
self._connect_delay: float = 0.02
|
||||
|
||||
def connect(self, config: MCPServerConfig) -> dict[str, Any]:
|
||||
with self._counter_lock:
|
||||
self.connect_call_count += 1
|
||||
# Sleep outside the counter lock to widen the race window.
|
||||
# This gives other threads time to also pass the _started check
|
||||
# and enter connect() before the first thread finishes.
|
||||
time.sleep(self._connect_delay)
|
||||
return super().connect(config)
|
||||
|
||||
def call(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
|
||||
if method == "tools/list":
|
||||
with self._counter_lock:
|
||||
self.discover_call_count += 1
|
||||
return super().call(method, params)
|
||||
|
||||
|
||||
# ── Given ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@given("an McpClient with a counting mock transport")
|
||||
def step_mcp_client_counting_transport(context: Context) -> None:
|
||||
"""Create an McpClient backed by a counting mock transport.
|
||||
|
||||
The client is configured with lazy_start=False, idle_timeout=0, and
|
||||
health_check_interval=0 so that only explicit start() calls trigger
|
||||
connect() and discover_tools().
|
||||
"""
|
||||
context.counting_transport = _CountingMockTransport()
|
||||
server_config = MCPServerConfig(
|
||||
name="test-server",
|
||||
transport="stdio",
|
||||
command="echo",
|
||||
)
|
||||
client_config = McpClientConfig(
|
||||
server=server_config,
|
||||
lazy_start=False,
|
||||
idle_timeout_seconds=0,
|
||||
health_check_interval_seconds=0,
|
||||
)
|
||||
context.race_client = McpClient(
|
||||
config=client_config,
|
||||
transport=context.counting_transport,
|
||||
)
|
||||
context.race_thread_errors: list[Exception] = []
|
||||
context.race_threads_completed: int = 0
|
||||
|
||||
|
||||
# ── When ──────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@when("{n:d} threads call start() concurrently through a barrier")
|
||||
def step_concurrent_start(context: Context, n: int) -> None:
|
||||
"""Launch *n* threads that all call start() simultaneously.
|
||||
|
||||
A threading.Barrier synchronises the threads so they all enter
|
||||
start() at approximately the same instant, maximising the chance
|
||||
of the race condition manifesting.
|
||||
|
||||
With the bug present, multiple threads will pass the _started check
|
||||
(still False) and each call connect() and discover_tools().
|
||||
With the fix applied, only one thread proceeds past the STARTING
|
||||
state check and the others return early.
|
||||
"""
|
||||
barrier = threading.Barrier(n)
|
||||
errors: list[Exception] = []
|
||||
completed: list[int] = [0]
|
||||
collect_lock = threading.Lock()
|
||||
|
||||
def worker() -> None:
|
||||
"""Thread worker: wait on barrier, then call start()."""
|
||||
try:
|
||||
barrier.wait(timeout=10)
|
||||
context.race_client.start()
|
||||
with collect_lock:
|
||||
completed[0] += 1
|
||||
except Exception as exc:
|
||||
with collect_lock:
|
||||
errors.append(exc)
|
||||
completed[0] += 1
|
||||
|
||||
threads = [threading.Thread(target=worker, daemon=True) for _ in range(n)]
|
||||
for t in threads:
|
||||
t.start()
|
||||
for t in threads:
|
||||
t.join(timeout=30)
|
||||
|
||||
# Verify no threads are still alive after join timeout
|
||||
for t in threads:
|
||||
assert not t.is_alive(), (
|
||||
f"Thread {t.name} is still alive after join timeout — "
|
||||
f"possible deadlock in McpClient.start()"
|
||||
)
|
||||
|
||||
barrier_errors = [
|
||||
err for err in errors if isinstance(err, threading.BrokenBarrierError)
|
||||
]
|
||||
assert not barrier_errors, (
|
||||
"Barrier synchronization failed while coordinating concurrent "
|
||||
"start() calls. This invalidates race setup. "
|
||||
f"Barrier errors: {barrier_errors}; all errors: {errors}"
|
||||
)
|
||||
|
||||
context.race_thread_errors = errors
|
||||
context.race_threads_completed = completed[0]
|
||||
context.race_thread_count = n
|
||||
|
||||
|
||||
# ── Then ──────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@then("connect() should have been called exactly once")
|
||||
def step_connect_called_once(context: Context) -> None:
|
||||
"""Assert connect() was called exactly once.
|
||||
|
||||
With the bug present (no STARTING state check inside the lock),
|
||||
multiple threads will enter the start() body and each call connect(),
|
||||
so the counter will be > 1.
|
||||
|
||||
This assertion therefore FAILS while the bug exists — which is the
|
||||
expected behavior for a @tdd_expected_fail test.
|
||||
|
||||
When the fix adds the _state == STARTING check inside the lock, only
|
||||
the first thread calls connect() and the count is exactly 1.
|
||||
"""
|
||||
count = context.counting_transport.connect_call_count
|
||||
assert count == 1, (
|
||||
f"connect() was called {count} time(s), expected exactly 1. "
|
||||
f"This confirms the race condition in McpClient.start() — multiple "
|
||||
f"threads passed the _started check because the STARTING state is "
|
||||
f"not checked inside the lock before proceeding. "
|
||||
f"Thread errors (if any): {context.race_thread_errors}"
|
||||
)
|
||||
|
||||
|
||||
@then("discover_tools() should have been called exactly once")
|
||||
def step_discover_tools_called_once(context: Context) -> None:
|
||||
"""Assert discover_tools() was called exactly once.
|
||||
|
||||
With the bug present, multiple threads call discover_tools() concurrently.
|
||||
With the fix, only one thread proceeds to call discover_tools().
|
||||
"""
|
||||
count = context.counting_transport.discover_call_count
|
||||
assert count == 1, (
|
||||
f"discover_tools() was called {count} time(s), expected exactly 1. "
|
||||
f"This confirms the race condition in McpClient.start() — multiple "
|
||||
f"threads proceeded past the idempotency check. "
|
||||
f"Thread errors (if any): {context.race_thread_errors}"
|
||||
)
|
||||
|
||||
|
||||
@then('the client state should be "{expected_state}"')
|
||||
def step_client_state_is(context: Context, expected_state: str) -> None:
|
||||
"""Assert the client is in the expected state after concurrent starts."""
|
||||
actual = context.race_client.state
|
||||
assert actual == expected_state, (
|
||||
f"Expected client state '{expected_state}', got '{actual}'. "
|
||||
f"Thread errors (if any): {context.race_thread_errors}"
|
||||
)
|
||||
|
||||
|
||||
@then("all {n:d} threads should have completed without hanging")
|
||||
def step_all_threads_completed_no_hang(context: Context, n: int) -> None:
|
||||
"""Verify that all *n* threads completed (either success or error).
|
||||
|
||||
The total of completed threads must equal the number launched.
|
||||
This confirms no threads silently hung or deadlocked.
|
||||
"""
|
||||
total = context.race_threads_completed
|
||||
assert total == n, (
|
||||
f"Expected {n} threads to complete but only {total} did. "
|
||||
"A mismatch indicates a hung thread or possible deadlock in "
|
||||
"McpClient.start(). "
|
||||
f"Thread errors (if any): {context.race_thread_errors}"
|
||||
)
|
||||
@@ -0,0 +1,27 @@
|
||||
# This test captures bug #10438 — McpClient.start() race condition.
|
||||
#
|
||||
# The @tdd_expected_fail tag inverts the test result: the underlying assertion
|
||||
# fails (proving the bug exists) but CI reports the scenario as passed.
|
||||
# When #10438 is fixed, the @tdd_expected_fail tag must be removed.
|
||||
#
|
||||
# See: https://git.cleverthis.com/cleveragents/cleveragents-core/issues/10438
|
||||
|
||||
@tdd_expected_fail @tdd_issue @tdd_issue_10438
|
||||
Feature: TDD Issue #10438 — McpClient.start() race condition allows concurrent double initialization
|
||||
McpClient.start() releases the threading.RLock after setting _state to STARTING
|
||||
but before calling connect() and discover_tools(). If two threads call start()
|
||||
concurrently, both can pass the _started idempotency check (which is still False)
|
||||
and both proceed to call connect() and discover_tools() simultaneously.
|
||||
|
||||
This test proves the race exists by launching multiple threads that call start()
|
||||
simultaneously via a threading.Barrier and verifying that connect() is called
|
||||
exactly once (the correct, thread-safe behavior). Without the bug present,
|
||||
connect() is called exactly once. With the bug, connect() is called multiple times.
|
||||
|
||||
Scenario: Concurrent start() calls must invoke connect() exactly once
|
||||
Given an McpClient with a counting mock transport
|
||||
When 5 threads call start() concurrently through a barrier
|
||||
Then connect() should have been called exactly once
|
||||
And discover_tools() should have been called exactly once
|
||||
And the client state should be "running"
|
||||
And all 5 threads should have completed without hanging
|
||||
Reference in New Issue
Block a user