fix(node): bound what a transcode may produce, and read ffmpeg's output off the loop

Two things about the same three functions, which write to a temp file with ffmpeg
and then read it back.

**The read was on the event loop.** These files are ffmpeg's own, under
`tempfile.mkstemp` on the system disk, so they are not a group root and there is
no spun-down platter to serialise against — which is why they go through
`asyncio.to_thread` and not `roots.off_disk`. But a whole transcode read inline is
still tens of megabytes of blocking read while nothing else in the node is served.
The AST guard now covers the module with no exemption at all, and a second check
refuses a direct call to the reading helper: passed to `to_thread` it appears in
the syntax tree as a name, called inline it appears as a call.

**The audio transcode had no size ceiling**, where the subtitle path beside it
has had one all along. The bound is the media cache's rather than memory's:
`put_thumb` writes one SQLite row, and the store is 512 MB with least-recently-used
eviction, sized for thumbnails, posters and short transcodes. At 192 kbit/s a
three-hour source is ~260 MB — one row that evicts most of the cache to fit and is
evicted again by the next few thumbnails. Not a size this store can hold usefully.

64 MiB, about forty-five minutes: past any track, any single piece, most sets.

It takes away nothing that worked. `AUDIO_TRANSCODE_TIMEOUT_SECS` is 120, so a
source long enough to reach this was already liable to be killed mid-transcode;
what changes is that the refusal now names the limit it met and the size that met
it. Serving audio of that length properly means streaming the conversion instead
of buffering it, which is a different feature — recorded in §9.8 rather than left
as an implied promise.

The stat comes before the read, so an oversized result costs a stat rather than
the read and the memory behind it.

Twelve `test_sticky_header.py[firefox]` setup errors again: Firefox is still open
on this machine, and its `[chrome]` half passes.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

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