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>

1 files changed, 55 insertions, 7 deletions

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