diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node')
5 files changed, 377 insertions, 338 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/__init__.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/__init__.py new file mode 100644 index 0000000..f52fda1 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/__init__.py @@ -0,0 +1 @@ +"""The WebRTC session, split by concern. `webrtc_server` assembles it.""" diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/channel.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/channel.py new file mode 100644 index 0000000..107a43e --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/channel.py @@ -0,0 +1,96 @@ +"""The data channel itself: framing, the reply correlation id, the peer's +address and DTLS fingerprint.""" + +import contextvars +import struct + +import msgpack +from aiortc import RTCPeerConnection + +from meshbay_node.transport.webrtc.limits import MAX_MSG + + +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"" + + +def _pack(obj: dict) -> bytes: + data = msgpack.packb(obj, use_bin_type=True) + return struct.pack(">I", len(data)) + data + + +# 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 "" diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/limits.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/limits.py new file mode 100644 index 0000000..89fe86b --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/limits.py @@ -0,0 +1,5 @@ +"""Limits read by more than one part of the WebRTC session.""" + + +CHUNK_SIZE = 1024 * 1024 +MAX_MSG = 64 * 1024 * 1024 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 new file mode 100644 index 0000000..2016ff4 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/media_tools.py @@ -0,0 +1,261 @@ +"""One-shot ffmpeg jobs: whole-file audio transcode, subtitle extraction, seek probe.""" + +import asyncio +import logging +import os +import tempfile +from pathlib import Path + +from meshbay_node import platform + +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. +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 + + +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) + + +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 `<track>` 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:<ordinal>` 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) 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 `<track>` 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:<ordinal>` 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. |