From 79b8f770ab319cf8d64132a56fc0036dcf0486f7 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Fri, 18 Sep 2026 15:37:28 +0200 Subject: fix(node): serve a file from a disk thread, never from the event loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A root that has spun down, or that lives on a network mount, answers its first syscall in seconds rather than microseconds. Made from the event loop, that stalls the whole node: no other group is served, no stream is fed, no chat message is delivered and the hub socket is not read, for as long as the platter takes to come back. It was found from the other end — a client's connection attempt timing out on a node with one member, while the disk woke up — and it is AV9's lesson with the disk in the place of the mail server. Every filesystem call on a group's content now goes through `off_disk`, onto a thread that belongs to that group's root set. The stat goes with the read: a stat is what *wakes* a sleeping disk, so offloading only the read would move the stall rather than remove it, and the read would then find the disk already awake. `_locate` is the one place allowed to call `entry_abs_path`, which is `Path.resolve()` and therefore syscalls too. Five handlers touched: a chunk request, an audio transcode, a subtitle request, a video stream and a delete. The delete became a coroutine, which its one caller already was. One worker per root set, not a pool and not one for the node. One worker for the same reason the indexer's executor has one — two interleaved reads of a spinning drive seek-thrash rather than go faster — and it keeps two threads from being inside the same file at once, which is what makes the `seek`/`read` pair safe without a lock. Per root set, because a node serves several groups and their roots are not all on the same volume: a single worker would put one group's sleeping USB drive in front of another's SSD, which is this symptom one level down. The thread is created on the first read, so a group nobody downloads from never starts one, and the daemon stops them all on the way out. The tests measure rather than read: a ticker counts its own wake-ups beside a request made slow on purpose, and a handler that blocks the loop takes every one of them with it. Put either call back inline and both tests report zero wake-ups, which was checked before they were trusted. QUIC still reads on its loop. Its handler is synchronous by construction, no client speaks it, and its distance from parity is already recorded in the design document (§15.3, L3); moving it is part of bringing it to parity, not of this. Co-Authored-By: Claude Opus 5 --- packages/meshbay-node/src/meshbay_node/daemon.py | 9 + packages/meshbay-node/src/meshbay_node/roots.py | 63 ++++++ .../src/meshbay_node/transport/webrtc_server.py | 74 ++++--- .../test_added_root_does_not_wait_for_its_scan.py | 2 +- .../meshbay-node/tests/test_disk_io_off_loop.py | 224 +++++++++++++++++++++ 5 files changed, 338 insertions(+), 34 deletions(-) create mode 100644 packages/meshbay-node/tests/test_disk_io_off_loop.py diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index fa9ef81..69c7e66 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -1789,6 +1789,15 @@ class NodeDaemon: for indexer in self._indexers: await indexer.stop() + # The thread each group's roots are read from (roots.py `io_executor`). + # Created on the first read, so a group nobody downloaded from never + # started one and this is a no-op for it. + groups = self._webrtc._ctx.get("groups") if self._webrtc else None + for group in (groups or {}).values(): + roots = group.get("roots") + if roots is not None: + roots.close_io() + if self._quic_server: await self._quic_server.stop() diff --git a/packages/meshbay-node/src/meshbay_node/roots.py b/packages/meshbay-node/src/meshbay_node/roots.py index 7854e28..67778ca 100644 --- a/packages/meshbay-node/src/meshbay_node/roots.py +++ b/packages/meshbay-node/src/meshbay_node/roots.py @@ -28,8 +28,10 @@ are one directory. from __future__ import annotations +import asyncio import logging import re +from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from pathlib import Path @@ -171,6 +173,49 @@ class RootSet: # like a deletion all over again on the pass after it. auto_ejected: list[str] = field(default_factory=list) + # The thread every blocking filesystem call on these roots is made from. + # + # **Why it exists at all.** A root that has spun down, or that lives on a + # network mount, answers its first syscall in seconds rather than + # microseconds. Made from the event loop that is the whole node: no other + # group is served, no stream is fed, no chat message is delivered and the + # hub socket is not read, for as long as the platter takes to come back. + # A client's connection attempt times out and has to be made again, which + # is what this was found by. It is `AV9`'s lesson with the disk in the + # place of the mail server. + # + # **Why one worker and not a pool.** The same reason the indexer's executor + # has one: two interleaved reads on a spinning drive seek-thrash against + # each other rather than go faster, measured there on a USB disk. One + # worker also keeps every read of these roots in the order it was asked + # for, which costs nothing — the protocol addresses a chunk by index, so + # no caller depends on that order — and leaves no way for two threads to + # be inside the same file at once. + # + # **Why per root set and not one for the node.** A node serves several + # groups, and their roots are not all on the same volume. A single worker + # would put the sleeping USB drive of one group in front of the SSD of + # another, which is the symptom this removes, one level down. + # + # Created on first use, so a RootSet that never reads anything — most of + # them, in tests — never starts a thread. Not compared and not printed: it + # is machinery, not part of what a root set *is*, and `daemon.py` compares + # root sets to decide whether a reload changed anything. + _io: ThreadPoolExecutor | None = field( + default=None, init=False, repr=False, compare=False) + + @property + def io_executor(self) -> ThreadPoolExecutor: + if self._io is None: + self._io = ThreadPoolExecutor(max_workers=1, thread_name_prefix="rootio") + return self._io + + def close_io(self) -> None: + """Stop the disk thread. Safe to call twice, and on a set that never read.""" + if self._io is not None: + self._io.shutdown(wait=False) + self._io = None + # ── Construction ───────────────────────────────────────────────────────── @classmethod @@ -401,6 +446,24 @@ def entry_abs_path(roots: RootSet, entry) -> Path | None: return (parent / entry.name) if parent else None +async def off_disk(roots: RootSet, fn, *args): + """ + Run one blocking filesystem call on the thread that serves `roots`. + + Every syscall against a group's content goes through here, `resolve()` and + `exists()` included: a `stat` is what *wakes* a sleeping disk, so a check + left on the event loop pays the spin-up in full and the read that follows + it finds the disk already awake. Offloading only the read would move the + stall, not remove it. + + `fn` must not touch anything the loop also touches — it runs on another + thread. Reading and encrypting a chunk qualifies; updating a session's + state does not. + """ + loop = asyncio.get_running_loop() + return await loop.run_in_executor(roots.io_executor, fn, *args) + + def _refuse_nesting(new: Root, existing: list[Root]) -> None: """ No root may contain another, compared case-insensitively. 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 d79674d..e748be9 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -139,7 +139,8 @@ from meshbay_node.media_probe import ( probe_video as _probe_video, ) from meshbay_node.roots import ( - ROOT_NOT_SERVED, RootSet, entry_abs_path, SAFE_UPLOAD_NAME, safe_subdir, _free_name, + ROOT_NOT_SERVED, RootSet, entry_abs_path, off_disk, SAFE_UPLOAD_NAME, safe_subdir, + _free_name, ) log = logging.getLogger(__name__) @@ -3877,12 +3878,9 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "File not found"}) return - file_path = entry_abs_path(ctx["roots"], entry) - if file_path is None: - self._send({"type": "error", "detail": ROOT_NOT_SERVED}) - return - if not file_path.exists(): - self._send({"type": "error", "detail": "File not on disk"}) + file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry) + if refusal is not None: + self._send({"type": "error", "detail": refusal}) return # A real index entry, asked for without a lease: browsing, or a client @@ -3913,7 +3911,8 @@ class WebRTCPeerSession: file_id[:12], chunk_index, getattr(self._channel, "bufferedAmount", "?")) file_hash = bytes.fromhex(entry.id) - chunk_data = _read_and_encrypt( + chunk_data = await off_disk( + ctx["roots"], _read_and_encrypt, ctx["gek"], file_path, chunk_index, file_hash, entry.id) # Backpressure. Without it the node hands the whole window to the # channel at once and the reader sees the first chunk, then nothing for @@ -4007,12 +4006,9 @@ class WebRTCPeerSession: if not entry: self._send({"type": "error", "detail": "File not found"}) return - file_path = entry_abs_path(ctx["roots"], entry) - if file_path is None: - self._send({"type": "error", "detail": ROOT_NOT_SERVED}) - return - if not file_path.exists(): - self._send({"type": "error", "detail": "File not on disk"}) + file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry) + if refusal is not None: + self._send({"type": "error", "detail": refusal}) return media_cache = self._ctx.get("media_cache") @@ -4074,12 +4070,9 @@ class WebRTCPeerSession: if not entry: self._send({"type": "error", "detail": "File not found"}) return - file_path = entry_abs_path(ctx["roots"], entry) - if file_path is None: - self._send({"type": "error", "detail": ROOT_NOT_SERVED}) - return - if not file_path.exists(): - self._send({"type": "error", "detail": "File not on disk"}) + file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry) + if refusal is not None: + self._send({"type": "error", "detail": refusal}) return media_cache = self._ctx.get("media_cache") @@ -5882,7 +5875,7 @@ class WebRTCPeerSession: self._audit("admin_auth_failed", f"file_delete:{file_id[:16]}") return - self._exec_file_delete(ctx, file_id, entry) + await self._exec_file_delete(ctx, file_id, entry) async def _admin_exec_invite_create( self, pending: dict, transcript: bytes, sig: bytes, @@ -5918,15 +5911,15 @@ class WebRTCPeerSession: "username": result.get("username", ""), }) - def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None: - file_path = entry_abs_path(ctx["roots"], entry) - if file_path is None: + async def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None: + file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry) + if refusal == ROOT_NOT_SERVED: # Frozen, not gone: removing the entry would lose a file that is # still on a drive the node cannot read right now. self._send({"type": "error", "detail": ROOT_NOT_SERVED}) return - if file_path.exists(): - file_path.unlink() + if file_path is not None: + await off_disk(ctx["roots"], file_path.unlink) log.info("File deleted: %s", entry.name) self._audit("file_delete", entry.name) @@ -6123,12 +6116,9 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "File not found"}) return - file_path = entry_abs_path(ctx["roots"], entry) - if file_path is None: - self._send({"type": "error", "detail": ROOT_NOT_SERVED}) - return - if not file_path.exists(): - self._send({"type": "error", "detail": "File not on disk"}) + file_path, refusal = await off_disk(ctx["roots"], _locate, ctx["roots"], entry) + if refusal is not None: + self._send({"type": "error", "detail": refusal}) return gek = ctx.get("gek") @@ -6581,6 +6571,24 @@ class WebRTCPeerSession: await self._pc.close() +def _locate(roots: RootSet, entry) -> tuple[Path | None, str | None]: + """ + Where an entry is, and whether it is readable — or the refusal to send. + + Both halves are syscalls: `resolve()` walks the path and `exists()` stats + it, and a stat is what *wakes* a sleeping disk. Leaving either on the event + loop and offloading only the read would move the stall rather than remove + it, and the read would then find the disk already awake. Blocking; called + through `off_disk`. + """ + path = entry_abs_path(roots, entry) + if path is None: + return None, ROOT_NOT_SERVED + if not path.exists(): + return None, "File not on disk" + return path, None + + def _read_and_encrypt( gek: bytes, file_path: Path, @@ -6588,7 +6596,7 @@ def _read_and_encrypt( file_hash: bytes, file_id: str = "", ) -> dict: - """Read one chunk off disk and encrypt it. Blocking; the caller keeps it short.""" + """Read one chunk off disk and encrypt it. Blocking; called through `off_disk`.""" with open(file_path, "rb") as f: f.seek(chunk_index * CHUNK_SIZE) plaintext = f.read(CHUNK_SIZE) diff --git a/packages/meshbay-node/tests/test_added_root_does_not_wait_for_its_scan.py b/packages/meshbay-node/tests/test_added_root_does_not_wait_for_its_scan.py index a80645a..4d4c4e8 100644 --- a/packages/meshbay-node/tests/test_added_root_does_not_wait_for_its_scan.py +++ b/packages/meshbay-node/tests/test_added_root_does_not_wait_for_its_scan.py @@ -235,6 +235,6 @@ async def test_a_file_request_is_refused_not_crashed(tmp_path): async def test_a_delete_is_refused_and_the_entry_kept(tmp_path): session, ctx, entry = await _served_without_two(tmp_path) - session._exec_file_delete(ctx, entry.id, entry) + await session._exec_file_delete(ctx, entry.id, entry) assert [m.get("detail") for m in session.sent] == [ROOT_NOT_SERVED] assert ctx["index"].get_entry(entry.id) is not None diff --git a/packages/meshbay-node/tests/test_disk_io_off_loop.py b/packages/meshbay-node/tests/test_disk_io_off_loop.py new file mode 100644 index 0000000..206b77b --- /dev/null +++ b/packages/meshbay-node/tests/test_disk_io_off_loop.py @@ -0,0 +1,224 @@ +""" +A slow disk must cost the caller that touched it, and nobody else. + +A root that has spun down, or that lives on a network mount, answers its first +syscall in seconds rather than microseconds. Made from the event loop, that +stalls the entire node: no other group is served, no stream is fed, no chat +message is delivered, and the hub socket is not read, for as long as the platter +takes to come back. It was found from the other end — a client's connection +attempt timing out on a node with one member, while the disk woke up. + +These tests do not read the source to check which thread a call is made from. +They make the disk slow and **measure whether the loop kept running**: a ticker +counts its own wake-ups beside the request, and a handler that blocks the loop +takes every one of those wake-ups with it. Put the call back inline and the +ticker count collapses to zero, which is the property being guarded. + +The third test is the one that is not about latency: one worker per root set +means two reads of the same file can never be inside it at once, and that is +what makes the single `f.seek()`/`f.read()` pair safe without a lock. +""" + +import asyncio +import re +import threading +import time +from pathlib import Path + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_common.crypto import generate_gek +from meshbay_common.protocol import MNP +from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes +from meshbay_node.indexer.indexer import DirectoryIndexer +from meshbay_node.roots import RootSet +from meshbay_node.transfers import LeaselessReads +from meshbay_node.transport import webrtc_server +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + +GROUP = "g" * 32 + +# Long enough that a blocked loop is unmistakable, short enough to keep the +# suite quick. The ticker below wakes every 5 ms, so a loop that stays free +# gets ~40 wake-ups inside one of these and a blocked one gets none. +SLOW_S = 0.2 +TICK_S = 0.005 +CONTENT = b"a file worth waking a disk for" * 400 + + +class _Channel: + readyState = "open" + bufferedAmount = 0 + + +class _Session(WebRTCPeerSession): + def __init__(self, ctx): + self._ctx = ctx + self._registry_key = "s1" + self._user_id = "alice" + self._username = "alice" + self._group_id = GROUP + self._channel = _Channel() + self._leaseless = LeaselessReads() + self._unleased_noted = False + self.sent: list[dict] = [] + + def _send(self, msg): + self.sent.append(msg) + + def _audit(self, event, detail=""): + pass + + def _spawn(self, coro): + coro.close() + return None + + +async def _served(tmp_path: Path): + """A session serving one real file out of one real root.""" + root = tmp_path / "films" + root.mkdir() + (root / "clip.bin").write_bytes(CONTENT) + roots = RootSet.build([{"path": str(root), "name": "films"}]) + gek = generate_gek() + idx = DirectoryIndexer(roots=roots, group_id=GROUP, + sk_node=Ed25519PrivateKey.generate(), gek=gek) + await idx.initial_scan() + entry = next(e for e in idx.index.entries if e.name == "clip.bin") + ctx = {"_peers": {}, "roots": roots, "index": idx.index, + "sk_node": idx.sk_node, "gek": gek} + return _Session(ctx), ctx, entry, gek + + +class _Ticker: + """Counts how many times the event loop came back to it.""" + + def __init__(self): + self.ticks = 0 + self._stop = False + self._task: asyncio.Task | None = None + + async def _run(self): + while not self._stop: + await asyncio.sleep(TICK_S) + self.ticks += 1 + + def __enter__(self): + self._task = asyncio.get_running_loop().create_task(self._run()) + return self + + def __exit__(self, *exc): + self._stop = True + self._task.cancel() + return False + + +def _slow(fn): + """`fn`, plus a sleep on whichever thread calls it.""" + + def wrapper(*args, **kwargs): + time.sleep(SLOW_S) + return fn(*args, **kwargs) + + return wrapper + + +async def test_a_slow_chunk_read_does_not_stop_the_loop(tmp_path, monkeypatch): + session, _, entry, gek = await _served(tmp_path) + monkeypatch.setattr(webrtc_server, "_read_and_encrypt", + _slow(webrtc_server._read_and_encrypt)) + + with _Ticker() as ticker: + await session._do_file_request( + {"type": MNP.FILE_REQUEST, "file_id": entry.id, "chunk_index": 0}) + + # The read really did take its time, and the loop really did keep running: + # both halves matter, because a wrapper that never ran would also leave the + # ticker free. + assert ticker.ticks > SLOW_S / TICK_S / 2, ( + f"the loop was blocked: {ticker.ticks} wake-ups during a {SLOW_S}s read") + + chunk = next(m for m in session.sent if m.get("type") == MNP.FILE_CHUNK) + key = chunk_key_aes(gek, bytes.fromhex(entry.id), 0) + assert decrypt_chunk_aes(key, chunk["nonce"], chunk["ct"]) == CONTENT + + +async def test_a_slow_stat_does_not_stop_the_loop(tmp_path, monkeypatch): + """ + The stat matters as much as the read: it is what *wakes* the disk. + + Offloading only the read would leave the spin-up on the loop and the read + would then find the disk already awake — the stall moved, not removed. + """ + session, _, entry, _ = await _served(tmp_path) + monkeypatch.setattr(webrtc_server, "_locate", _slow(webrtc_server._locate)) + + with _Ticker() as ticker: + await session._do_file_request( + {"type": MNP.FILE_REQUEST, "file_id": entry.id, "chunk_index": 0}) + + assert ticker.ticks > SLOW_S / TICK_S / 2, ( + f"the loop was blocked: {ticker.ticks} wake-ups during a {SLOW_S}s stat") + assert any(m.get("type") == MNP.FILE_CHUNK for m in session.sent) + + +async def test_one_root_set_reads_one_chunk_at_a_time(tmp_path): + """ + Two reads of the same root are never in flight together. + + Not a performance choice: interleaved reads of one spinning drive seek-thrash + (the indexer's executor carries the measurement), and a second thread inside + the same `open`/`seek`/`read` sequence is a correctness question this avoids + having to answer. A pool would reopen both. + """ + session, _, entry, _ = await _served(tmp_path) + inside = 0 + peak = 0 + guard = threading.Lock() + real = webrtc_server._read_and_encrypt + + def counting(*args, **kwargs): + nonlocal inside, peak + with guard: + inside += 1 + peak = max(peak, inside) + try: + time.sleep(0.02) + return real(*args, **kwargs) + finally: + with guard: + inside -= 1 + + webrtc_server._read_and_encrypt = counting + try: + await asyncio.gather(*[ + session._do_file_request( + {"type": MNP.FILE_REQUEST, "file_id": entry.id, "chunk_index": 0}) + for _ in range(6) + ]) + finally: + webrtc_server._read_and_encrypt = real + + assert peak == 1, f"{peak} reads of one root set were inside the disk at once" + + +def test_no_handler_resolves_a_path_on_the_loop(): + """ + `entry_abs_path` is `Path.resolve()`, which is syscalls — it belongs on the + disk thread with everything else. + + This reads the source because there is nothing else to read: a handler added + later that resolves an entry inline would pass every test above, since those + only exercise the handlers that exist today. `_locate` is the one place + allowed to call it, and `off_disk` is how `_locate` is reached. + """ + src = Path(webrtc_server.__file__).read_text() + # Every call site, not the first one: a guard that stops at the first + # occurrence stops guarding the moment a new call is inserted above it. + calls = [m.start() for m in re.finditer(r"\bentry_abs_path\(", src)] + body = re.search(r"\ndef _locate\(.*?\n(?=\n\ndef |\n\nclass )", src, re.S) + assert body, "_locate is gone or has been renamed — this guard needs rewriting" + allowed = range(body.start(), body.end()) + stray = [c for c in calls if c not in allowed] + assert not stray, ( + f"{len(stray)} call(s) to entry_abs_path outside _locate: " + f"resolve a path through `off_disk(roots, _locate, ...)` instead") -- cgit v1.2.3