Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,7 @@ All notable changes to `mcp/sdk` will be documented in this file.
* Fix `Client::isConnected()` staying true after the stdio server process exited, which left later calls failing against the dead process.
* Fix the server's `StdioTransport` answering a request on an idle connection only after ~50ms: it now waits on the input with `stream_select()` instead of sleeping.
* Fix a client request that timed out (elicitation, sampling) staying in the session's pending requests: `Protocol::checkResponse()` now expires it, answering with the `Request timed out` error, so the transports no longer implement the timeout themselves.
* Fix a handshake-era Streamable HTTP stream pinning its PHP worker until the elicitation (or other client request) times out after the client disconnected: the idle stream now sends a keep-alive comment and stops once the connection is aborted.

0.8.1
-----
Expand Down
25 changes: 25 additions & 0 deletions src/Server/Transport/StreamableHttpTransport.php
Original file line number Diff line number Diff line change
Expand Up @@ -293,6 +293,8 @@ protected function createStreamedResponse(): ResponseInterface
flush();
}

$lastKeepAlive = null;

while ($fiber->isSuspended()) {
$this->flushOutgoingMessages($this->sessionId);

Expand All @@ -319,6 +321,24 @@ protected function createStreamedResponse(): ResponseInterface
}

if (!$resumed) {
// PHP only notices a dropped peer when it writes, so the
// idle stream writes, or a vanished client would pin this
// worker until the pending request times out. Paced like
// the stateless keep-alive, not at every tick.
$now = microtime(true);
if (null === $lastKeepAlive || $now - $lastKeepAlive >= 0.25) {
$lastKeepAlive = $now;
echo ": keep-alive\n\n";
@ob_flush();
flush();
}

if ($this->isConnectionAborted()) {
$this->logger->info('SSE: Client disconnected, dropping the suspended request.');

return;
Comment on lines +336 to +339
}

usleep(100000);
} // Prevent tight loop
}
Expand All @@ -344,6 +364,11 @@ protected function createStreamedResponse(): ResponseInterface
return $response;
}

protected function isConnectionAborted(): bool
{
return 1 === connection_aborted();
}

/**
* @param McpFiber $fiber
*/
Expand Down
51 changes: 51 additions & 0 deletions tests/Unit/Server/Transport/StreamableHttpTransportTest.php
Original file line number Diff line number Diff line change
Expand Up @@ -448,6 +448,57 @@ public function testPollingLoopResumesFiberWithTimeoutErrorFromResponseFinder():
$this->assertInstanceOf(Error::class, $received);
}

#[TestDox('the polling loop stops once the client has disconnected')]
public function testPollingLoopStopsOnceTheClientHasDisconnected(): void
{
$request = $this->factory
->createServerRequest('POST', 'http://localhost/')
->withHeader('Host', 'localhost')
->withBody($this->factory->createStream('{"jsonrpc":"2.0","id":1,"method":"ping"}'));

$transport = new class($request, $this->factory, $this->factory) extends StreamableHttpTransport {
protected function isConnectionAborted(): bool
{
return true;
}
};

$received = null;
$fiber = new \Fiber(static function () use (&$received) {
$received = \Fiber::suspend();

return null;
});
$fiber->start();

$transport->onMessage(static function (TransportInterface $transport) use ($fiber): void {
$transport->attachFiberToSession($fiber, Uuid::v4());
});
$transport->setOutgoingMessagesProvider(static fn (): array => []);
$transport->setResponseFinder(static fn () => null);
$transport->setPendingRequestsProvider(static fn (): array => [
['request_id' => 1, 'timestamp' => time(), 'timeout' => 120],
]);

$response = $transport->listen();

$output = '';
ob_start(static function (string $chunk) use (&$output): string {
$output .= $chunk;

return '';
});
try {
$response->getBody()->getContents();
} finally {
ob_end_flush();
}

$this->assertSame(": keep-alive\n\n", $output);
$this->assertTrue($fiber->isSuspended(), 'The abandoned call must not be resumed.');
$this->assertNull($received);
}

/**
* @return iterable<string, array{bool}>
*/
Expand Down
Loading