diff --git a/src/Console/Commands/StartServer.php b/src/Console/Commands/StartServer.php index 8f77099..86f72ca 100644 --- a/src/Console/Commands/StartServer.php +++ b/src/Console/Commands/StartServer.php @@ -88,6 +88,25 @@ class StartServer extends Command */ protected string $cacheStore = 'file'; + /** + * Seconds the soft shutdown waits after closing connections, so Ratchet can + * run the handler's onClose (channel cleanup) before the loop stops. + */ + protected const SOFT_SHUTDOWN_GRACE_SECONDS = 1.0; + + /** + * Hard cap on a soft shutdown. Whatever state the connections are in, the + * loop stops after this long, so a stuck close can never leave the server + * declining every new connection indefinitely. + */ + protected const SOFT_SHUTDOWN_DEADLINE_SECONDS = 5.0; + + /** + * Whether a soft shutdown already stopped the loop (the grace timer, the + * error path and the deadline may all try). + */ + protected bool $softShutdownFinished = false; + /** * Initialize the command. * @@ -998,34 +1017,65 @@ class StartServer extends Command \Log::channel('websocket')->debug('Declining new connections...'); $channelManager->declineNewConnections(); - // Get all local connections and close them. They will - // be automatically be unsubscribed from all channels. + // Whatever happens below, the loop stops within the deadline. + $this->loop->addTimer(static::SOFT_SHUTDOWN_DEADLINE_SECONDS, function () { + $this->finishSoftShutdown('Soft shutdown deadline reached, stopping loop...'); + }); + + // Close all local connections. Ratchet runs the handler's onClose (which + // unsubscribes the connection from its channels) once each stream closes, + // so the handler must not be called here as well. Handler::onClose returns + // void; chaining ->then() on it threw, the error was lost as a promise + // rejection and the loop never stopped. \Log::channel('websocket')->debug('Getting local connections to close...'); $channelManager->getLocalConnections() ->then(function ($connections) { \Log::channel('websocket')->debug('Closing connections...', ['count' => count($connections)]); - return all(collect($connections)->map(function ($connection) { - return app('websockets.handler') - ->onClose($connection) - ->then(function () use ($connection) { - $connection->close(); - }); - })->toArray()); + + foreach ($connections as $connection) { + try { + $connection->close(); + } catch (\Throwable $e) { + \Log::channel('websocket')->warning('Closing a connection failed during soft shutdown', [ + 'error' => $e->getMessage(), + ]); + } + } + + $this->loop->addTimer(static::SOFT_SHUTDOWN_GRACE_SECONDS, function () { + $this->finishSoftShutdown('All connections closed, stopping loop...'); + }); }) - ->then(function () { - \Log::channel('websocket')->debug('All connections closed, stopping loop...'); - - // Stop the broadcast socket server - $this->stopBroadcastSocket(); - - // Reap any already-exited fork children before we exit (see - // triggerHardShutdown). - $this->reapChildren(); - - $this->loop->stop(); + ->then(null, function ($reason) { + \Log::channel('websocket')->warning('Soft shutdown could not list connections', [ + 'error' => $reason instanceof \Throwable ? $reason->getMessage() : (string) $reason, + ]); + $this->finishSoftShutdown('Stopping loop after soft shutdown error...'); }); } + /** + * Stop the loop once at the end of a soft shutdown. + */ + protected function finishSoftShutdown(string $message): void + { + if ($this->softShutdownFinished) { + return; + } + $this->softShutdownFinished = true; + + \Log::channel('websocket')->debug($message); + + // Stop the broadcast socket server + $this->stopBroadcastSocket(); + + // Reap any already-exited fork children before we exit (see + // triggerHardShutdown). + $this->reapChildren(); + + $this->loop->stop(); + } + /** * Stop the broadcast socket server if running. * diff --git a/src/Websocket/ChannelManager.php b/src/Websocket/ChannelManager.php index d09c14e..1164ecb 100644 --- a/src/Websocket/ChannelManager.php +++ b/src/Websocket/ChannelManager.php @@ -96,6 +96,12 @@ class ChannelManager extends LocalChannelManager */ public function unsubscribeFromAllChannels(ConnectionInterface $connection) : PromiseInterface { + // A connection declined in onOpen (e.g. during a soft shutdown) never got + // an app and joined no channels. Same guard as LocalChannelManager. + if (! isset($connection->app)) { + return Helpers::createFulfilledPromise(false); + } + return $this->getGlobalChannels($connection->app->id) ->then(function ($channels) use ($connection) { foreach ($channels as $channel) { diff --git a/tests/ClosedBeforeOpenTest.php b/tests/ClosedBeforeOpenTest.php new file mode 100644 index 0000000..d2f290f --- /dev/null +++ b/tests/ClosedBeforeOpenTest.php @@ -0,0 +1,29 @@ +newInstanceWithoutConstructor(); + + $result = null; + $manager->unsubscribeFromAllChannels(new Mocks\Connection) + ->then(function ($value) use (&$result) { + $result = $value; + }); + + $this->assertFalse($result); + } +} diff --git a/tests/Commands/SoftShutdownTest.php b/tests/Commands/SoftShutdownTest.php new file mode 100644 index 0000000..e961b48 --- /dev/null +++ b/tests/Commands/SoftShutdownTest.php @@ -0,0 +1,104 @@ +then() on Handler::onClose() (which returns void): the error was dropped as + * a promise rejection, the loop never stopped, and the server kept declining + * every new connection until a hard restart. Driven on a fresh loop with fake + * connections so nothing else can stop the loop for it. + */ +class SoftShutdownTest extends TestCase +{ + private function runSoftShutdown(object $channelManager): array + { + $loop = LoopFactory::create(); + $this->app->instance(ChannelManager::class, $channelManager); + + $command = $this->app->make(StartServer::class); + $command->setLaravel($this->app); + $command->setOutput(new OutputStyle(new ArrayInput([]), new BufferedOutput)); + (new \ReflectionProperty(StartServer::class, 'loop'))->setValue($command, $loop); + + $stoppedBySafetyNet = false; + $loop->addTimer(8, function () use (&$stoppedBySafetyNet, $loop) { + $stoppedBySafetyNet = true; + $loop->stop(); + }); + + $trigger = new \ReflectionMethod(StartServer::class, 'triggerSoftShutdown'); + $loop->futureTick(fn () => $trigger->invoke($command)); + + $started = microtime(true); + $loop->run(); + + return [$stoppedBySafetyNet, microtime(true) - $started]; + } + + private function fakeConnection(): object + { + return new class { + public bool $closed = false; + + public function close(): void + { + $this->closed = true; + } + }; + } + + public function test_soft_shutdown_closes_connections_and_stops_the_loop() + { + $connections = [$this->fakeConnection(), $this->fakeConnection()]; + $manager = new class($connections) { + public bool $declined = false; + + public function __construct(private array $connections) {} + + public function declineNewConnections(): void + { + $this->declined = true; + } + + public function getLocalConnections(): PromiseInterface + { + return Helpers::createFulfilledPromise($this->connections); + } + }; + + [$stoppedBySafetyNet, $seconds] = $this->runSoftShutdown($manager); + + $this->assertFalse($stoppedBySafetyNet, 'The soft shutdown never stopped the event loop.'); + $this->assertTrue($manager->declined); + $this->assertTrue($connections[0]->closed && $connections[1]->closed, 'Every connection is closed.'); + // The loop waits the grace period so Ratchet can run onClose for each connection. + $this->assertGreaterThanOrEqual(0.9, $seconds); + } + + public function test_soft_shutdown_stops_the_loop_even_when_listing_connections_fails() + { + $manager = new class { + public function declineNewConnections(): void {} + + public function getLocalConnections(): PromiseInterface + { + return \React\Promise\reject(new \RuntimeException('redis unavailable')); + } + }; + + [$stoppedBySafetyNet] = $this->runSoftShutdown($manager); + + $this->assertFalse($stoppedBySafetyNet, 'A failed connection listing left the loop running.'); + } +} diff --git a/tests/Commands/StartServerTest.php b/tests/Commands/StartServerTest.php index 4e9aef8..fc2966a 100644 --- a/tests/Commands/StartServerTest.php +++ b/tests/Commands/StartServerTest.php @@ -48,4 +48,47 @@ class StartServerTest extends TestCase $this->assertTrue(true); } + + public function test_soft_shutdown_stops_the_loop_on_its_own() + { + // The soft shutdown chained ->then() on Handler::onClose(), which returns + // void. The resulting error became a silently dropped promise rejection, + // loop->stop() never ran, and the server kept accepting and closing every + // new socket until a hard restart. Unlike the tests above, nothing here + // stops the loop for the server. + $stoppedBySafetyNet = false; + $errors = []; + \Illuminate\Support\Facades\Event::listen(\Illuminate\Log\Events\MessageLogged::class, function ($event) use (&$errors) { + if (in_array($event->level, ['error', 'critical', 'alert', 'emergency'], true)) { + $errors[] = $event->message . ' ' . (($event->context['exception'] ?? null)?->getMessage() ?? ''); + } + }); + + $this->loop->futureTick(function () { + // The server rebinds the channel manager on boot, so connect through + // its own handler; connections made via $this->wsHandler would live + // in the test's manager and the shutdown would see none. + $handler = app('websockets.handler'); + for ($i = 0; $i < 2; $i++) { + $connection = $this->newConnection(); + $handler->onOpen($connection); + $handler->onMessage($connection, new \BlaxSoftware\LaravelWebSockets\Test\Mocks\Message([ + 'event' => 'websocket.subscribe', + 'data' => ['channel' => 'public-channel'], + ])); + } + + posix_kill(posix_getpid(), SIGTERM); + }); + + $this->loop->addTimer(8, function () use (&$stoppedBySafetyNet) { + $stoppedBySafetyNet = true; + $this->loop->stop(); + }); + + $this->artisan('websockets:serve', ['--loop' => $this->loop, '--debug' => true, '--port' => 6004, '--soft' => true]); + + $this->assertSame([], $errors, 'The soft shutdown logged errors.'); + $this->assertFalse($stoppedBySafetyNet, 'The soft shutdown never stopped the event loop.'); + } }