From 7ee9e7153701aade0e767cb69708a03cc01520c6 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 23:50:59 +0000 Subject: [PATCH] fix(users): publish FPM worker pids so Redis reaping spares LB viewers A connection row records the PHP-FPM worker serving it (getmypid() in live.php, vod.php and timeshift.php), but the watchdog published as servers.php_pids the contents of bin/php/sockets/*.pid: the FPM master pids. In Redis mode MAIN's cron:users reaps every node's connections and, for a remote PHP-served connection older than the node's last heartbeat, checks the worker pid against that list. It never matched, so the connection was closed and the LB told to kill its worker: every VOD, timeshift and PHP-served live viewer on an LB was dropped about a minute after connecting. MySQL mode was unaffected (each node probes its own pids locally). - ProcessManager::phpFpmWorkerPIDs() lists the processes titled "php-fpm: pool xc_vm" (sorted, unique); findProcessPIDs() takes an optional procfs root so it can run against a fixture tree. - WatchdogCommand publishes that list instead of the pid files. - ServerRepository::getAll() drops php_pids from the rows it returns and caches, so the per-request servers cache does not carry every node's worker list. CacheCronJob's unset($rServers['php_pids']) meant to do that but hit a top-level key of an id-keyed map; it is removed. - UsersCronJob reads php_pids with its own SELECT `id`, `php_pids`. The `- 1` heartbeat margin is unchanged. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01TQHeypQpfAXTdCysmi7b6a --- src/Cli/Commands/WatchdogCommand.php | 10 +-- src/Cli/CronJobs/CacheCronJob.php | 1 - src/Cli/CronJobs/UsersCronJob.php | 25 ++++++-- src/Core/Process/ProcessManager.php | 25 ++++++-- src/Domain/Server/ServerRepository.php | 3 + tests/Unit/PhpFpmWorkerPidsTest.php | 65 +++++++++++++++++++ tests/Unit/ServerRepositoryPhpPidsTest.php | 75 ++++++++++++++++++++++ tests/Unit/UsersCronPhpPidsTest.php | 40 ++++++++++++ 8 files changed, 227 insertions(+), 17 deletions(-) create mode 100644 tests/Unit/PhpFpmWorkerPidsTest.php create mode 100644 tests/Unit/ServerRepositoryPhpPidsTest.php create mode 100644 tests/Unit/UsersCronPhpPidsTest.php diff --git a/src/Cli/Commands/WatchdogCommand.php b/src/Cli/Commands/WatchdogCommand.php index e1f260e4..ff84786c 100644 --- a/src/Cli/Commands/WatchdogCommand.php +++ b/src/Cli/Commands/WatchdogCommand.php @@ -134,13 +134,9 @@ class WatchdogCommand implements CommandInterface { $rStats['fanout'] = FanoutClient::status(); // ── PHP PIDs ───────────────────────────────────────── - $rPHPPIDs = []; - foreach (glob(MAIN_HOME . 'bin/php/sockets/*.pid') ?: [] as $rPidFile) { - $rPid = trim(@file_get_contents($rPidFile) ?: ''); - if (is_numeric($rPid) && 0 < intval($rPid)) { - $rPHPPIDs[] = intval($rPid); - } - } + // FPM worker pids, which connection rows record; the pool pid + // files hold only the masters. + $rPHPPIDs = ProcessManager::phpFpmWorkerPIDs(); // ── Update servers table ───────────────────────────── $rConnections = $rUsers = 0; diff --git a/src/Cli/CronJobs/CacheCronJob.php b/src/Cli/CronJobs/CacheCronJob.php index e185e50d..85cc61d1 100644 --- a/src/Cli/CronJobs/CacheCronJob.php +++ b/src/Cli/CronJobs/CacheCronJob.php @@ -99,7 +99,6 @@ class CacheCronJob implements CommandInterface { FileCache::setCache('settings', SettingsRepository::getAll(true)); FileCache::setCache('bouquets', BouquetService::getAll(true)); $rServers = ServerRepository::getAll(true); - unset($rServers['php_pids']); FileCache::setCache('servers', $rServers); FileCache::setCache('proxy_servers', BlocklistService::getProxyIPs(true)); FileCache::setCache('blocked_servers', BlocklistService::getBlockedServers(true)); diff --git a/src/Cli/CronJobs/UsersCronJob.php b/src/Cli/CronJobs/UsersCronJob.php index 47b86ed1..dd2022f9 100644 --- a/src/Cli/CronJobs/UsersCronJob.php +++ b/src/Cli/CronJobs/UsersCronJob.php @@ -126,11 +126,7 @@ class UsersCronJob implements CommandInterface { if (SettingsManager::getBool('redis_handler') && $this->rServers[SERVER_ID]['is_main']) { $this->rServers = ServerRepository::getAll(true); - - foreach ($this->rServers as $rServer) { - $rDecodedPids = json_decode($rServer['php_pids'] ?? '', true); - $this->rPHPPIDs[$rServer['id']] = is_array($rDecodedPids) ? array_map('intval', $rDecodedPids) : []; - } + $this->rPHPPIDs = $this->loadPHPPIDs(); } $this->loadCron(); @@ -138,6 +134,25 @@ class UsersCronJob implements CommandInterface { return 0; } + /** + * Every node's live PHP-FPM worker pids, as its watchdog last published + * them. Read here because ServerRepository::getAll() leaves php_pids out. + * + * @return array Worker pids keyed by server id ([] = unknown). + */ + private function loadPHPPIDs(): array { + global $db; + $rPHPPIDs = []; + + $db->query('SELECT `id`, `php_pids` FROM `servers`;'); + foreach ($db->get_rows() ?: [] as $rRow) { + $rDecodedPids = json_decode($rRow['php_pids'] ?? '', true); + $rPHPPIDs[intval($rRow['id'])] = is_array($rDecodedPids) ? array_map('intval', $rDecodedPids) : []; + } + + return $rPHPPIDs; + } + private function processDeletions($rDelete, $rDelStream = []) { $rRedis = SettingsManager::getBool('redis_handler'); global $db; diff --git a/src/Core/Process/ProcessManager.php b/src/Core/Process/ProcessManager.php index cabc1493..25f3d927 100644 --- a/src/Core/Process/ProcessManager.php +++ b/src/Core/Process/ProcessManager.php @@ -604,14 +604,15 @@ class ProcessManager { * match unrelated processes (e.g. ffmpeg's -thread_queue_size satisfied * the "queue" daemon check, so the encode queue was never revived). * - * @param array $rTerms Cmdline substrings to match (exact, case-sensitive) - * @param int $rLimit Stop after this many matches (0 = no limit) + * @param array $rTerms Cmdline substrings to match (exact, case-sensitive) + * @param int $rLimit Stop after this many matches (0 = no limit) + * @param string $rProcRoot procfs mount point (overridable for tests) * @return array Matching PIDs (own PID excluded) */ - public static function findProcessPIDs(array $rTerms, int $rLimit = 0) { + public static function findProcessPIDs(array $rTerms, int $rLimit = 0, string $rProcRoot = '/proc') { $rPIDs = []; $rSelf = getmypid(); - foreach (glob('/proc/[0-9]*/cmdline') ?: [] as $rCmdFile) { + foreach (glob($rProcRoot . '/[0-9]*/cmdline') ?: [] as $rCmdFile) { $rPID = intval(basename(dirname($rCmdFile))); if ($rPID == $rSelf) { continue; @@ -644,4 +645,20 @@ class ProcessManager { public static function isAnyProcessRunning(array $rTerms) { return count(self::findProcessPIDs($rTerms, 1)) > 0; } + + /** + * PIDs of the live PHP-FPM workers of the xc_vm pool — the pids that + * connection rows record (getmypid() in the stream endpoints). The watchdog + * publishes them as servers.php_pids. FPM titles a worker "php-fpm: pool + * xc_vm" and its master "php-fpm: master process (...)", so the pool pid + * files (masters only) would never match a connection. + * + * @param string $rProcRoot procfs mount point (overridable for tests) + * @return array Worker PIDs, ascending + */ + public static function phpFpmWorkerPIDs(string $rProcRoot = '/proc') { + $rPIDs = array_unique(self::findProcessPIDs(['php-fpm: pool xc_vm'], 0, $rProcRoot)); + sort($rPIDs); + return $rPIDs; + } } diff --git a/src/Domain/Server/ServerRepository.php b/src/Domain/Server/ServerRepository.php index 064d1db0..57bea1dd 100644 --- a/src/Domain/Server/ServerRepository.php +++ b/src/Domain/Server/ServerRepository.php @@ -103,6 +103,9 @@ class ServerRepository { if (!isset($rRow['order'])) { $rRow['order'] = 0; } + // One pid per live PHP-FPM worker; only cron:users needs it and it + // reads the column itself. Keep it out of the per-request cache. + unset($rRow['php_pids']); $rServers[intval($rRow['id'])] = $rRow; } diff --git a/tests/Unit/PhpFpmWorkerPidsTest.php b/tests/Unit/PhpFpmWorkerPidsTest.php new file mode 100644 index 00000000..8635e7a2 --- /dev/null +++ b/tests/Unit/PhpFpmWorkerPidsTest.php @@ -0,0 +1,65 @@ +rRoot = sys_get_temp_dir() . '/xcvm_proc_' . uniqid('', true); + mkdir($this->rRoot, 0755, true); + } + + protected function tearDown(): void { + foreach (glob($this->rRoot . '/*/cmdline') ?: [] as $rFile) { + unlink($rFile); + rmdir(dirname($rFile)); + } + @rmdir($this->rRoot); + } + + private function process(int $rPID, string $rCmdline): void { + mkdir($this->rRoot . '/' . $rPID); + file_put_contents($this->rRoot . '/' . $rPID . '/cmdline', $rCmdline); + } + + public function testListsTheXcVmPoolWorkersOnly(): void { + $this->process(4300001, "php-fpm: master process (/home/xc_vm/bin/php/etc/1.conf)\0"); + // FPM overwrites the worker's argv with its title and pads the rest with NULs. + $this->process(4300002, 'php-fpm: pool xc_vm' . str_repeat("\0", 40)); + $this->process(4300003, 'php-fpm: pool xc_vm'); + $this->process(4300004, "/home/xc_vm/bin/php/bin/php\0console.php\0watchdog\0"); + $this->process(4300005, 'php-fpm: pool www'); + // Globbed before 4300002 (string order); listed after it (numeric order). + $this->process(43000040, 'php-fpm: pool xc_vm'); + + $this->assertSame([4300002, 4300003, 43000040], ProcessManager::phpFpmWorkerPIDs($this->rRoot)); + } + + public function testNoWorkersYieldsAnEmptyList(): void { + $this->process(4300001, "php-fpm: master process (/home/xc_vm/bin/php/etc/1.conf)\0"); + + $this->assertSame([], ProcessManager::phpFpmWorkerPIDs($this->rRoot)); + } + + public function testFindProcessPIDsReadsTheGivenProcRoot(): void { + $this->assertSame([], ProcessManager::findProcessPIDs(['x'], 0, $this->rRoot)); + + $this->process(4300007, "ffmpeg\0-i\0x\0"); + $this->assertSame([4300007], ProcessManager::findProcessPIDs(['ffmpeg'], 0, $this->rRoot)); + } +} diff --git a/tests/Unit/ServerRepositoryPhpPidsTest.php b/tests/Unit/ServerRepositoryPhpPidsTest.php new file mode 100644 index 00000000..26b7007f --- /dev/null +++ b/tests/Unit/ServerRepositoryPhpPidsTest.php @@ -0,0 +1,75 @@ +rSettingsBackup = $GLOBALS['rSettings'] ?? null; + $this->rSchemeBackup = $_SERVER['REQUEST_SCHEME'] ?? null; + $GLOBALS['rSettings'] = ['live_streaming_pass' => 'x']; + + $this->db = new TestDb(); + $this->db->exec( + 'CREATE TABLE servers (id INTEGER PRIMARY KEY, server_type INTEGER, is_main INTEGER, enabled INTEGER, status INTEGER, last_check_ago INTEGER, parent_id TEXT, domain_name TEXT, server_ip TEXT, private_ip TEXT, enable_https INTEGER, http_broadcast_port INTEGER, https_broadcast_port INTEGER, rtmp_port INTEGER, geoip_countries TEXT, isp_names TEXT, watchdog_data TEXT, php_pids TEXT, `order` INTEGER);' + ); + $this->db->query( + 'INSERT INTO servers (id, server_type, is_main, enabled, status, last_check_ago, parent_id, domain_name, server_ip, private_ip, enable_https, http_broadcast_port, https_broadcast_port, rtmp_port, geoip_countries, isp_names, watchdog_data, php_pids, `order`) VALUES + (7, 0, 0, 1, 1, ?, NULL, "", "10.0.0.7", "", 0, 80, 443, 8880, "", "", "{}", "[11,12]", 0);', + time() + ); + ServerRepository::setDb($this->db); + } + + protected function tearDown(): void { + FileCache::delCache('servers'); + $GLOBALS['rSettings'] = $this->rSettingsBackup; + if ($this->rSchemeBackup === null) { + unset($_SERVER['REQUEST_SCHEME']); + } else { + $_SERVER['REQUEST_SCHEME'] = $this->rSchemeBackup; + } + } + + public function testReturnedRowsCarryNoPhpPids(): void { + $rServers = ServerRepository::getAll(true); + + $this->assertArrayHasKey(7, $rServers); + $this->assertSame('10.0.0.7', $rServers[7]['server_ip'], 'the row itself is still loaded'); + $this->assertArrayNotHasKey('php_pids', $rServers[7]); + } + + public function testCachedRowsCarryNoPhpPids(): void { + ServerRepository::getAll(true); + + $rCache = FileCache::getCache('servers'); + $this->assertIsArray($rCache); + $this->assertArrayHasKey(7, $rCache); + $this->assertArrayNotHasKey('php_pids', $rCache[7]); + } +} diff --git a/tests/Unit/UsersCronPhpPidsTest.php b/tests/Unit/UsersCronPhpPidsTest.php new file mode 100644 index 00000000..1d0eae98 --- /dev/null +++ b/tests/Unit/UsersCronPhpPidsTest.php @@ -0,0 +1,40 @@ +rDbBackup = $GLOBALS['db'] ?? null; + + $rDb = new TestDb(); + $rDb->exec('CREATE TABLE servers (id INTEGER PRIMARY KEY, php_pids TEXT);'); + $rDb->exec('INSERT INTO servers (id, php_pids) VALUES (1, \'[11,12]\'), (2, NULL), (3, \'not json\'), (4, \'["13"]\');'); + $GLOBALS['db'] = $rDb; + } + + protected function tearDown(): void { + $GLOBALS['db'] = $this->rDbBackup; + } + + public function testReadsEveryNodesWorkerPidsFromTheServersTable(): void { + $rMethod = new ReflectionMethod(UsersCronJob::class, 'loadPHPPIDs'); + $rMethod->setAccessible(true); + + $this->assertSame( + [1 => [11, 12], 2 => [], 3 => [], 4 => [13]], + $rMethod->invoke($rMethod->isStatic() ? null : new UsersCronJob()) + ); + } +}