From 15e7084bad34edb60a5999994731a8a2ed1dc6c8 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Thu, 24 Sep 2026 01:35:54 +0200 Subject: refactor(node): move framing, shared limits and ffmpeg jobs out of webrtc_server transport/webrtc/channel.py, limits.py and media_tools.py, cut from webrtc_server.py as text; the facade imports them back. Co-Authored-By: Claude Opus 5.5 --- .../src/meshbay_node/transport/webrtc_server.py | 352 +-------------------- 1 file changed, 14 insertions(+), 338 deletions(-) (limited to 'packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py') 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 af90dfc..66a0f3c 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -24,19 +24,15 @@ Signaling flow (handled externally by the hub): import asyncio import base64 -import contextvars import logging import os import re -import struct -import tempfile import time import uuid from pathlib import Path from typing import Any import blake3 -import msgpack from aiortc import RTCDataChannel, RTCPeerConnection, RTCSessionDescription from cryptography.hazmat.primitives.asymmetric.ed25519 import ( Ed25519PrivateKey, @@ -151,13 +147,24 @@ 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.channel import ( + _REPLY_TO, + _DataChannelBuffer, + _extract_dtls_fingerprint, + _get_remote_ip, + _pack, +) +from meshbay_node.transport.webrtc.limits import CHUNK_SIZE, MAX_MSG +from meshbay_node.transport.webrtc.media_tools import ( + _extract_subtitle_to_webvtt, + _seek_lands_at, + _subtitle_timeout_for, + _transcode_audio_to_aac, +) from meshbay_node.transport.wire import index_sync_message log = logging.getLogger(__name__) -CHUNK_SIZE = 1024 * 1024 -MAX_MSG = 64 * 1024 * 1024 - # Per-account blobs (docs/playlists.md §4.3). These are an unbounded write # primitive pointed at somebody else's disk, so every one of them is checked — # and every check **refuses**, never truncates. A truncating cap silently loses @@ -320,62 +327,6 @@ MAX_CONCURRENT_TRANSCODES = 8 # own, not something that varies by how the file happens to be encoded # inside. BROWSER_INCOMPATIBLE_AUDIO_EXTS = frozenset({".wma", ".mpc"}) -# 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 above) 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 -# pathological container rather than against the work itself, and it is short -# next to the audio one because nothing here decodes a media stream. -SUBTITLE_EXTRACT_TIMEOUT_SECS = 60 -# Extracting a subtitle demuxes the whole container, so the cost is set by the -# file and not by the subtitle: measured at **9.8 s per GB** on a library held -# on an external disk — 36 s for a 3.9 GB title, 71 s for a 7.3 GB one. A flat -# 60 s therefore worked on most of a library and failed on the big films, with -# nothing to distinguish that from a broken feature. The allowance is three -# times the measured rate so a slower disk, or one being read by a stream at -# the same time, still finishes. -SUBTITLE_EXTRACT_SECS_PER_GB = 30 -SUBTITLE_EXTRACT_TIMEOUT_MAX_SECS = 900 - - -def _subtitle_timeout_for(size_bytes: int) -> float: - """How long this file is allowed to take. See the constants above.""" - gb = max(0.0, size_bytes) / 1_000_000_000 - return min(SUBTITLE_EXTRACT_TIMEOUT_MAX_SECS, - max(SUBTITLE_EXTRACT_TIMEOUT_SECS, - SUBTITLE_EXTRACT_SECS_PER_GB * gb)) -# Probing where a copied seek lands costs 0.06–0.07 s on a real title, so this -# bounds a pathological container rather than the work itself. -SEEK_PROBE_TIMEOUT_SECS = 10 -# An index seek backing off further than this did not find a keyframe gap; it -# measured something other than the stream about to be served, and the old -# label beats a fabricated one. The largest real gap seen was under 10 s. -SEEK_PROBE_MAX_BACKOFF_SECS = 60 -# A subtitle file is text; a film's is ~96 KB. Anything past this is not a -# subtitle track, it is an ffmpeg that found something else to write, and it -# would sit in the media cache for ever. -SUBTITLE_MAX_BYTES = 8 * 1024 * 1024 - -# What a whole-file audio transcode may produce. The output is AAC at 192 kbit/s, -# so this is about forty-five minutes of source — past any track, any single -# piece, most sets. -# -# The bound is the media cache's, not memory's. `put_thumb` writes one SQLite row -# and the store is 512 MB with least-recently-used eviction, sized for what it -# holds: thumbnails, posters, subtitle tracks, short transcodes. A three-hour -# audiobook at this bitrate is ~260 MB — a single row that would evict most of -# the cache to make room for itself, and be evicted in turn by the next few -# thumbnails. It is not a size this store can hold usefully. -# -# It does not take away something that worked: `AUDIO_TRANSCODE_TIMEOUT_SECS` is -# 120, so a source long enough to reach this cap was already liable to be killed -# mid-transcode. What changes is that the refusal now says which limit was met. -# Serving audio of that length properly is streaming the transcode rather than -# buffering it, which is a different feature from this one. -AUDIO_TRANSCODE_MAX_BYTES = 64 * 1024 * 1024 # 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 @@ -393,15 +344,6 @@ JOIN_FAILURE_WINDOW = 600 # seconds # nobody could read, or files scattered wherever someone happened to be looking. -def _extract_dtls_fingerprint(sdp: str) -> bytes: - """Extract the DTLS SHA-256 fingerprint from SDP as raw 32 bytes.""" - for line in sdp.splitlines(): - if line.startswith("a=fingerprint:sha-256 "): - hex_str = line.split(" ", 1)[1].replace(":", "") - return bytes.fromhex(hex_str) - return b"" - - 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 @@ -422,11 +364,6 @@ STREAM_CREDIT_POLL = 3 TRANSFER_SWEEP_SECS = 15 -def _pack(obj: dict) -> bytes: - data = msgpack.packb(obj, use_bin_type=True) - return struct.pack(">I", len(data)) + data - - # Opt-in, off by default: a per-session heartbeat log (message count, time # since the last message, ICE state) and ICE-state-change logging, on top of # the connectionstatechange logging that already runs unconditionally. Added @@ -440,77 +377,6 @@ def _pack(obj: dict) -> bytes: _WEBRTC_TRACE = os.environ.get("MESHBAY_WEBRTC_TRACE") == "1" _WEBRTC_TRACE_INTERVAL_S = 30.0 -# The request this session is currently answering, as (session, req_id). -# -# MNP has never carried a correlation id: a reply named its own type and -# nothing else, so a client with more than one request outstanding had to guess -# which one a message answered — by arrival order, for every reply the client -# could not key off a field of its own. The guess is wrong whenever two replies -# reorder, and catastrophically wrong for the replies that name *nothing*: this -# module sends `{"type": "error"}` from 240 places and two of them name what -# they are about. A refusal therefore reached no caller at all, and the request -# it belonged to waited out the client's 30s timeout while some unrelated -# request was resolved with the refusal instead. Live symptom, found 2026-09-06: -# the Chat composer is disabled while a send is in flight, so a chat message -# whose reply went astray froze the tab for 30 seconds. -# -# `req_id` closes it: whatever the caller put on the request is stamped on the -# reply. A ContextVar rather than a parameter because the alternative is -# threading an argument through all 240 send sites — and asyncio copies the -# current context into a task, so a handler that `_spawn`s its real work still -# answers under the id of the request that started it. -# -# The session is held alongside the id because a handler may send to *other* -# sessions as well as its own (a chat broadcast, an index push): those are not -# replies to anything and must not be stamped. _send checks the owner. -_REPLY_TO: contextvars.ContextVar[tuple] = contextvars.ContextVar( - "meshbay_reply_to", default=(None, None)) - - -class _DataChannelBuffer: - """ - Accumulate DataChannel messages and extract length-prefixed msgpack. - - Finding H6: the limit was a flat 64 MB applied even before the handshake, so an - unauthenticated peer could announce a 64 MB frame and dribble bytes into it, - holding that much memory per connection. Until a peer has proved GEK - possession it gets a small budget; the large one is for file uploads. - """ - - def __init__(self, max_message: int = MAX_MSG): - self._buf = bytearray() - self.max_message = max_message - - def feed(self, data: bytes): - self._buf.extend(data) - - def messages(self): - while len(self._buf) >= 4: - length = struct.unpack(">I", self._buf[:4])[0] - if length > self.max_message: - raise ValueError(f"Message too large: {length}") - if len(self._buf) < 4 + length: - break - msg_bytes = bytes(self._buf[4:4 + length]) - del self._buf[:4 + length] - yield msgpack.unpackb(msg_bytes, raw=False) - - -def _get_remote_ip(pc: RTCPeerConnection) -> str: - """Best-effort extraction of the remote peer IP from the ICE transport.""" - try: - dtls = pc.sctp and pc.sctp.transport - ice = dtls and dtls.transport - conn = ice and ice._connection - if conn and hasattr(conn, '_nominated') and conn._nominated: - for pair in conn._nominated.values(): - return pair.remote_candidate.host - if conn and conn.remote_candidates: - return conn.remote_candidates[0].host - except Exception: - pass - return "" - class WebRTCPeerSession: """One WebRTC peer connection, handling MNP over a DataChannel.""" @@ -6924,30 +6790,6 @@ def _locate(roots: RootSet, entry) -> tuple[Path | None, str | None]: return path, None -def _read_scratch_capped(tmp_path: Path, cap: int, what: str) -> bytes: - """ - Stat ffmpeg's output, refuse it if it is too big, read it. Blocking. - - Run through `asyncio.to_thread` and not `off_disk`: this file is ffmpeg's - own, under `tempfile.mkstemp` on the system disk, so it is not a group root - and there is no spun-down platter to serialise against — it only has to be - off the event loop. A whole transcode read inline is tens of megabytes of - blocking read while nothing else in the node is served. - - The size is checked before the bytes are asked for, so an oversized result - costs a stat rather than the read *and* the memory. - """ - size = tmp_path.stat().st_size - if size > cap: - raise RuntimeError(f"{what} is {size} bytes, over the {cap} cap") - return tmp_path.read_bytes() - - -async def _discard_scratch(tmp_path: Path) -> None: - """Remove one of ffmpeg's temp files, off the loop like the read of it.""" - await asyncio.to_thread(tmp_path.unlink, True) - - def _append_chunk(tmp_path: Path, chunk_bytes: bytes, first: bool) -> None: """Add one chunk to a partial upload. Blocking; called through `off_disk`.""" with open(tmp_path, "wb" if first else "ab") as f: @@ -6968,172 +6810,6 @@ def _read_and_encrypt( return file_chunk_wire(gek, plaintext, chunk_index, file_hash, file_id) -async def _transcode_audio_to_aac(file_path: Path) -> bytes: - """ - One-shot, whole-file transcode to AAC in an M4A container — no live - piping, no seeking, unlike `_stream_video_inner`'s fMP4 segments: a - WMA/Musepack source here is a few MB at most, so there is nothing to - gain from streaming it and a real cost to the added complexity - (fragmented output needs `-movflags empty_moov` and its own - client-side reassembly). A plain temp file lets ffmpeg write a normal, - fully-seekable M4A container instead. `-vn` drops any attached-picture - "video" stream some taggers embed as cover art — without it, ffmpeg's - mp4 muxer has been seen treating that picture as a video track to - encode, which is not what this is for; cover art still comes from the - ordinary embedded/sibling-file path (enrich_audio.py), never from here. - """ - fd, tmp_name = tempfile.mkstemp(suffix=".m4a") - os.close(fd) - tmp_path = Path(tmp_name) - try: - proc = await asyncio.create_subprocess_exec( - platform.ffmpeg_cmd(), "-hide_banner", "-loglevel", "error", "-y", - "-i", str(file_path), - "-vn", "-c:a", "aac", "-ac", "2", "-b:a", "192k", - "-f", "ipod", str(tmp_path), - stdout=asyncio.subprocess.DEVNULL, - stderr=asyncio.subprocess.PIPE, - ) - try: - _, stderr = await asyncio.wait_for( - proc.communicate(), timeout=AUDIO_TRANSCODE_TIMEOUT_SECS) - except TimeoutError: - proc.kill() - await proc.wait() - raise RuntimeError(f"ffmpeg timed out after {AUDIO_TRANSCODE_TIMEOUT_SECS}s") - if proc.returncode != 0: - raise RuntimeError( - f"ffmpeg exited {proc.returncode}: {stderr.decode(errors='replace')[:300]}") - return await asyncio.to_thread( - _read_scratch_capped, tmp_path, AUDIO_TRANSCODE_MAX_BYTES, - "transcoded audio") - finally: - await _discard_scratch(tmp_path) - - -async def _seek_lands_at(file_path: Path, t: float, map_args: list[str]) -> float | None: - """Where an index seek to `t` actually puts this stream, in source time. - - Measured, not predicted. Copied video can only begin on a keyframe, and - the obvious way to find that keyframe — scan ffprobe's key frames and take - the last one at or before `t` — is wrong twice over. Matroska's Cues index - only some keyframes, so the seek backs off to an indexed one that can be - much earlier; and the landing point depends on **which streams are - mapped**, because the container is positioned where every mapped stream - has data. Measured on a real title: a seek to 4913.7 s landed at 4909.863 - with video alone and at 4907.236 with the second audio track mapped - alongside it. A prediction from the frame list gave the first number and - the stream delivered the second, which is 2.65 s of subtitles standing - away from the voice. - - So ffmpeg is asked instead: the same seek, the same mapping, one copied - frame, `-copyts` to keep the source's own timestamps, and the answer read - back off the result. Measured at 0.06–0.07 s, which is cheaper than the - frame scan it replaces. - - Returns None if anything about the probe fails, and the caller then keeps - the old label: a number that is wrong by a few seconds is worth much less - than a stream that does not start. - """ - fd, tmp_name = tempfile.mkstemp(suffix=".mp4") - os.close(fd) - tmp_path = Path(tmp_name) - try: - proc = await asyncio.create_subprocess_exec( - platform.ffmpeg_cmd(), "-hide_banner", "-loglevel", "error", "-y", - "-copyts", "-noaccurate_seek", "-ss", f"{t:.3f}", - "-i", str(file_path), - *map_args, "-c", "copy", "-frames:v", "1", - "-f", "mp4", str(tmp_path), - stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL, - ) - await asyncio.wait_for(proc.wait(), timeout=SEEK_PROBE_TIMEOUT_SECS) - if proc.returncode != 0: - return None - probe = await asyncio.create_subprocess_exec( - platform.ffprobe_cmd(), "-v", "error", - "-select_streams", "v:0", "-show_entries", "stream=start_time", - "-of", "csv=p=0", str(tmp_path), - stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL, - ) - stdout, _ = await asyncio.wait_for( - probe.communicate(), timeout=SEEK_PROBE_TIMEOUT_SECS) - except (TimeoutError, OSError) as e: - log.warning("stream: seek probe failed at %.1fs: %r", t, e) - return None - finally: - await _discard_scratch(tmp_path) - text = stdout.decode(errors="replace").strip().rstrip(",") - try: - landed = float(text) - except ValueError: - return None - # A seek never lands after what was asked for, and a landing point wildly - # before it is a probe that measured something else — a chapter track, an - # attachment. Either way the old label beats a fabricated one. - if not 0 <= landed <= t + 0.5 or t - landed > SEEK_PROBE_MAX_BACKOFF_SECS: - log.warning("stream: seek probe at %.1fs answered %.3f — ignored", t, landed) - return None - return landed - - -async def _extract_subtitle_to_webvtt(file_path: Path, ordinal: int, - timeout: float = SUBTITLE_EXTRACT_TIMEOUT_SECS) -> bytes: - """ - One subtitle track out of a container, whole, as WebVTT. - - Whole-file rather than following the stream, which is what makes the - result reusable: the cues carry the source's own absolute timestamps, so - the same extraction serves every seek, every audio-language change and - every later viewing, and the `` the client attaches never has to be - rebuilt. It is also the only shape the cache makes sense in — a segment - keyed on a seek position would be a different blob every time. - - `-map 0:s:` counts subtitle streams (see media_probe.py), and - `-c:s webvtt` converts subrip/ass to text; a bitmap codec reaching here - would produce an empty file rather than an error, which is why the caller - checks membership of the probed text list first and never a range. - - Written to a temp file rather than read off a pipe: the caller wants one - complete blob to hash and cache, and there is nothing to gain from - streaming a hundred kilobytes. - """ - fd, tmp_name = tempfile.mkstemp(suffix=".vtt") - os.close(fd) - tmp_path = Path(tmp_name) - try: - proc = await asyncio.create_subprocess_exec( - platform.ffmpeg_cmd(), "-hide_banner", "-loglevel", "error", "-y", - "-i", str(file_path), - "-map", f"0:s:{ordinal}", "-c:s", "webvtt", - "-f", "webvtt", str(tmp_path), - stdout=asyncio.subprocess.DEVNULL, - stderr=asyncio.subprocess.PIPE, - ) - try: - _, stderr = await asyncio.wait_for( - proc.communicate(), timeout=timeout) - except TimeoutError: - proc.kill() - await proc.wait() - raise RuntimeError(f"ffmpeg timed out after {timeout:.0f}s") - if proc.returncode != 0: - raise RuntimeError( - f"ffmpeg exited {proc.returncode}: {stderr.decode(errors='replace')[:300]}") - blob = await asyncio.to_thread( - _read_scratch_capped, tmp_path, SUBTITLE_MAX_BYTES, "subtitle track") - # A WebVTT file that is only its header has no cues in it. That is what - # a bitmap track extracted by mistake produces, and what a text track - # whose stream is empty produces; either way there is nothing to show, - # and an empty track attached to the player is worse than none — it - # appears in the menu and does nothing when picked. - if len(blob.strip()) <= len(b"WEBVTT"): - raise RuntimeError("extracted subtitle contains no cues") - return blob - finally: - await _discard_scratch(tmp_path) - - class WebRTCTransport: """ Manages WebRTC peer connections for browser clients. -- cgit v1.2.3