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")