diff options
Diffstat (limited to 'packages')
4 files changed, 298 insertions, 33 deletions
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/video-player.js b/packages/meshbay-hub/src/meshbay_hub/static/video-player.js index 42842d2..82e1116 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/video-player.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/video-player.js @@ -200,16 +200,26 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose, onDownload }) { const evictBehind = useCallback(() => { const sb = sbRef.current; const v = videoRef.current; - if (!sb || !v || sb.updating || !sb.buffered.length) return false; - const keepFrom = Math.max(0, v.currentTime - BUFFER_BEHIND_S); - // The range being watched, not the first one: after a seek backwards the - // first range is somewhere else entirely, and removing from its start to - // just behind the playhead would take out everything in between — - // including what is playing. - const range = currentRange(); - const start = range ? range[0] : sb.buffered.start(0); - if (keepFrom - start < 10) return false; + if (!sb || !v) return false; + // `.buffered`/`.updating` throw InvalidStateError once the SourceBuffer + // has been removed from its MediaSource — same reasoning as currentRange + // and describeRanges just above, which already guard the same read. This + // one did not, and an uncaught throw here skips flushQueue right after it + // in pump() too, since nothing between them catches it. Found live: + // fired on every incoming segment once the SourceBuffer went stale, + // which pump() runs on a 1s timer regardless of whether new data is + // arriving — an unguarded read here is not a rare corner, it repeats + // forever. try { + if (sb.updating || !sb.buffered.length) return false; + const keepFrom = Math.max(0, v.currentTime - BUFFER_BEHIND_S); + // The range being watched, not the first one: after a seek backwards the + // first range is somewhere else entirely, and removing from its start to + // just behind the playhead would take out everything in between — + // including what is playing. + const range = currentRange(); + const start = range ? range[0] : sb.buffered.start(0); + if (keepFrom - start < 10) return false; sb.remove(start, keepFrom); return true; } catch { @@ -253,8 +263,22 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose, onDownload }) { } queueRef.current.shift(); console.error('[MSE] appendBuffer error:', e); + // Anything else here does not get better by retrying: a SourceBuffer + // removed from its MediaSource stays removed. Discarding the segment + // and continuing left pump()'s credit grant running unchecked — it is + // driven by how much is successfully buffered, which never grows when + // nothing is actually appending — so the node kept sending and this + // kept discarding, forever. Found live: over 2 GB and 8000+ segments + // fetched for a picture that never appeared. Stop asking instead of + // spinning. + queueRef.current = []; + endedRef.current = true; + const transport = transportRef.current; + if (transport) transport.stopStream(); + setError(t('video.err_transport')); + setPhase('error'); } - }, [evictBehind]); + }, [evictBehind, transportRef]); /** * Decide whether the node may send more, and keep the pipeline moving. @@ -591,11 +615,32 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose, onDownload }) { setPhase('streaming'); flushQueue(); }); + // Neither fires for the ordinary end-of-stream (that's `endOfStream()` + // succeeding, no event needed) — only for the browser's own decoder + // giving up on what was appended. Logged, not acted on: by the time + // this fires the MediaSource is already unusable and every SourceBuffer + // call from here on throws, which the existing catches already handle. + ms.addEventListener('sourceclose', () => { + console.error('[MSE] sourceclose — MediaSource left "open" on its own', + 'readyState:', ms.readyState); + }); if (videoRef.current) { videoRef.current.src = url; videoRef.current.addEventListener('seeking', onSeeking); videoRef.current.addEventListener('timeupdate', pump); + videoRef.current.addEventListener('error', () => { + const err = videoRef.current && videoRef.current.error; + console.error('[MSE] video element error, code:', err && err.code, + 'message:', err && err.message); + if (transportRef.current) { + transportRef.current.sendStreamDiag({ + event: 'video-element-error', + code: err ? err.code : null, + message: err ? err.message : null, + }); + } + }); // The element's own verdict. "buffering" on screen is this, and it // is the one thing the node cannot infer from a stream it is feeding. videoRef.current.addEventListener('waiting', onStarved); diff --git a/packages/meshbay-hub/tests/test_upload_controls_hidden.py b/packages/meshbay-hub/tests/test_upload_controls_hidden.py index 9290a94..955b47d 100644 --- a/packages/meshbay-hub/tests/test_upload_controls_hidden.py +++ b/packages/meshbay-hub/tests/test_upload_controls_hidden.py @@ -109,8 +109,12 @@ def test_the_notice_still_answers_the_operators_own_request(app): caused it — returning early on it would leave that request hanging until it timed out.""" transport = TRANSPORT.read_text(encoding="utf-8") + # Scoped to member_upload_ack's own handler, not everything up to the next + # occurrence of "index_sync" — other handlers with their own, legitimate + # early `return` (index_progress, set_scan_settings_ack: neither is ever a + # reply anyone awaits) now sit between the two in the file. block = transport[transport.index("member_upload_ack"):] - block = block[:block.index("index_sync")] + block = block[:block.index("apps_enabled_ack")] assert "return" not in block 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 fa6c3e9..aaf1ecf 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -166,8 +166,21 @@ STREAM_CREDIT_POLL = 3 _H264_PROFILES = {"Baseline": "42", "Main": "4d", "High": "64", "High 10": "6e"} -async def _probe_video(path: str) -> tuple[str | None, float]: - """Probe video file with ffprobe, return (MSE codec string, duration).""" +async def _probe_video(path: str) -> tuple[str | None, float, bool]: + """ + Probe video file with ffprobe, return (MSE codec string, duration, + has_audio). + + The audio half of the codec string is always "mp4a.40.2" (AAC-LC) or + absent — never the source's real audio codec — because _stream_video_ + inner always transcodes audio to AAC and never copies it: MSE in every + mainstream browser only decodes AAC/Opus, and a source codec outside + that (AC-3, E-AC-3, DTS, ...) is at best silently unplayable and at + worst, for E-AC-3 at least, makes ffmpeg itself refuse to write the + fragmented MP4 header ("Cannot write moov atom before EAC3 packets + parsed" — reproduced against a real 5.1 E-AC-3 WEB-DL). Video stays + whatever it actually is: it is always copied, never transcoded. + """ import json as _json proc = await asyncio.create_subprocess_exec( "ffprobe", "-v", "error", @@ -180,7 +193,8 @@ async def _probe_video(path: str) -> tuple[str | None, float]: info = _json.loads(stdout) duration = float(info.get("format", {}).get("duration", 0)) - v_codec = a_codec = "" + v_codec = "" + has_audio = False for s in info.get("streams", []): if s.get("codec_type") == "video" and not v_codec: cn = s.get("codec_name", "") @@ -194,23 +208,13 @@ async def _probe_video(path: str) -> tuple[str | None, float]: v_codec = "vp09.00.10.08" elif cn == "av1": v_codec = "av01.0.01M.08" - elif s.get("codec_type") == "audio" and not a_codec: - cn = s.get("codec_name", "") - if cn == "aac": - a_codec = "mp4a.40.2" - elif cn in ("mp3", "mp2"): - a_codec = "mp4a.6b" - elif cn == "opus": - a_codec = "opus" - elif cn == "ac3": - a_codec = "ac-3" - elif cn == "flac": - a_codec = "flac" + elif s.get("codec_type") == "audio": + has_audio = True if not v_codec: - return None, duration - codec = f"{v_codec},{a_codec}" if a_codec else v_codec - return codec, duration + return None, duration, has_audio + codec = f"{v_codec},mp4a.40.2" if has_audio else v_codec + return codec, duration, has_audio def _pack(obj: dict) -> bytes: @@ -3005,7 +3009,7 @@ class WebRTCPeerSession: file_hash = bytes.fromhex(entry.id) try: - codec_str, duration = await _probe_video(str(file_path)) + codec_str, duration, has_audio = await _probe_video(str(file_path)) except Exception as e: self._send({"type": "error", "detail": f"Probe failed: {e}"}) return @@ -3034,11 +3038,31 @@ class WebRTCPeerSession: # what every streaming player does, and why the client is told the # value used rather than left to assume its own. seek_args = ["-ss", f"{start:.3f}"] if start > 0 else [] + # Video is always copied — re-encoding it is the expensive thing this + # pipeline exists to avoid, and H264/HEVC/VP9/AV1 already decode fine + # in-browser. 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. + map_args = ["-map", "0:v:0"] + codec_args = ["-c:v", "copy"] + if has_audio: + map_args += ["-map", "0:a:0"] + # 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. + codec_args += ["-c:a", "aac", "-ac", "2", "-b:a", "192k"] proc = await asyncio.create_subprocess_exec( "ffmpeg", "-hide_banner", "-loglevel", "error", *seek_args, "-i", str(file_path), - "-c", "copy", + *map_args, + *codec_args, "-movflags", "frag_keyframe+empty_moov+default_base_moof", "-f", "mp4", "pipe:1", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, @@ -3125,11 +3149,14 @@ class WebRTCPeerSession: # 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: - await asyncio.wait_for(pipe.read(), timeout=2) + drained = await asyncio.wait_for(pipe.read(), timeout=2) + if pipe is proc.stderr: + stderr_output = drained except Exception: pass try: @@ -3138,7 +3165,22 @@ class WebRTCPeerSession: log.warning("stream: ffmpeg did not reap in 5s — " "releasing the slot regardless") - if not self._stream_stopped: + # 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, diff --git a/packages/meshbay-node/tests/test_stream_audio_transcode.py b/packages/meshbay-node/tests/test_stream_audio_transcode.py new file mode 100644 index 0000000..5b1fc46 --- /dev/null +++ b/packages/meshbay-node/tests/test_stream_audio_transcode.py @@ -0,0 +1,174 @@ +""" +Audio in a streamed video is always transcoded to AAC, never copied. + +Found live against a real 5.1 E-AC-3 WEB-DL (season 2 of a show whose season 1 +was AAC and played fine): "-c copy" on an E-AC-3 track makes ffmpeg itself +refuse to write the fragmented MP4 header at all — "Cannot write moov atom +before EAC3 packets parsed" — and even for the codecs that don't make ffmpeg +outright refuse (plain AC-3, DTS, ...), no mainstream browser's MSE decodes +them, so the viewer would see a stream that starts and immediately ends with +no picture and no error. Video is untouched here — it is always copied, +still H264 in every fixture below, since re-encoding it is the expensive +thing this pipeline exists to avoid. + +These tests spawn real ffmpeg/ffprobe against small synthetic files (lavfi +test sources, ~1s) rather than asserting against the source text — a source +match cannot tell a working remux from one ffmpeg silently refuses to write. +""" + +import shutil +import subprocess +from pathlib import Path + +import pytest +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 conftest import one_root + +_HAVE_FFMPEG = shutil.which("ffmpeg") and shutil.which("ffprobe") +pytestmark = [ + pytest.mark.asyncio, + pytest.mark.skipif(not _HAVE_FFMPEG, reason="ffmpeg/ffprobe not installed"), +] + + +def _make_clip(path: Path, *, acodec: str, channels: int = 2) -> None: + """~1s of H264 video + `acodec` audio, muxed into an .mkv — a minimal + stand-in for a real WEB-DL with that audio codec.""" + subprocess.run( + ["ffmpeg", "-hide_banner", "-loglevel", "error", "-y", + "-f", "lavfi", "-i", "testsrc=size=320x240:rate=25:duration=1", + "-f", "lavfi", "-i", f"sine=frequency=440:duration=1:sample_rate=48000", + "-ac", str(channels), + "-c:v", "libx264", "-preset", "ultrafast", "-c:a", acodec, + str(path)], + check=True, capture_output=True, + ) + + +def _session(tmp_path: Path, video_path: Path, gek: bytes) -> WebRTCPeerSession: + import blake3 + file_bytes = video_path.read_bytes() + file_id = blake3.blake3(file_bytes).hexdigest() + + sk_node = Ed25519PrivateKey.generate() + index = GroupIndex(group_id="g" * 32, sk_node=sk_node, gek=gek) + from meshbay_common.protocol import IndexEntry + index.add_entry(IndexEntry( + id=file_id, name=video_path.name, path=video_path.parent.name, + size=len(file_bytes), type="video", added_at=0)) + + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = { + "roots": one_root(video_path.parent), + "index": index, + "gek": gek, + "sk_node": sk_node, + "max_concurrent_streams": 4, + } + session._group_id = None + session._user_id = "tester" + session._stream_stopped = False + session._stream_keepalives = 0 + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + return session, file_id + + +def _reassemble(sent: list[dict], gek: bytes, file_id: str) -> bytes: + file_hash = bytes.fromhex(file_id) + segments = sorted( + (m for m in sent if m.get("type") == "stream_data"), + key=lambda m: m["segment_index"]) + out = b"" + for m in segments: + key = chunk_key_aes(gek, file_hash, m["segment_index"]) + out += decrypt_chunk_aes(key, m["nonce"], m["ct"]) + return out + + +async def test_eac3_audio_is_transcoded_not_copied(tmp_path): + """The exact reproduction: ffmpeg refuses "-c copy" on E-AC-3 outright.""" + clip = tmp_path / "clip.mkv" + _make_clip(clip, acodec="eac3", channels=6) + gek = generate_gek() + session, file_id = _session(tmp_path, clip, gek) + + await session._stream_video_inner({"file_id": file_id, "start": 0, "credits": 0}) + + errors = [m for m in session.sent if m.get("type") == "error"] + assert not errors, f"streaming must not fail: {errors}" + + init = next(m for m in session.sent if m.get("type") == "stream_init") + assert "mp4a.40.2" in init["codec"], \ + "the reported codec must be AAC, never the source's eac3" + + data_segments = [m for m in session.sent if m.get("type") == "stream_data"] + assert data_segments, "no video data was ever sent" + assert any(m.get("type") == "stream_end" for m in session.sent) + + remuxed = _reassemble(session.sent, gek, file_id) + out_path = tmp_path / "out.mp4" + out_path.write_bytes(remuxed) + probe = subprocess.run( + ["ffprobe", "-v", "error", "-show_entries", "stream=codec_name,codec_type,channels", + "-of", "csv=p=0", str(out_path)], + check=True, capture_output=True, text=True) + rows = [line.split(",") for line in probe.stdout.strip().splitlines()] + codecs = {r[1]: r[0] for r in rows} + channels = {r[1]: r[2] for r in rows if len(r) > 2} + assert codecs.get("audio") == "aac", f"audio must be AAC on the wire: {codecs}" + assert codecs.get("video") == "h264", f"video must still be H264: {codecs}" + assert channels.get("audio") == "2", ( + "audio must be downmixed to stereo: some browsers' MSE decoder " + f"silently rejects multichannel AAC once real fragments are " + f"appended, even though ffprobe/VLC accept it fine: {channels}") + + +async def test_already_aac_audio_still_streams(tmp_path): + """Season-1-shaped file (already AAC): must keep working exactly as before.""" + clip = tmp_path / "clip.mkv" + _make_clip(clip, acodec="aac", channels=2) + gek = generate_gek() + session, file_id = _session(tmp_path, clip, gek) + + await session._stream_video_inner({"file_id": file_id, "start": 0, "credits": 0}) + + assert not [m for m in session.sent if m.get("type") == "error"] + assert any(m.get("type") == "stream_data" for m in session.sent) + assert any(m.get("type") == "stream_end" for m in session.sent) + + +async def test_probe_video_reports_aac_regardless_of_source_audio_codec(tmp_path): + clip = tmp_path / "clip.mkv" + _make_clip(clip, acodec="eac3", channels=6) + + codec, duration, has_audio = await _probe_video(str(clip)) + + assert has_audio is True + assert duration > 0 + assert codec is not None + assert "eac3" not in codec and "ec-3" not in codec + assert "mp4a.40.2" in codec + + +async def test_probe_video_handles_no_audio_track(tmp_path): + clip = tmp_path / "silent.mkv" + subprocess.run( + ["ffmpeg", "-hide_banner", "-loglevel", "error", "-y", + "-f", "lavfi", "-i", "testsrc=size=320x240:rate=25:duration=1", + "-c:v", "libx264", "-preset", "ultrafast", "-an", str(clip)], + check=True, capture_output=True, + ) + + codec, duration, has_audio = await _probe_video(str(clip)) + + assert has_audio is False + assert codec is not None and "," not in codec, \ + "no audio track must not produce a dangling ',' or a fake audio codec" |