diff --git a/docs/adr/0004-cluster-api.md b/docs/adr/0004-cluster-api.md index 1ebab05d..c4ac8859 100644 --- a/docs/adr/0004-cluster-api.md +++ b/docs/adr/0004-cluster-api.md @@ -1,6 +1,6 @@ # ADR 0004 — Cluster API between MAIN and load balancers: the panel's contract -- **Status:** Accepted. Phase 0 (seams), Phase 1 (crypto contract, schema, settings) and Phase 2's API, Go agent and SSH enrolment of new LBs (below) are implemented. Enrolling existing LBs over SSH (`server:enrol`), `token_rekey` and enrolment by code are too. The admin page *Servers → Cluster Nodes* and `cron:cluster` are too. Phase 3 is under way: authoritative telemetry is in, the 1 s liveness loop is not. Phases 4–11 are not. +- **Status:** Accepted. Phase 0 (seams), Phase 1 (crypto contract, schema, settings) and Phase 2's API, Go agent and SSH enrolment of new LBs (below) are implemented. Enrolling existing LBs over SSH (`server:enrol`), `token_rekey` and enrolment by code are too. The admin page *Servers → Cluster Nodes* and `cron:cluster` are too. Phase 3's authoritative telemetry and 1 s liveness loop are in; its MAIN endpoint-change handling is not. Phases 4–11 are not. - **Date:** 2026-09-25 - **Plan:** `docs/superpowers/specs/2026-09-21-main-lb-api-communication-design.md` (MAIN ↔ LB API communication, revision 3 plus corrections). - **Extension side:** `xcvm_core` ADR-002, "Cluster API: the extension's half of MAIN ↔ LB communication", cluster API version 1. @@ -232,6 +232,24 @@ The node learns its mode and flows from MAIN's authenticated replies. The agent A stopped agent removes `flows.json`, so the node falls back to the legacy paths. +### Liveness (Phase 3) + +`LivenessService::tick()` runs every second in MAIN's signals daemon, with `cron:cluster` as the per-minute fallback. It judges the active nodes whose TELEMETRY flow is on by their silence (`NodeHealth`): + +| State | When | Effect on routing | +| --- | --- | --- | +| `ok` | heard within 10 s | online, whatever the legacy `last_check_ago` says | +| `suspect` | silent over 10 s | still online; capacity weight doubled | +| `offline` | silent over `cluster_offline_after_sec` (30 s) | offline | + +Other rules: + +- A node never heard from counts as offline only once the loop has been up for that long. +- The result goes to `tmp/cluster/health.json`. `Core\Cluster\ClusterHealth` is how `ServerRepository::getAll()` (`server_online`, `cluster_health`) and `ConnectionTracker::getCapacity()` read it. +- Each transition rewrites the servers cache at once and is audited (`node.health`). +- **Fleet silence guard:** when over half of those nodes, and at least two, are silent together, MAIN suspects itself. It holds every node at its last published state instead of marking any offline. It audits `cluster.fleet_silence`, and the Cluster Nodes page shows an alert until the silence clears. +- Nodes without the flow keep the legacy 90 s rule. The Phase 6 orphan purge at `cluster_orphan_conn_ttl_sec` is not part of this loop yet. + ### Extension updates `console.php xcvm_core` rolls back an update whose cluster API falls outside the panel's range when the installed one was inside it. `console.php xcvm_core status` reports what is loaded. Installing an exact pinned version needs versioned paths in the binaries repo; that prerequisite is still open. diff --git a/src/Cli/Commands/SignalsCommand.php b/src/Cli/Commands/SignalsCommand.php index ce321677..76f5a0af 100644 --- a/src/Cli/Commands/SignalsCommand.php +++ b/src/Cli/Commands/SignalsCommand.php @@ -4,7 +4,9 @@ namespace XcVm\Cli\Commands; use XcVm\Cli\CommandInterface; use XcVm\Cli\DaemonTrait; +use XcVm\Core\Cluster\NodeRole; use XcVm\Core\Config\SettingsManager; +use XcVm\Domain\Cluster\LivenessService; use XcVm\Domain\Server\ServerRepository; use XcVm\Domain\Stream\StreamProcess; use XcVm\Infrastructure\Redis\RedisManager; @@ -52,6 +54,8 @@ class SignalsCommand implements CommandInterface { $rServers = ServerRepository::getAll(); $rLastReconcile = 0; + $rLastLiveness = 0; + $rIsMain = NodeRole::isMain(); while ($db && $db->ping()) { if (!$this->refreshOrBreak()) { @@ -79,6 +83,19 @@ class SignalsCommand implements CommandInterface { StreamProcess::reconcileSupervised(); } + // MAIN's liveness loop for nodes whose agent reports telemetry: + // once a second, and each transition rewrites the servers cache. + if ($rIsMain && time() !== $rLastLiveness && SettingsManager::get('cluster_api_enabled')) { + $rLastLiveness = time(); + try { + if (LivenessService::tick(max(10, min(300, intval(SettingsManager::get('cluster_offline_after_sec') ?: 30)))) !== []) { + $rServers = $this->refreshServers(); + } + } catch (\Throwable $rE) { + echo 'Liveness: ' . $rE->getMessage() . "\n"; + } + } + // ── Kill-сигналы из БД ────────────────────────────── if ($db->query('SELECT `signal_id`, `pid`, `rtmp` FROM `signals` WHERE `server_id` = ? AND `pid` IS NOT NULL ORDER BY `signal_id` ASC LIMIT 100', SERVER_ID)) { if ($db->num_rows() > 0) { diff --git a/src/Cli/CronJobs/ClusterCronJob.php b/src/Cli/CronJobs/ClusterCronJob.php index 6c96759b..749a3805 100644 --- a/src/Cli/CronJobs/ClusterCronJob.php +++ b/src/Cli/CronJobs/ClusterCronJob.php @@ -8,15 +8,18 @@ use XcVm\Core\Cluster\NodeRole; use XcVm\Core\Config\SettingsManager; use XcVm\Domain\Cluster\ClusterAudit; use XcVm\Domain\Cluster\EnrolCodeService; +use XcVm\Domain\Cluster\LivenessService; use XcVm\Domain\Cluster\NonceStore; use XcVm\Domain\Cluster\TokenService; +use XcVm\Domain\Server\ServerRepository; /** * ClusterCronJob — MAIN's cluster API housekeeping, every minute: * * - expired token epochs are deleted, which erases their `z`; * - the replay cache and single-use challenges past their 180 s go; - * - enrolment codes nobody used, and decided requests after a day, go. + * - enrolment codes nobody used, and decided requests after a day, go; + * - the liveness loop runs once (the signals daemon runs it every second). * * The crontab row (`cluster`, role `main`) is copied to load balancers with * the rest; there the job returns before touching anything, as the cluster @@ -55,6 +58,12 @@ class ClusterCronJob implements CommandInterface { 'epochs' => static fn() => TokenService::prune(), 'nonces' => static fn() => NonceStore::purge(), 'enrol_codes' => static fn() => EnrolCodeService::prune(), + // The signals daemon runs this every second; the minute is its fallback. + 'liveness' => static function () { + if (LivenessService::tick(max(10, min(300, intval(SettingsManager::get('cluster_offline_after_sec') ?: 30)))) !== []) { + ServerRepository::getAll(true); + } + }, ] as $rStep => $rRun) { try { $rRun(); diff --git a/src/Core/Cluster/ClusterHealth.php b/src/Core/Cluster/ClusterHealth.php new file mode 100644 index 00000000..d1f56d76 --- /dev/null +++ b/src/Core/Cluster/ClusterHealth.php @@ -0,0 +1,75 @@ +, guard: bool}|null */ + private static ?array $rCache = null; + + private static float $rReadAt = 0.0; + + private static ?string $rPath = null; + + /** @return 'ok'|'suspect'|'offline'|null null when the node is not judged by the loop. */ + public static function state(int $rServerID): ?string { + return self::read()['states'][$rServerID] ?? null; + } + + /** Capacity weight: 2 for a suspect node, else 1. */ + public static function weight(int $rServerID): float { + return self::state($rServerID) === 'suspect' ? 2.0 : 1.0; + } + + /** @return array{states: array, guard: bool} */ + public static function read(): array { + if (self::$rCache === null || microtime(true) - self::$rReadAt >= 1.0) { + $rDoc = json_decode((string) @file_get_contents(self::path()), true); + $rStates = []; + foreach ((is_array($rDoc['states'] ?? null) ? $rDoc['states'] : []) as $rID => $rState) { + if (in_array($rState, ['ok', 'suspect', 'offline'], true)) { + $rStates[(int) $rID] = $rState; + } + } + self::$rCache = ['states' => $rStates, 'guard' => !empty($rDoc['guard'])]; + self::$rReadAt = microtime(true); + } + return self::$rCache; + } + + /** @param array $rStates */ + public static function write(array $rStates, bool $rGuard): void { + $rPath = self::path(); + if (!is_dir(dirname($rPath))) { + @mkdir(dirname($rPath), 0750, true); + } + $rTmp = $rPath . '.tmp'; + if (@file_put_contents($rTmp, (string) json_encode(['states' => $rStates, 'guard' => $rGuard]), LOCK_EX) !== false) { + @rename($rTmp, $rPath); + } + self::$rCache = ['states' => $rStates, 'guard' => $rGuard]; + self::$rReadAt = microtime(true); + } + + /** Tests: use another file, and forget what was read. */ + public static function usePath(?string $rPath): void { + self::$rPath = $rPath; + self::$rCache = null; + } + + private static function path(): string { + return self::$rPath ?? ((defined('TMP_PATH') ? TMP_PATH : sys_get_temp_dir() . '/') . 'cluster/health.json'); + } +} diff --git a/src/Core/Localization/lang/ar.ini b/src/Core/Localization/lang/ar.ini index e9b20b10..4791028a 100644 --- a/src/Core/Localization/lang/ar.ini +++ b/src/Core/Localization/lang/ar.ini @@ -1881,6 +1881,7 @@ cluster_flow_off = "معطّل" cluster_telemetry_help = "تدفق القياس عن بُعد: عند تفعيله يأخذ MAIN إحصاءات هذه العقدة من وكيلها، ويتوقف watchdog وإحصاءات cron:servers و network.py على العقدة." cluster_telemetry_on_done = "تم تفعيل تدفق القياس: وكيل العقدة يرسل الإحصاءات الآن." cluster_telemetry_off_done = "تم تعطيل تدفق القياس: يعود watchdog العقدة إلى الإرسال." +cluster_fleet_silence = "صمتت معظم العقد في وقت واحد: يشك MAIN في شبكته أو في مجمع الواجهة، ويحتفظ بحالة كل عقدة دون وضعها غير متصلة حتى ينتهي الصمت." restreamer_bypass_proxy = "وكيل تجاوز Restreamer" channel_sorting_type = "نوع فرز القناة" sort_vod_by_date = "فرز VOD حسب التاريخ" diff --git a/src/Core/Localization/lang/bg.ini b/src/Core/Localization/lang/bg.ini index 041cc9e5..6dac7b52 100644 --- a/src/Core/Localization/lang/bg.ini +++ b/src/Core/Localization/lang/bg.ini @@ -1603,6 +1603,7 @@ cluster_flow_off = "Изкл." cluster_telemetry_help = "Поток на телеметрията: при включване MAIN взема статистиките на възела от агента му, а watchdog, статистиките на cron:servers и network.py на възела спират." cluster_telemetry_on_done = "Потокът на телеметрията е включен: агентът на възела вече отчита статистиките." cluster_telemetry_off_done = "Потокът на телеметрията е изключен: watchdog на възела отново отчита." +cluster_fleet_silence = "Повечето възли замлъкнаха едновременно: MAIN подозира собствената си мрежа или API пула и запазва състоянието на всеки възел, без да го маркира офлайн, докато тишината отмине." fanout_debug = "Дебъг дневник Fanout" fanout_debug_off = "Изкл." fanout_debug_all = "Всички категории" diff --git a/src/Core/Localization/lang/de.ini b/src/Core/Localization/lang/de.ini index a78eb1c8..daf6a633 100644 --- a/src/Core/Localization/lang/de.ini +++ b/src/Core/Localization/lang/de.ini @@ -1602,6 +1602,7 @@ cluster_flow_off = "Aus" cluster_telemetry_help = "Telemetrie-Flow: an, MAIN bezieht die Statistiken dieses Knotens von seinem Agenten; Watchdog, cron:servers-Statistiken und network.py des Knotens pausieren." cluster_telemetry_on_done = "Telemetrie-Flow an: Der Agent des Knotens meldet jetzt seine Statistiken." cluster_telemetry_off_done = "Telemetrie-Flow aus: Der Watchdog des Knotens meldet wieder." +cluster_fleet_silence = "Die meisten Knoten sind gleichzeitig verstummt: MAIN vermutet sein eigenes Netz oder den API-Pool und hält den Zustand jedes Knotens, statt ihn offline zu setzen, bis die Stille endet." fanout_debug = "Fanout-Debug-Protokoll" fanout_debug_off = "Aus" fanout_debug_all = "Alle Kategorien" diff --git a/src/Core/Localization/lang/en.ini b/src/Core/Localization/lang/en.ini index ef26f318..5a8e5e4a 100644 --- a/src/Core/Localization/lang/en.ini +++ b/src/Core/Localization/lang/en.ini @@ -1892,6 +1892,7 @@ cluster_flow_off = "Off" cluster_telemetry_help = "Telemetry flow: when on, MAIN takes this node's stats from its agent and the node's watchdog, cron:servers stats and network.py stand down." cluster_telemetry_on_done = "Telemetry flow on: the node's agent now reports its stats." cluster_telemetry_off_done = "Telemetry flow off: the node's watchdog reports again." +cluster_fleet_silence = "Most nodes went silent at once: MAIN suspects its own network or API pool, and holds every node's state instead of marking it offline until the silence clears." fanout_debug = "Fanout Debug Log" fanout_debug_off = "Off" fanout_debug_all = "All categories" diff --git a/src/Core/Localization/lang/es.ini b/src/Core/Localization/lang/es.ini index ccc11400..ca24591f 100644 --- a/src/Core/Localization/lang/es.ini +++ b/src/Core/Localization/lang/es.ini @@ -1609,6 +1609,7 @@ cluster_flow_off = "Desactivada" cluster_telemetry_help = "Flujo de telemetría: activado, MAIN toma las estadísticas de este nodo de su agente y el watchdog, las estadísticas de cron:servers y network.py del nodo se detienen." cluster_telemetry_on_done = "Flujo de telemetría activado: el agente del nodo informa ahora sus estadísticas." cluster_telemetry_off_done = "Flujo de telemetría desactivado: el watchdog del nodo vuelve a informar." +cluster_fleet_silence = "La mayoría de los nodos quedaron en silencio a la vez: MAIN sospecha de su propia red o del pool de la API y mantiene el estado de cada nodo, sin marcarlos offline, hasta que pase el silencio." fanout_debug = "Registro de depuración Fanout" fanout_debug_off = "Desactivado" fanout_debug_all = "Todas las categorías" diff --git a/src/Core/Localization/lang/fr.ini b/src/Core/Localization/lang/fr.ini index 01b16d3c..7899f197 100644 --- a/src/Core/Localization/lang/fr.ini +++ b/src/Core/Localization/lang/fr.ini @@ -1602,6 +1602,7 @@ cluster_flow_off = "Désactivée" cluster_telemetry_help = "Flux de télémétrie : activé, MAIN reçoit les statistiques de ce nœud de son agent, et le watchdog, les statistiques de cron:servers et network.py du nœud s'arrêtent." cluster_telemetry_on_done = "Flux de télémétrie activé : l'agent du nœud rapporte désormais ses statistiques." cluster_telemetry_off_done = "Flux de télémétrie désactivé : le watchdog du nœud rapporte de nouveau." +cluster_fleet_silence = "La plupart des nœuds se sont tus en même temps : MAIN soupçonne son propre réseau ou le pool de l'API et conserve l'état de chaque nœud, sans les marquer hors ligne, jusqu'à la fin du silence." fanout_debug = "Journal de débogage Fanout" fanout_debug_off = "Désactivé" fanout_debug_all = "Toutes les catégories" diff --git a/src/Core/Localization/lang/pt.ini b/src/Core/Localization/lang/pt.ini index 9b0b22e8..bb7a3034 100644 --- a/src/Core/Localization/lang/pt.ini +++ b/src/Core/Localization/lang/pt.ini @@ -1602,6 +1602,7 @@ cluster_flow_off = "Desligada" cluster_telemetry_help = "Fluxo de telemetria: ligado, o MAIN recebe as estatísticas deste nó do agente, e o watchdog, as estatísticas do cron:servers e o network.py do nó param." cluster_telemetry_on_done = "Fluxo de telemetria ligado: o agente do nó passa a reportar as estatísticas." cluster_telemetry_off_done = "Fluxo de telemetria desligado: o watchdog do nó volta a reportar." +cluster_fleet_silence = "A maioria dos nós ficou em silêncio ao mesmo tempo: o MAIN suspeita da sua própria rede ou do pool da API e mantém o estado de cada nó, sem os marcar como offline, até o silêncio passar." fanout_debug = "Registo de depuração Fanout" fanout_debug_off = "Desligado" fanout_debug_all = "Todas as categorias" diff --git a/src/Core/Localization/lang/ru.ini b/src/Core/Localization/lang/ru.ini index 9f7b69db..a83649aa 100644 --- a/src/Core/Localization/lang/ru.ini +++ b/src/Core/Localization/lang/ru.ini @@ -1602,6 +1602,7 @@ cluster_flow_off = "Выкл." cluster_telemetry_help = "Поток телеметрии: при включении MAIN берёт статистику узла от его агента, а watchdog, статистика cron:servers и network.py узла останавливаются." cluster_telemetry_on_done = "Поток телеметрии включён: статистику узла теперь передаёт агент." cluster_telemetry_off_done = "Поток телеметрии выключен: статистику снова передаёт watchdog узла." +cluster_fleet_silence = "Большинство узлов замолчали одновременно: MAIN подозревает свою сеть или пул API и сохраняет состояние каждого узла, не помечая их offline, пока тишина не закончится." fanout_debug = "Отладочный лог Fanout" fanout_debug_off = "Выкл." fanout_debug_all = "Все категории" diff --git a/src/Domain/Cluster/LivenessService.php b/src/Domain/Cluster/LivenessService.php new file mode 100644 index 00000000..c8b38d0a --- /dev/null +++ b/src/Domain/Cluster/LivenessService.php @@ -0,0 +1,75 @@ + [from, to]; + * the caller rewrites the servers cache when there are any. + * + * @return array + */ + public static function tick(int $rOfflineAfterSec): array { + $rReady = ClusterMeta::readyAtMs(); + $rNow = ClusterClock::nowMs(); + self::db()->query("SELECT `server_id`, `last_seen_at` FROM `cluster_nodes` WHERE `state` = 'active' AND `mode` >= 1 AND (`flows` & ?) <> 0;", NodeRegistry::FLOW_TELEMETRY); + $rJudged = []; + foreach (self::db()->get_rows() as $rRow) { + $rState = NodeHealth::state($rRow['last_seen_at'] === null ? null : (int) $rRow['last_seen_at'], $rReady, $rNow, $rOfflineAfterSec); + // A node that has not spoken since the flow went on is judged offline + // only once the loop has been ready long enough to have heard it. + if ($rState === 'unknown') { + $rState = $rNow - $rReady > $rOfflineAfterSec * 1000 ? 'offline' : 'suspect'; + } + $rJudged[(int) $rRow['server_id']] = $rState; + } + + $rPrev = ClusterHealth::read(); + $rSilent = count(array_filter($rJudged, static fn($rState) => $rState !== 'ok')); + $rGuard = count($rJudged) >= 2 && $rSilent >= 2 && $rSilent * 2 > count($rJudged); + if ($rGuard) { + foreach ($rJudged as $rID => $rState) { + if ($rState === 'offline' && ($rPrev['states'][$rID] ?? null) !== 'offline') { + $rJudged[$rID] = $rPrev['states'][$rID] ?? 'suspect'; + } + } + } + if ($rGuard !== $rPrev['guard']) { + ClusterAudit::log($rGuard ? 'cluster.fleet_silence' : 'cluster.fleet_silence_clear', null, ['silent' => $rSilent, 'nodes' => count($rJudged)], 'liveness'); + } + + $rTransitions = []; + foreach ($rJudged + $rPrev['states'] as $rID => $rUnused) { + $rFrom = $rPrev['states'][$rID] ?? null; + $rTo = $rJudged[$rID] ?? null; + if ($rFrom !== $rTo) { + $rTransitions[$rID] = [$rFrom, $rTo]; + if ($rTo !== null && $rFrom !== null) { + ClusterAudit::log('node.health', $rID, ['from' => $rFrom, 'to' => $rTo], 'liveness'); + } + } + } + if ($rTransitions !== [] || $rGuard !== $rPrev['guard']) { + ksort($rJudged); + ClusterHealth::write($rJudged, $rGuard); + } + return $rTransitions; + } +} diff --git a/src/Domain/Server/ServerRepository.php b/src/Domain/Server/ServerRepository.php index 601e9e3e..6cbf41c9 100644 --- a/src/Domain/Server/ServerRepository.php +++ b/src/Domain/Server/ServerRepository.php @@ -4,6 +4,7 @@ namespace XcVm\Domain\Server; use XcVm\Core\Backup\BackupService; use XcVm\Core\Cache\FileCache; +use XcVm\Core\Cluster\ClusterHealth; use XcVm\Core\Cluster\NodeRpc; use XcVm\Core\Config\SettingsManager; use XcVm\Domain\Stream\ConnectionTracker; @@ -100,6 +101,12 @@ class ServerRepository { $rRow['watchdog'] = json_decode($rRow['watchdog_data'], true); $rRow['server_online'] = $rRow['enabled'] && in_array($rRow['status'], $rOnlineStatus) && time() - $rRow['last_check_ago'] <= $rLastCheckTime || SERVER_ID == $rRow['id']; + // A node whose agent reports its telemetry is judged by MAIN's + // liveness loop instead (offline after 30 s of silence, not 90 s). + $rRow['cluster_health'] = ClusterHealth::state(intval($rRow['id'])); + if ($rRow['cluster_health'] !== null && SERVER_ID != $rRow['id']) { + $rRow['server_online'] = $rRow['enabled'] && in_array($rRow['status'], $rOnlineStatus) && $rRow['cluster_health'] !== 'offline'; + } if (!isset($rRow['order'])) { $rRow['order'] = 0; } diff --git a/src/Domain/Stream/ConnectionTracker.php b/src/Domain/Stream/ConnectionTracker.php index dfcab4d1..db59dbdf 100644 --- a/src/Domain/Stream/ConnectionTracker.php +++ b/src/Domain/Stream/ConnectionTracker.php @@ -2,6 +2,7 @@ namespace XcVm\Domain\Stream; +use XcVm\Core\Cluster\ClusterHealth; use XcVm\Core\Cluster\SignalDispatcher; use XcVm\Core\Config\SettingsManager; use XcVm\Core\Process\ProcessManager; @@ -130,6 +131,16 @@ class ConnectionTracker { } } + // A node MAIN's liveness loop finds silent for over 10 s (suspect) weighs + // double, so the balancer sends it fewer new viewers. + if (!$rProxy) { + foreach ($rRows as $rServerID => $rRow) { + if (isset($rRow['capacity'])) { + $rRows[$rServerID]['capacity'] = $rRow['capacity'] * ClusterHealth::weight((int) $rServerID); + } + } + } + if (defined('CACHE_TMP_PATH') && is_dir(CACHE_TMP_PATH) && is_writable(CACHE_TMP_PATH)) { file_put_contents(CACHE_TMP_PATH . $rFile, json_encode($rRows), LOCK_EX); } diff --git a/src/Public/Views/admin/cluster_nodes.php b/src/Public/Views/admin/cluster_nodes.php index fc80843a..e7b564b3 100644 --- a/src/Public/Views/admin/cluster_nodes.php +++ b/src/Public/Views/admin/cluster_nodes.php @@ -8,6 +8,7 @@ * (ClusterNodesController → Domain\Cluster\ClusterAdmin). */ +use XcVm\Core\Cluster\ClusterHealth; use XcVm\Core\Util\LayoutRenderer; $rBadge = static fn(string $rState): string => match ($rState) { @@ -41,6 +42,10 @@ $rWhen = static fn(?int $rTs): string => $rTs ? gmdate('Y-m-d H:i:s', $rTs) . ' + + + +
diff --git a/tests/Unit/ClusterLivenessTest.php b/tests/Unit/ClusterLivenessTest.php new file mode 100644 index 00000000..0f7717f0 --- /dev/null +++ b/tests/Unit/ClusterLivenessTest.php @@ -0,0 +1,114 @@ +rDb = new TestDb(); + $this->rDb->exec((string) preg_replace(['/^--.*$/m', '/,\s*(UNIQUE )?KEY `\w+` \([^)]*\)/', '/ unsigned| COLLATE \w+/', '/\) ENGINE=[^;]*;/'], ['', '', '', ');'], (string) file_get_contents(dirname(__DIR__, 2) . '/src/migrations/database/up/029_create_cluster_nodes.sql'))); + $this->rDb->exec('CREATE TABLE `cluster_audit` (`id` INTEGER PRIMARY KEY AUTOINCREMENT, `time` int, `server_id` int, `actor` varchar(64), `event` varchar(64), `detail` text, `ip` varchar(64))'); + DatabaseFactory::set($this->rDb); + ClusterClock::fix($this->rT0); + \XcVm\Domain\Cluster\ClusterMeta::set('ready_at', (string) ($this->rT0 - 600000)); + foreach ([5, 6, 7] as $rID) { + NodeRegistry::startEnrolment($rID, sprintf('00000000-0000-4000-a000-%012d', $rID), str_repeat("\1", 32), str_repeat("\2", 32), 1); + NodeRegistry::update($rID, ['state' => 'active', 'flows' => NodeRegistry::FLOW_TELEMETRY, 'last_seen_at' => $this->rT0]); + } + $this->rHealth = sys_get_temp_dir() . '/health_' . bin2hex(random_bytes(4)) . '.json'; + ClusterHealth::usePath($this->rHealth); + } + + protected function tearDown(): void { + ClusterClock::fix(null); + DatabaseFactory::reset(); + @unlink($this->rHealth); + ClusterHealth::usePath(null); + } + + private function at(int $rMs): array { + ClusterClock::fix($this->rT0 + $rMs); + return LivenessService::tick(30); + } + + private function audit(string $rEvent): int { + $this->rDb->query('SELECT COUNT(*) AS `n` FROM `cluster_audit` WHERE `event` = ?', $rEvent); + return (int) $this->rDb->get_row()['n']; + } + + public function testSilenceMakesANodeSuspectThenOffline(): void { + $this->assertSame([5 => [null, 'ok'], 6 => [null, 'ok'], 7 => [null, 'ok']], $this->at(0)); + $this->assertSame([], $this->at(1000), 'no transition, no rewrite'); + + // 5 and 6 keep heartbeating; 7 falls silent. + $rBeat = fn(int $rMs) => $this->rDb->query('UPDATE `cluster_nodes` SET `last_seen_at` = ? WHERE `server_id` IN (5, 6)', $this->rT0 + $rMs); + $rBeat(10000); + $this->assertSame([], $this->at(10000)); + $rBeat(11000); + $this->assertSame([7 => ['ok', 'suspect']], $this->at(11000)); + $this->assertSame('suspect', ClusterHealth::state(7)); + $this->assertSame(2.0, ClusterHealth::weight(7), 'a suspect node weighs double'); + $rBeat(31000); + $this->assertSame([7 => ['suspect', 'offline']], $this->at(31000)); + $this->assertSame(1.0, ClusterHealth::weight(7)); + + // It speaks again. + $this->rDb->query('UPDATE `cluster_nodes` SET `last_seen_at` = ? WHERE `server_id` = 7', $this->rT0 + 32000); + $rBeat(32000); + $this->assertSame([7 => ['offline', 'ok']], $this->at(32000)); + $this->assertSame(3, $this->audit('node.health'), 'ok→suspect, suspect→offline, offline→ok; the first publication is not a transition'); + } + + public function testFleetSilenceHoldsNodesInsteadOfMarkingThemOffline(): void { + $this->at(0); + // 6 and 7 of 3 fall silent together: MAIN suspects itself. + $this->rDb->query('UPDATE `cluster_nodes` SET `last_seen_at` = ? WHERE `server_id` = 5', $this->rT0 + 40000); + $this->assertSame([], $this->at(40000), 'held at ok, not marked offline'); + $this->assertTrue(ClusterHealth::read()['guard']); + $this->assertSame(1, $this->audit('cluster.fleet_silence')); + $this->assertSame('ok', ClusterHealth::state(7)); + + // They come back: the guard clears. + $this->rDb->query('UPDATE `cluster_nodes` SET `last_seen_at` = ? WHERE `server_id` IN (5, 6, 7)', $this->rT0 + 45000); + $this->at(45000); + $this->assertFalse(ClusterHealth::read()['guard']); + $this->assertSame(1, $this->audit('cluster.fleet_silence_clear')); + } + + public function testOnlyNodesWithTheFlowAreJudged(): void { + NodeRegistry::update(6, ['flows' => 0]); + NodeRegistry::update(7, ['state' => 'quarantined']); + $this->assertSame([5 => [null, 'ok']], $this->at(0)); + $this->assertNull(ClusterHealth::state(6), 'legacy rule for a node without the flow'); + + // Turning the flow off drops the node from the published states. + NodeRegistry::update(5, ['flows' => 0]); + $this->assertSame([5 => ['ok', null]], $this->at(1000)); + $this->assertNull(ClusterHealth::state(5)); + } + + public function testANodeNeverHeardIsOfflineOnceTheLoopHasWaited(): void { + $this->rDb->query('UPDATE `cluster_nodes` SET `last_seen_at` = NULL WHERE `server_id` = 5'); + $this->at(0); + $this->assertSame('offline', ClusterHealth::state(5)); + $this->rDb->query("UPDATE `cluster_meta` SET `value` = ? WHERE `name` = 'ready_at'", (string) ($this->rT0 + 5000)); + $this->rDb->query("UPDATE `cluster_nodes` SET `last_seen_at` = ? WHERE `server_id` IN (6, 7)", $this->rT0 + 10000); + $this->at(10000); + $this->assertSame('suspect', ClusterHealth::state(5), 'MAIN just restarted: not yet'); + } +} diff --git a/tests/Unit/ServerRepositoryPhpPidsTest.php b/tests/Unit/ServerRepositoryPhpPidsTest.php index 15ce584f..c7256b8d 100644 --- a/tests/Unit/ServerRepositoryPhpPidsTest.php +++ b/tests/Unit/ServerRepositoryPhpPidsTest.php @@ -105,4 +105,26 @@ final class ServerRepositoryPhpPidsTest extends TestCase { public function testGetByIdKeepsTheFullRow(): void { $this->assertSame('[11,12]', ServerRepository::getById(7)['php_pids']); } + + public function testClusterHealthDecidesOnlineForAgentNodes(): void { + $rPath = sys_get_temp_dir() . '/health_' . bin2hex(random_bytes(4)) . '.json'; + \XcVm\Core\Cluster\ClusterHealth::usePath($rPath); + try { + // Fresh last_check_ago, but MAIN's liveness loop says offline. + \XcVm\Core\Cluster\ClusterHealth::write([7 => 'offline'], false); + $this->assertFalse(ServerRepository::getAll(true)[7]['server_online']); + // A stale last_check_ago (legacy: offline after 90 s), but the loop says suspect. + $this->db->query('UPDATE servers SET last_check_ago = ? WHERE id = 7', time() - 600); + \XcVm\Core\Cluster\ClusterHealth::write([7 => 'suspect'], false); + $rServers = ServerRepository::getAll(true); + $this->assertTrue($rServers[7]['server_online']); + $this->assertSame('suspect', $rServers[7]['cluster_health']); + // Not judged by the loop: the legacy rule. + \XcVm\Core\Cluster\ClusterHealth::write([], false); + $this->assertFalse(ServerRepository::getAll(true)[7]['server_online']); + } finally { + @unlink($rPath); + \XcVm\Core\Cluster\ClusterHealth::usePath(null); + } + } }