refactor(node): move video streaming out of webrtc_server

StreamingMixin in transport/webrtc/apps/streaming.py: stream credit,
handover, the transcode slots and _stream_video_inner, moved unchanged.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

1 files changed, 3 insertions, 631 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index 5b7f700..5024ffd 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -113,13 +113,12 @@ from meshbay_common.join import (
 from meshbay_common.protocol import (
     MNP,
     UPLOAD_PROBE_INDEX,
-    chunk_ciphertext,
     file_chunk_wire,
     file_upload_ack_wire,
     file_upload_payload,
 )
 
-from meshbay_node import hwaccel, linkpreview, ops, platform
+from meshbay_node import linkpreview, ops
 from meshbay_node import transfers as transfers_mod
 from meshbay_node import uploads as uploads_mod
 from meshbay_node.chat import FORMAT_SEALED_V1, ReplayedMessage
@@ -130,12 +129,6 @@ from meshbay_node.indexer.indexer import DirectoryIndexer
 # this module still refers to it as `_probe_video`. The implementation lives
 # in media_probe.py so the indexer package (imported just above) can call it
 # too, for index-time enrichment, without a circular import.
-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 (
     ROOT_NOT_SERVED,
     SAFE_UPLOAD_NAME,
@@ -147,6 +140,7 @@ from meshbay_node.roots import (
 from meshbay_node.roster import KIND_ACCOUNT, KIND_LINK
 from meshbay_node.transfers import TransferSlots
 from meshbay_node.transport.webrtc.apps.music import MusicMixin
+from meshbay_node.transport.webrtc.apps.streaming import StreamingMixin
 from meshbay_node.transport.webrtc.apps.subtitles import SubtitlesMixin
 from meshbay_node.transport.webrtc.apps.video_meta import VideoMetaMixin
 from meshbay_node.transport.webrtc.channel import (
@@ -158,9 +152,6 @@ from meshbay_node.transport.webrtc.channel import (
 )
 from meshbay_node.transport.webrtc.disk import _locate
 from meshbay_node.transport.webrtc.limits import CHUNK_SIZE, MAX_MSG
-from meshbay_node.transport.webrtc.media_tools import (
-    _seek_lands_at,
-)
 from meshbay_node.transport.wire import index_sync_message
 
 log = logging.getLogger(__name__)
@@ -287,22 +278,6 @@ PRE_HANDSHAKE_MAX_MSG = 64 * 1024
 MAX_PEER_SESSIONS = 64
 UNAUTHENTICATED_SESSION_TIMEOUT = 60   # seconds
 
-# 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
 # Bundle fetches are served in the pre-proof window (C4). Bounded and audited
 # until the native client removes remote keypair bundles entirely.
 MAX_PRE_PROOF_FETCHES = 4
@@ -320,19 +295,11 @@ JOIN_FAILURE_WINDOW = 600  # seconds
 # nobody could read, or files scattered wherever someone happened to be looking.
 
 
-STREAM_SEGMENT_SIZE = 256 * 1024
 # A chunk is a megabyte and the browser keeps eight in flight, so answering them
 # as they arrive queues 8 MB on the channel with nothing watching. On a LAN that
 # drains before anyone notices; on a phone that is also uploading, it is minutes
 # of head-of-line delay for the reader. Above this, wait for room.
 DOWNLOAD_BUFFER_HIGH = 2 * 1024 * 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
 # How often transfer leases are swept. Nothing depends on it being
 # prompt -- the session teardown is the reclaim that matters and is
 # immediate; this catches peers that vanished without the connection
@@ -354,7 +321,7 @@ _WEBRTC_TRACE = os.environ.get("MESHBAY_WEBRTC_TRACE") == "1"
 _WEBRTC_TRACE_INTERVAL_S = 30.0
 
 
-class WebRTCPeerSession(VideoMetaMixin, MusicMixin, SubtitlesMixin):
+class WebRTCPeerSession(StreamingMixin, VideoMetaMixin, MusicMixin, SubtitlesMixin):
     """One WebRTC peer connection, handling MNP over a DataChannel."""
 
     def __init__(self, pc: RTCPeerConnection, node_ctx: dict, peer_id: str = ""):
@@ -5014,601 +4981,6 @@ class WebRTCPeerSession(VideoMetaMixin, MusicMixin, SubtitlesMixin):
             "file_id": file_id,
         })
 
-    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 _send(self, obj: dict) -> None:
         # Stamp the reply with the id of the request being answered, so the
         # caller never has to guess. Only for this session's own replies: a