path: root/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py

"""Video streaming for the Videos app: the credit a viewer grants, the one
stream a session holds, and the ffmpeg pipeline behind it."""

import asyncio
import logging
import time

from meshbay_common import MNP_VERSION
from meshbay_common.protocol import MNP, chunk_ciphertext

from meshbay_node import hwaccel, platform
from meshbay_node.media_probe import BROWSER_INCOMPATIBLE_VIDEO_CODECS
from meshbay_node.media_probe import probe_video as _probe_video
from meshbay_node.roots import off_disk
from meshbay_node.transport.webrtc.disk import _locate
from meshbay_node.transport.webrtc.media_tools import _seek_lands_at

log = logging.getLogger("meshbay_node.transport.webrtc_server")


# ffmpeg is spawned per stream request; without a cap any member can fork-bomb
# the node by requesting many streams at once (H6).
#
# Two was sized when a stream was a burst: the client took segments as fast as
# it could append them, so a slot was held for the minute it took to push the
# file and then came back. Now that the client only pulls ninety seconds ahead
# of the playhead, a slot is held for as long as the film runs — so two slots
# means two people can watch anything at all, and the third is refused for the
# next hour and a half. The work behind a slot has not changed and is small:
# ffmpeg runs `-c copy`, a remux with no encoding in it, and spends most of the
# film blocked on a pipe nobody is reading.
#
# This is the default, not the policy: the right number depends on the machine,
# so the operator sets `max_concurrent_streams` under [node] in node.toml. This
# value applies when they have said nothing.
MAX_CONCURRENT_TRANSCODES = 8

STREAM_SEGMENT_SIZE = 256 * 1024

# What a client may ask for in one go, and how long the node waits for it to ask
# again before deciding nobody is watching any more.
STREAM_MAX_CREDIT = 256
STREAM_CREDIT_TIMEOUT = 120
# How often that budget is re-examined. A viewer who left stops being
# charged for a slot within this, rather than within the timeout.
STREAM_CREDIT_POLL = 3


class StreamingMixin:
    def _grant_stream_credit(self, msg: dict) -> None:
        """
        The client has room for more segments.

        `n` of zero is a keepalive, not a no-op: a viewer whose buffer is
        already a minute and a half ahead of the playhead deliberately grants
        nothing, and must still be able to say it is there. Without that, the
        stall timeout below cannot tell a paused film from a closed tab.
        """
        log.debug("stream credit +%s (had %d, sent %d)",
                  msg.get("n"), self._stream_credit, self._stream_segments)
        try:
            n = int(msg.get("n", 1))
        except (TypeError, ValueError):
            n = 1
        if n == 0:
            # The fingerprint of a client that bounds its read-ahead. A client
            # that never sends one is granting credit per append — which is
            # what fills the browser's buffer ceiling and wedges the player.
            self._stream_keepalives += 1
            if self._stream_keepalives == 1:
                log.info("stream: peer is pacing itself (first keepalive at "
                         "%d segments)", self._stream_segments)
        self._stream_credit += max(0, min(n, STREAM_MAX_CREDIT))
        self._stream_heard_at = time.monotonic()
        self._stream_credit_evt.set()

    def _stop_stream(self) -> None:
        """
        The viewer was closed. Stop transcoding and let go of the slot.

        Without this the only thing that ended a stream was the credit timeout,
        so ffmpeg kept running and held one of the node's two transcode slots
        for two minutes after nobody was watching — which is how closing a video
        made the next one answer "server busy".
        """
        self._stream_stopped = True
        self._stream_credit_evt.set()

    async def _await_stream_credit(self) -> bool:
        """
        Block until the client has room. False if it stopped asking.

        Without this the node hands ffmpeg's entire output to the channel as
        fast as it is produced, and the browser holds a four gigabyte film in a
        JavaScript array while MediaSource consumes it a segment at a time.
        """
        # Measured from the last thing the peer said, not from the start of the
        # wait: a viewer that is buffered well ahead sends keepalives and grants
        # nothing for minutes at a time, and that is a watched film, not a
        # stalled one.
        self._stream_heard_at = time.monotonic()
        waiting_since = 0.0
        while self._stream_credit <= 0:
            if waiting_since == 0.0:
                waiting_since = time.monotonic()
                # Debug: a paced viewer runs out of credit between every
                # window, so this is one line per eight segments — hundreds
                # per film. It is worth having, but not by default.
                log.debug("stream: out of credit at %d segments (%.0f MB) — "
                          "waiting for the peer",
                          self._stream_segments,
                          self._stream_segments * STREAM_SEGMENT_SIZE / 1048576)
            if self._stream_stopped:
                return False
            # Checked before the wait as well as after it: a peer that vanishes
            # sends no credit and fires no event, so waiting the full timeout
            # on a channel that is already shut is pure dead time on a slot.
            if self._channel is None or self._channel.readyState != "open":
                return False
            self._stream_credit_evt.clear()
            try:
                # In slices rather than one long sleep, so a connection that
                # dies mid-wait is noticed in seconds instead of minutes. The
                # total budget is unchanged.
                await asyncio.wait_for(self._stream_credit_evt.wait(),
                                       timeout=STREAM_CREDIT_POLL)
            except TimeoutError:
                silent = time.monotonic() - self._stream_heard_at
                if silent >= STREAM_CREDIT_TIMEOUT:
                    log.info("Stream stalled: nothing from peer=%s for %.0fs",
                             (self._user_id or "?")[:8], silent)
                    return False
                continue
            if self._stream_stopped:
                return False
            if self._channel is None or self._channel.readyState != "open":
                return False
        if waiting_since:
            waited_for = time.monotonic() - waiting_since
            # Only a wait long enough to be a symptom. Normal pacing puts a
            # gap of a few seconds between windows; a minute means the viewer
            # is buffered right up and playing, or has stopped watching.
            level = log.info if waited_for >= 10 else log.debug
            level("stream: credit arrived after %.1fs", waited_for)
        self._stream_credit -= 1
        return True

    async def _replace_stream(self, msg: dict) -> None:
        """Retire this session's previous stream before starting another.

        A viewer plays one film at a time, so a second request means the first
        one is finished whatever the client managed to tell us. Relying on
        `stream_stop` alone was not enough: a browser that is backgrounded,
        reloaded or simply loses the message never sends it, and the only other
        thing that ends a stream is STREAM_CREDIT_TIMEOUT — two minutes during
        which ffmpeg keeps running and holds one of the node's two transcode
        slots.

        That is the reported failure exactly: first video fine, second fine,
        third answered "Server busy" because the first two were still holding
        both slots. The client shows that as "buffering" forever.

        Waiting for the old task is what makes the slot available: it is the
        exit of its `async with sem` that releases it.
        """
        prev = self._stream_task
        if prev is not None and not prev.done():
            t0 = time.monotonic()
            log.info("stream: retiring previous stream")
            self._stop_stream()
            try:
                await asyncio.wait_for(asyncio.shield(prev), timeout=15)
                log.info("stream: previous stream ended in %.1fs",
                         time.monotonic() - t0)
            except TimeoutError:
                log.warning("stream: previous stream STILL RUNNING after 15s")
            except Exception:
                pass   # it failed on its own; the slot is free either way
        self._stream_task = asyncio.current_task()
        await self._stream_video(msg)

    def _transcode_semaphore(self) -> asyncio.Semaphore:
        """The node's stream budget, shared across every peer.

        One ffmpeg per request with no cap lets any member exhaust the node's
        CPU and process table (H6). The semaphore lives on the transport
        context rather than the session so that it counts the node's viewers
        and not one browser's, and it is created once: rebuilding it per call
        would hand every caller its own budget and cap nothing at all.
        """
        sem = self._ctx.get("_transcode_sem")
        if sem is None:
            n = self._ctx.get("max_concurrent_streams") or MAX_CONCURRENT_TRANSCODES
            sem = asyncio.Semaphore(n)
            self._ctx["_transcode_sem"] = sem
            log.info("stream: %d concurrent viewers allowed", n)
        return sem

    async def _stream_video(self, msg: dict) -> None:
        """Stream a video file as fMP4 segments via MSE-compatible output."""
        sem = self._transcode_semaphore()
        if sem.locked() and sem._value <= 0:
            self._send({"type": "error", "detail": "Server busy, retry shortly"})
            return
        ctx = self._ctx
        log.info("stream: waiting for a slot (%d of %d in use)",
                 ctx.get("_streams_in_flight", 0), self._stream_capacity())
        async with sem:
            # Counted here rather than read back out of the semaphore's private
            # `_value`: `set_capacity` needs to know how many slots are held in
            # order to resize without letting the pool overshoot, and a number
            # this code maintains itself is one that survives the semaphore
            # object being replaced underneath it.
            ctx["_streams_in_flight"] = ctx.get("_streams_in_flight", 0) + 1
            log.info("stream: slot acquired (%d of %d in use)",
                     ctx["_streams_in_flight"], self._stream_capacity())
            try:
                await self._stream_video_inner(msg)
            finally:
                ctx["_streams_in_flight"] = max(
                    0, ctx.get("_streams_in_flight", 1) - 1)
                log.info("stream: slot released (%d of %d in use)",
                         ctx["_streams_in_flight"], self._stream_capacity())

    def _stream_capacity(self) -> int:
        return self._ctx.get("max_concurrent_streams") or MAX_CONCURRENT_TRANSCODES

    async def _stream_video_inner(self, msg: dict) -> None:
        ctx = self._group_ctx()
        file_id = msg.get("file_id", "")
        entry = ctx["index"].get_entry(file_id)
        if not entry:
            self._send({"type": "error", "detail": "File not found"})
            return

        file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry)
        if refusal is not None:
            self._send({"type": "error", "detail": refusal})
            return

        gek = ctx.get("gek")
        file_hash = bytes.fromhex(entry.id)

        try:
            probe = await _probe_video(str(file_path))
        except Exception as e:
            self._send({"type": "error", "detail": f"Probe failed: {e}"})
            return
        codec_str = probe.codec
        duration = probe.duration
        has_audio = probe.has_audio
        raw_video_codec = probe.raw_codec_name

        # No video stream at all is the only thing this path cannot serve, and
        # it is the only thing refused here. A source with no MSE codec string
        # is emphatically not that — it is the case the re-encode below exists
        # for, and refusing it here (as "Unsupported video codec") is what this
        # fixed.
        if not raw_video_codec:
            self._send({"type": "error", "detail": "No video stream in this file"})
            return

        # Where to begin. Seeking is a stream restarted somewhere else: the
        # viewer moves the scrubber, this session's previous stream is retired
        # by _replace_stream, and ffmpeg is spawned again with -ss.
        try:
            start = float(msg.get("start", 0) or 0)
        except (TypeError, ValueError):
            start = 0.0
        # Past the end would produce an empty stream and a player waiting for
        # segments that are never coming.
        if duration and start >= duration - 1:
            start = max(0.0, duration - 5)
        start = max(0.0, start)

        # Video is copied whenever the browser can decode it directly —
        # re-encoding it is the expensive thing this pipeline exists to avoid,
        # and H264/VP9/AV1 already decode fine in-browser. HEVC is the one
        # exception (BROWSER_INCOMPATIBLE_VIDEO_CODECS, media_probe.py): found
        # live, a real HEVC/EAC3 WEB-DL reported "Codec not supported for
        # streaming" from MediaSource.isTypeSupported even though ffprobe/VLC
        # play it fine — Chrome has no HEVC decoder on most non-Apple
        # platforms. The operator can turn this fallback off (node.toml
        # transcode_incompatible_video = false) for a client fleet they know
        # already decodes HEVC, since it is real CPU cost, unlike the copy
        # path. Audio is always transcoded to AAC, never copied — see
        # _probe_video for why "copy" there is not an option, not even for a
        # codec that sounds close enough (plain AC-3 has the same in-browser
        # decode problem as E-AC-3, just without ffmpeg also refusing to mux
        # it). Transcoding audio is cheap; it does not change the cost model
        # the transcode-slot semaphore is sized around.
        #
        # Two kinds of source cannot be copied, and both re-encode:
        #
        #  - one whose MSE codec string is real but that no mainstream browser
        #    decodes — HEVC, BROWSER_INCOMPATIBLE_VIDEO_CODECS;
        #  - one with **no MSE codec string at all**: MPEG-4 Part 2 (Xvid,
        #    DivX), MPEG-2, VC-1, WMV, Theora. `stream_init` has to carry a
        #    string the client puts through MediaSource.isTypeSupported, and
        #    `probe_video` returns None for these precisely because no browser
        #    has a MediaSource decoder for them, so there is none to carry.
        #    This second kind used to be refused outright with "Unsupported
        #    video codec" — which named the source's problem and not the
        #    node's answer to it, since ffmpeg re-encodes these in real time on
        #    any machine that can run this daemon. Reported live against an
        #    Xvid/MP3 .avi. `transcode_incompatible_video`'s own documentation
        #    (docs/MESHBAY_DESIGN.md §6.8) already said "HEVC *and other browser-
        #    incompatible video codecs*"; only HEVC was ever wired up.
        can_copy = (bool(codec_str)
                    and raw_video_codec not in BROWSER_INCOMPATIBLE_VIDEO_CODECS)
        allow_transcode = self._ctx.get("transcode_incompatible_video", True)
        transcode_video = not can_copy and allow_transcode
        if not can_copy and not allow_transcode and not codec_str:
            # The operator turned the fallback off and there is nothing to fall
            # back *to*: a stream_init with no codec string is one the client
            # refuses before the first byte arrives. Which of the two it is
            # matters — "unsupported codec" sends the reader to look at the
            # file, and the file is fine.
            log.info("stream: %s is %s, which needs a re-encode, and "
                     "transcode_incompatible_video is off — refusing",
                     entry.name, raw_video_codec)
            self._send({"type": "error",
                        "detail": "This video needs transcoding, which the "
                                  "operator has turned off"})
            return
        # Seeking, and the trap that made a seek on a copied stream unwatchable.
        #
        # -ss BEFORE -i seeks by the container index rather than by decoding up
        # to the point: milliseconds on a 500 MB film instead of tens of
        # seconds. It lands on the keyframe at or before `start`, so the
        # picture can begin a few seconds earlier than asked — which is what
        # every streaming player does.
        #
        # **`-accurate_seek` is on by default, and it trims what it can.** It
        # cannot trim copied video, which has to begin on a keyframe; it does
        # trim the re-encoded audio, to exactly `start`. So the output began
        # with video from the keyframe and audio from `start` — correct
        # timestamps, both streams honestly placed, and **a hole in the audio
        # one whole GOP wide**. Measured on a real film with a 10 s keyframe
        # interval: seeking to 609 s against a keyframe at 599.104 s left
        # 9.979 s of silence, after which sound and picture were a GOP apart
        # for the rest of the film.
        #
        # Nothing downstream could see it. Every timestamp check passes — the
        # first PTS of each stream, their durations, their spans, the browser's
        # own A/V delta through MediaSource — because the timestamps were never
        # wrong. Only the *content* at a given instant was, which is why this
        # was found by decoding the output and comparing it against the source:
        # the first frame is byte-identical to the source frame at the
        # keyframe, and with the fix the audio's energy envelope matches the
        # source at that same instant (r = 0.97) instead of one GOP later.
        #
        # This is also why re-encoded video never showed the fault, and why a
        # library's HEVC files looked like the only ones that worked: video
        # that is re-encoded *can* start exactly at `start`, so accurate
        # seeking is right there and stays on.
        #
        # **`start` is rewritten on the copy path to where the seek actually
        # lands**, which is measured below rather than predicted — see
        # `_seek_lands_at`. From the rewrite on, it is where the picture really
        # begins and not where the viewer dragged to. The distinction was
        # invisible while only the scrubber read the number; it stopped being
        # invisible when subtitles did, since their cues carry the source's
        # absolute times and every second of disagreement puts a line on
        # screen a second away from the voice saying it.
        requested = start
        seek_args: list[str] = []
        if requested > 0:
            seek_args = ["-ss", f"{requested:.3f}"] if transcode_video else [
                "-noaccurate_seek", "-ss", f"{requested:.3f}"]
        map_args = ["-map", "0:v:0"]
        if transcode_video:
            log.info("stream: re-encoding %s (%s) to H264", entry.name,
                     raw_video_codec)
            # Where the re-encode runs, and with which arguments — both live
            # in hwaccel.py now, including the 8-bit downsampling a 10-bit HDR
            # source needs before either encoder will take it. `modes_for` has
            # measured this machine by encoding on it and returns the ladder to
            # try, always ending in libx264: a node with no usable VA-API does
            # exactly what it did before this existed, and a Celeron with an
            # iGPU stops being a machine where `transcode_incompatible_video`
            # has to be turned off to keep streaming watchable.
            modes = await hwaccel.modes_for(raw_video_codec)
            hw = await hwaccel.encoder()
            # Must match "-profile:v high -level 4.1" byte-for-byte (avc1.<profile
            # hex><constraint><level hex>) — the client checks this string with
            # MediaSource.isTypeSupported before trusting a single byte of the
            # stream, so a mismatch here fails exactly the check this exists to
            # pass. Both encoders get those two arguments, spelled the same way,
            # from hwaccel._PROFILE_ARGS — one place, so they cannot drift.
            codec_str = "avc1.640029,mp4a.40.2" if has_audio else "avc1.640029"
        else:
            modes, hw = [hwaccel.SW], None

        def video_args(mode: str) -> list[str]:
            return (hwaccel.codec_args(mode, hw) if transcode_video
                    else ["-c:v", "copy"])

        # The audio half does not change with the video encoder, and is never a
        # copy — see _probe_video for why.
        audio_args: list[str] = []
        # Which audio track. A dubbed film carries several and the first one is
        # not a neutral default — it is whatever the person who muxed the file
        # happened to put first, which across a real library is overwhelmingly
        # one language. Out of range falls back to the first rather than
        # refusing: the client's list comes from a `stream_init` that may
        # predate the file being replaced on disk, and a viewer who asked for
        # the second track of a file that now has one wants the film, not an
        # error. `stream_init` says which track was actually used, the same way
        # it says which `start` was actually used and for the same reason.
        try:
            audio_track = int(msg.get("audio_track", 0) or 0)
        except (TypeError, ValueError):
            audio_track = 0
        if not 0 <= audio_track < len(probe.audio_tracks):
            audio_track = 0
        if has_audio:
            map_args += ["-map", f"0:a:{audio_track}"]
            # Downmixed to stereo: a WEB-DL's 5.1 track becomes 6-channel AAC
            # with no "-ac", which ffprobe and VLC accept fine but which some
            # browsers' MSE decoder rejects outright once real fragments are
            # appended — isTypeSupported() only checks the codec string, so
            # the failure doesn't surface until playback, as a SourceBuffer
            # forced out of its MediaSource with no further explanation.
            audio_args = ["-c:a", "aac", "-ac", "2", "-b:a", "192k"]
        # Where that seek lands, measured with the mapping this stream will
        # use. It has to be here rather than beside `seek_args` above: the
        # landing point depends on which audio track is mapped, because the
        # container is seeked to a position that serves *every* mapped stream
        # — on a real title, video alone landed at 4909.863 s and the same
        # seek with the second audio track landed at 4907.236 s. The `-ss`
        # argument is deliberately left at the request, so the bytes served
        # are exactly the ones served before; only the number naming them
        # changes.
        if requested > 0 and not transcode_video:
            landed = await _seek_lands_at(file_path, requested, map_args)
            if landed is not None:
                start = landed
        # One spawn per mode, and only ever more than one when hwaccel.py found
        # a working GPU. **What a mode is tried against is the file itself**:
        # a test encode proves the encoder, and nothing proves the GPU can
        # decode *this* source until it is asked to — iHD has no MPEG-4 Part 2
        # decoder at all, so an Xvid .avi fails the full-hardware mode and
        # nothing about the machine could have predicted it.
        #
        # The failure is silent and instant: ffmpeg writes its complaint to
        # stderr and exits, so stdout reaches EOF with nothing on it. That is
        # the signal read here, before `stream_init` is sent and therefore
        # before the client has been told anything it would have to be told
        # again. The first segment is kept and handed to the loop below rather
        # than re-read, since the process it came from is still running.
        #
        # The last mode is spawned and trusted, which is what keeps the
        # single-mode path — every node without a GPU, and every copied stream
        # — byte-for-byte what it was: no extra read, no extra wait.
        first_segment = b""
        for attempt, mode in enumerate(modes):
            proc = await asyncio.create_subprocess_exec(
                platform.ffmpeg_cmd(), "-hide_banner", "-loglevel", "error",
                *hwaccel.input_args(mode, hw),
                *seek_args,
                "-i", str(file_path),
                *map_args,
                *video_args(mode), *audio_args,
                "-movflags", "frag_keyframe+empty_moov+default_base_moof",
                "-f", "mp4", "pipe:1",
                stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
            )
            if attempt == len(modes) - 1:
                break
            first_segment = await proc.stdout.read(STREAM_SEGMENT_SIZE)
            if first_segment:
                break
            err = (await proc.stderr.read()).decode("utf-8", "replace").strip()
            await proc.wait()
            hwaccel.demote(raw_video_codec, mode,
                           err.splitlines()[0] if err else "no output")

        self._send({
            "type": MNP.STREAM_INIT,
            "v": MNP_VERSION,
            "file_id": file_id,
            "codec": codec_str,
            "duration": duration,
            # ffmpeg restarts its timestamps at zero whatever we seek to, so
            # this is what the client adds back (`SourceBuffer.timestampOffset`)
            # to put the fragments where they belong on the timeline.
            "start": start,
            # The track list is how a client discovers that this node can
            # switch language at all — there is no version check anywhere in
            # the player. A node that does not send it gets no selector, and
            # the client then never sends `audio_track` to a peer that would
            # ignore it and serve the wrong language without saying so.
            "audio_tracks": [
                {
                    "i": tr.ordinal,
                    "lang": tr.language,
                    "title": tr.title,
                    "codec": tr.codec_name,
                    "ch": tr.channels,
                }
                for tr in probe.audio_tracks
            ],
            "audio_track": audio_track if has_audio else None,
            # Same discovery-from-the-answer shape as `audio_tracks`: a node
            # too old to enumerate sends no list, the client shows no selector
            # and never sends `subtitle_req` to a peer that would answer
            # "unknown message type". Text tracks only — a bitmap one has no
            # WebVTT to offer (media_probe.py), so it is absent here rather
            # than present and unplayable.
            "subtitle_tracks": [
                {
                    "i": tr.ordinal,
                    "lang": tr.language,
                    "title": tr.title,
                    "codec": tr.codec_name,
                    # What tells a full translation from signage-only. Without
                    # it the two are the same menu entry, and picking the
                    # forced one shows nothing for minutes at a time — which
                    # reads as a broken feature and was reported as one.
                    "forced": tr.forced,
                    "sdh": tr.hearing_impaired,
                }
                for tr in probe.subtitle_tracks
            ],
        })

        # A client that says nothing gets the old behaviour, which is why this
        # defaults to unlimited rather than to zero: a stream that waits for
        # credit from a peer that will never send any is a stream that hangs.
        try:
            self._stream_credit = int(msg.get("credits", 0) or 0)
        except (TypeError, ValueError):
            self._stream_credit = 0
        paced = self._stream_credit > 0
        self._stream_stopped = False

        index = 0
        self._stream_started_at = time.monotonic()
        self._stream_segments = 0
        reason = "eof"
        log.info("stream: stream_init sent file=%s paced=%s credits=%d start=%.1fs "
                 "audio=%s/%d",
                 file_id[:12], paced, self._stream_credit, start,
                 audio_track if has_audio else "-", len(probe.audio_tracks))
        try:
            while True:
                if paced and not await self._await_stream_credit():
                    reason = "no-credit-or-gone"
                    break
                if self._stream_stopped:
                    reason = "stopped-by-peer"
                    log.info("Stream stopped by peer=%s after %d segments",
                             (self._user_id or "?")[:8], index)
                    break
                if first_segment:
                    data, first_segment = first_segment, b""
                else:
                    data = await proc.stdout.read(STREAM_SEGMENT_SIZE)
                if not data:
                    break
                # Same derivation as a file chunk, indexed by segment: one
                # implementation, in `meshbay_common.protocol`.
                nonce, ct = chunk_ciphertext(gek, data, index, file_hash)
                self._send({
                    "type": MNP.STREAM_DATA,
                    "v": MNP_VERSION,
                    "file_id": file_id,
                    "segment_index": index,
                    "nonce": nonce,
                    "ct": ct,
                    "plaintext_size": len(data),
                })
                index += 1
                self._stream_segments = index
                if index % 100 == 0:
                    # A stream that stops shows up here as a last line, and the
                    # numbers on it say which side stopped it.
                    log.info("stream: %d segments (%.0f MB), credit=%d, "
                             "keepalives=%d, %.0fs in",
                             index, index * STREAM_SEGMENT_SIZE / 1048576,
                             self._stream_credit, self._stream_keepalives,
                             time.monotonic() - self._stream_started_at)
                await asyncio.sleep(0)
        except Exception as e:
            log.error("Stream error: %s", e)
        finally:
            try:
                proc.kill()
            except ProcessLookupError:
                pass
            # `await proc.wait()` on its own is the deadlock the asyncio docs
            # warn about: ffmpeg fills the stdout pipe we have stopped reading,
            # and the transport cannot finish closing until that buffer is
            # drained. Measured on 2026-08-16 with stream: — a viewer closed
            # the player after 99 segments (25 MB) and the task sat here past
            # the 15 s handover timeout, holding a transcode slot. The node has
            # two, so the next video waited and the one after was refused.
            #
            # Drain first, then wait with a bound. The slot must come back even
            # if the process is being stubborn: it has already had SIGKILL, and
            # the OS will reap it whether or not we are still watching.
            stderr_output = b""
            for pipe in (proc.stdout, proc.stderr):
                if pipe is None:
                    continue
                try:
                    drained = await asyncio.wait_for(pipe.read(), timeout=2)
                    if pipe is proc.stderr:
                        stderr_output = drained
                except Exception:
                    pass
            try:
                await asyncio.wait_for(proc.wait(), timeout=5)
            except Exception:
                log.warning("stream: ffmpeg did not reap in 5s — "
                            "releasing the slot regardless")

        # A positive returncode is ffmpeg exiting on its own with an error,
        # before we ever killed it (a kill shows up as a negative signal
        # number instead) — zero segments in that case is a real failure,
        # not a normal end, and saying nothing here is indistinguishable
        # from "the file is just this short". Found live against a real
        # 5.1 E-AC-3 WEB-DL that ffmpeg refused to even start muxing.
        # Detail stays server-side (L3: never hand a peer raw stderr).
        if index == 0 and proc.returncode is not None and proc.returncode > 0:
            log.error("stream: ffmpeg exited rc=%s before any output — %s",
                      proc.returncode,
                      stderr_output.decode(errors="replace").strip().splitlines()[-1:]
                      or "(no stderr)")
            if not self._stream_stopped:
                self._send({"type": "error",
                           "detail": "Could not stream this file"})
        elif not self._stream_stopped:
            self._send({
                "type": MNP.STREAM_END,
                "v": MNP_VERSION,
                "file_id": file_id,
            })
        log.info("stream: stream ended reason=%s segments=%d after %.1fs",
                 reason, index, time.monotonic() - self._stream_started_at)
        log.info("Streamed %s: %d segments", entry.name, index)
        self._audit("stream_video", entry.name)

    def _do_stream_request(self, msg: dict) -> None:
        sem = self._ctx.get("_transcode_sem")
        log.info("stream: req file=%s credits=%s slots_free=%s prev=%s",
                 str(msg.get("file_id"))[:12], msg.get("credits"),
                 getattr(sem, "_value", "?"),
                 "alive" if (self._stream_task and
                             not self._stream_task.done()) else "none")
        self._spawn(self._replace_stream(msg))

    def _do_client_diag(self, msg: dict) -> None:
        # Diagnostics only. The node acts on none of it — it writes it
        # next to its own view of the same stream, which is the only
        # place the two halves can be compared when the client is a
        # phone with no console.
        # Every field is peer-controlled, so each is stringified and
        # cut short: this is a log line, not a channel for writing
        # whatever one likes into the operator's file.
        def _f(key: str, n: int = 24) -> str:
            return str(msg.get(key))[:n].replace("\n", " ")
        if msg.get("event"):
            # Once per stream or per seek, not once per five seconds —
            # and a seek nobody asked for looks exactly like a viewer
            # dragging the scrubber from this side, so it has to be
            # visible without turning DEBUG on.
            log.info(
                "stream: client %s target=%s t=%ss offset=%s ready=%s "
                "duration=%s ranges=[%s]",
                _f("event", 16), _f("target"), _f("t"), _f("offset"),
                _f("ready"), _f("duration"), _f("ranges", 120))
        # Debug: one line every five seconds per viewer. Run the daemon
        # with --log-level debug to see inside a player that is
        # misbehaving — it is the only view of the browser there is
        # when the browser is a phone.
        else:
            # `ahead` on its own cannot say whether a short buffer is
            # the player's own gate holding or the network failing to
            # keep up, and those two want opposite answers. `limit` is
            # what the gate is set to for this film and `budget` the
            # byte budget it was derived from, so the three read as one
            # sentence.
            log.debug(
                "stream: client t=%ss ahead=%ss/%ss budget=%sMB "
                "ready=%s paused=%s "
                "stalled=%s q=%s inflight=%s appending=%s updating=%s "
                "quota=%s ms=%s err=%s ranges=[%s] (sent=%d)",
                _f("t"), _f("ahead"), _f("limit"), _f("budgetMB"),
                _f("ready"), _f("paused"),
                _f("stalled"), _f("q"), _f("inflight"), _f("appending"),
                _f("updating"), _f("quota"), _f("ms"), _f("err", 80),
                _f("ranges", 120), self._stream_segments)

    def _do_stream_stop(self) -> None:
        age = (time.monotonic() - self._stream_started_at
               if self._stream_started_at else -1)
        log.info("stream: stop received %.1fs after start, %d segments sent",
                 age, self._stream_segments)
        self._stop_stream()