diff options
15 files changed, 669 insertions, 658 deletions
@@ -886,7 +886,7 @@ here are kept only where they are a rule about *editing* the code. | WebRTC transport and every MNP handler | `meshbay_node/transport/webrtc_server.py` | the big one; framing, shared limits and one-shot ffmpeg jobs are in `transport/webrtc/`, handlers for one app only in `transport/webrtc/apps/` | | QUIC transport | `meshbay_node/transport/quic_server.py` | off by default; **must stay at parity with the WebRTC handlers** | | Background tasks (node) | `webrtc_server.py` — `_spawn()` | the only way to start one; a bare `ensure_future` can be collected | -| Stream handover, backpressure | `webrtc_server.py` — `_replace_stream`, `shutdown_tasks`, `DOWNLOAD_BUFFER_HIGH` | §8.5 | +| Stream handover, backpressure | `transport/webrtc/apps/streaming.py` — `_replace_stream`; `webrtc_server.py` — `shutdown_tasks`, `DOWNLOAD_BUFFER_HIGH` | §8.5 | | Stream diagnosis | `webrtc_server.py` — `client_diag` at DEBUG | the player's own view in the node's log; the only window into a phone | | Chat store and paging | `meshbay_node/chat/store.py` | `get_recent`/`get_before`/`has_before`; `get_messages` pages *forwards* and is not what a chat opens with | | Chat epochs | `meshbay_node/ops.py` — `open_chat_epoch`, `ensure_chat_epoch`, `chat_epoch_keys` | §4.5 | diff --git a/docs/MESHBAY_NODE_PROTOCOL.md b/docs/MESHBAY_NODE_PROTOCOL.md index c3db019..4e0983d 100644 --- a/docs/MESHBAY_NODE_PROTOCOL.md +++ b/docs/MESHBAY_NODE_PROTOCOL.md @@ -2307,12 +2307,12 @@ LP(x) = uint32be(len(x)) || x every field, no exceptions | `MAX_LEASELESS_IN_FLIGHT` / `LEASELESS_IDLE_SECS` | 12 files / 60 s | ” | | `TRANSFER_SWEEP_SECS` | 15 s | `webrtc_server.py` | | Lease watchdog (client) | 60 s, then re-ask | `transport.js` | -| `STREAM_SEGMENT_SIZE` | 256 KiB | `webrtc_server.py` | +| `STREAM_SEGMENT_SIZE` | 256 KiB | `webrtc/apps/streaming.py` | | `STREAM_MAX_CREDIT` | 256 | ” | | `STREAM_CREDIT_TIMEOUT` / `_POLL` | 120 s / 3 s | ” | | Client stream credits | 24 | `transport.js` | -| `MAX_CONCURRENT_TRANSCODES` | 8 | `webrtc_server.py` | -| Link preview rate | 15/conn, 60/node per 60 s; cache 1 h × 256 | ” | +| `MAX_CONCURRENT_TRANSCODES` | 8 | `webrtc/apps/streaming.py` | +| Link preview rate | 15/conn, 60/node per 60 s; cache 1 h × 256 | `webrtc_server.py` | | ICE gathering deadline | 4 s | `transport.js` | | Signaling: max SDP, pending per user, rate | 16 KiB, 3, 30/min, 15 s answer timeout | `api/signaling.py` | | GEK | 256-bit, node CSPRNG | `crypto.py` | diff --git a/docs/transfers-v1.md b/docs/transfers-v1.md index f99424f..13e4869 100644 --- a/docs/transfers-v1.md +++ b/docs/transfers-v1.md @@ -67,7 +67,7 @@ So there is already a resume story here, it is just not reachable — and there is already a leak. **Video streaming is capped, and its cap is a good model.** `_stream_video` -(`webrtc_server.py:5230`) takes `ctx["_transcode_sem"]`, refuses with +(`webrtc/apps/streaming.py`) takes `ctx["_transcode_sem"]`, refuses with `"Server busy, retry shortly"` when it is empty, and every hard-won lesson in this repo about slots — `_replace_stream`, `shutdown_tasks()`, `_spawn()` and the garbage-collected task, `await proc.wait()` after `kill()` — is a lesson @@ -76,7 +76,7 @@ about *not losing a slot*. That history is why §5 exists. **One bug found while reading, in scope for this work.** `ops.py:1362-1365` hot-swaps the stream cap by assigning `webrtc._stream_sem` — an attribute that does not exist. The real semaphore is `ctx["_transcode_sem"]` -(`webrtc_server.py:5220`), and `hasattr(webrtc, '_stream_sem')` is always +(`_transcode_semaphore`, `webrtc/apps/streaming.py`), and `hasattr(webrtc, '_stream_sem')` is always False, so **changing `max_concurrent_streams` from the Node page has never taken effect without a restart**, contrary to §2.11 of draft v6. Two more hot-swappable caps are about to be added next to it; fix it with one shared diff --git a/packages/meshbay-node/src/meshbay_node/config.py b/packages/meshbay-node/src/meshbay_node/config.py index 437f629..563e395 100644 --- a/packages/meshbay-node/src/meshbay_node/config.py +++ b/packages/meshbay-node/src/meshbay_node/config.py @@ -178,7 +178,7 @@ class NodeConfig: # How many people may watch a video at the same time. One ffmpeg runs per # viewer for as long as they watch, so this is the knob that decides when # the node answers "server busy" — see MAX_CONCURRENT_TRANSCODES in - # transport/webrtc_server.py for what one costs. + # transport/webrtc/apps/streaming.py for what one costs. max_concurrent_streams: int = 8 # How many transfers run at once on this node, across every group — # separate pools, because a download and an upload cost different things 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 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 diff --git a/packages/meshbay-node/tests/test_stream_audio_track_selection.py b/packages/meshbay-node/tests/test_stream_audio_track_selection.py index b9e2105..f17592b 100644 --- a/packages/meshbay-node/tests/test_stream_audio_track_selection.py +++ b/packages/meshbay-node/tests/test_stream_audio_track_selection.py @@ -26,7 +26,8 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshbay_common.crypto import generate_gek from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes from meshbay_node.indexer.group_index import GroupIndex -from meshbay_node.transport.webrtc_server import WebRTCPeerSession, _probe_video +from meshbay_node.media_probe import probe_video as _probe_video +from meshbay_node.transport.webrtc_server import WebRTCPeerSession from conftest import needs_subprocess, one_root diff --git a/packages/meshbay-node/tests/test_stream_audio_transcode.py b/packages/meshbay-node/tests/test_stream_audio_transcode.py index 6c9bde7..8411d25 100644 --- a/packages/meshbay-node/tests/test_stream_audio_transcode.py +++ b/packages/meshbay-node/tests/test_stream_audio_transcode.py @@ -25,7 +25,8 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshbay_common.crypto import generate_gek from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes from meshbay_node.indexer.group_index import GroupIndex -from meshbay_node.transport.webrtc_server import WebRTCPeerSession, _probe_video +from meshbay_node.media_probe import probe_video as _probe_video +from meshbay_node.transport.webrtc_server import WebRTCPeerSession from conftest import needs_subprocess, one_root diff --git a/packages/meshbay-node/tests/test_stream_capacity.py b/packages/meshbay-node/tests/test_stream_capacity.py index 7ce6bf6..aecee6f 100644 --- a/packages/meshbay-node/tests/test_stream_capacity.py +++ b/packages/meshbay-node/tests/test_stream_capacity.py @@ -17,11 +17,8 @@ find. import asyncio import pytest -from meshbay_node.transport.webrtc_server import ( - MAX_CONCURRENT_TRANSCODES, - WebRTCPeerSession, - WebRTCTransport, -) +from meshbay_node.transport.webrtc.apps.streaming import MAX_CONCURRENT_TRANSCODES +from meshbay_node.transport.webrtc_server import WebRTCPeerSession, WebRTCTransport def _pool(transport) -> asyncio.Semaphore: diff --git a/packages/meshbay-node/tests/test_stream_capacity_config.py b/packages/meshbay-node/tests/test_stream_capacity_config.py index c788ba6..5514712 100644 --- a/packages/meshbay-node/tests/test_stream_capacity_config.py +++ b/packages/meshbay-node/tests/test_stream_capacity_config.py @@ -25,11 +25,8 @@ from pathlib import Path import pytest from meshbay_node.config import load_config from meshbay_node.roots import RootSet -from meshbay_node.transport.webrtc_server import ( - MAX_CONCURRENT_TRANSCODES, - WebRTCPeerSession, - WebRTCTransport, -) +from meshbay_node.transport.webrtc.apps.streaming import MAX_CONCURRENT_TRANSCODES +from meshbay_node.transport.webrtc_server import WebRTCPeerSession, WebRTCTransport def _cfg(tmp_path: Path, body: str): diff --git a/packages/meshbay-node/tests/test_stream_seek_audio_alignment.py b/packages/meshbay-node/tests/test_stream_seek_audio_alignment.py index 5d78c00..3a10464 100644 --- a/packages/meshbay-node/tests/test_stream_seek_audio_alignment.py +++ b/packages/meshbay-node/tests/test_stream_seek_audio_alignment.py @@ -222,7 +222,7 @@ async def test_a_transcoded_video_keeps_accurate_seeking(tmp_path): session, file_id = _session(tmp_path, clip, gek) # Force the re-encode branch the way a browser-incompatible source would. session._ctx["transcode_incompatible_video"] = True - import meshbay_node.transport.webrtc_server as ws + import meshbay_node.transport.webrtc.apps.streaming as ws real = ws.BROWSER_INCOMPATIBLE_VIDEO_CODECS ws.BROWSER_INCOMPATIBLE_VIDEO_CODECS = frozenset({"h264"}) try: diff --git a/packages/meshbay-node/tests/test_stream_seek_reports_the_keyframe.py b/packages/meshbay-node/tests/test_stream_seek_reports_the_keyframe.py index c575572..af6ea71 100644 --- a/packages/meshbay-node/tests/test_stream_seek_reports_the_keyframe.py +++ b/packages/meshbay-node/tests/test_stream_seek_reports_the_keyframe.py @@ -38,10 +38,8 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshbay_common.crypto import generate_gek from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes from meshbay_node.indexer.group_index import GroupIndex -from meshbay_node.transport.webrtc_server import ( - WebRTCPeerSession, - _seek_lands_at, -) +from meshbay_node.transport.webrtc.media_tools import _seek_lands_at +from meshbay_node.transport.webrtc_server import WebRTCPeerSession from conftest import needs_subprocess, one_root diff --git a/packages/meshbay-node/tests/test_stream_subtitle_tracks.py b/packages/meshbay-node/tests/test_stream_subtitle_tracks.py index 32b503b..fc93dc9 100644 --- a/packages/meshbay-node/tests/test_stream_subtitle_tracks.py +++ b/packages/meshbay-node/tests/test_stream_subtitle_tracks.py @@ -37,8 +37,9 @@ from meshbay_common.crypto import generate_gek from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes from meshbay_node.indexer.group_index import GroupIndex from meshbay_node.media_probe import TEXT_SUBTITLE_CODECS +from meshbay_node.media_probe import probe_video as _probe_video from meshbay_node.transport.webrtc.media_tools import _subtitle_timeout_for -from meshbay_node.transport.webrtc_server import WebRTCPeerSession, _probe_video +from meshbay_node.transport.webrtc_server import WebRTCPeerSession from conftest import needs_subprocess, one_root diff --git a/packages/meshbay-node/tests/test_stream_video_transcode.py b/packages/meshbay-node/tests/test_stream_video_transcode.py index b60d546..87e4130 100644 --- a/packages/meshbay-node/tests/test_stream_video_transcode.py +++ b/packages/meshbay-node/tests/test_stream_video_transcode.py @@ -36,7 +36,8 @@ from meshbay_common.crypto import generate_gek from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes from meshbay_node import hwaccel from meshbay_node.indexer.group_index import GroupIndex -from meshbay_node.transport.webrtc_server import WebRTCPeerSession, _probe_video +from meshbay_node.media_probe import probe_video as _probe_video +from meshbay_node.transport.webrtc_server import WebRTCPeerSession from conftest import needs_subprocess, one_root |