From 269bee01482b592182c21d7d2afeb3c6e840f02a Mon Sep 17 00:00:00 2001 From: root Date: Fri, 11 Sep 2026 10:30:07 +0000 Subject: [PATCH] feat(streams): say which producer runs a stream, and never hand remux to an old daemon MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three things an operator could not see, all in the file they already open: - the command handed to the supervisor is recorded beside the stream's files the way the self-launched path records its ffmpeg line — _.fanout for the native remuxer, _.ffmpeg for ffmpeg (in auto, both: the second is the fallback). A supervised stream used to leave no record at all. - when the native backend is on and a stream runs ffmpeg anyway, the reason is appended to .errors ("[panel] ffmpeg runs this stream: Generate PTS is on"). isNativeEligible() becomes nativeRefusal(), returning that sentence instead of a bare false, because the answer is always one of these settings. - the panel only composes `xc_fanout remux` when the node's daemon advertises it (features in GET /monitors/state, FanoutClient::supportsRemux). An older binary does not reject that command, it misparses it — "remux" reads as a positional argument, the process tries to become a second daemon on sockets the running one holds, and the stream never starts. On a node whose panel was updated first, streams now keep running ffmpeg and say so. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01K48c64npichw9ZCZDU16ja --- docs/en/development/streaming-subsystem.md | 12 ++- src/Domain/Stream/StreamProcess.php | 104 ++++++++++++++++---- src/Streaming/Fanout/FanoutClient.php | 27 ++++- tests/Unit/StreamProcessSupervisionTest.php | 18 +++- 4 files changed, 136 insertions(+), 25 deletions(-) diff --git a/docs/en/development/streaming-subsystem.md b/docs/en/development/streaming-subsystem.md index 424636da..70ac16f1 100644 --- a/docs/en/development/streaming-subsystem.md +++ b/docs/en/development/streaming-subsystem.md @@ -240,9 +240,17 @@ database write; the daemon runs what it is handed. built by `StreamProcess::buildNativeLive()` beside `buildLive()`: it reads the source natively (MPEG-TS over http(s), HLS with TS segments, udp/rtp) and writes the same on-disk HLS and daemon feed as ffmpeg's `-f tee` line, with no ffmpeg. Which streams qualify is - `StreamProcess::isNativeEligible()` / `isNativeSource()`; `fanout_source_backend` decides: + `StreamProcess::nativeRefusal()` / `isNativeSource()`; `fanout_source_backend` decides: `auto` = remuxer with the ffmpeg command as `fallback_cmd` (used when the remuxer exits 3, - "cannot serve this source"), `native` = remuxer only, `ffmpeg` = ffmpeg only. + "cannot serve this source"), `native` = remuxer only, `ffmpeg` = ffmpeg only. The panel only + writes a remuxer command when the node's daemon advertises it (`features` in + `GET /monitors/state`, `FanoutClient::supportsRemux()`) — an older binary would misparse it. +- **Which producer ran, and why** — the command handed over is recorded beside the stream's + files like the self-launched path's `_.ffmpeg`: `_.fanout` for the remuxer, + `_.ffmpeg` for ffmpeg (in `auto`, both). When the native backend is on and a stream runs + ffmpeg anyway, `StreamProcess::nativeRefusal()`'s reason is appended to `.errors` + (`[panel] ffmpeg runs this stream: Generate PTS is on`), the same file the producer's stderr + goes to. - **Reconcile** — the daemon cannot write the database, so `StreamProcess::reconcileSupervised()` copies its state into `streams_servers` (status, pid, current source, codecs, resolution, measured bitrate): every `cron:streams` pass, and every 5 s from the `signals` daemon. A diff --git a/src/Domain/Stream/StreamProcess.php b/src/Domain/Stream/StreamProcess.php index 27e4ac7e..a92ff340 100644 --- a/src/Domain/Stream/StreamProcess.php +++ b/src/Domain/Stream/StreamProcess.php @@ -1045,41 +1045,51 @@ class StreamProcess { } /** - * Whether a live stream's configuration is a pure copy the native remuxer - * reproduces: one MPEG-TS source passed through, with nothing configured that - * needs ffmpeg in the path. PURE. Anything unrecognised falls to ffmpeg — a - * refusal costs an ffmpeg process, a wrong acceptance a channel served wrong. + * Why the native remuxer cannot serve this live stream — or null when it can: + * one MPEG-TS source passed through, with nothing configured that needs ffmpeg + * in the path. PURE. Anything unrecognised falls to ffmpeg — a refusal costs + * an ffmpeg process, a wrong acceptance a channel served wrong. + * + * The reason is a sentence, not a flag, because it is written to the stream's + * log: "this channel runs ffmpeg" is the question operators ask of a panel + * with the native backend on, and the answer is always one of these settings. * * @param array $rStreamInfo streams ⨝ streams_types row. * @param array $rArgs Stream arguments keyed by argument_key. - * @return bool + * @return string|null */ - private static function isNativeEligible(array $rStreamInfo, array $rArgs): bool { + private static function nativeRefusal(array $rStreamInfo, array $rArgs): ?string { if (($rStreamInfo['type_key'] ?? '') !== 'live_streams') { - return false; // radio and created channels are ffmpeg's + return 'not a live channel'; // radio and created channels are ffmpeg's } - if (intval($rStreamInfo['enable_transcode'] ?? 0) === 1 || !empty($rStreamInfo['custom_ffmpeg'])) { - return false; // re-encoding + if (intval($rStreamInfo['enable_transcode'] ?? 0) === 1) { + return 'transcoding is enabled'; + } + if (!empty($rStreamInfo['custom_ffmpeg'])) { + return 'the stream has a custom ffmpeg command'; } if (!empty($rStreamInfo['custom_map'])) { - return false; // a stream selection the remuxer (which copies every PID) cannot honour + return 'the stream maps specific tracks'; // the remuxer copies every PID } if (intval($rStreamInfo['rtmp_output'] ?? 0) === 1) { - return false; // FLV output + return 'RTMP (FLV) output is enabled'; } $rPush = json_decode((string) ($rStreamInfo['external_push'] ?? ''), true); if (is_array($rPush) && !empty($rPush[SERVER_ID])) { - return false; // external RTMP pushes + return 'the stream is pushed to an external server'; } // Asked for timestamp repair or realtime pacing: the source is not a // clean live feed, and passing its bytes through unchanged is not enough. - if (intval($rStreamInfo['gen_timestamps'] ?? 0) === 1 || intval($rStreamInfo['read_native'] ?? 0) === 1) { - return false; + if (intval($rStreamInfo['gen_timestamps'] ?? 0) === 1) { + return 'Generate PTS is on'; + } + if (intval($rStreamInfo['read_native'] ?? 0) === 1) { + return 'Read Native is on'; } if (!empty($rArgs['force_input_acodec']['value'])) { - return false; // re-interprets the audio: an ffmpeg input option + return 'an input audio codec is forced'; // re-interprets the audio } - return true; + return null; } /** @@ -1228,7 +1238,21 @@ class StreamProcess { $rArgsByKey[$rArg['argument_key']] = $rArg; } $rBackend = (string) ($rSettings['fanout_source_backend'] ?? 'auto'); - $rNativeStream = $rBackend !== 'ffmpeg' && self::isNativeEligible($rInfo, $rArgsByKey); + $rNativeStream = false; + if ($rBackend !== 'ffmpeg') { + $rRefusal = self::nativeRefusal($rInfo, $rArgsByKey); + if ($rRefusal === null && !FanoutClient::supportsRemux()) { + // The node's daemon predates `xc_fanout remux`. Handing it the + // command would not fail cleanly — it would start a process that + // tries to be a second daemon — so this stream stays on ffmpeg + // until the binary is updated. + $rRefusal = 'this node\'s xc_fanout has no native remuxer (update the daemon binary)'; + } + $rNativeStream = $rRefusal === null; + if ($rRefusal !== null) { + self::noteProducer($rStreamID, 'ffmpeg runs this stream: ' . $rRefusal); + } + } $rPriority = !empty($rSettings['priority_backup']) && count($rSources) > 1 && !$rLoopback; $rSpecSources = array(); @@ -1260,6 +1284,9 @@ class StreamProcess { )); $rEntry = array('label' => $rLabels[$i], 'cmd' => $rFFMPEG); + if ($rNativeStream && !self::isNativeSource($rStreamSource)) { + self::noteProducer($rStreamID, 'ffmpeg runs source #' . $i . ': ' . strtolower((string) parse_url($rStreamSource, PHP_URL_SCHEME)) . ':// is not a scheme the remuxer reads'); + } if ($rNativeStream && self::isNativeSource($rStreamSource)) { $rNativeArgs = array(); foreach ($rArguments as $rArg) { @@ -1410,10 +1437,53 @@ class StreamProcess { FanoutClient::release($rStreamID); return false; } + self::recordCommand($rStreamID, $rSpec); self::updateStream($rStreamID); return true; } + /** + * Record the command(s) handed to the supervisor beside the stream's files, + * the way the self-launched path records its ffmpeg line in `_.ffmpeg`: + * the native remuxer's goes to `_.fanout`, ffmpeg's (the command itself, + * or the fallback the supervisor switches to in `auto`) to `_.ffmpeg`. + * Purely a forensic record — nothing reads these back — but it is the first + * thing anyone opens when a channel misbehaves, and a supervised stream used + * to leave none. Both are removed with the rest of `_*` when it stops. + */ + private static function recordCommand(int $rStreamID, array $rSpec): void { + $rCmd = (string) ($rSpec['sources'][0]['cmd'] ?? ''); + $rFanout = STREAMS_PATH . $rStreamID . '_.fanout'; + $rFFMPEG = STREAMS_PATH . $rStreamID . '_.ffmpeg'; + if (self::isRemuxCommand($rCmd)) { + @file_put_contents($rFanout, $rCmd); + $rFallback = (string) ($rSpec['sources'][0]['fallback_cmd'] ?? ''); + if ($rFallback !== '') { + @file_put_contents($rFFMPEG, $rFallback); + } else { + @unlink($rFFMPEG); + } + return; + } + @file_put_contents($rFFMPEG, $rCmd); + @unlink($rFanout); + } + + /** + * Append a panel-side line to the stream's error log — the file the producer's + * own stderr goes to, and the one an operator opens. Used for the decisions + * that happen before any producer exists, above all "why is this channel on + * ffmpeg when the native backend is on". + */ + private static function noteProducer(int $rStreamID, string $rLine): void { + @file_put_contents(STREAMS_PATH . $rStreamID . '.errors', date('Y/m/d H:i:s') . ' [panel] ' . $rLine . "\n", FILE_APPEND); + } + + /** Whether a supervisor command line is the daemon's native remuxer. */ + private static function isRemuxCommand(string $rCmd): bool { + return strpos($rCmd, ' remux ') !== false && strpos($rCmd, FanoutClient::binaryPath()) !== false; + } + /** * End this stream's running producer, if it has one — or, with $rKeepAdoptable, * only if the daemon could not adopt it: adoption needs the producer's command diff --git a/src/Streaming/Fanout/FanoutClient.php b/src/Streaming/Fanout/FanoutClient.php index 906513e1..a1a9774f 100644 --- a/src/Streaming/Fanout/FanoutClient.php +++ b/src/Streaming/Fanout/FanoutClient.php @@ -28,6 +28,9 @@ class FanoutClient { /** Default source User-Agent, matching ProxyCommand::startProxy(). */ private const DEFAULT_UA = 'Mozilla/5.0'; + /** What the last monitorStates() call reported the daemon can be handed. */ + private static ?array $features = null; + /** * Build the daemon source config from a `streams` row and its keyed * `streams_arguments` (as ProxyCommand reads them). Pure function — no I/O — @@ -495,7 +498,29 @@ class FanoutClient { if (!is_array($rData) || !isset($rData['streams']) || !is_array($rData['streams'])) { return null; } - return array('accepting' => !empty($rData['accepting']), 'daemon_pid' => intval($rData['daemon_pid'] ?? 0), 'streams' => $rData['streams']); + $rFeatures = (isset($rData['features']) && is_array($rData['features'])) ? array_map('strval', $rData['features']) : array(); + self::$features = $rFeatures; + return array('accepting' => !empty($rData['accepting']), 'daemon_pid' => intval($rData['daemon_pid'] ?? 0), 'features' => $rFeatures, 'streams' => $rData['streams']); + } + + /** + * Whether the running daemon understands `xc_fanout remux` — the native + * remuxer command the panel composes for a copy-only stream. + * + * A daemon from before it does not reject such a command, it MISPARSES it: + * `remux` reads as a positional argument to the daemon's own flag set, the + * process tries to become a second daemon on sockets the running one holds, + * and the stream never starts. So a panel that is newer than the node's + * binary must ask first — on a half-upgraded node the streams simply keep + * running ffmpeg, which is the whole point of asking. + * + * @return bool False when the daemon has not been asked yet, or says no. + */ + public static function supportsRemux(): bool { + if (self::$features === null) { + self::monitorStates(); + } + return is_array(self::$features) && in_array('remux', self::$features, true); } /** diff --git a/tests/Unit/StreamProcessSupervisionTest.php b/tests/Unit/StreamProcessSupervisionTest.php index 10a4e346..1aa96182 100644 --- a/tests/Unit/StreamProcessSupervisionTest.php +++ b/tests/Unit/StreamProcessSupervisionTest.php @@ -6,7 +6,7 @@ use XcVm\Domain\Stream\StreamProcess; /** * The pure parts of handing a live stream to the xc_fanout supervisor: the * native remuxer command (buildNativeLive), which streams may use it - * (isNativeEligible / isNativeSource), the policy and health the PHP monitor + * (nativeRefusal / isNativeSource), the policy and health the PHP monitor * obeyed (supervisorPolicy / supervisorHealth), and what a supervisor state * means for streams_servers (supervisedRowUpdate). */ @@ -111,7 +111,7 @@ final class StreamProcessSupervisionTest extends TestCase { } public function testPlainCopyStreamIsNativeEligible(): void { - $this->assertTrue(self::call('isNativeEligible', $this->plainStream(), [])); + $this->assertNull(self::call('nativeRefusal', $this->plainStream(), [])); } public function testAnythingNeedingFfmpegIsNotNativeEligible(): void { @@ -126,10 +126,18 @@ final class StreamProcessSupervisionTest extends TestCase { 'radio' => ['type_key' => 'radio_streams'], 'created channel' => ['type_key' => 'created_live'], ] as $rWhy => $rOverride) { - $this->assertFalse(self::call('isNativeEligible', $this->plainStream($rOverride), []), $rWhy); + // The refusal is written to the stream's log, so every one of them + // must say something an operator can act on. + $this->assertIsString(self::call('nativeRefusal', $this->plainStream($rOverride), []), $rWhy); } - $this->assertTrue(self::call('isNativeEligible', $this->plainStream(['external_push' => json_encode([7 => ['rtmp://push/x']])]), []), "another server's push is not this one's"); - $this->assertFalse(self::call('isNativeEligible', $this->plainStream(), ['force_input_acodec' => ['value' => 'ac3']]), 'forced input codec'); + $this->assertNull(self::call('nativeRefusal', $this->plainStream(['external_push' => json_encode([7 => ['rtmp://push/x']])]), []), "another server's push is not this one's"); + $this->assertIsString(self::call('nativeRefusal', $this->plainStream(), ['force_input_acodec' => ['value' => 'ac3']]), 'forced input codec'); + } + + public function testRemuxCommandIsRecognisedForTheCommandRecord(): void { + $rCmd = $this->native(['binary' => \XcVm\Streaming\Fanout\FanoutClient::binaryPath()]); + $this->assertTrue(self::call('isRemuxCommand', $rCmd), 'the remuxer command is recorded as _.fanout'); + $this->assertFalse(self::call('isRemuxCommand', '/bin/ffmpeg40 -i x -c copy -f tee y'), 'an ffmpeg command is not'); } public function testNativeSources(): void {