laravel-websockets/tests/Server/ChildProcessesTest.php

<?php

declare(strict_types=1);

namespace BlaxSoftware\LaravelWebSockets\Test\Server;

use BlaxSoftware\LaravelWebSockets\Server\ChildProcesses;
use PHPUnit\Framework\TestCase;
use React\EventLoop\StreamSelectLoop;
use React\Socket\Server as LegacyServer;
use React\Socket\SocketServer;
use React\Socket\TcpServer;

/**
 * Regression guard for #2465: a fork-per-message child that outlived the
 * server kept :6001 bound, so the restarted server failed with EADDRINUSE.
 *
 * Uses real forks, real signals and real sockets. Every child is killed and
 * reaped in tearDown even when an assertion fails.
 */
class ChildProcessesTest extends TestCase
{
    /** @var int[] */
    private array $spawned = [];

    protected function setUp(): void
    {
        parent::setUp();

        foreach (['pcntl_fork', 'pcntl_waitpid', 'posix_kill', 'pcntl_signal'] as $function) {
            if (! function_exists($function)) {
                $this->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);
    }
}