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