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 --- .../src/meshbay_node/transport/webrtc_server.py | 74 ++++++++++++---------- 1 file changed, 41 insertions(+), 33 deletions(-) (limited to 'packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py') 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) -- cgit v1.2.3