From 35ed3202024eec02b7930911410e50041fdb5b53 Mon Sep 17 00:00:00 2001 From: "Fabian @ Blax Software" Date: Wed, 14 Oct 2026 10:38:11 +0200 Subject: [PATCH] fix(fork): children never outlive the server or hold its port (#2465) A fork-per-message child inherits the server's listening socket. A child that outlived the server (a long-running handler such as an engine loop) kept the port bound, so the restarted server failed with EADDRINUSE until that child died. - Track every forked child (Server\ChildProcesses). On hard and soft shutdown, send SIGTERM to the ones still running, SIGKILL them after websockets.shutdown.child_grace_seconds (default 2) and reap them. A PID is signalled only while waitpid() confirms it is still an unreaped child, with SIGCHLD blocked, so a recycled PID is never hit. Opt out with websockets.shutdown.terminate_children = false. - Right after fork the child closes its copy of the listening socket, so even a stray child cannot hold the port. Only the fd is closed: the React server is not closed, because removing it from an ext-ev/ext-event loop would edit the epoll set the child shares with the parent. - The child restores default SIGTERM/SIGINT/SIGCHLD handling. Under ext-ev the inherited libev handlers swallowed SIGTERM in the child. - A child that dies of a fatal error (out of memory, E_ERROR) now sends :error from a shutdown function (Server\ChildFatalErrorGuard, #1462). A child that dies without any reply (killed by a signal, or the shutdown function never ran) gets :error from the parent, decided once both its stream and its exit status are in (Websocket\ChildMessageOutcome). - Tests: real forks, signals and sockets for the shutdown, the port release and the fatal reply; end to end through Handler::forkWithSocketPair(). The test TestCase now stops a leftover broadcast socket, which made ws_available() true for later tests. Co-Authored-By: Claude Opus 5.5 (1M context) --- AGENTS.md | 14 +- config/websockets.php | 20 + src/Console/Commands/StartServer.php | 52 +- src/Server/ChildFatalErrorGuard.php | 116 +++++ src/Server/ChildProcesses.php | 466 ++++++++++++++++++ src/Websocket/ChildMessageOutcome.php | 107 ++++ src/Websocket/Handler.php | 128 ++++- tests/Commands/ShutdownStopsChildrenTest.php | 138 ++++++ tests/Server/ChildFatalErrorGuardTest.php | 81 +++ tests/Server/ChildProcessesTest.php | 335 +++++++++++++ tests/Server/Fixtures/fatal_child.php | 97 ++++ tests/TestCase.php | 7 + tests/Websocket/ChildMessageOutcomeTest.php | 115 +++++ .../ChildCrash/ChildCrashController.php | 43 ++ .../DeferredCallbackCollectionStub.php | 18 + tests/Websocket/HandlerChildLifecycleTest.php | 204 ++++++++ 16 files changed, 1927 insertions(+), 14 deletions(-) create mode 100644 src/Server/ChildFatalErrorGuard.php create mode 100644 src/Server/ChildProcesses.php create mode 100644 src/Websocket/ChildMessageOutcome.php create mode 100644 tests/Commands/ShutdownStopsChildrenTest.php create mode 100644 tests/Server/ChildFatalErrorGuardTest.php create mode 100644 tests/Server/ChildProcessesTest.php create mode 100644 tests/Server/Fixtures/fatal_child.php create mode 100644 tests/Websocket/ChildMessageOutcomeTest.php create mode 100644 tests/Websocket/Fixtures/ChildCrash/ChildCrashController.php create mode 100644 tests/Websocket/Fixtures/DeferredCallbackCollectionStub.php create mode 100644 tests/Websocket/HandlerChildLifecycleTest.php diff --git a/AGENTS.md b/AGENTS.md index e5e6bb2..e6967f0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -30,6 +30,17 @@ Consequences — internalize these, they change how you write handlers: cache. Each child gets a *fresh* DB connection and purged Redis/cache singletons. - **Concurrency is capped** by `config('websockets.max_concurrent_children')` (protects MySQL `max_connections`); excess messages queue in memory and run as children free up. +- **A child never outlives the server.** Right after fork the child closes its copy of the + listening socket and gets default signal handling back; on shutdown the server SIGTERMs + running children, SIGKILLs them after `websockets.shutdown.child_grace_seconds` (default 2) + and reaps them (`src/Server/ChildProcesses.php`). A handler that blocks for minutes (an + engine loop) is killed on every restart, so keep its state in Redis/DB and let a later + message take over. Opt out with `websockets.shutdown.terminate_children = false`. +- **A child that dies still answers.** A fatal error (out of memory, E_ERROR) sends + `:error` with `meta.fatal = true` from a shutdown function + (`src/Server/ChildFatalErrorGuard.php`); a child that dies with no reply at all (killed by a + signal, or the shutdown function never ran) gets `:error` from the parent + (`src/Websocket/ChildMessageOutcome.php`). --- @@ -324,7 +335,8 @@ migrated to real queued **Jobs** (which also get retry semantics + show up in to | Out-of-band broadcast/whisper + tracking | `src/Services/WebsocketService.php`, `src/Broadcast/BroadcastClient.php` | | Per-connection store | `src/Websocket/ConnectionSession.php` | | Canonical handler example | `src/Websocket/Controllers/ExampleController.php` | -| Config (introspection, hot_reload, max_concurrent_children, broadcast_socket) | `config/websockets.php` | +| Config (introspection, hot_reload, max_concurrent_children, shutdown, broadcast_socket) | `config/websockets.php` | +| Child lifecycle: shutdown kill, listener close, fatal-error reply | `src/Server/ChildProcesses.php`, `src/Server/ChildFatalErrorGuard.php`, `src/Websocket/ChildMessageOutcome.php` | **Frontend counterpart:** the client side of this protocol (how `:progress`/`:response`/`:error` are consumed) is documented in `@blax-software/networking`'s `AGENTS.md`. diff --git a/config/websockets.php b/config/websockets.php index fa7ba0b..106ac72 100644 --- a/config/websockets.php +++ b/config/websockets.php @@ -99,6 +99,26 @@ return [ */ 'max_concurrent_children' => (int) env('WEBSOCKET_MAX_CHILDREN', 30), + /* + |-------------------------------------------------------------------------- + | Shutdown: running fork children + |-------------------------------------------------------------------------- + | + | When the server stops (SIGTERM, websockets:restart, websocket:steer + | restart), message children that are still running get SIGTERM, then + | SIGKILL once the grace period is over, and are reaped. A child that + | outlived the server used to keep running as an orphan (and, before + | children closed their copy of the listening socket, kept the port bound). + | + | Long-running handlers that must clean up can install their own SIGTERM + | handler inside the child; everything else simply dies on SIGTERM. + | + */ + 'shutdown' => [ + 'terminate_children' => env('WEBSOCKET_SHUTDOWN_TERMINATE_CHILDREN', true), + 'child_grace_seconds' => (float) env('WEBSOCKET_SHUTDOWN_CHILD_GRACE', 2.0), + ], + /* |-------------------------------------------------------------------------- | Broadcast Socket Settings diff --git a/src/Console/Commands/StartServer.php b/src/Console/Commands/StartServer.php index 86f72ca..c9498a0 100644 --- a/src/Console/Commands/StartServer.php +++ b/src/Console/Commands/StartServer.php @@ -11,6 +11,7 @@ use BlaxSoftware\LaravelWebSockets\Ipc\SocketPairIpc; use BlaxSoftware\LaravelWebSockets\Server\Loggers\ConnectionLogger; use BlaxSoftware\LaravelWebSockets\Server\Loggers\HttpLogger; use BlaxSoftware\LaravelWebSockets\Server\Loggers\WebSocketsLogger; +use BlaxSoftware\LaravelWebSockets\Server\ChildProcesses; use BlaxSoftware\LaravelWebSockets\ServerFactory; use BlaxSoftware\LaravelWebSockets\Services\WebsocketService; use Illuminate\Console\Command; @@ -520,12 +521,40 @@ class StartServer extends Command return 0; } - $reaped = 0; - while (($pid = pcntl_waitpid(-1, $status, WNOHANG)) > 0) { - $reaped++; + // Also drops each reaped child from the list terminateChildren() stops. + return ChildProcesses::reapExited(); + } + + /** + * Stop the fork-per-message children that are still running. + * + * A child outlives the server otherwise: the ATC engine in learn-atc blocks + * its child for up to two hours, kept the listening port bound and made the + * restarted server fail with EADDRINUSE (#2465). Running children get + * SIGTERM, then SIGKILL after the grace period, and are reaped. Blocks the + * loop for at most the grace period plus one second; only called while + * shutting down. + */ + protected function terminateChildren(): void + { + if (! config('websockets.shutdown.terminate_children', true)) { + return; } - return $reaped; + try { + $grace = max(0.0, (float) config('websockets.shutdown.child_grace_seconds', 2.0)); + $running = ChildProcesses::count(); + $stats = ChildProcesses::terminateAll($grace); + + if ($stats['terminated'] > 0 || $stats['killed'] > 0 || $stats['remaining'] > 0) { + \Log::channel('websocket')->info('Stopped fork children on shutdown', $stats + [ + 'tracked' => $running, + 'grace_seconds' => $grace, + ]); + } + } catch (\Throwable $e) { + \Log::channel('websocket')->warning('Stopping fork children on shutdown failed: ' . $e->getMessage()); + } } /** @@ -907,6 +936,12 @@ class StartServer extends Command ->setConsoleOutput($this->output) ->createServer(); + // Forked message children close their inherited copy of this + // socket, so none of them can keep the port bound (#2465). + if (! ChildProcesses::registerListener($this->server->socket)) { + \Log::channel('websocket')->warning('Could not find the listening socket; forked children will keep it open.'); + } + break; } catch (\RuntimeException $e) { $isAddrInUse = str_contains($e->getMessage(), 'EADDRINUSE') @@ -992,10 +1027,10 @@ class StartServer extends Command // Stop the broadcast socket server $this->stopBroadcastSocket(); - // Reap any already-exited fork children before we exit so they are not - // handed to PID 1 as orphans (still-running children are unavoidably - // reparented and reaped by the container init/supervisord). + // Reap the children that already exited, then stop and reap the ones + // still running, so none is handed to PID 1 as an orphan. $this->reapChildren(); + $this->terminateChildren(); $this->loop->stop(); } @@ -1069,9 +1104,10 @@ class StartServer extends Command // Stop the broadcast socket server $this->stopBroadcastSocket(); - // Reap any already-exited fork children before we exit (see + // Reap and stop the fork children before we exit (see // triggerHardShutdown). $this->reapChildren(); + $this->terminateChildren(); $this->loop->stop(); } diff --git a/src/Server/ChildFatalErrorGuard.php b/src/Server/ChildFatalErrorGuard.php new file mode 100644 index 0000000..ae0c0bd --- /dev/null +++ b/src/Server/ChildFatalErrorGuard.php @@ -0,0 +1,116 @@ +onFatal = \Closure::fromCallable($onFatal); + $this->reserve = $reserveBytes > 0 ? str_repeat("\0", $reserveBytes) : null; + } + + /** + * Register the guard in the current (child) process. It starts armed. + * + * @param callable(array{type: int, message: string, file: string, line: int}): void $onFatal + */ + public static function register(callable $onFatal, int $reserveBytes = 32768): self + { + $guard = new self($onFatal, $reserveBytes); + + register_shutdown_function([$guard, 'handleShutdown']); + + return $guard; + } + + /** + * Stop reacting to fatal errors, for example once the handler has replied + * or a catch block has already sent its own error frame. + */ + public function disarm(): void + { + $this->armed = false; + $this->reserve = null; + } + + public function isArmed(): bool + { + return $this->armed; + } + + /** + * @internal called by PHP at shutdown + */ + public function handleShutdown(): void + { + if (! $this->armed) { + return; + } + + $error = error_get_last(); + + if ($error === null || ! in_array($error['type'], self::FATAL_ERROR_TYPES, true)) { + return; + } + + $this->armed = false; + $this->reserve = null; + + if (str_starts_with((string) $error['message'], 'Allowed memory size')) { + $this->raiseMemoryLimit(); + } + + try { + ($this->onFatal)($error); + } catch (\Throwable) { + // The child is dying; there is nobody left to tell. + } + } + + private function raiseMemoryLimit(): void + { + $limit = (string) ini_get('memory_limit'); + + if ($limit === '' || $limit === '-1') { + return; + } + + @ini_set('memory_limit', (string) (memory_get_usage(true) + self::OOM_HEADROOM_BYTES)); + } +} diff --git a/src/Server/ChildProcesses.php b/src/Server/ChildProcesses.php new file mode 100644 index 0000000..d71e91c --- /dev/null +++ b/src/Server/ChildProcesses.php @@ -0,0 +1,466 @@ + + */ + private static array $children = []; + + /** + * Listening sockets a forked child must close. + * + * @var resource[] + */ + private static array $listeners = []; + + /** + * Set once terminateAll() starts, so exit callbacks can tell a child + * stopped by the shutdown from one that crashed. + */ + private static bool $terminating = false; + + /** + * Signals the server process installs handlers for. A forked child gets + * the default dispositions back (see restoreDefaultSignals()). + */ + private const SERVER_SIGNALS = ['SIGTERM', 'SIGINT', 'SIGCHLD']; + + /** + * Record a freshly forked child. + * + * $onExit is called once with the raw wait status when the child is + * reaped, or with null when it was reaped somewhere this registry could + * not see (its status is then unknown). + */ + public static function track(int $pid, ?string $event = null, ?callable $onExit = null): void + { + if ($pid <= 0) { + return; + } + + self::$children[$pid] = [ + 'event' => $event, + 'started_at' => microtime(true), + 'on_exit' => $onExit, + ]; + } + + /** + * PIDs of the tracked children. + * + * @return int[] + */ + public static function pids(): array + { + return array_keys(self::$children); + } + + public static function count(): int + { + return count(self::$children); + } + + /** + * Reap every child that has already exited, without blocking. + * + * Safe from a SIGCHLD handler, a timer or the shutdown path: WNOHANG only + * collects children that are gone. Untracked children are reaped too. + * + * @return int number of children reaped + */ + public static function reapExited(): int + { + if (! function_exists('pcntl_waitpid')) { + return 0; + } + + $reaped = 0; + while (($pid = pcntl_waitpid(-1, $status, WNOHANG)) > 0) { + $reaped++; + self::finish($pid, $status); + } + + return $reaped; + } + + /** + * Stop every tracked child that is still running. + * + * SIGTERM first, then poll for up to $graceSeconds, then SIGKILL whatever + * is left and wait up to $killWaitSeconds for the kernel to finish it off. + * Every child that exits is reaped. Blocks for at most + * $graceSeconds + $killWaitSeconds, so call it only on shutdown. + * + * A PID is signalled only while waitpid() confirms it is still an unreaped + * child of this process, and SIGCHLD is blocked meanwhile so no handler can + * reap it between that check and the kill. A recycled PID that belongs to + * another process is therefore never signalled. + * + * @return array{terminated: int, killed: int, remaining: int} children sent + * SIGTERM, children sent SIGKILL, children still alive at the end + */ + public static function terminateAll(float $graceSeconds = 2.0, float $killWaitSeconds = 1.0): array + { + $stats = ['terminated' => 0, 'killed' => 0, 'remaining' => 0]; + + if (! function_exists('pcntl_waitpid')) { + return $stats; + } + + self::$terminating = true; + + $blocked = self::blockSigchld(); + + try { + self::reapExited(); + + $running = self::stillRunning(self::pids()); + if ($running === []) { + return $stats; + } + + foreach ($running as $pid) { + if (self::signal($pid, SIGTERM)) { + $stats['terminated']++; + } + } + + $running = self::waitForExit($running, $graceSeconds); + + if ($running !== []) { + foreach ($running as $pid) { + if (self::signal($pid, SIGKILL)) { + $stats['killed']++; + } + } + + $running = self::waitForExit($running, $killWaitSeconds); + } + + $stats['remaining'] = count($running); + + return $stats; + } finally { + self::restoreSigmask($blocked); + } + } + + /** + * Register a listening server whose socket forked children must close. + * + * Accepts a stream resource or a ReactPHP server object (Server, + * SocketServer, SecureServer, TcpServer, UnixServer); for objects the + * underlying stream is found through their private properties. + * + * @param mixed $server + * @return bool whether a socket was found and registered + */ + public static function registerListener($server): bool + { + $stream = self::findStream($server); + + if ($stream === null) { + return false; + } + + self::$listeners[(int) $stream] = $stream; + + return true; + } + + public static function listenerCount(): int + { + return count(self::$listeners); + } + + /** + * Whether terminateAll() has started in this process. + */ + public static function isTerminating(): bool + { + return self::$terminating; + } + + /** + * Run in a freshly forked child, before any handler code. + * + * Closes the child's copy of every registered listening socket, forgets + * the parent's child bookkeeping (siblings are not this child's children) + * and gives the server's signals their default dispositions back. + * + * @return int number of listening sockets closed + */ + public static function prepareChild(?LoopInterface $loop = null): int + { + $closed = self::closeInheritedListeners($loop); + + self::$children = []; + self::$terminating = false; + + self::restoreDefaultSignals(); + + return $closed; + } + + /** + * Close this process's copy of every registered listening socket. + * + * After fork() the child holds its own descriptor for the same kernel + * socket. fclose() releases only that descriptor: the parent keeps + * listening, and once the parent exits the port is free even if this + * child is still running. + * + * The ReactPHP server object is deliberately NOT closed: close() pauses it, + * which removes the stream from the event loop. The ext-ev / ext-event / + * ext-uv loops keep their watch list in a kernel epoll set that a forked + * child shares with the parent, so removing the socket there would stop + * the PARENT from accepting connections. StreamSelectLoop keeps its watch + * list in plain PHP arrays, so the stream is dropped from it first; a + * closed stream left in it would break stream_select() should the child + * ever run that loop. + * + * @return int number of sockets closed + */ + public static function closeInheritedListeners(?LoopInterface $loop = null): int + { + $closed = 0; + + foreach (self::$listeners as $stream) { + if (! is_resource($stream)) { + continue; + } + + if ($loop instanceof StreamSelectLoop) { + $loop->removeReadStream($stream); + } + + if (@fclose($stream)) { + $closed++; + } + } + + self::$listeners = []; + + return $closed; + } + + /** + * Give the server's signals their default dispositions back. + * + * The child inherits the server's handlers. Under ext-ev they are libev C + * handlers that only flag the signal for a loop the child never runs, so a + * SIGTERM sent to the child did nothing. Under StreamSelectLoop with async + * signals they ran the parent's shutdown closure inside the child. + */ + public static function restoreDefaultSignals(): void + { + if (! function_exists('pcntl_signal')) { + return; + } + + $signals = []; + foreach (self::SERVER_SIGNALS as $name) { + if (defined($name)) { + $signals[] = constant($name); + } + } + + foreach ($signals as $signal) { + @pcntl_signal($signal, SIG_DFL); + } + + if (function_exists('pcntl_sigprocmask') && $signals !== []) { + @pcntl_sigprocmask(SIG_UNBLOCK, $signals); + } + } + + /** + * Whether a raw wait status describes an abnormal exit: killed by a + * signal, or exited with a non-zero code (PHP exits with 255 after a + * fatal error). + */ + public static function exitedAbnormally(int $status): bool + { + if (pcntl_wifsignaled($status)) { + return true; + } + + return pcntl_wifexited($status) && pcntl_wexitstatus($status) !== 0; + } + + /** + * Forget all state. For tests. + */ + public static function flush(): void + { + self::$children = []; + self::$listeners = []; + self::$terminating = false; + } + + /** + * @param int[] $pids + * @return int[] the PIDs that are still running + */ + private static function stillRunning(array $pids): array + { + $running = []; + + foreach ($pids as $pid) { + $result = pcntl_waitpid($pid, $status, WNOHANG); + + if ($result === 0) { + $running[] = $pid; + } elseif ($result === $pid) { + self::finish($pid, $status); + } else { + // -1: not (any longer) a child of this process, for example + // reaped by code that bypassed this registry. + self::finish($pid, null); + } + } + + return $running; + } + + /** + * @param int[] $pids + * @return int[] the PIDs still running when the time ran out + */ + private static function waitForExit(array $pids, float $seconds): array + { + $deadline = microtime(true) + max(0.0, $seconds); + + while (true) { + $pids = self::stillRunning($pids); + + if ($pids === [] || microtime(true) >= $deadline) { + return $pids; + } + + usleep(20_000); + } + } + + private static function finish(int $pid, ?int $status): void + { + $entry = self::$children[$pid] ?? null; + unset(self::$children[$pid]); + + if ($entry === null || $entry['on_exit'] === null) { + return; + } + + try { + ($entry['on_exit'])($status); + } catch (\Throwable) { + // An exit callback must never break reaping or the shutdown. + } + } + + private static function signal(int $pid, int $signal): bool + { + if (function_exists('posix_kill')) { + return @posix_kill($pid, $signal); + } + + return false; + } + + /** + * @return array|null the previous mask, or null when nothing was blocked + */ + private static function blockSigchld(): ?array + { + if (! function_exists('pcntl_sigprocmask') || ! defined('SIGCHLD')) { + return null; + } + + $previous = []; + + return @pcntl_sigprocmask(SIG_BLOCK, [SIGCHLD], $previous) ? $previous : null; + } + + private static function restoreSigmask(?array $previous): void + { + if ($previous !== null) { + @pcntl_sigprocmask(SIG_SETMASK, $previous); + } + } + + /** + * Find the stream resource behind a ReactPHP server object. + * + * @param mixed $server + * @return resource|null + */ + private static function findStream($server, int $depth = 0) + { + if (is_resource($server)) { + return get_resource_type($server) === 'stream' ? $server : null; + } + + if (! is_object($server) || $depth > 4) { + return null; + } + + // TcpServer/UnixServer: $master. Server/SocketServer: $server. + // SecureServer: $tcp. Ratchet's IoServer: $socket. + foreach (['master', 'server', 'tcp', 'socket'] as $name) { + $class = new \ReflectionObject($server); + + while ($class !== false && ! $class->hasProperty($name)) { + $class = $class->getParentClass(); + } + + if ($class === false) { + continue; + } + + $property = $class->getProperty($name); + if (PHP_VERSION_ID < 80100) { + $property->setAccessible(true); + } + + if (! $property->isInitialized($server)) { + continue; + } + + $found = self::findStream($property->getValue($server), $depth + 1); + + if ($found !== null) { + return $found; + } + } + + return null; + } +} diff --git a/src/Websocket/ChildMessageOutcome.php b/src/Websocket/ChildMessageOutcome.php new file mode 100644 index 0000000..41d022e --- /dev/null +++ b/src/Websocket/ChildMessageOutcome.php @@ -0,0 +1,107 @@ +:response` or + * `:error` frame, the client would wait forever. The crash callback + * then sends the error frame in the child's place. + * + * The verdict waits for BOTH the end of the IPC stream and the reaped exit + * status. A SIGCHLD reaper can collect the child before the loop has read the + * child's last frames, so deciding at reap time alone would send an `:error` + * ahead of a `:response` that is still in the socket buffer. + */ +final class ChildMessageOutcome +{ + /** @var string[] */ + private array $finalFramePrefixes; + + private bool $replied = false; + + private bool $streamClosed = false; + + private bool $exited = false; + + private ?int $status = null; + + private bool $settled = false; + + /** @var \Closure(int): void */ + private \Closure $onCrash; + + /** + * @param callable(int): void $onCrash receives the raw wait status + */ + public function __construct(string $event, callable $onCrash) + { + $this->finalFramePrefixes = [ + '{"event":' . json_encode($event . ':response'), + '{"event":' . json_encode($event . ':error'), + ]; + $this->onCrash = \Closure::fromCallable($onCrash); + } + + /** + * Inspect a frame the child sent. Final frames for this message are + * encoded by the controller with "event" as the first key. + */ + public function frameReceived(mixed $data): void + { + if ($this->replied || ! is_string($data)) { + return; + } + + foreach ($this->finalFramePrefixes as $prefix) { + if (str_starts_with($data, $prefix)) { + $this->replied = true; + + return; + } + } + } + + public function streamClosed(): void + { + $this->streamClosed = true; + $this->settle(); + } + + /** + * @param int|null $status raw wait status, or null when unknown + */ + public function exited(?int $status): void + { + $this->exited = true; + $this->status = $status; + $this->settle(); + } + + public function replied(): bool + { + return $this->replied; + } + + private function settle(): void + { + if ($this->settled || ! $this->streamClosed || ! $this->exited) { + return; + } + + $this->settled = true; + + if ($this->replied || $this->status === null || ! ChildProcesses::exitedAbnormally($this->status)) { + return; + } + + ($this->onCrash)($this->status); + } +} diff --git a/src/Websocket/Handler.php b/src/Websocket/Handler.php index aea1775..a69d807 100644 --- a/src/Websocket/Handler.php +++ b/src/Websocket/Handler.php @@ -12,6 +12,8 @@ use BlaxSoftware\LaravelWebSockets\Contracts\ChannelManager; use BlaxSoftware\LaravelWebSockets\Events\ConnectionClosed; use BlaxSoftware\LaravelWebSockets\Events\NewConnection; use BlaxSoftware\LaravelWebSockets\Ipc\SocketPairIpc; +use BlaxSoftware\LaravelWebSockets\Server\ChildFatalErrorGuard; +use BlaxSoftware\LaravelWebSockets\Server\ChildProcesses; use BlaxSoftware\LaravelWebSockets\Websocket\MockConnectionSocketPair; use BlaxSoftware\LaravelWebSockets\Server\Exceptions\ConnectionsOverCapacity; use BlaxSoftware\LaravelWebSockets\Server\Exceptions\OriginNotAllowed; @@ -581,6 +583,91 @@ class Handler implements MessageComponentInterface app()->forgetInstance('session.store'); } + /** + * Arm a shutdown-time reply in this message child for a fatal error (#1462). + * + * Fatal errors (out of memory, E_ERROR, ...) skip the catch block in + * forkWithSocketPair(), so the child used to exit without any frame and the + * client waited forever. The frame is encoded now, while memory is still + * available; at shutdown the guard only writes it. Same shape as the + * controller's catch-all error frame (Controller::send_error()). + */ + protected function guardChildAgainstFatalErrors(SocketPairIpc $ipc, Channel $channel, array $message): ChildFatalErrorGuard + { + $frame = static::fatalErrorFrame($message, $channel->getName()); + + return ChildFatalErrorGuard::register(static function () use ($ipc, $frame): void { + $ipc->sendToParent($frame); + }); + } + + /** + * The `:error` frame a message child sends when it dies of a fatal error. + */ + protected static function fatalErrorFrame(array $message, ?string $channelName): string + { + return json_encode([ + 'event' => ($message['event'] ?? 'unknown') . ':error', + 'data' => [ + 'message' => 'Internal server error', + 'meta' => [ + // The application's own fatal handler (Laravel's + // HandleExceptions) runs before the guard and reports it. + 'reported' => true, + 'fatal' => true, + ], + ], + 'channel' => $message['channel'] ?? $channelName, + ]); + } + + /** + * Answer the client for a message child that died without replying: killed + * by a signal (kernel OOM killer, server shutdown) or a fatal error its own + * guard could not report (#1462). Runs in the event loop: must not throw. + */ + protected function replyForCrashedChild( + ConnectionInterface $connection, + Channel $channel, + array $message, + int $pid, + int $status + ): void { + $event = $message['event'] ?? 'unknown'; + $stoppedByShutdown = ChildProcesses::isTerminating(); + + try { + $exit = pcntl_wifsignaled($status) + ? 'signal ' . pcntl_wtermsig($status) + : 'exit code ' . pcntl_wexitstatus($status); + + $context = ['event' => $event, 'pid' => $pid, 'exit' => $exit]; + + if ($stoppedByShutdown) { + Log::channel('websocket')->info('Message child stopped by the server shutdown before replying', $context); + } else { + Log::channel('websocket')->error('Message child died without replying', $context); + } + } catch (\Throwable) { + // Logging must never break the event loop. + } + + try { + $connection->send(json_encode([ + 'event' => $event . ':error', + 'data' => [ + 'message' => $stoppedByShutdown ? 'Server is restarting' : 'Internal server error', + 'meta' => [ + 'reported' => false, + ], + ], + 'channel' => $message['channel'] ?? $channel->getName(), + ])); + } catch (\Throwable) { + // The connection may already be gone. + } + } + /** * Fork with event-driven socket pair IPC (no polling!) * Parent is notified INSTANTLY when child sends data @@ -627,6 +714,16 @@ class Handler implements MessageComponentInterface // === CHILD PROCESS === $ipc->setupChild(); + // Close this child's copy of the server's listening socket (a child + // that outlived the server kept the port bound, #2465), forget the + // parent's child bookkeeping and restore default signal handling, + // so the parent's shutdown can SIGTERM this child. + ChildProcesses::prepareChild($this->channelManager->loop); + + // A fatal error skips every catch block below; answer the client + // from a shutdown function instead (#1462). + $fatalGuard = $this->guardChildAgainstFatalErrors($ipc, $channel, $message); + // Hot reload: clear all caches for fresh code loading (only in dev mode) $this->hotReloadChild(); @@ -691,6 +788,11 @@ class Handler implements MessageComponentInterface $this->channelManager ); + // The controller has answered (or chose not to). A fatal error + // in the deferred callbacks or the session save below must not + // send a second, contradicting frame. + $fatalGuard->disarm(); + \Illuminate\Container\Container::getInstance() ->make(\Illuminate\Support\Defer\DeferredCallbackCollection::class) ->invokeWhen(fn($callback) => true); @@ -698,6 +800,9 @@ class Handler implements MessageComponentInterface // Persist session changes to Redis before exit $session->save(); } catch (\Throwable $e) { + // This catch sends its own error frame. + $fatalGuard->disarm(); + // Send error via socket pair $ipc->sendToParent(json_encode([ 'event' => $message['event'] . ':error', @@ -755,6 +860,16 @@ class Handler implements MessageComponentInterface } // === PARENT PROCESS === + // Track the child so the shutdown can stop it (#2465), and answer the + // client in its place should it die without replying (#1462). + $outcome = new ChildMessageOutcome( + (string) ($message['event'] ?? ''), + function (int $status) use ($connection, $channel, $message, $pid) { + $this->replyForCrashedChild($connection, $channel, $message, $pid, $status); + } + ); + ChildProcesses::track($pid, $message['event'] ?? null, [$outcome, 'exited']); + // Setup event-driven reading - NO POLLING! $startTime = microtime(true); @@ -764,8 +879,9 @@ class Handler implements MessageComponentInterface // uncaught throwable here would propagate up through ExtEvLoop and // crash the entire WebSocket server (supervisor would then restart // it, dropping every connected client). We must catch and log. - function ($data) use ($connection, $message, $startTime) { + function ($data) use ($connection, $message, $startTime, $outcome) { try { + $outcome->frameReceived($data); $this->handleChildData($connection, $message, $data); } catch (\Throwable $e) { Log::channel('websocket')->error('handleChildData failed: ' . $e->getMessage(), [ @@ -802,17 +918,19 @@ class Handler implements MessageComponentInterface }, // onClose callback - child process ended. // Same isolation rules apply: must not throw out of the loop. - function () { + function () use ($outcome) { try { + // Every frame the child sent has been handled by now. + $outcome->streamClosed(); + // Reap ALL exited fork children, not just one. onClose fires on // socket EOF, which can precede the child's actual process exit — // so this call often reaps the *previous* message's child (or // nothing yet) rather than this one. Draining in a loop keeps // zombies from accumulating between messages; StartServer's // SIGCHLD + periodic reaper is the idle/straggler backstop. - while (pcntl_waitpid(-1, $status, WNOHANG) > 0) { - // reaped one child - } + // Each reap also clears the child from the shutdown list. + ChildProcesses::reapExited(); // Free up a child slot and process any queued messages. // NOTE: this decrement is per-child-SOCKET (one onClose per diff --git a/tests/Commands/ShutdownStopsChildrenTest.php b/tests/Commands/ShutdownStopsChildrenTest.php new file mode 100644 index 0000000..263c5ae --- /dev/null +++ b/tests/Commands/ShutdownStopsChildrenTest.php @@ -0,0 +1,138 @@ +markTestSkipped('pcntl and posix are required'); + } + + ChildProcesses::flush(); + } + + public function tearDown(): void + { + foreach ($this->spawned as $pid) { + @posix_kill($pid, SIGKILL); + @pcntl_waitpid($pid, $status); + } + ChildProcesses::flush(); + + parent::tearDown(); + } + + public function test_hard_shutdown_stops_running_children() + { + $pid = $this->spawnTrackedChild(); + + $command = $this->command(LoopFactory::create()); + (new \ReflectionMethod(StartServer::class, 'triggerHardShutdown'))->invoke($command); + + $this->assertFalse(@posix_kill($pid, 0), 'The child must not outlive the server.'); + $this->assertSame(0, ChildProcesses::count()); + } + + public function test_soft_shutdown_stops_running_children() + { + $pid = $this->spawnTrackedChild(); + + $loop = LoopFactory::create(); + $this->app->instance(ChannelManager::class, new class { + public function declineNewConnections(): void {} + + public function getLocalConnections(): PromiseInterface + { + return Helpers::createFulfilledPromise([]); + } + }); + $command = $this->command($loop); + + $loop->addTimer(8, fn () => $loop->stop()); + $loop->futureTick(fn () => (new \ReflectionMethod(StartServer::class, 'triggerSoftShutdown'))->invoke($command)); + $loop->run(); + + $this->assertFalse(@posix_kill($pid, 0), 'The child must not outlive the server.'); + $this->assertSame(0, ChildProcesses::count()); + } + + public function test_the_server_registers_its_listening_socket_for_children_to_close() + { + // No broadcast socket file left behind in /tmp by this test. + config(['websockets.broadcast_socket_enabled' => false]); + + $registered = null; + $this->loop->futureTick(function () use (&$registered) { + $registered = ChildProcesses::listenerCount(); + $this->loop->stop(); + }); + + $this->artisan('websockets:serve', ['--loop' => $this->loop, '--host' => '127.0.0.1', '--port' => 0]); + + $this->assertSame(1, $registered); + } + + public function test_children_are_left_alone_when_disabled() + { + config(['websockets.shutdown.terminate_children' => false]); + $pid = $this->spawnTrackedChild(); + + $command = $this->command(LoopFactory::create()); + (new \ReflectionMethod(StartServer::class, 'triggerHardShutdown'))->invoke($command); + + $this->assertTrue(posix_kill($pid, 0), 'Opted out: the child keeps running.'); + } + + private function command($loop): StartServer + { + $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); + + return $command; + } + + private function spawnTrackedChild(): int + { + $pid = pcntl_fork(); + $this->assertNotSame(-1, $pid); + + if ($pid === 0) { + // A busy message child: default SIGTERM disposition, as after + // ChildProcesses::prepareChild(). Never returns into PHPUnit. + ChildProcesses::restoreDefaultSignals(); + while (true) { + usleep(50_000); + } + } + + $this->spawned[] = $pid; + ChildProcesses::track($pid, 'child-crash.hold'); + + return $pid; + } +} diff --git a/tests/Server/ChildFatalErrorGuardTest.php b/tests/Server/ChildFatalErrorGuardTest.php new file mode 100644 index 0000000..2b681fd --- /dev/null +++ b/tests/Server/ChildFatalErrorGuardTest.php @@ -0,0 +1,81 @@ +markTestSkipped('pcntl and sockets are required'); + } + } + + public function test_a_child_that_runs_out_of_memory_still_sends_the_error_frame() + { + $result = $this->runChild('oom'); + + $this->assertSame(255, $result['exit_code'], 'PHP exits with 255 after a fatal error.'); + $this->assertCount(1, $result['frames']); + + $frame = json_decode($result['frames'][0], true); + $this->assertSame('demo.crash:error', $frame['event']); + $this->assertSame('Internal server error', $frame['data']['message']); + $this->assertTrue($frame['data']['meta']['fatal']); + $this->assertTrue($result['replied'], 'The parent recognises the frame as the final reply.'); + $this->assertStringContainsString('app handler saw: Allowed memory size', $result['stderr'], 'The earlier app handler still ran first.'); + } + + public function test_a_disarmed_guard_stays_silent() + { + $result = $this->runChild('oom-disarmed'); + + $this->assertSame(255, $result['exit_code']); + $this->assertSame([], $result['frames']); + $this->assertFalse($result['replied']); + } + + public function test_a_normal_exit_sends_nothing() + { + $this->assertSame([], $this->runChild('clean-exit')['frames']); + $this->assertSame([], $this->runChild('warning-then-exit')['frames'], 'A non-fatal last error is not a crash.'); + } + + /** + * @return array{frames: string[], exit_code: ?int, replied: bool, stderr: string} + */ + private function runChild(string $mode): array + { + $process = proc_open( + [PHP_BINARY, '-d', 'display_errors=stderr', '-d', 'memory_limit=128M', __DIR__ . '/Fixtures/fatal_child.php', $mode], + [1 => ['pipe', 'w'], 2 => ['pipe', 'w']], + $pipes + ); + $this->assertIsResource($process); + + $stdout = stream_get_contents($pipes[1]); + $stderr = stream_get_contents($pipes[2]); + fclose($pipes[1]); + fclose($pipes[2]); + proc_close($process); + + $result = json_decode((string) $stdout, true); + $this->assertIsArray($result, "Driver output was not JSON.\nstdout: {$stdout}\nstderr: {$stderr}"); + + return $result + ['stderr' => $stderr]; + } +} diff --git a/tests/Server/ChildProcessesTest.php b/tests/Server/ChildProcessesTest.php new file mode 100644 index 0000000..30e3d5c --- /dev/null +++ b/tests/Server/ChildProcessesTest.php @@ -0,0 +1,335 @@ +markTestSkipped("{$function}() is required"); + } + } + + ChildProcesses::flush(); + } + + protected function tearDown(): void + { + foreach ($this->spawned as $pid) { + @posix_kill($pid, SIGKILL); + @pcntl_waitpid($pid, $status); + } + + ChildProcesses::flush(); + + parent::tearDown(); + } + + public function test_shutdown_sigterms_a_running_child_and_reaps_it() + { + $statuses = []; + + // The child inherits a SIGTERM handler, as message children inherit the + // server's. prepareChild() must restore the default, otherwise SIGTERM + // is swallowed and only the SIGKILL fallback ends the child. + $pid = $this->forkChild(function () { + pcntl_signal(SIGTERM, function () { + // inherited handler that would swallow the signal + }); + ChildProcesses::prepareChild(); + }); + + ChildProcesses::track($pid, 'atc-traffic.subscribe', function (?int $status) use (&$statuses) { + $statuses[] = $status; + }); + + $stats = ChildProcesses::terminateAll(2.0); + + $this->assertSame(['terminated' => 1, 'killed' => 0, 'remaining' => 0], $stats); + $this->assertSame(0, ChildProcesses::count()); + $this->assertCount(1, $statuses, 'The exit callback runs once, with the wait status.'); + $this->assertTrue(pcntl_wifsignaled($statuses[0])); + $this->assertSame(SIGTERM, pcntl_wtermsig($statuses[0])); + $this->assertTrue(ChildProcesses::exitedAbnormally($statuses[0])); + $this->assertFalse(@posix_kill($pid, 0), 'The child is gone and reaped.'); + } + + public function test_shutdown_sigkills_a_child_that_ignores_sigterm_after_the_grace_period() + { + $statuses = []; + + $pid = $this->forkChild(function () { + pcntl_signal(SIGTERM, SIG_IGN); + }); + + ChildProcesses::track($pid, null, function (?int $status) use (&$statuses) { + $statuses[] = $status; + }); + + $started = microtime(true); + $stats = ChildProcesses::terminateAll(0.3, 1.0); + $elapsed = microtime(true) - $started; + + $this->assertSame(['terminated' => 1, 'killed' => 1, 'remaining' => 0], $stats); + $this->assertGreaterThanOrEqual(0.3, $elapsed, 'SIGKILL waits for the grace period.'); + $this->assertLessThan(1.5, $elapsed, 'The shutdown wait is bounded.'); + $this->assertSame(SIGKILL, pcntl_wtermsig($statuses[0])); + $this->assertFalse(@posix_kill($pid, 0)); + } + + public function test_shutdown_never_signals_a_process_that_is_not_its_child() + { + // A grandchild is re-parented away from this process, exactly like a + // PID that was reaped elsewhere and recycled by an unrelated process. + $grandchild = $this->forkGrandchild(); + $this->spawned[] = $grandchild; + + $exitCalls = []; + ChildProcesses::track($grandchild, null, function (?int $status) use (&$exitCalls) { + $exitCalls[] = $status; + }); + + $stats = ChildProcesses::terminateAll(0.2, 0.2); + + $this->assertSame(['terminated' => 0, 'killed' => 0, 'remaining' => 0], $stats); + $this->assertTrue(posix_kill($grandchild, 0), 'A process that is not our child must survive.'); + $this->assertSame([null], $exitCalls, 'It is forgotten with an unknown status.'); + $this->assertSame(0, ChildProcesses::count()); + } + + public function test_reap_exited_reports_the_exit_status_to_the_callback() + { + $statuses = []; + + foreach ([0, 3] as $code) { + $pid = pcntl_fork(); + if ($pid === 0) { + $this->exitChild($code); + } + $this->spawned[] = $pid; + + ChildProcesses::track($pid, null, function (?int $status) use (&$statuses, $code) { + $statuses[$code] = $status; + }); + } + + $deadline = microtime(true) + 5; + while (count($statuses) < 2 && microtime(true) < $deadline) { + ChildProcesses::reapExited(); + usleep(10_000); + } + + $this->assertCount(2, $statuses); + $this->assertSame(0, pcntl_wexitstatus($statuses[0])); + $this->assertFalse(ChildProcesses::exitedAbnormally($statuses[0])); + $this->assertSame(3, pcntl_wexitstatus($statuses[3])); + $this->assertTrue(ChildProcesses::exitedAbnormally($statuses[3])); + $this->assertSame(0, ChildProcesses::count()); + } + + public function test_a_child_that_ran_prepare_child_does_not_keep_the_port_bound() + { + $server = $this->listen($port); + $this->assertTrue(ChildProcesses::registerListener($server)); + + $pid = $this->forkChild(function () { + ChildProcesses::prepareChild(); + }); + + // The parent still accepts on its own copy of the socket. + $client = stream_socket_client("tcp://127.0.0.1:{$port}", $errno, $errstr, 2); + $this->assertIsResource($client, $errstr); + $accepted = @stream_socket_accept($server, 2); + $this->assertIsResource($accepted, 'The parent must keep accepting after the child closed its copy.'); + fclose($accepted); + fclose($client); + + // The server goes away while the child is still running: the port + // must be free for its successor at once. + fclose($server); + $this->assertTrue(posix_kill($pid, 0), 'The child is still alive.'); + + $successor = @stream_socket_server("tcp://127.0.0.1:{$port}", $errno, $errstr); + $this->assertIsResource($successor, "Rebinding failed while the child lives: {$errstr}"); + fclose($successor); + } + + public function test_a_child_that_did_not_close_the_socket_keeps_the_port_bound() + { + // The failure mode the fix removes: proves the test above is meaningful. + $server = $this->listen($port); + + $this->forkChild(function () { + // inherits the listening socket and keeps it + }); + + fclose($server); + + $successor = @stream_socket_server("tcp://127.0.0.1:{$port}", $errno, $errstr); + $this->assertFalse($successor, 'The port should still be held by the child.'); + $this->assertStringContainsStringIgnoringCase('in use', $errstr); + } + + public function test_it_finds_the_listening_socket_inside_react_servers_and_keeps_the_loop_usable() + { + $loop = new StreamSelectLoop(); + $tcp = new TcpServer('127.0.0.1:0', $loop); + $socketServer = new SocketServer('127.0.0.1:0', [], $loop); + // What ServerFactory builds today (the deprecated wrapper). + $legacyServer = new LegacyServer('127.0.0.1:0', $loop); + + $this->assertTrue(ChildProcesses::registerListener($tcp)); + $this->assertTrue(ChildProcesses::registerListener($socketServer)); + $this->assertTrue(ChildProcesses::registerListener($legacyServer)); + $this->assertFalse(ChildProcesses::registerListener(new \stdClass())); + $this->assertSame(3, ChildProcesses::listenerCount()); + + $ports = [ + (int) parse_url((string) $tcp->getAddress(), PHP_URL_PORT), + (int) parse_url((string) $socketServer->getAddress(), PHP_URL_PORT), + (int) parse_url((string) $legacyServer->getAddress(), PHP_URL_PORT), + ]; + + $this->assertSame(3, ChildProcesses::closeInheritedListeners($loop)); + $this->assertSame(0, ChildProcesses::listenerCount()); + + // Both ports are free again. + foreach ($ports as $port) { + $rebound = @stream_socket_server("tcp://127.0.0.1:{$port}", $errno, $errstr); + $this->assertIsResource($rebound, $errstr); + fclose($rebound); + } + + // StreamSelectLoop no longer selects on the closed streams; a closed + // stream left in its list makes stream_select() throw. + $ticked = false; + $loop->futureTick(function () use ($loop, &$ticked) { + $ticked = true; + $loop->stop(); + }); + $loop->addTimer(1, fn () => $loop->stop()); + $loop->run(); + + $this->assertTrue($ticked); + } + + /** + * Fork a child that runs $setup, tells the parent it is ready, then waits + * to be killed. + */ + private function forkChild(callable $setup): int + { + $pair = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP); + + $pid = pcntl_fork(); + $this->assertNotSame(-1, $pid, 'fork failed'); + + if ($pid === 0) { + // Never return into PHPUnit from a forked child: it would go on + // running the rest of the suite. + try { + fclose($pair[0]); + $setup(); + fwrite($pair[1], "ready\n"); + } catch (\Throwable) { + $this->exitChild(1); + } + + while (true) { + usleep(50_000); + } + } + + $this->spawned[] = $pid; + fclose($pair[1]); + stream_set_timeout($pair[0], 5); + $this->assertSame("ready\n", fgets($pair[0]), 'The child never became ready.'); + fclose($pair[0]); + + return $pid; + } + + /** + * Fork a child that forks a long-sleeping grandchild, reports its PID + * and exits, so the grandchild is no longer a child of this process. + */ + private function forkGrandchild(): int + { + $pair = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP); + + $pid = pcntl_fork(); + $this->assertNotSame(-1, $pid, 'fork failed'); + + if ($pid === 0) { + fclose($pair[0]); + $grandchild = pcntl_fork(); + + if ($grandchild === 0) { + fclose($pair[1]); + ChildProcesses::restoreDefaultSignals(); + sleep(30); + $this->exitChild(0); + } + + fwrite($pair[1], $grandchild . "\n"); + $this->exitChild(0); + } + + fclose($pair[1]); + stream_set_timeout($pair[0], 5); + $grandchild = (int) trim((string) fgets($pair[0])); + fclose($pair[0]); + pcntl_waitpid($pid, $status); + + $this->assertGreaterThan(0, $grandchild); + + return $grandchild; + } + + /** + * @return resource + */ + private function listen(?int &$port) + { + $server = stream_socket_server('tcp://127.0.0.1:0', $errno, $errstr); + $this->assertIsResource($server, $errstr); + $port = (int) parse_url('tcp://' . stream_socket_get_name($server, false), PHP_URL_PORT); + + return $server; + } + + /** + * Leave a forked child without flushing PHPUnit's buffered output twice. + */ + private function exitChild(int $code): void + { + while (ob_get_level() > 0) { + ob_end_clean(); + } + + exit($code); + } +} diff --git a/tests/Server/Fixtures/fatal_child.php b/tests/Server/Fixtures/fatal_child.php new file mode 100644 index 0000000..032b570 --- /dev/null +++ b/tests/Server/Fixtures/fatal_child.php @@ -0,0 +1,97 @@ + + */ + +declare(strict_types=1); + +use BlaxSoftware\LaravelWebSockets\Ipc\SocketPairIpc; +use BlaxSoftware\LaravelWebSockets\Server\ChildFatalErrorGuard; +use BlaxSoftware\LaravelWebSockets\Websocket\ChildMessageOutcome; +use React\EventLoop\StreamSelectLoop; + +require __DIR__ . '/../../../vendor/autoload.php'; + +$mode = $argv[1] ?? 'oom'; + +$loop = new StreamSelectLoop(); +$ipc = SocketPairIpc::create($loop); + +$pid = pcntl_fork(); + +if ($pid === 0) { + $ipc->setupChild(); + + // Stand-in for Laravel's HandleExceptions: registered earlier (in the + // server process), so it runs before the guard and sees the error first. + register_shutdown_function(function () { + $error = error_get_last(); + if ($error !== null) { + fwrite(STDERR, 'app handler saw: ' . $error['message'] . "\n"); + } + }); + + $frame = json_encode([ + 'event' => 'demo.crash:error', + 'data' => ['message' => 'Internal server error', 'meta' => ['reported' => true, 'fatal' => true]], + 'channel' => 'test', + ]); + + $guard = ChildFatalErrorGuard::register(function () use ($ipc, $frame) { + $ipc->sendToParent($frame); + }); + + if ($mode === 'clean-exit') { + exit(0); + } + + if ($mode === 'warning-then-exit') { + @trigger_error('just a warning', E_USER_WARNING); + exit(0); + } + + if ($mode === 'oom-disarmed') { + $guard->disarm(); + } + + ini_set('memory_limit', '32M'); + $hog = []; + while (true) { + $hog[] = str_repeat('x', 1024 * 1024); + } +} + +$frames = []; +$outcome = new ChildMessageOutcome('demo.crash', function () { +}); + +$ipc->setupParent( + function ($data) use (&$frames, $outcome) { + $frames[] = $data; + $outcome->frameReceived($data); + }, + function () use ($loop) { + $loop->stop(); + } +); + +$loop->addTimer(10, function () use ($loop) { + $loop->stop(); +}); +$loop->run(); + +pcntl_waitpid($pid, $status); + +echo json_encode([ + 'frames' => $frames, + 'exit_code' => pcntl_wifexited($status) ? pcntl_wexitstatus($status) : null, + 'replied' => $outcome->replied(), +]); diff --git a/tests/TestCase.php b/tests/TestCase.php index c98aac8..91a77d8 100644 --- a/tests/TestCase.php +++ b/tests/TestCase.php @@ -132,6 +132,13 @@ abstract class TestCase extends Orchestra protected function tearDown(): void { + // A test that ran websockets:serve and stopped its loop directly never + // reached the shutdown, so the broadcast socket file stayed in /tmp and + // made ws_available() true for every later test. + if ($this->app && $this->app->bound(\BlaxSoftware\LaravelWebSockets\Broadcast\BroadcastSocketServer::class)) { + $this->app->make(\BlaxSoftware\LaravelWebSockets\Broadcast\BroadcastSocketServer::class)->stop(); + } + parent::tearDown(); if ($this->server) { diff --git a/tests/Websocket/ChildMessageOutcomeTest.php b/tests/Websocket/ChildMessageOutcomeTest.php new file mode 100644 index 0000000..d9f8e52 --- /dev/null +++ b/tests/Websocket/ChildMessageOutcomeTest.php @@ -0,0 +1,115 @@ +crashes = []; + + return new ChildMessageOutcome($event, function (int $status) { + $this->crashes[] = $status; + }); + } + + public function test_a_killed_child_without_reply_is_answered_once_both_signals_are_in() + { + foreach ([[true, false], [false, true]] as [$streamFirst]) { + $outcome = $this->outcome(); + $outcome->frameReceived('{"event":"atc-traffic.subscribe[x1]:progress","data":[]}'); + + if ($streamFirst) { + $outcome->streamClosed(); + $this->assertSame([], $this->crashes, 'No verdict before the exit status is known.'); + $outcome->exited(self::KILLED); + } else { + $outcome->exited(self::KILLED); + $this->assertSame([], $this->crashes, 'No verdict while frames may still be unread.'); + $outcome->streamClosed(); + } + + $this->assertSame([self::KILLED], $this->crashes); + } + } + + public function test_a_fatal_exit_without_reply_is_answered() + { + $outcome = $this->outcome(); + $outcome->streamClosed(); + $outcome->exited(self::EXIT_FATAL); + + $this->assertSame([self::EXIT_FATAL], $this->crashes); + } + + public function test_a_child_that_replied_is_left_alone_even_if_it_crashed_afterwards() + { + foreach (['response', 'error'] as $suffix) { + $outcome = $this->outcome(); + $outcome->frameReceived('{"event":"atc-traffic.subscribe[x1]:' . $suffix . '","data":{"ok":true},"channel":"websocket"}'); + $outcome->streamClosed(); + $outcome->exited(self::EXIT_FATAL); + + $this->assertTrue($outcome->replied()); + $this->assertSame([], $this->crashes, "A {$suffix} frame counts as the reply."); + } + } + + public function test_frames_for_other_events_or_broadcasts_are_not_a_reply() + { + $outcome = $this->outcome('dashboard.index'); + $outcome->frameReceived('B:{"event":"dashboard.index:response"}'); + $outcome->frameReceived('{"event":"dashboard.indexer:response"}'); + $outcome->frameReceived('{"data":1,"event":"dashboard.index:response"}'); + $outcome->frameReceived(null); + + $this->assertFalse($outcome->replied()); + } + + public function test_event_names_are_matched_in_their_json_encoded_form() + { + $event = 'admin/user.show[a"b]'; + $outcome = $this->outcome($event); + $outcome->frameReceived(json_encode(['event' => $event . ':response', 'data' => null])); + + $this->assertTrue($outcome->replied()); + } + + public function test_a_clean_exit_or_an_unknown_status_sends_nothing() + { + foreach ([self::EXIT_OK, null] as $status) { + $outcome = $this->outcome(); + $outcome->streamClosed(); + $outcome->exited($status); + + $this->assertSame([], $this->crashes); + } + } + + public function test_the_verdict_is_given_only_once() + { + $outcome = $this->outcome(); + $outcome->streamClosed(); + $outcome->exited(self::KILLED); + $outcome->streamClosed(); + $outcome->exited(self::KILLED); + + $this->assertCount(1, $this->crashes); + } +} diff --git a/tests/Websocket/Fixtures/ChildCrash/ChildCrashController.php b/tests/Websocket/Fixtures/ChildCrash/ChildCrashController.php new file mode 100644 index 0000000..3e2b2d9 --- /dev/null +++ b/tests/Websocket/Fixtures/ChildCrash/ChildCrashController.php @@ -0,0 +1,43 @@ + true]; + } + + /** + * Dies without replying, like a child taken by the kernel OOM killer. + */ + #[Websocket] + public function killed(): array + { + posix_kill(getmypid(), SIGKILL); + sleep(5); + + return ['unreachable' => true]; + } + + /** + * Blocks like a long-running engine until the server stops it. + */ + #[Websocket] + public function hold(): array + { + sleep(10); + + return ['held' => true]; + } +} diff --git a/tests/Websocket/Fixtures/DeferredCallbackCollectionStub.php b/tests/Websocket/Fixtures/DeferredCallbackCollectionStub.php new file mode 100644 index 0000000..5a57fad --- /dev/null +++ b/tests/Websocket/Fixtures/DeferredCallbackCollectionStub.php @@ -0,0 +1,18 @@ +markTestSkipped('pcntl, posix and sockets are required'); + } + + if (! class_exists(\Illuminate\Support\Defer\DeferredCallbackCollection::class)) { + require_once __DIR__ . '/Fixtures/DeferredCallbackCollectionStub.php'; + } + + ChildProcesses::flush(); + ControllerResolver::clearCache(); + EventRegistry::clear(); + EventRegistry::setSearchPaths([ + __DIR__ . '/Fixtures/ChildCrash' => self::FIXTURES_NS, + ]); + } + + public function tearDown(): void + { + // Never leave a forked child behind, even after a failed assertion. + ChildProcesses::terminateAll(0.0, 1.0); + ChildProcesses::flush(); + + EventRegistry::clear(); + (new \ReflectionProperty(EventRegistry::class, 'searchPaths'))->setValue(null, null); + + parent::tearDown(); + } + + public function test_a_child_killed_before_replying_is_answered_by_the_parent() + { + $connection = $this->send('child-crash.killed'); + + $this->runLoopUntil(fn () => $this->sentEvent($connection, 'child-crash.killed:error') !== null); + + $error = $this->sentEvent($connection, 'child-crash.killed:error'); + $this->assertNotNull($error, 'The client must not wait forever for a child that died.'); + $this->assertSame('Internal server error', $error['data']['message']); + $this->assertFalse($error['data']['meta']['reported']); + $this->assertSame(0, ChildProcesses::count(), 'The dead child was reaped and forgotten.'); + } + + public function test_a_child_that_replied_gets_no_extra_error_frame() + { + $connection = $this->send('child-crash.ok'); + + $this->runLoopUntil( + fn () => ChildProcesses::count() === 0 && $this->sentEvent($connection, 'child-crash.ok:response') !== null, + 5.0, + 0.2 + ); + + $response = $this->sentEvent($connection, 'child-crash.ok:response'); + $this->assertNotNull($response); + $this->assertSame(['ok' => true], $response['data']); + $connection->assertNotSentEvent('child-crash.ok:error'); + } + + public function test_a_running_child_neither_holds_the_port_nor_survives_the_shutdown() + { + $server = stream_socket_server('tcp://127.0.0.1:0', $errno, $errstr); + $this->assertIsResource($server, $errstr); + $port = (int) parse_url('tcp://' . stream_socket_get_name($server, false), PHP_URL_PORT); + $this->assertTrue(ChildProcesses::registerListener($server)); + + $connection = $this->send('child-crash.hold'); + $this->assertSame(1, ChildProcesses::count()); + [$pid] = ChildProcesses::pids(); + + // The server closes its socket while the child is still busy: the port + // frees as soon as the child has dropped its inherited copy. + fclose($server); + $successor = false; + $deadline = microtime(true) + 5; + while ($successor === false && microtime(true) < $deadline) { + $successor = @stream_socket_server("tcp://127.0.0.1:{$port}", $errno, $errstr); + if ($successor === false) { + usleep(20_000); + } + } + $this->assertIsResource($successor, "The busy child kept port {$port} bound: {$errstr}"); + fclose($successor); + $this->assertTrue(posix_kill($pid, 0), 'The child is still running its handler.'); + + // Server shutdown: SIGTERM does end the child (default disposition + // restored after fork), no SIGKILL needed. + $stats = ChildProcesses::terminateAll(2.0); + $this->assertSame(['terminated' => 1, 'killed' => 0, 'remaining' => 0], $stats); + $this->assertFalse(@posix_kill($pid, 0)); + + // Its client is told instead of waiting for a reply that never comes. + $this->runLoopUntil(fn () => $this->sentEvent($connection, 'child-crash.hold:error') !== null); + $error = $this->sentEvent($connection, 'child-crash.hold:error'); + $this->assertNotNull($error); + $this->assertSame('Server is restarting', $error['data']['message']); + } + + public function test_the_fatal_error_frame_is_recognised_as_the_reply() + { + $method = new \ReflectionMethod(Handler::class, 'fatalErrorFrame'); + if (PHP_VERSION_ID < 80100) { + $method->setAccessible(true); + } + + $message = ['event' => 'child-crash.ok[u1]', 'data' => [], 'channel' => 'crash-channel']; + $frame = $method->invoke(null, $message, 'crash-channel'); + + // Same shape as the controller's catch-all error frame. + $this->assertSame([ + 'event' => 'child-crash.ok[u1]:error', + 'data' => ['message' => 'Internal server error', 'meta' => ['reported' => true, 'fatal' => true]], + 'channel' => 'crash-channel', + ], json_decode($frame, true)); + + // The parent must count it as the reply, or it would add its own + // error frame on top of it. + $outcome = new ChildMessageOutcome($message['event'], fn () => null); + $outcome->frameReceived($frame); + + $this->assertTrue($outcome->replied()); + } + + private function send(string $event): Mocks\Connection + { + $connection = $this->newActiveConnection(['crash-channel']); + $connection->resetEvents(); + + $this->wsHandler->onMessage($connection, new Mocks\Message([ + 'event' => $event, + 'data' => [], + 'channel' => 'crash-channel', + ])); + + return $connection; + } + + /** + * Run the loop with StartServer's reaper until $done or the timeout, then + * keep it running for $settle seconds. + */ + private function runLoopUntil(callable $done, float $timeout = 5.0, float $settle = 0.0): void + { + $finishedAt = null; + + $tick = $this->loop->addPeriodicTimer(0.02, function () use ($done, $settle, &$finishedAt) { + ChildProcesses::reapExited(); + + if ($finishedAt === null && $done()) { + $finishedAt = microtime(true); + } + + if ($finishedAt !== null && microtime(true) - $finishedAt >= $settle) { + $this->loop->stop(); + } + }); + $timeoutTimer = $this->loop->addTimer($timeout, fn () => $this->loop->stop()); + + $this->loop->run(); + + $this->loop->cancelTimer($tick); + $this->loop->cancelTimer($timeoutTimer); + } + + private function sentEvent(Mocks\Connection $connection, string $event): ?array + { + foreach ($connection->sentData as $frame) { + if (($frame['event'] ?? null) === $event) { + return $frame; + } + } + + return null; + } +}