From 5fa158fab709d3d24a33318b3d910f75c051af2e Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Fri, 18 Sep 2026 15:59:24 +0200 Subject: fix(node): take the availability poll and every upload write off the loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The rest of AV9's disk half. Serving a file left the loop in the commit before this one; two paths were still on it. **The availability poll.** `RootSet.refresh_availability` stats every root, and eleven call sites reached it from `async def` — the reconcile loop among them, on a timer. On a sleeping disk that is a stall once per tick, and the stat is also what keeps the disk awake, so a node paid spin-up for a library nobody was reading. All eleven now go through `off_disk`, `Root.is_live` included. **The upload write.** `open`/`write`, and the resolve, the stat, the free-name search, the rename and the unlink around it. This one could not simply be awaited: the handler was synchronous, so nothing could come between the `chunk_index != state.next_index` check and the `advance` that answers it, and that is the whole of the chunk-ordering rule. Awaiting the write opens the gap — chunk 1 arriving while chunk 0 is in the disk thread reads a position that has not moved and is refused as out of order, so an upload would fail on a slow disk and nowhere else. Verified, not assumed: without the lock the new ordering test refuses three chunks of four. So the check, the write and the advance are one critical section again, under a lock held **per group**. Not per session: `partial_uploads` lives in the group context so a reconnecting client finds its upload where it left it, which means two sessions of one member share the position of one `.part` file. Arrival order is preserved by construction — the dispatcher creates one task per message as it arrives, tasks start in creation order, and the lock is the first thing each one waits on, so its waiters queue in arrival order too. `_do_file_upload` is a coroutine now, which is why forty-two test call sites gain an `await`. Their outcomes are unchanged, file by file, against the run before the change. `test_ops.py` asked which public coroutines `ops` exposes and got `off_disk`, imported rather than defined there. It now asks for the ones written in the module, which is what its own docstring means; all forty-three operations are still checked. Co-Authored-By: Claude Opus 5 --- .../meshbay-node/tests/test_disk_io_off_loop.py | 98 +++++++++++++++++++++- 1 file changed, 97 insertions(+), 1 deletion(-) (limited to 'packages/meshbay-node/tests/test_disk_io_off_loop.py') 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 206b77b..2811836 100644 --- a/packages/meshbay-node/tests/test_disk_io_off_loop.py +++ b/packages/meshbay-node/tests/test_disk_io_off_loop.py @@ -29,12 +29,15 @@ 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.group_index import GroupIndex from meshbay_node.indexer.indexer import DirectoryIndexer -from meshbay_node.roots import RootSet +from meshbay_node.roots import Root, RootSet from meshbay_node.transfers import LeaselessReads from meshbay_node.transport import webrtc_server from meshbay_node.transport.webrtc_server import WebRTCPeerSession +from conftest import one_root, sealed_upload + GROUP = "g" * 32 # Long enough that a blocked loop is unmistakable, short enough to keep the @@ -222,3 +225,96 @@ def test_no_handler_resolves_a_path_on_the_loop(): assert not stray, ( f"{len(stray)} call(s) to entry_abs_path outside _locate: " f"resolve a path through `off_disk(roots, _locate, ...)` instead") + + +async def test_the_availability_poll_does_not_stop_the_loop(tmp_path, monkeypatch): + """ + The poll is the one that runs whether anybody asked for anything. + + `refresh_availability` stats every root, and reconcile calls it on a timer. + On the loop, a node with a sleeping disk stalls once per tick for as long as + the spin-up takes — and the stat is also what keeps waking the disk, so the + node pays for a library nobody is reading. + """ + root = tmp_path / "films" + root.mkdir() + (root / "clip.bin").write_bytes(CONTENT) + roots = RootSet.build([{"path": str(root), "name": "films"}]) + monkeypatch.setattr(Root, "is_live", _slow(Root.is_live)) + + idx = DirectoryIndexer(roots=roots, group_id=GROUP, + sk_node=Ed25519PrivateKey.generate(), gek=None) + try: + with _Ticker() as ticker: + await idx.initial_scan() + finally: + await idx.stop() + roots.close_io() + + assert ticker.ticks > SLOW_S / TICK_S / 2, ( + f"the loop was blocked: {ticker.ticks} wake-ups during a {SLOW_S}s poll") + + +def _upload_session(tmp_path): + """One connection into a group with one writable root, as a node has.""" + shared = tmp_path / "shared" + shared.mkdir(exist_ok=True) + ctx = {"roots": one_root(shared), + "index": GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate()), + "gek": generate_gek()} + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = ctx + session._group_id = GROUP + session._user_id = "user-1" + session._pk_user = "" + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + return session, shared + + +async def test_a_slow_upload_write_does_not_stop_the_loop(tmp_path, monkeypatch): + session, shared = _upload_session(tmp_path) + monkeypatch.setattr(webrtc_server, "_append_chunk", + _slow(webrtc_server._append_chunk)) + + with _Ticker() as ticker: + await session._do_file_upload(sealed_upload( + session, filename="clip.bin", data=CONTENT)) + + assert ticker.ticks > SLOW_S / TICK_S / 2, ( + f"the loop was blocked: {ticker.ticks} wake-ups during a {SLOW_S}s write") + assert (shared / "clip.bin").read_bytes() == CONTENT + + +async def test_chunks_of_one_upload_keep_their_order_under_a_slow_disk(tmp_path, + monkeypatch): + """ + The rule that used to hold for free. + + `chunk_index != state.next_index` is refused, and nothing could come between + that check and the `advance` answering it while the handler was synchronous. + Awaiting the write opens the gap: chunk 1 arriving while chunk 0 is still in + the disk thread reads a position that has not moved yet and is refused as + out of order — an upload that fails on a slow disk and nowhere else. The + lock closes it, and the disk made slow here is what makes the gap wide + enough to fall into. + """ + session, shared = _upload_session(tmp_path) + pieces = [b"first-", b"second-", b"third-", b"fourth"] + + monkeypatch.setattr(webrtc_server, "_append_chunk", + _slow(webrtc_server._append_chunk)) + + # Fired together and in order, which is what the dispatcher does: it creates + # one task per message as it arrives. + await asyncio.gather(*[ + session._do_file_upload(sealed_upload( + session, filename="clip.bin", data=piece, + chunk_index=i, total_chunks=len(pieces))) + for i, piece in enumerate(pieces) + ]) + + 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) -- cgit v1.2.3