diff options
Diffstat (limited to 'packages')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 58 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_disk_io_off_loop.py | 62 |
2 files changed, 105 insertions, 15 deletions
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 773d3dc..d2ec1de 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -339,6 +339,24 @@ SEEK_PROBE_MAX_BACKOFF_SECS = 60 # 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 @@ -6721,6 +6739,30 @@ 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: @@ -6777,9 +6819,11 @@ async def _transcode_audio_to_aac(file_path: Path) -> bytes: if proc.returncode != 0: raise RuntimeError( f"ffmpeg exited {proc.returncode}: {stderr.decode(errors='replace')[:300]}") - return tmp_path.read_bytes() + return await asyncio.to_thread( + _read_scratch_capped, tmp_path, AUDIO_TRANSCODE_MAX_BYTES, + "transcoded audio") finally: - tmp_path.unlink(missing_ok=True) + await _discard_scratch(tmp_path) async def _seek_lands_at(file_path: Path, t: float, map_args: list[str]) -> float | None: @@ -6833,7 +6877,7 @@ async def _seek_lands_at(file_path: Path, t: float, map_args: list[str]) -> floa log.warning("stream: seek probe failed at %.1fs: %r", t, e) return None finally: - tmp_path.unlink(missing_ok=True) + await _discard_scratch(tmp_path) text = stdout.decode(errors="replace").strip().rstrip(",") try: landed = float(text) @@ -6891,10 +6935,8 @@ async def _extract_subtitle_to_webvtt(file_path: Path, ordinal: int, if proc.returncode != 0: raise RuntimeError( f"ffmpeg exited {proc.returncode}: {stderr.decode(errors='replace')[:300]}") - size = tmp_path.stat().st_size - if size > SUBTITLE_MAX_BYTES: - raise RuntimeError(f"subtitle track is {size} bytes, over the {SUBTITLE_MAX_BYTES} cap") - blob = tmp_path.read_bytes() + 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, @@ -6904,7 +6946,7 @@ async def _extract_subtitle_to_webvtt(file_path: Path, ordinal: int, raise RuntimeError("extracted subtitle contains no cues") return blob finally: - tmp_path.unlink(missing_ok=True) + await _discard_scratch(tmp_path) class WebRTCTransport: diff --git a/packages/meshbay-node/tests/test_disk_io_off_loop.py b/packages/meshbay-node/tests/test_disk_io_off_loop.py index 2179e66..ceaf565 100644 --- a/packages/meshbay-node/tests/test_disk_io_off_loop.py +++ b/packages/meshbay-node/tests/test_disk_io_off_loop.py @@ -25,6 +25,7 @@ import threading import time from pathlib import Path +import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshbay_common.crypto import generate_gek from meshbay_common.protocol import MNP @@ -252,13 +253,11 @@ def test_no_handler_touches_the_disk_on_the_loop(): on_the_disk_thread = {"_locate", "_append_chunk", "_read_and_encrypt", "_mkdir_if_absent", "_is_empty_dir", "_rmdir_if_empty", "safe_subdir"} - # ffmpeg's own output, under `tempfile.mkstemp` on the system disk — not a - # group root, so not what spins down. Listed rather than silently allowed: - # these still read a whole transcode into memory from the loop, and the day - # that matters it is a different measurement from this one. - ffmpeg_scratch = {"_transcode_audio_to_aac", "_seek_lands_at", - "_extract_subtitle_to_webvtt"} - allowed = on_the_disk_thread | ffmpeg_scratch + # ffmpeg's own output goes through `_read_scratch_capped` and + # `_discard_scratch` on a worker thread — `asyncio.to_thread` and not + # `off_disk`, because a temp file is not a group root and has no platter to + # serialise against. Nothing is exempt here any more. + allowed = on_the_disk_thread | {"_read_scratch_capped"} found = [] @@ -354,3 +353,52 @@ async def test_chunks_of_one_upload_keep_their_order_under_a_slow_disk(tmp_path, refusals = [m for m in session.sent if m.get("type") == "error"] assert not refusals, f"a chunk was refused: {refusals}" assert (shared / "clip.bin").read_bytes() == b"".join(pieces) + + +def test_the_scratch_read_is_only_ever_reached_on_a_thread(): + """ + `_read_scratch_capped` blocks by design, so the guard above allows it — and + that allowance is worth nothing if somebody calls it straight from a + handler. Passed to `asyncio.to_thread` it appears in the syntax tree as a + name; called inline it appears as a call, which is what this refuses. + """ + tree = ast.parse(Path(webrtc_server.__file__).read_text()) + direct = [n.lineno for n in ast.walk(tree) + if isinstance(n, ast.Call) + and isinstance(n.func, ast.Name) + and n.func.id == "_read_scratch_capped"] + assert not direct, ( + f"_read_scratch_capped is called directly at line(s) {direct} — hand it " + "to `asyncio.to_thread` instead, or the cap is paid for on the loop") + + +async def test_ffmpeg_output_over_the_cap_is_refused_before_it_is_read(tmp_path): + """ + The stat comes first, so an oversized result costs a stat rather than the + read and the memory. The number in the message is the one that was measured, + not the cap, because an operator reading a log wants to know by how much. + """ + scratch = tmp_path / "out.m4a" + scratch.write_bytes(b"x" * 5000) + + with pytest.raises(RuntimeError, match=r"5000 bytes, over the 1024 cap"): + webrtc_server._read_scratch_capped(scratch, 1024, "transcoded audio") + + # And under the cap it simply reads. + assert webrtc_server._read_scratch_capped(scratch, 8192, "x") == b"x" * 5000 + + +async def test_a_slow_scratch_read_does_not_stop_the_loop(tmp_path, monkeypatch): + """Measured like the others: the loop keeps its wake-ups during the read.""" + scratch = tmp_path / "out.vtt" + scratch.write_bytes(CONTENT) + monkeypatch.setattr(webrtc_server, "_read_scratch_capped", + _slow(webrtc_server._read_scratch_capped)) + + with _Ticker() as ticker: + blob = await asyncio.to_thread( + webrtc_server._read_scratch_capped, scratch, 1 << 20, "subtitle track") + + assert blob == CONTENT + assert ticker.ticks > SLOW_S / TICK_S / 2, ( + f"the loop was blocked: {ticker.ticks} wake-ups during a {SLOW_S}s read") |