From ef800c242dea7b664da56804e53e7779da816895 Mon Sep 17 00:00:00 2001 From: Christopher Hertel Date: Wed, 14 Oct 2026 01:34:41 +0200 Subject: [PATCH 1/6] [Server] Drop pending client requests of a vanished stream --- CHANGELOG.md | 1 + src/Server/Protocol.php | 32 ++++++++++++++++++++++++++ tests/Unit/Server/ProtocolTest.php | 37 ++++++++++++++++++++++++++++++ 3 files changed, 70 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 17c2c2a7..1b570397 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -85,6 +85,7 @@ All notable changes to `mcp/sdk` will be documented in this file. * `FileSessionStore::gc()` also removes the lock files of expired sessions a `FileSessionLock` in the same directory left behind, each only while no request holds it. * Resume a suspended handler with an error when the transport fails to store what it sends to the client next, instead of resuming it with `null`; `ClientGateway::notify()` then throws a `ClientException`. * Fix session writes of a handler after it resumed from an elicitation, sampling request or notification being lost. +* Fix pending client requests of a vanished stream, and their late answers, staying in the session until it expires: they are dropped once a minute past their timeout. 0.8.1 ----- diff --git a/src/Server/Protocol.php b/src/Server/Protocol.php index f17a4e5c..7463b0f0 100644 --- a/src/Server/Protocol.php +++ b/src/Server/Protocol.php @@ -80,6 +80,12 @@ class Protocol private const SESSION_BUSY_MESSAGE = 'Another request of this session is still being processed.'; private const SESSION_LOCK_UNAVAILABLE_MESSAGE = 'The session lock is unavailable.'; + /** + * Seconds past its timeout after which a pending client request counts as abandoned by its stream. + * Until then, the stream that sent it expires it itself and reports the timeout. + */ + private const ABANDONED_REQUEST_GRACE = 60; + /** * The client request each suspended fiber waits on. Pending requests are stored in the session, * which concurrent calls share, so a fiber must only be resumed with the answer to the one it sent. @@ -559,6 +565,8 @@ private function registerRequest(Request $request, int $timeout, SessionInterfac 'method' => $request::getMethod(), ]); + $this->dropAbandonedRequests($session); + // Per entry, so saving does not write back what other streams changed in the others. $session->set(self::SESSION_PENDING_REQUESTS.".{$id}", [ 'request_id' => $id, @@ -807,6 +815,30 @@ private function expireRequest(Uuid $requestId, Uuid $sessionId): ?Error return Error::forInternalError('Request timed out', $id); } + /** + * Drops the pending client requests, and their unconsumed answers, that no stream waits for anymore. + * + * Only the stream that sent a request clears it, so one whose stream is gone, through a client + * disconnect or a dead worker, would stay until the session expires. + */ + private function dropAbandonedRequests(SessionInterface $session): void + { + foreach ($session->get(self::SESSION_PENDING_REQUESTS, []) as $id => $pending) { + if (!\is_array($pending) || time() < (int) ($pending['timestamp'] ?? 0) + (int) ($pending['timeout'] ?? 120) + self::ABANDONED_REQUEST_GRACE) { + continue; + } + + $this->logger->debug('Dropping abandoned client request.', [ + 'request_id' => $id, + 'session_id' => $session->getId()->toRfc4122(), + ]); + + // Per entry, so saving does not write back what other streams changed in the others. + $session->forget(self::SESSION_PENDING_REQUESTS.".{$id}"); + $session->forget(self::SESSION_RESPONSES.".{$id}"); + } + } + /** * @param array $pending */ diff --git a/tests/Unit/Server/ProtocolTest.php b/tests/Unit/Server/ProtocolTest.php index c1fc4823..c21cc715 100644 --- a/tests/Unit/Server/ProtocolTest.php +++ b/tests/Unit/Server/ProtocolTest.php @@ -2291,6 +2291,43 @@ public function testAnswerAfterReportedTimeoutIsDropped(): void $this->assertNull($protocol->checkResponse(Uuid::fromString('00000000-0000-4000-8000-000000001000'), $session->getId())); } + #[TestDox('A client request its stream abandoned is dropped with its answer once another one is sent')] + public function testAbandonedRequestIsDroppedWhenAnotherRequestIsSent(): void + { + [$protocol, $sessionManager, $session] = $this->createProtocolWithPendingRequests([ + '00000000-0000-4000-8000-000000001000' => ['request_id' => '00000000-0000-4000-8000-000000001000', 'timeout' => 5, 'timestamp' => time() - 100], + ]); + $session->set('_mcp.responses.00000000-0000-4000-8000-000000001000', ['jsonrpc' => '2.0', 'id' => '00000000-0000-4000-8000-000000001000', 'result' => []]); + $session->save(); + + $session = $sessionManager->createWithId($session->getId()); + $requestId = $protocol->sendRequest(new PingRequest(), 5, $session); + $session->save(); + + $session = $sessionManager->createWithId($session->getId()); + $this->assertSame([$requestId->toRfc4122()], array_keys($session->get('_mcp.pending_requests'))); + $this->assertNull($session->get('_mcp.responses.00000000-0000-4000-8000-000000001000')); + } + + #[TestDox('A timed out client request within the grace period is left to its own stream')] + public function testTimedOutRequestWithinGracePeriodIsKeptWhenAnotherRequestIsSent(): void + { + [$protocol, $sessionManager, $session] = $this->createProtocolWithPendingRequests([ + '00000000-0000-4000-8000-000000001000' => ['request_id' => '00000000-0000-4000-8000-000000001000', 'timeout' => 5, 'timestamp' => time() - 10], + ]); + + $session = $sessionManager->createWithId($session->getId()); + $protocol->sendRequest(new PingRequest(), 5, $session); + $session->save(); + + $this->assertArrayHasKey('00000000-0000-4000-8000-000000001000', $protocol->getPendingRequests($session->getId())); + + $result = $protocol->checkResponse(Uuid::fromString('00000000-0000-4000-8000-000000001000'), $session->getId()); + + $this->assertInstanceOf(Error::class, $result); + $this->assertSame('Request timed out', $result->message); + } + #[TestDox('ClientResponseEvent::getId() is null when the error has no id')] public function testClientResponseEventGetIdCanBeNull(): void { From 6c6df7c0c95887b69a66ba87445df58064334e54 Mon Sep 17 00:00:00 2001 From: Christopher Hertel Date: Wed, 14 Oct 2026 01:48:48 +0200 Subject: [PATCH 2/6] [Server] Overwrite the answer of an abandoned request instead of forgetting it --- src/Server/Protocol.php | 5 +++-- tests/Unit/Server/ProtocolTest.php | 20 ++++++++++++++++++++ 2 files changed, 23 insertions(+), 2 deletions(-) diff --git a/src/Server/Protocol.php b/src/Server/Protocol.php index 7463b0f0..b840ed49 100644 --- a/src/Server/Protocol.php +++ b/src/Server/Protocol.php @@ -833,9 +833,10 @@ private function dropAbandonedRequests(SessionInterface $session): void 'session_id' => $session->getId()->toRfc4122(), ]); - // Per entry, so saving does not write back what other streams changed in the others. + // Per entry, so saving does not write back what other streams changed in the others. The answer + // may have been stored after this session was loaded, so it is overwritten, not forgotten. $session->forget(self::SESSION_PENDING_REQUESTS.".{$id}"); - $session->forget(self::SESSION_RESPONSES.".{$id}"); + $session->set(self::SESSION_RESPONSES.".{$id}", null); } } diff --git a/tests/Unit/Server/ProtocolTest.php b/tests/Unit/Server/ProtocolTest.php index c21cc715..4f7133ea 100644 --- a/tests/Unit/Server/ProtocolTest.php +++ b/tests/Unit/Server/ProtocolTest.php @@ -2309,6 +2309,26 @@ public function testAbandonedRequestIsDroppedWhenAnotherRequestIsSent(): void $this->assertNull($session->get('_mcp.responses.00000000-0000-4000-8000-000000001000')); } + #[TestDox('An answer to an abandoned client request is dropped even if it arrived after the session was loaded')] + public function testAnswerStoredAfterLoadingIsDroppedWithAbandonedRequest(): void + { + [$protocol, $sessionManager, $session] = $this->createProtocolWithPendingRequests([ + '00000000-0000-4000-8000-000000001000' => ['request_id' => '00000000-0000-4000-8000-000000001000', 'timeout' => 5, 'timestamp' => time() - 100], + ]); + + $snapshot = $sessionManager->createWithId($session->getId()); + $snapshot->get('_mcp.pending_requests'); + + $other = $sessionManager->createWithId($session->getId()); + $other->set('_mcp.responses.00000000-0000-4000-8000-000000001000', ['jsonrpc' => '2.0', 'id' => '00000000-0000-4000-8000-000000001000', 'result' => []]); + $other->save(); + + $protocol->sendRequest(new PingRequest(), 5, $snapshot); + $snapshot->save(); + + $this->assertNull($sessionManager->createWithId($session->getId())->get('_mcp.responses.00000000-0000-4000-8000-000000001000')); + } + #[TestDox('A timed out client request within the grace period is left to its own stream')] public function testTimedOutRequestWithinGracePeriodIsKeptWhenAnotherRequestIsSent(): void { From 8318cfeee303445b9f7a9140788ac9a2842c5185 Mon Sep 17 00:00:00 2001 From: Christopher Hertel Date: Thu, 15 Oct 2026 01:05:50 +0200 Subject: [PATCH 3/6] [Server] Address review: keep the pending requests a fiber of this process still waits on --- src/Server/Protocol.php | 8 +++++- tests/Unit/Server/ProtocolTest.php | 40 ++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+), 1 deletion(-) diff --git a/src/Server/Protocol.php b/src/Server/Protocol.php index b840ed49..d51f15e3 100644 --- a/src/Server/Protocol.php +++ b/src/Server/Protocol.php @@ -823,8 +823,14 @@ private function expireRequest(Uuid $requestId, Uuid $sessionId): ?Error */ private function dropAbandonedRequests(SessionInterface $session): void { + // A fiber of this process still waits on these, e.g. on stdio while another handler blocked the loop. + $awaited = []; + foreach ($this->awaitedRequestIds as $requestId) { + $awaited[$requestId->toRfc4122()] = true; + } + foreach ($session->get(self::SESSION_PENDING_REQUESTS, []) as $id => $pending) { - if (!\is_array($pending) || time() < (int) ($pending['timestamp'] ?? 0) + (int) ($pending['timeout'] ?? 120) + self::ABANDONED_REQUEST_GRACE) { + if (isset($awaited[$id]) || !\is_array($pending) || time() < (int) ($pending['timestamp'] ?? 0) + (int) ($pending['timeout'] ?? 120) + self::ABANDONED_REQUEST_GRACE) { continue; } diff --git a/tests/Unit/Server/ProtocolTest.php b/tests/Unit/Server/ProtocolTest.php index 4f7133ea..c33f4015 100644 --- a/tests/Unit/Server/ProtocolTest.php +++ b/tests/Unit/Server/ProtocolTest.php @@ -2329,6 +2329,46 @@ public function testAnswerStoredAfterLoadingIsDroppedWithAbandonedRequest(): voi $this->assertNull($sessionManager->createWithId($session->getId())->get('_mcp.responses.00000000-0000-4000-8000-000000001000')); } + #[TestDox('A client request a suspended fiber of the same process still waits on is not dropped as abandoned')] + public function testRequestAwaitedByLocalFiberIsKeptPastGracePeriod(): void + { + $handler = $this->createMock(RequestHandlerInterface::class); + $handler->method('supports')->willReturn(true); + $handler->method('handle')->willReturnCallback(static function (Request $request, SessionInterface $session): Response { + \Fiber::suspend(new RequestSuspension(new PingRequest(), $session->getId()->toRfc4122(), 5)); + + return new Response($request->getId(), []); + }); + + $sessionManager = new SessionManager(new InMemorySessionStore()); + $protocol = new Protocol( + requestHandlers: [$handler], + notificationHandlers: [], + messageFactory: MessageFactory::make(), + sessionManager: $sessionManager, + ); + + $session = $sessionManager->create(); + $session->save(); + $sessionId = $session->getId(); + + $stream = new PollingLoopTransport(); + $protocol->connect($stream); + $protocol->processInput($stream, '{"jsonrpc": "2.0", "id": 1, "method": "ping"}', $sessionId); + $this->assertCount(1, $pendingRequestIds = $stream->getPendingRequestIds()); + + // The loop that would poll the fiber was blocked well past the request's timeout. + $session = $sessionManager->createWithId($sessionId); + $session->set("_mcp.pending_requests.{$pendingRequestIds[0]}.timestamp", time() - 100); + $session->save(); + + $session = $sessionManager->createWithId($sessionId); + $protocol->sendRequest(new PingRequest(), 5, $session); + $session->save(); + + $this->assertSame($pendingRequestIds, $stream->getPendingRequestIds()); + } + #[TestDox('A timed out client request within the grace period is left to its own stream')] public function testTimedOutRequestWithinGracePeriodIsKeptWhenAnotherRequestIsSent(): void { From 585572e9105cef4ea4f7a97cec780ee464064b7f Mon Sep 17 00:00:00 2001 From: Christopher Hertel Date: Thu, 15 Oct 2026 01:13:39 +0200 Subject: [PATCH 4/6] [Server] Address review: remove the answer key of an abandoned request instead of storing null --- src/Server/Protocol.php | 3 ++- tests/Unit/Server/ProtocolTest.php | 4 ++-- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/src/Server/Protocol.php b/src/Server/Protocol.php index d51f15e3..b926626a 100644 --- a/src/Server/Protocol.php +++ b/src/Server/Protocol.php @@ -840,9 +840,10 @@ private function dropAbandonedRequests(SessionInterface $session): void ]); // Per entry, so saving does not write back what other streams changed in the others. The answer - // may have been stored after this session was loaded, so it is overwritten, not forgotten. + // may have been stored after this session was loaded: set() makes forget() record its removal. $session->forget(self::SESSION_PENDING_REQUESTS.".{$id}"); $session->set(self::SESSION_RESPONSES.".{$id}", null); + $session->forget(self::SESSION_RESPONSES.".{$id}"); } } diff --git a/tests/Unit/Server/ProtocolTest.php b/tests/Unit/Server/ProtocolTest.php index c33f4015..3432ccad 100644 --- a/tests/Unit/Server/ProtocolTest.php +++ b/tests/Unit/Server/ProtocolTest.php @@ -2306,7 +2306,7 @@ public function testAbandonedRequestIsDroppedWhenAnotherRequestIsSent(): void $session = $sessionManager->createWithId($session->getId()); $this->assertSame([$requestId->toRfc4122()], array_keys($session->get('_mcp.pending_requests'))); - $this->assertNull($session->get('_mcp.responses.00000000-0000-4000-8000-000000001000')); + $this->assertArrayNotHasKey('00000000-0000-4000-8000-000000001000', $session->get('_mcp.responses', [])); } #[TestDox('An answer to an abandoned client request is dropped even if it arrived after the session was loaded')] @@ -2326,7 +2326,7 @@ public function testAnswerStoredAfterLoadingIsDroppedWithAbandonedRequest(): voi $protocol->sendRequest(new PingRequest(), 5, $snapshot); $snapshot->save(); - $this->assertNull($sessionManager->createWithId($session->getId())->get('_mcp.responses.00000000-0000-4000-8000-000000001000')); + $this->assertArrayNotHasKey('00000000-0000-4000-8000-000000001000', $sessionManager->createWithId($session->getId())->get('_mcp.responses', [])); } #[TestDox('A client request a suspended fiber of the same process still waits on is not dropped as abandoned')] From 57800e35811acc03d7bd65129c5798501b6778f1 Mon Sep 17 00:00:00 2001 From: Christopher Hertel Date: Thu, 15 Oct 2026 01:24:24 +0200 Subject: [PATCH 5/6] [Server] Address review: time out an awaited request another worker dropped as abandoned --- src/Server/Protocol.php | 43 ++++++++++++++++++- tests/Unit/Server/ProtocolSessionLockTest.php | 35 +++++++++++++++ 2 files changed, 77 insertions(+), 1 deletion(-) diff --git a/src/Server/Protocol.php b/src/Server/Protocol.php index b926626a..2790a9fe 100644 --- a/src/Server/Protocol.php +++ b/src/Server/Protocol.php @@ -102,6 +102,14 @@ class Protocol */ private \WeakMap $fiberOutgoingMessages; + /** + * Awaited requests another worker dropped as abandoned, e.g. while this one could not get the session + * lock to expire them: their fiber is still owed the timeout. + * + * @var array + */ + private array $droppedAwaitedRequestIds = []; + /** * @param array>> $requestHandlers * @param array $notificationHandlers @@ -744,6 +752,16 @@ public function checkResponse(Uuid $requestId, Uuid $sessionId): Response|Error| { $id = $requestId->toRfc4122(); + if (isset($this->droppedAwaitedRequestIds[$id])) { + unset($this->droppedAwaitedRequestIds[$id]); + $this->logger->warning('Client request timed out.', [ + 'request_id' => $id, + 'session_id' => $sessionId->toRfc4122(), + ]); + + return Error::forInternalError('Request timed out', $id); + } + if (!$this->tryAcquireSessionLockForPoll($sessionId)) { return null; } @@ -766,6 +784,7 @@ public function checkResponse(Uuid $requestId, Uuid $sessionId): Response|Error| $session->set(self::SESSION_RESPONSES.".{$id}", null); $session->forget(self::SESSION_PENDING_REQUESTS.".{$id}"); $session->save(); + $this->settleAwaitedRequest($id); } finally { $this->releaseSessionLock($sessionId); } @@ -811,6 +830,7 @@ private function expireRequest(Uuid $requestId, Uuid $sessionId): ?Error $session->forget(self::SESSION_PENDING_REQUESTS.".{$id}"); $session->save(); + $this->settleAwaitedRequest($id); return Error::forInternalError('Request timed out', $id); } @@ -968,6 +988,18 @@ private function trackAwaitedRequest(\Fiber $fiber, ?Uuid $requestId): void $this->awaitedRequestIds[$fiber] = $requestId; } + /** + * The fiber that awaited this request gets its answer now, so a later poll must not take it for one dropped elsewhere. + */ + private function settleAwaitedRequest(string $id): void + { + foreach ($this->awaitedRequestIds as $fiber => $requestId) { + if ($requestId->toRfc4122() === $id) { + unset($this->awaitedRequestIds[$fiber]); + } + } + } + /** * @param McpFiber $fiber * @@ -980,7 +1012,16 @@ private function getAwaitedPendingRequests(\Fiber $fiber, Uuid $sessionId): arra return []; } - return array_intersect_key($this->getPendingRequests($sessionId), [$requestId->toRfc4122() => true]); + $id = $requestId->toRfc4122(); + $pending = $this->getPendingRequests($sessionId)[$id] ?? null; + + if (null === $pending) { + $this->droppedAwaitedRequestIds[$id] = true; + + return [$id => ['request_id' => $id]]; + } + + return [$id => $pending]; } /** diff --git a/tests/Unit/Server/ProtocolSessionLockTest.php b/tests/Unit/Server/ProtocolSessionLockTest.php index ca62a625..47b360c5 100644 --- a/tests/Unit/Server/ProtocolSessionLockTest.php +++ b/tests/Unit/Server/ProtocolSessionLockTest.php @@ -26,6 +26,7 @@ use Mcp\Server\Session\SessionManagerInterface; use Mcp\Server\Suspension\RequestSuspension; use Mcp\Server\Transport\InMemoryTransport; +use Mcp\Tests\Unit\Fixtures\PollingLoopTransport; use Mcp\Tests\Unit\Fixtures\RecordingTransport; use Mcp\Tests\Unit\Server\Session\Fixture\RecordingLogger; use Mcp\Tests\Unit\Server\Session\Fixture\RecordingSessionLock; @@ -512,6 +513,40 @@ public function testSessionDestroyedWhileTheHandlerWaitedStaysDestroyed(): void $this->assertSame([], $lock->heldSessions()); } + #[TestDox('a request another worker dropped as abandoned while the lock kept its stream from expiring it still times out')] + public function testAwaitedRequestDroppedByAnotherWorkerTimesOut(): void + { + $lock = new RecordingSessionLock($this->log); + $waiting = $this->resumingProtocol($lock); + $sessionId = $this->existingSession(); + $stream = new PollingLoopTransport(); + $waiting->connect($stream); + $waiting->processInput($stream, '{"jsonrpc":"2.0","id":1,"method":"ping"}', $sessionId); + $this->assertCount(1, $pendingRequestIds = $stream->getPendingRequestIds()); + $requestId = Uuid::fromString($pendingRequestIds[0]); + + $session = new Session($this->store, $sessionId); + $session->set("_mcp.pending_requests.{$pendingRequestIds[0]}.timestamp", time() - 100); + $session->save(); + + // Another worker holds the session past the request's timeout, then sends a request of its own. + $lock->acquirable = false; + $this->assertNull($waiting->checkResponse($requestId, $sessionId)); + $lock->acquirable = true; + + $sweeping = $this->protocol($lock); + $session = new Session($this->store, $sessionId); + $sweeping->sendRequest(new PingRequest(), 5, $session); + $session->save(); + $this->assertArrayNotHasKey($pendingRequestIds[0], (new Session($this->store, $sessionId))->get('_mcp.pending_requests')); + + $this->assertSame($pendingRequestIds, $stream->getPendingRequestIds()); + $result = $waiting->checkResponse($requestId, $sessionId); + $this->assertInstanceOf(Error::class, $result); + $this->assertSame('Request timed out', $result->message); + $this->assertNull($waiting->checkResponse($requestId, $sessionId), 'The timeout is reported only once.'); + } + private function resumingProtocol(RecordingSessionLock $lock): Protocol { $handler = new class implements RequestHandlerInterface { From 63ab5389896f1370fc9487e5f05c025c9bb07def Mon Sep 17 00:00:00 2001 From: Christopher Hertel Date: Thu, 15 Oct 2026 01:30:31 +0200 Subject: [PATCH 6/6] [Server] Address review: settle a dropped awaited request once its timeout is reported --- src/Server/Protocol.php | 1 + tests/Unit/Server/ProtocolSessionLockTest.php | 1 + tests/Unit/Server/ProtocolTest.php | 1 + 3 files changed, 3 insertions(+) diff --git a/src/Server/Protocol.php b/src/Server/Protocol.php index 2790a9fe..7c8a7be0 100644 --- a/src/Server/Protocol.php +++ b/src/Server/Protocol.php @@ -754,6 +754,7 @@ public function checkResponse(Uuid $requestId, Uuid $sessionId): Response|Error| if (isset($this->droppedAwaitedRequestIds[$id])) { unset($this->droppedAwaitedRequestIds[$id]); + $this->settleAwaitedRequest($id); $this->logger->warning('Client request timed out.', [ 'request_id' => $id, 'session_id' => $sessionId->toRfc4122(), diff --git a/tests/Unit/Server/ProtocolSessionLockTest.php b/tests/Unit/Server/ProtocolSessionLockTest.php index 47b360c5..039fca4d 100644 --- a/tests/Unit/Server/ProtocolSessionLockTest.php +++ b/tests/Unit/Server/ProtocolSessionLockTest.php @@ -544,6 +544,7 @@ public function testAwaitedRequestDroppedByAnotherWorkerTimesOut(): void $result = $waiting->checkResponse($requestId, $sessionId); $this->assertInstanceOf(Error::class, $result); $this->assertSame('Request timed out', $result->message); + $this->assertSame([], $stream->getPendingRequestIds()); $this->assertNull($waiting->checkResponse($requestId, $sessionId), 'The timeout is reported only once.'); } diff --git a/tests/Unit/Server/ProtocolTest.php b/tests/Unit/Server/ProtocolTest.php index 3432ccad..24d169c6 100644 --- a/tests/Unit/Server/ProtocolTest.php +++ b/tests/Unit/Server/ProtocolTest.php @@ -2366,6 +2366,7 @@ public function testRequestAwaitedByLocalFiberIsKeptPastGracePeriod(): void $protocol->sendRequest(new PingRequest(), 5, $session); $session->save(); + $this->assertArrayHasKey($pendingRequestIds[0], $protocol->getPendingRequests($sessionId)); $this->assertSame($pendingRequestIds, $stream->getPendingRequestIds()); }