From b29ec6126ac795c576bd6344aa7028d0160259e1 Mon Sep 17 00:00:00 2001 From: loks0n <22452787+loks0n@users.noreply.github.com> Date: Wed, 9 Sep 2026 12:31:19 +0100 Subject: [PATCH 1/4] fix(websocket): time out stalled sends and reset failed connections --- packages/websocket/README.md | 15 +++ packages/websocket/src/WebSocket/Adapter.php | 2 +- .../src/WebSocket/Adapter/Swoole.php | 27 ++++- packages/websocket/tests/E2E/AdapterTest.php | 104 ++++++++++++++++++ .../tests/Fixtures/Swoole/server.php | 19 +++- 5 files changed, 162 insertions(+), 5 deletions(-) diff --git a/packages/websocket/README.md b/packages/websocket/README.md index ddc42f862..9bcdcd214 100644 --- a/packages/websocket/README.md +++ b/packages/websocket/README.md @@ -49,6 +49,21 @@ $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 before starting the server: + +```php +$adapter = new WebSocket\Adapter\Swoole(); +$adapter->setSendTimeout(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. diff --git a/packages/websocket/src/WebSocket/Adapter.php b/packages/websocket/src/WebSocket/Adapter.php index 81d3db8f8..b2fc0cfe1 100644 --- a/packages/websocket/src/WebSocket/Adapter.php +++ b/packages/websocket/src/WebSocket/Adapter.php @@ -7,7 +7,7 @@ abstract class Adapter { /** - * @var array + * @var array */ protected array $config = []; diff --git a/packages/websocket/src/WebSocket/Adapter/Swoole.php b/packages/websocket/src/WebSocket/Adapter/Swoole.php index d05b62332..171319923 100644 --- a/packages/websocket/src/WebSocket/Adapter/Swoole.php +++ b/packages/websocket/src/WebSocket/Adapter/Swoole.php @@ -11,6 +11,8 @@ class Swoole extends Adapter { + public const DEFAULT_SEND_TIMEOUT = 5.0; + protected Server $server; protected string $host; @@ -25,6 +27,7 @@ public function __construct(string $host = '0.0.0.0', int $port = 80) // Set maximum connections to Swoole's limit of 1 Million $this->config['max_connection'] = 1_000_000; + $this->config['send_timeout'] = self::DEFAULT_SEND_TIMEOUT; } public function start(): void @@ -47,13 +50,19 @@ public function send(array $connections, string $message): void foreach ($connections as $connection) { go(function () use ($connection, $message, $flags): void { - if ($this->server->exist($connection) && $this->server->isEstablished($connection)) { - $this->server->push( + if ($this->server->isEstablished($connection)) { + $pushed = $this->server->push( $connection, $message, SWOOLE_WEBSOCKET_OPCODE_TEXT, $flags, ); + + if (!$pushed && $this->server->exist($connection)) { + // Discard queued output: a graceful close would keep + // waiting for the same client to drain its buffer. + $this->server->close($connection, true); + } } else { $this->server->close($connection); } @@ -153,6 +162,20 @@ public function setWorkerNumber(int $num): self return $this; } + /** + * Sets the timeout in seconds for each wait on a full output buffer. + */ + public function setSendTimeout(float $seconds): self + { + if (!is_finite($seconds) || $seconds <= 0) { + throw new \InvalidArgumentException('Send timeout must be a finite positive number of seconds'); + } + + $this->config['send_timeout'] = $seconds; + + return $this; + } + public function getNative(): Server { return $this->server; diff --git a/packages/websocket/tests/E2E/AdapterTest.php b/packages/websocket/tests/E2E/AdapterTest.php index 0254ad32e..faa746dc9 100644 --- a/packages/websocket/tests/E2E/AdapterTest.php +++ b/packages/websocket/tests/E2E/AdapterTest.php @@ -5,6 +5,8 @@ namespace Utopia\WebSocket\Tests; use PHPUnit\Framework\TestCase; +use Swoole\Coroutine; +use Swoole\Coroutine\Http\Client as HttpClient; use function Swoole\Coroutine\run; @@ -29,6 +31,108 @@ 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); + $client->send('flood'); + $this->waitForPendingSends($http, $baseline['coroutines']); + + // Resume reading before the send timeout. Every frame must arrive. + for ($i = 0; $i < 300; $i++) { + $frame = $client->receive(); + $this->assertSame(65536, \strlen($frame)); + $this->assertSame(\sprintf('%06d:', $i), substr($frame, 0, 7)); + } + $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->getWebsocket('127.0.0.1', 18081); + $healthy = $this->getWebsocket('127.0.0.1', 18081); + $http = new HttpClient('127.0.0.1', 18081); + $http->set(['timeout' => 2]); + + try { + $slow->connect(); + $healthy->connect(); + $baseline = $this->getInfo($http); + $slow->send('flood'); + $this->waitForPendingSends($http, $baseline['coroutines']); + + if ($closePeer) { + // Close the TCP socket with unread frames, as in the upstream repro. + $slow->close(); + } + + $deadline = microtime(true) + 6; + do { + Coroutine::sleep(0.05); + $info = $this->getInfo($http); + } while (($info['connections'] !== 1 || $info['native_connections'] !== 2 + || $info['coroutines'] !== $baseline['coroutines']) && microtime(true) < $deadline); + + $this->assertSame(1, $info['connections'], 'The stalled client must be disconnected'); + // The native server also counts this HTTP request. Checking it catches + // graceful closes that run onClose but retain the undrained socket. + $this->assertSame(2, $info['native_connections'], 'The stalled socket must be released'); + $this->assertSame($baseline['coroutines'], $info['coroutines'], 'Pending sends must finish'); + $healthy->send('ping'); + $this->assertSame('pong', $healthy->receive()); + } finally { + $slow->close(); + $healthy->close(); + $http->close(); + } + }); + } + + private function waitForPendingSends(HttpClient $http, int $baseline): void + { + $deadline = microtime(true) + 2; + do { + Coroutine::sleep(0.01); + $info = $this->getInfo($http); + } while ((!$info['flood_complete'] || $info['coroutines'] <= $baseline + 10) && microtime(true) < $deadline); + + $this->assertTrue($info['flood_complete']); + $this->assertGreaterThan($baseline + 10, $info['coroutines'], 'The test must exercise suspended sends'); + } + + /** + * @return array{connections: int, native_connections: int, coroutines: int, flood_complete: bool} + */ + 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 { diff --git a/packages/websocket/tests/Fixtures/Swoole/server.php b/packages/websocket/tests/Fixtures/Swoole/server.php index 0f4a0ecb0..f7e3a90ca 100644 --- a/packages/websocket/tests/Fixtures/Swoole/server.php +++ b/packages/websocket/tests/Fixtures/Swoole/server.php @@ -4,17 +4,22 @@ require_once __DIR__ . '/../../../vendor/autoload.php'; +use Swoole\Coroutine; use Swoole\Http\Request; use Swoole\Http\Response; use Utopia\WebSocket; $adapter = new WebSocket\Adapter\Swoole('127.0.0.1', 18081); $adapter->setWorkerNumber(1); // Important for tests +$adapter->setSendTimeout(1.0); +// 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 $connections */ $connections = []; +$floodComplete = false; $server ->onWorkerStart(function (int $workerId): void { @@ -31,10 +36,17 @@ unset($connections[$connection]); echo 'disconnected ', $connection, PHP_EOL; }) - ->onMessage(function (int $connection, string $message) use ($server, &$connections): void { + ->onMessage(function (int $connection, string $message) use ($server, &$connections, &$floodComplete): void { echo $message, PHP_EOL; switch ($message) { + case 'flood': + $floodComplete = false; + for ($i = 0; $i < 300; $i++) { + $server->send([$connection], str_pad(sprintf('%06d:', $i), 65536, 'x')); + } + $floodComplete = true; + break; case 'ping': $server->send([$connection], 'pong'); break; @@ -50,7 +62,7 @@ break; } }) - ->onRequest(function (Request $request, Response $response) use (&$connections): void { + ->onRequest(function (Request $request, Response $response) use ($adapter, &$connections, &$floodComplete): void { echo 'HTTP request received: ', $request->server['request_uri'], PHP_EOL; if ($request->server['request_uri'] === '/health') { @@ -63,6 +75,9 @@ $response->end(json_encode([ 'server' => 'Swoole WebSocket', 'connections' => count($connections), + 'native_connections' => $adapter->getNative()->stats()['connection_num'], + 'coroutines' => Coroutine::stats()['coroutine_num'], + 'flood_complete' => $floodComplete, 'timestamp' => time(), ])); } else { From 00541246abf79232255ba2c78b70060f18ce6139 Mon Sep 17 00:00:00 2001 From: loks0n <22452787+loks0n@users.noreply.github.com> Date: Wed, 9 Sep 2026 12:47:02 +0100 Subject: [PATCH 2/4] refactor(websocket): configure send timeout in constructor --- packages/websocket/README.md | 5 ++-- .../src/WebSocket/Adapter/Swoole.php | 30 ++++++++----------- .../tests/Fixtures/Swoole/server.php | 3 +- 3 files changed, 16 insertions(+), 22 deletions(-) diff --git a/packages/websocket/README.md b/packages/websocket/README.md index 9bcdcd214..ee226ae7e 100644 --- a/packages/websocket/README.md +++ b/packages/websocket/README.md @@ -53,11 +53,10 @@ $server->start(); 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 before starting the server: +Configure the timeout in the constructor: ```php -$adapter = new WebSocket\Adapter\Swoole(); -$adapter->setSendTimeout(2.0); // Seconds; must be finite and greater than zero. +$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. diff --git a/packages/websocket/src/WebSocket/Adapter/Swoole.php b/packages/websocket/src/WebSocket/Adapter/Swoole.php index 171319923..2ec77adb8 100644 --- a/packages/websocket/src/WebSocket/Adapter/Swoole.php +++ b/packages/websocket/src/WebSocket/Adapter/Swoole.php @@ -19,15 +19,25 @@ class Swoole extends Adapter 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'] = self::DEFAULT_SEND_TIMEOUT; + $this->config['send_timeout'] = $sendTimeout; } public function start(): void @@ -162,20 +172,6 @@ public function setWorkerNumber(int $num): self return $this; } - /** - * Sets the timeout in seconds for each wait on a full output buffer. - */ - public function setSendTimeout(float $seconds): self - { - if (!is_finite($seconds) || $seconds <= 0) { - throw new \InvalidArgumentException('Send timeout must be a finite positive number of seconds'); - } - - $this->config['send_timeout'] = $seconds; - - return $this; - } - public function getNative(): Server { return $this->server; diff --git a/packages/websocket/tests/Fixtures/Swoole/server.php b/packages/websocket/tests/Fixtures/Swoole/server.php index f7e3a90ca..800a632ff 100644 --- a/packages/websocket/tests/Fixtures/Swoole/server.php +++ b/packages/websocket/tests/Fixtures/Swoole/server.php @@ -9,9 +9,8 @@ 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 -$adapter->setSendTimeout(1.0); // Fill the buffer quickly without allocating production-sized backlogs. $adapter->getNative()->ports[0]->set(['socket_buffer_size' => 65536]); From d626679282c4e9e5ad75390c0b41159a04d806f1 Mon Sep 17 00:00:00 2001 From: loks0n <22452787+loks0n@users.noreply.github.com> Date: Wed, 9 Sep 2026 13:05:32 +0100 Subject: [PATCH 3/4] test(websocket): verify transport failure and payload recovery --- .../src/WebSocket/Adapter/Swoole.php | 16 ++-- packages/websocket/tests/E2E/AdapterTest.php | 84 ++++++++++++++----- .../tests/Fixtures/Swoole/server.php | 15 ++-- 3 files changed, 80 insertions(+), 35 deletions(-) diff --git a/packages/websocket/src/WebSocket/Adapter/Swoole.php b/packages/websocket/src/WebSocket/Adapter/Swoole.php index 2ec77adb8..2f286b7a8 100644 --- a/packages/websocket/src/WebSocket/Adapter/Swoole.php +++ b/packages/websocket/src/WebSocket/Adapter/Swoole.php @@ -58,23 +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->isEstablished($connection)) { + foreach ($connections as $sessionId) { + go(function () use ($sessionId, $message, $flags): void { + if ($this->server->isEstablished($sessionId)) { $pushed = $this->server->push( - $connection, + $sessionId, $message, SWOOLE_WEBSOCKET_OPCODE_TEXT, $flags, ); - if (!$pushed && $this->server->exist($connection)) { + // 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($connection, true); + $this->server->close($sessionId, true); } } else { - $this->server->close($connection); + $this->server->close($sessionId); } }); } diff --git a/packages/websocket/tests/E2E/AdapterTest.php b/packages/websocket/tests/E2E/AdapterTest.php index faa746dc9..2b48f5928 100644 --- a/packages/websocket/tests/E2E/AdapterTest.php +++ b/packages/websocket/tests/E2E/AdapterTest.php @@ -10,6 +10,8 @@ use function Swoole\Coroutine\run; +use Swoole\Coroutine\Socket; +use Swoole\WebSocket\Server as NativeServer; use Utopia\WebSocket\Client; final class AdapterTest extends TestCase @@ -52,7 +54,7 @@ public function testSwooleRecoversFromBriefStall(): void $client->connect(); $baseline = $this->getInfo($http); $client->send('flood'); - $this->waitForPendingSends($http, $baseline['coroutines']); + $this->waitForFlood($http, $baseline['floods'] + 1); // Resume reading before the send timeout. Every frame must arrive. for ($i = 0; $i < 300; $i++) { @@ -72,59 +74,103 @@ public function testSwooleRecoversFromBriefStall(): void private function testPendingSends(bool $closePeer): void { run(function () use ($closePeer): void { - $slow = $this->getWebsocket('127.0.0.1', 18081); + $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 { - $slow->connect(); $healthy->connect(); $baseline = $this->getInfo($http); - $slow->send('flood'); - $this->waitForPendingSends($http, $baseline['coroutines']); + $request = NativeServer::pack('flood', WEBSOCKET_OPCODE_TEXT, SWOOLE_WEBSOCKET_FLAG_FIN | SWOOLE_WEBSOCKET_FLAG_MASK); + $this->assertSame(\strlen($request), $slow->sendAll($request)); + $this->waitForFlood($http, $baseline['floods'] + 1); if ($closePeer) { - // Close the TCP socket with unread frames, as in the upstream repro. + // 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['connections'] !== 1 || $info['native_connections'] !== 2 - || $info['coroutines'] !== $baseline['coroutines']) && microtime(true) < $deadline); - - $this->assertSame(1, $info['connections'], 'The stalled client must be disconnected'); - // The native server also counts this HTTP request. Checking it catches - // graceful closes that run onClose but retain the undrained socket. - $this->assertSame(2, $info['native_connections'], 'The stalled socket must be released'); - $this->assertSame($baseline['coroutines'], $info['coroutines'], 'Pending sends must finish'); + } 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(); } }); } - private function waitForPendingSends(HttpClient $http, int $baseline): void + 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 waitForFlood(HttpClient $http, int $expected): void { $deadline = microtime(true) + 2; do { Coroutine::sleep(0.01); $info = $this->getInfo($http); - } while ((!$info['flood_complete'] || $info['coroutines'] <= $baseline + 10) && microtime(true) < $deadline); + } while ($info['floods'] < $expected && microtime(true) < $deadline); - $this->assertTrue($info['flood_complete']); - $this->assertGreaterThan($baseline + 10, $info['coroutines'], 'The test must exercise suspended sends'); + $this->assertSame($expected, $info['floods'], 'The fixture must finish submitting the burst'); } /** - * @return array{connections: int, native_connections: int, coroutines: int, flood_complete: bool} + * @return array{memory_used: int, floods: int} */ private function getInfo(HttpClient $http): array { diff --git a/packages/websocket/tests/Fixtures/Swoole/server.php b/packages/websocket/tests/Fixtures/Swoole/server.php index 800a632ff..3ac10ea53 100644 --- a/packages/websocket/tests/Fixtures/Swoole/server.php +++ b/packages/websocket/tests/Fixtures/Swoole/server.php @@ -4,7 +4,6 @@ require_once __DIR__ . '/../../../vendor/autoload.php'; -use Swoole\Coroutine; use Swoole\Http\Request; use Swoole\Http\Response; use Utopia\WebSocket; @@ -18,7 +17,7 @@ /** @var array $connections */ $connections = []; -$floodComplete = false; +$floods = 0; $server ->onWorkerStart(function (int $workerId): void { @@ -35,16 +34,15 @@ unset($connections[$connection]); echo 'disconnected ', $connection, PHP_EOL; }) - ->onMessage(function (int $connection, string $message) use ($server, &$connections, &$floodComplete): void { + ->onMessage(function (int $connection, string $message) use ($server, &$connections, &$floods): void { echo $message, PHP_EOL; switch ($message) { case 'flood': - $floodComplete = false; for ($i = 0; $i < 300; $i++) { $server->send([$connection], str_pad(sprintf('%06d:', $i), 65536, 'x')); } - $floodComplete = true; + $floods++; break; case 'ping': $server->send([$connection], 'pong'); @@ -61,7 +59,7 @@ break; } }) - ->onRequest(function (Request $request, Response $response) use ($adapter, &$connections, &$floodComplete): void { + ->onRequest(function (Request $request, Response $response) use (&$connections, &$floods): void { echo 'HTTP request received: ', $request->server['request_uri'], PHP_EOL; if ($request->server['request_uri'] === '/health') { @@ -74,9 +72,8 @@ $response->end(json_encode([ 'server' => 'Swoole WebSocket', 'connections' => count($connections), - 'native_connections' => $adapter->getNative()->stats()['connection_num'], - 'coroutines' => Coroutine::stats()['coroutine_num'], - 'flood_complete' => $floodComplete, + 'memory_used' => memory_get_usage(), + 'floods' => $floods, 'timestamp' => time(), ])); } else { From c1ef13d1e8f5c7f6970baecdc931b589dda427f4 Mon Sep 17 00:00:00 2001 From: loks0n <22452787+loks0n@users.noreply.github.com> Date: Wed, 9 Sep 2026 13:15:00 +0100 Subject: [PATCH 4/4] test(websocket): drive recovery with caller-provided messages --- packages/websocket/tests/E2E/AdapterTest.php | 56 ++++++++++++++----- .../tests/Fixtures/Swoole/server.php | 31 ++++++---- 2 files changed, 62 insertions(+), 25 deletions(-) diff --git a/packages/websocket/tests/E2E/AdapterTest.php b/packages/websocket/tests/E2E/AdapterTest.php index 2b48f5928..e30a21557 100644 --- a/packages/websocket/tests/E2E/AdapterTest.php +++ b/packages/websocket/tests/E2E/AdapterTest.php @@ -53,14 +53,13 @@ public function testSwooleRecoversFromBriefStall(): void try { $client->connect(); $baseline = $this->getInfo($http); - $client->send('flood'); - $this->waitForFlood($http, $baseline['floods'] + 1); + $messages = $this->createMessages(); + $this->sendBatch($client, $messages); + $this->waitForBatch($http, $baseline['batches_sent'] + 1); // Resume reading before the send timeout. Every frame must arrive. - for ($i = 0; $i < 300; $i++) { - $frame = $client->receive(); - $this->assertSame(65536, \strlen($frame)); - $this->assertSame(\sprintf('%06d:', $i), substr($frame, 0, 7)); + foreach ($messages as $message) { + $this->assertSame($message, $client->receive()); } $client->send('ping'); $this->assertSame('pong', $client->receive()); @@ -83,9 +82,8 @@ private function testPendingSends(bool $closePeer): void try { $healthy->connect(); $baseline = $this->getInfo($http); - $request = NativeServer::pack('flood', WEBSOCKET_OPCODE_TEXT, SWOOLE_WEBSOCKET_FLAG_FIN | SWOOLE_WEBSOCKET_FLAG_MASK); - $this->assertSame(\strlen($request), $slow->sendAll($request)); - $this->waitForFlood($http, $baseline['floods'] + 1); + $this->sendBatch($slow, $this->createMessages()); + $this->waitForBatch($http, $baseline['batches_sent'] + 1); if ($closePeer) { // Close with unread frames and connect another client before @@ -137,6 +135,38 @@ private function testPendingSends(bool $closePeer): void }); } + /** @return list */ + private function createMessages(): array + { + $messages = []; + for ($i = 0; $i < 300; $i++) { + $messages[] = $i . ':' . str_repeat('x', 65536); + } + + return $messages; + } + + /** @param list $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); @@ -158,19 +188,19 @@ private function connectSlowClient(): Socket return $socket; } - private function waitForFlood(HttpClient $http, int $expected): void + private function waitForBatch(HttpClient $http, int $expected): void { $deadline = microtime(true) + 2; do { Coroutine::sleep(0.01); $info = $this->getInfo($http); - } while ($info['floods'] < $expected && microtime(true) < $deadline); + } while ($info['batches_sent'] < $expected && microtime(true) < $deadline); - $this->assertSame($expected, $info['floods'], 'The fixture must finish submitting the burst'); + $this->assertSame($expected, $info['batches_sent'], 'The fixture must finish submitting the burst'); } /** - * @return array{memory_used: int, floods: int} + * @return array{memory_used: int, batches_sent: int} */ private function getInfo(HttpClient $http): array { diff --git a/packages/websocket/tests/Fixtures/Swoole/server.php b/packages/websocket/tests/Fixtures/Swoole/server.php index 3ac10ea53..99ad3a20a 100644 --- a/packages/websocket/tests/Fixtures/Swoole/server.php +++ b/packages/websocket/tests/Fixtures/Swoole/server.php @@ -17,7 +17,9 @@ /** @var array $connections */ $connections = []; -$floods = 0; +/** @var array> $batches */ +$batches = []; +$batchesSent = 0; $server ->onWorkerStart(function (int $workerId): void { @@ -30,19 +32,24 @@ $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, &$floods): 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) { - case 'flood': - for ($i = 0; $i < 300; $i++) { - $server->send([$connection], str_pad(sprintf('%06d:', $i), 65536, 'x')); + switch ($command) { + case 'buffer': + $batches[$connection][] = $payload; + break; + case 'flush': + $batch = $batches[$connection] ?? []; + unset($batches[$connection]); + foreach ($batch as $payload) { + $server->send([$connection], $payload); } - $floods++; + $batchesSent++; break; case 'ping': $server->send([$connection], 'pong'); @@ -59,7 +66,7 @@ break; } }) - ->onRequest(function (Request $request, Response $response) use (&$connections, &$floods): 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') { @@ -73,7 +80,7 @@ 'server' => 'Swoole WebSocket', 'connections' => count($connections), 'memory_used' => memory_get_usage(), - 'floods' => $floods, + 'batches_sent' => $batchesSent, 'timestamp' => time(), ])); } else {