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>

2 files changed, 644 insertions, 1 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py
new file mode 100644
index 0000000..05ab9e7
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py
@@ -0,0 +1,643 @@
+"""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)
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/media_tools.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/media_tools.py
index 2016ff4..7ecb0a5 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/media_tools.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/media_tools.py
@@ -14,7 +14,7 @@ log = logging.getLogger("meshbay_node.transport.webrtc_server")
 # A whole audio file is small enough to transcode in one shot rather than
 # live-piped like video's fMP4 segments — a few seconds of ffmpeg at most,
 # bounded generously so one slow/huge outlier can't pin a transcode slot
-# (shared with video, MAX_CONCURRENT_TRANSCODES in webrtc_server.py) indefinitely.
+# (shared with video, MAX_CONCURRENT_TRANSCODES in apps/streaming.py) indefinitely.
 AUDIO_TRANSCODE_TIMEOUT_SECS = 120
 # Extracting one subtitle track is a demux and a text conversion, not an
 # encode: measured at ~1.2 s for a full film. The bound is generous against a