[Server] Stop the handshake-era SSE loop once the client disconnected

Author: chr-hertel

CHANGELOG.md

@@ -59,6 +59,7 @@ All notable changes to `mcp/sdk` will be documented in this file.
 * Fix a client `onProgress` callback receiving progress for an earlier request that timed out: progress now only reaches the callback of the request whose `progressToken` it carries.
 * Fix `Client::isConnected()` staying true after the stdio server process exited, which left later calls failing against the dead process.
 * 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
 -----

src/Server/Transport/StreamableHttpTransport.php

@@ -293,6 +293,8 @@ protected function createStreamedResponse(): ResponseInterface
                     flush();
                 }
 
+                $lastKeepAlive = null;
+
                 while ($fiber->isSuspended()) {
                     $this->flushOutgoingMessages($this->sessionId);
 
@@ -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;
+                        }
+
                         usleep(100000);
                     } // Prevent tight loop
                 }
@@ -344,6 +364,11 @@ protected function createStreamedResponse(): ResponseInterface
         return $response;
     }
 
+    protected function isConnectionAborted(): bool
+    {
+        return 1 === connection_aborted();
+    }
+
     /**
      * @param McpFiber $fiber
      */

tests/Unit/Server/Transport/StreamableHttpTransportTest.php

@@ -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}>
      */