fix(websocket): time out stalled sends and reset failed connections

packages/websocket/README.md

@@ -49,6 +49,20 @@ $server->onClose(function (int $connection) {
 $server->start();
 ```
 
+## Slow clients
+
+The Swoole adapter allows sends to wait for a full output buffer to drain, with a five-second timeout for each wait. This prevents pending sends from waiting indefinitely after a client disconnects, including the failure described in [Swoole issue #6196](https://github.com/swoole/swoole-src/issues/6196).
+
+Configure the timeout in the constructor:
+
+```php
+$adapter = new WebSocket\Adapter\Swoole(sendTimeout: 2.0); // Seconds; must be finite and greater than zero.
+```
+
+When a push fails, including on timeout, the adapter resets the connection and discards queued output. Clients must reconnect and refresh application state to recover missed events. Brief stalls can recover if the buffer drains before the timeout.
+
+The timeout applies to each Swoole wait, not the total lifetime of a send; a retry can start another wait. It does not cap pending bytes or change Swoole's output buffer size. Applications sending large bursts should also limit queued work or reduce update frequency.
+
 ## System requirements
 
 Utopia Framework requires PHP 8.0 or later. We recommend using the latest PHP version whenever possible.

packages/websocket/src/WebSocket/Adapter.php

@@ -7,7 +7,7 @@
 abstract class Adapter
 {
     /**
-    * @var array<int|string,bool|int|string>
+    * @var array<int|string,bool|int|float|string>
     */
 
     protected array $config = [];

packages/websocket/src/WebSocket/Adapter/Swoole.php

@@ -11,20 +11,33 @@
 
 class Swoole extends Adapter
 {
+    public const DEFAULT_SEND_TIMEOUT = 5.0;
+
     protected Server $server;
 
     protected string $host;
 
     protected int $port;
 
-    public function __construct(string $host = '0.0.0.0', int $port = 80)
-    {
+    /**
+     * @param float $sendTimeout Positive finite seconds for each wait on a full output buffer.
+     */
+    public function __construct(
+        string $host = '0.0.0.0',
+        int $port = 80,
+        float $sendTimeout = self::DEFAULT_SEND_TIMEOUT,
+    ) {
+        if (!is_finite($sendTimeout) || $sendTimeout <= 0) {
+            throw new \InvalidArgumentException('Send timeout must be a finite positive number of seconds');
+        }
+
         parent::__construct($host, $port);
 
         $this->server = new Server($this->host, $this->port);
 
         // Set maximum connections to Swoole's limit of 1 Million
         $this->config['max_connection'] = 1_000_000;
+        $this->config['send_timeout'] = $sendTimeout;
     }
 
     public function start(): void
@@ -45,17 +58,25 @@ public function send(array $connections, string $message): void
             $flags |= SWOOLE_WEBSOCKET_FLAG_COMPRESS;
         }
 
-        foreach ($connections as $connection) {
-            go(function () use ($connection, $message, $flags): void {
-                if ($this->server->exist($connection) && $this->server->isEstablished($connection)) {
-                    $this->server->push(
-                        $connection,
+        foreach ($connections as $sessionId) {
+            go(function () use ($sessionId, $message, $flags): void {
+                if ($this->server->isEstablished($sessionId)) {
+                    $pushed = $this->server->push(
+                        $sessionId,
                         $message,
                         SWOOLE_WEBSOCKET_OPCODE_TEXT,
                         $flags,
                     );
+
+                    // push() can yield. Swoole verifies the session ID here,
+                    // even if another client has reused the underlying socket fd.
+                    if (!$pushed && $this->server->exist($sessionId)) {
+                        // Discard queued output: a graceful close would keep
+                        // waiting for the same client to drain its buffer.
+                        $this->server->close($sessionId, true);
+                    }
                 } else {
-                    $this->server->close($connection);
+                    $this->server->close($sessionId);
                 }
             });
         }

packages/websocket/tests/E2E/AdapterTest.php

@@ -5,9 +5,13 @@
 namespace Utopia\WebSocket\Tests;
 
 use PHPUnit\Framework\TestCase;
+use Swoole\Coroutine;
+use Swoole\Coroutine\Http\Client as HttpClient;
 
 use function Swoole\Coroutine\run;
 
+use Swoole\Coroutine\Socket;
+use Swoole\WebSocket\Server as NativeServer;
 use Utopia\WebSocket\Client;
 
 final class AdapterTest extends TestCase
@@ -29,6 +33,182 @@ public function testWorkerman(): void
         $this->testServer('127.0.0.1', 18082);
     }
 
+    public function testSwooleTimesOutStalledSends(): void
+    {
+        $this->testPendingSends(false);
+    }
+
+    public function testSwooleReleasesSendsAfterPeerClose(): void
+    {
+        $this->testPendingSends(true);
+    }
+
+    public function testSwooleRecoversFromBriefStall(): void
+    {
+        run(function (): void {
+            $client = $this->getWebsocket('127.0.0.1', 18081);
+            $http = new HttpClient('127.0.0.1', 18081);
+            $http->set(['timeout' => 2]);
+
+            try {
+                $client->connect();
+                $baseline = $this->getInfo($http);
+                $messages = $this->createMessages();
+                $this->sendBatch($client, $messages);
+                $this->waitForBatch($http, $baseline['batches_sent'] + 1);
+
+                // Resume reading before the send timeout. Every frame must arrive.
+                foreach ($messages as $message) {
+                    $this->assertSame($message, $client->receive());
+                }
+                $client->send('ping');
+                $this->assertSame('pong', $client->receive());
+            } finally {
+                $client->close();
+                $http->close();
+            }
+        });
+    }
+
+    private function testPendingSends(bool $closePeer): void
+    {
+        run(function () use ($closePeer): void {
+            $slow = $this->connectSlowClient();
+            $healthy = $this->getWebsocket('127.0.0.1', 18081);
+            $replacement = null;
+            $http = new HttpClient('127.0.0.1', 18081);
+            $http->set(['timeout' => 2]);
+
+            try {
+                $healthy->connect();
+                $baseline = $this->getInfo($http);
+                $this->sendBatch($slow, $this->createMessages());
+                $this->waitForBatch($http, $baseline['batches_sent'] + 1);
+
+                if ($closePeer) {
+                    // Close with unread frames and connect another client before
+                    // the old sends time out. Its session must stay usable.
+                    $slow->close();
+                    $replacement = $this->getWebsocket('127.0.0.1', 18081);
+                    $replacement->connect();
+                    $replacement->send('ping');
+                    $this->assertSame('pong', $replacement->receive());
+                } else {
+                    // Probe for a transport error without draining the output. A
+                    // graceful close keeps the socket alive behind queued frames.
+                    $ping = NativeServer::pack('ping', WEBSOCKET_OPCODE_TEXT, SWOOLE_WEBSOCKET_FLAG_FIN | SWOOLE_WEBSOCKET_FLAG_MASK);
+                    $deadline = microtime(true) + 6;
+                    do {
+                        Coroutine::sleep(0.05);
+                        if ($slow->sendAll($ping, 0.2) === false) {
+                            $socketError = $slow->errCode;
+                            break;
+                        }
+                        $socketError = $slow->getOption(SOL_SOCKET, SO_ERROR);
+                    } while ($socketError === 0 && microtime(true) < $deadline);
+                    $this->assertContains($socketError, [SOCKET_ECONNRESET, SOCKET_EPIPE], 'The stalled peer must observe disconnection');
+                }
+
+                // The regression is retained payload memory, not a particular
+                // coroutine count. Allow 8 MiB for runtime/bookkeeping variation;
+                // the burst contains roughly 19 MiB of distinct payloads alone.
+                $budget = $baseline['memory_used'] + 8 * 1024 * 1024;
+                $deadline = microtime(true) + 6;
+                do {
+                    Coroutine::sleep(0.05);
+                    $info = $this->getInfo($http);
+                } while ($info['memory_used'] > $budget && microtime(true) < $deadline);
+                $this->assertLessThanOrEqual($budget, $info['memory_used'], 'Disconnected clients must release retained payloads');
+
+                $healthy->send('ping');
+                $this->assertSame('pong', $healthy->receive());
+                if ($replacement instanceof \Utopia\WebSocket\Client) {
+                    $replacement->send('ping');
+                    $this->assertSame('pong', $replacement->receive());
+                }
+            } finally {
+                $slow->close();
+                $healthy->close();
+                $replacement?->close();
+                $http->close();
+            }
+        });
+    }
+
+    /** @return list<string> */
+    private function createMessages(): array
+    {
+        $messages = [];
+        for ($i = 0; $i < 300; $i++) {
+            $messages[] = $i . ':' . str_repeat('x', 65536);
+        }
+
+        return $messages;
+    }
+
+    /** @param list<string> $messages */
+    private function sendBatch(Client|Socket $client, array $messages): void
+    {
+        foreach ($messages as $message) {
+            $this->sendMessage($client, 'buffer:' . $message);
+        }
+        $this->sendMessage($client, 'flush');
+    }
+
+    private function sendMessage(Client|Socket $client, string $message): void
+    {
+        if ($client instanceof Client) {
+            $client->send($message);
+
+            return;
+        }
+
+        $frame = NativeServer::pack($message, WEBSOCKET_OPCODE_TEXT, SWOOLE_WEBSOCKET_FLAG_FIN | SWOOLE_WEBSOCKET_FLAG_MASK);
+        $this->assertSame(\strlen($frame), $client->sendAll($frame));
+    }
+
+    private function connectSlowClient(): Socket
+    {
+        $socket = new Socket(AF_INET, SOCK_STREAM, IPPROTO_IP);
+        $this->assertTrue($socket->setOption(SOL_SOCKET, SO_RCVBUF, 1024));
+        $this->assertTrue($socket->connect('127.0.0.1', 18081, 2));
+        $request = "GET / HTTP/1.1\r\nHost: localhost\r\nUpgrade: websocket\r\n"
+            . "Connection: Upgrade\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n"
+            . "Sec-WebSocket-Version: 13\r\n\r\n";
+        $this->assertSame(\strlen($request), $socket->sendAll($request));
+        $headers = '';
+        while (!str_contains($headers, "\r\n\r\n")) {
+            $chunk = $socket->recv(4096, 2);
+            $this->assertNotFalse($chunk);
+            $this->assertNotSame('', $chunk);
+            $headers .= $chunk;
+        }
+        $this->assertStringContainsString('101 Switching Protocols', $headers);
+
+        return $socket;
+    }
+
+    private function waitForBatch(HttpClient $http, int $expected): void
+    {
+        $deadline = microtime(true) + 2;
+        do {
+            Coroutine::sleep(0.01);
+            $info = $this->getInfo($http);
+        } while ($info['batches_sent'] < $expected && microtime(true) < $deadline);
+
+        $this->assertSame($expected, $info['batches_sent'], 'The fixture must finish submitting the burst');
+    }
+
+    /**
+     * @return array{memory_used: int, batches_sent: int}
+     */
+    private function getInfo(HttpClient $http): array
+    {
+        $this->assertTrue($http->get('/info'));
+
+        return json_decode($http->body, true, flags: JSON_THROW_ON_ERROR);
+    }
+
     private function testServer(string $host, int $port): void
     {
         run(function () use ($host, $port): void {

packages/websocket/tests/Fixtures/Swoole/server.php

@@ -8,13 +8,18 @@
 use Swoole\Http\Response;
 use Utopia\WebSocket;
 
-$adapter = new WebSocket\Adapter\Swoole('127.0.0.1', 18081);
+$adapter = new WebSocket\Adapter\Swoole('127.0.0.1', 18081, sendTimeout: 1.0);
 $adapter->setWorkerNumber(1); // Important for tests
+// Fill the buffer quickly without allocating production-sized backlogs.
+$adapter->getNative()->ports[0]->set(['socket_buffer_size' => 65536]);
 
 $server = new WebSocket\Server($adapter);
 
 /** @var array<int,bool> $connections */
 $connections = [];
+/** @var array<int,list<string>> $batches */
+$batches = [];
+$batchesSent = 0;
 
 $server
     ->onWorkerStart(function (int $workerId): void {
@@ -27,14 +32,25 @@
         $connections[$connection] = true;
         echo 'connected ', $connection, PHP_EOL;
     })
-    ->onClose(function (int $connection) use (&$connections): void {
-        unset($connections[$connection]);
+    ->onClose(function (int $connection) use (&$connections, &$batches): void {
+        unset($connections[$connection], $batches[$connection]);
         echo 'disconnected ', $connection, PHP_EOL;
     })
-    ->onMessage(function (int $connection, string $message) use ($server, &$connections): void {
-        echo $message, PHP_EOL;
+    ->onMessage(function (int $connection, string $message) use ($server, &$connections, &$batches, &$batchesSent): void {
+        [$command, $payload] = explode(':', $message, 2) + [1 => ''];
 
-        switch ($message) {
+        switch ($command) {
+            case 'buffer':
+                $batches[$connection][] = $payload;
+                break;
+            case 'flush':
+                $batch = $batches[$connection] ?? [];
+                unset($batches[$connection]);
+                foreach ($batch as $payload) {
+                    $server->send([$connection], $payload);
+                }
+                $batchesSent++;
+                break;
             case 'ping':
                 $server->send([$connection], 'pong');
                 break;
@@ -50,7 +66,7 @@
                 break;
         }
     })
-    ->onRequest(function (Request $request, Response $response) use (&$connections): void {
+    ->onRequest(function (Request $request, Response $response) use (&$connections, &$batchesSent): void {
         echo 'HTTP request received: ', $request->server['request_uri'], PHP_EOL;
 
         if ($request->server['request_uri'] === '/health') {
@@ -63,6 +79,8 @@
             $response->end(json_encode([
                 'server' => 'Swoole WebSocket',
                 'connections' => count($connections),
+                'memory_used' => memory_get_usage(),
+                'batches_sent' => $batchesSent,
                 'timestamp' => time(),
             ]));
         } else {