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_partial_uploads.py | 48 +++++++++++----------- 1 file changed, 24 insertions(+), 24 deletions(-) (limited to 'packages/meshbay-node/tests/test_partial_uploads.py') diff --git a/packages/meshbay-node/tests/test_partial_uploads.py b/packages/meshbay-node/tests/test_partial_uploads.py index f5b6602..0270a69 100644 --- a/packages/meshbay-node/tests/test_partial_uploads.py +++ b/packages/meshbay-node/tests/test_partial_uploads.py @@ -305,7 +305,7 @@ def _errors(session): return [m for m in session.sent if m.get("type") == "error"] -def test_an_upload_survives_the_connection_that_started_it(tmp_path): +async def test_an_upload_survives_the_connection_that_started_it(tmp_path): """The defect this stage exists to fix. The state used to live on the session, so the second connection saw no @@ -315,14 +315,14 @@ def test_an_upload_survives_the_connection_that_started_it(tmp_path): """ ctx = _group_ctx(tmp_path) first = _peer(ctx) - first._do_file_upload(sealed_upload(first, filename="film.mkv", + await first._do_file_upload(sealed_upload(first, filename="film.mkv", data=b"first-half", chunk_index=0, total_chunks=2)) assert _errors(first) == [] # The link drops; the client comes back on a new connection and carries on. second = _peer(ctx) - second._do_file_upload(sealed_upload(second, filename="film.mkv", + await second._do_file_upload(sealed_upload(second, filename="film.mkv", data=b"second-half", chunk_index=1, total_chunks=2)) assert _errors(second) == [], _errors(second) @@ -331,31 +331,31 @@ def test_an_upload_survives_the_connection_that_started_it(tmp_path): assert (root.path / "film.mkv").read_bytes() == b"first-halfsecond-half" -def test_another_member_cannot_continue_somebody_elses_upload(tmp_path): +async def test_another_member_cannot_continue_somebody_elses_upload(tmp_path): """The key includes the member for a reason. Without it, a second person sending the same name into the same folder would append their chunks to the first person's file — which a shared folder makes an ordinary accident, not only an attack.""" ctx = _group_ctx(tmp_path) alice = _peer(ctx, "alice") - alice._do_file_upload(sealed_upload(alice, filename="IMG_1234.jpg", + await alice._do_file_upload(sealed_upload(alice, filename="IMG_1234.jpg", data=b"hers", chunk_index=0, total_chunks=2)) assert _errors(alice) == [] bob = _peer(ctx, "bob") - bob._do_file_upload(sealed_upload(bob, filename="IMG_1234.jpg", + await bob._do_file_upload(sealed_upload(bob, filename="IMG_1234.jpg", data=b"his", chunk_index=1, total_chunks=2)) assert [m.get("code") for m in _errors(bob)] == ["not_started"] -def test_an_upload_in_flight_is_known_to_the_reaper(tmp_path): +async def test_an_upload_in_flight_is_known_to_the_reaper(tmp_path): """The two halves of this stage meeting: the state the node keeps is what stops the janitor deleting a file somebody is still sending.""" ctx = _group_ctx(tmp_path) peer = _peer(ctx) - peer._do_file_upload(sealed_upload(peer, filename="film.mkv", + await peer._do_file_upload(sealed_upload(peer, filename="film.mkv", data=b"half", chunk_index=0, total_chunks=2)) live = ctx["partial_uploads"].live_paths() @@ -379,38 +379,38 @@ def _probe(session, filename: str) -> dict: chunk_index=UPLOAD_PROBE_INDEX, total_chunks=1) -def test_a_probe_for_an_unknown_file_says_start_at_the_beginning(tmp_path): +async def test_a_probe_for_an_unknown_file_says_start_at_the_beginning(tmp_path): ctx = _group_ctx(tmp_path) peer = _peer(ctx) - peer._do_file_upload(_probe(peer, "film.mkv")) + await peer._do_file_upload(_probe(peer, "film.mkv")) assert _errors(peer) == [] assert _acks(peer, ctx)[0]["resume_from"] == 0 -def test_a_probe_reports_what_the_node_already_holds(tmp_path): +async def test_a_probe_reports_what_the_node_already_holds(tmp_path): """The point of the whole stage: the client learns it has 2 chunks there and sends the third, instead of sending a film again.""" ctx = _group_ctx(tmp_path) first = _peer(ctx) for i in range(2): - first._do_file_upload(sealed_upload(first, filename="film.mkv", + await first._do_file_upload(sealed_upload(first, filename="film.mkv", data=b"xxxx", chunk_index=i, total_chunks=5)) assert _errors(first) == [] reconnected = _peer(ctx) - reconnected._do_file_upload(_probe(reconnected, "film.mkv")) + await reconnected._do_file_upload(_probe(reconnected, "film.mkv")) ack = _acks(reconnected, ctx)[0] assert ack["resume_from"] == 2 assert ack["stored_as"] == "film.mkv" -def test_a_probe_writes_nothing_and_reserves_nothing(tmp_path): +async def test_a_probe_writes_nothing_and_reserves_nothing(tmp_path): """It has to be free of consequence: a client that asks and goes away must leave no file, no state and no name taken.""" ctx = _group_ctx(tmp_path) peer = _peer(ctx) - peer._do_file_upload(_probe(peer, "film.mkv")) + await peer._do_file_upload(_probe(peer, "film.mkv")) root = ctx["roots"].roots[0] assert list(root.path.iterdir()) == [] assert len(ctx.get("partial_uploads") or []) == 0 @@ -418,36 +418,36 @@ def test_a_probe_writes_nothing_and_reserves_nothing(tmp_path): assert _acks(peer, ctx)[0]["stored_as"] == "" -def test_a_probe_answers_only_about_the_member_who_asks(tmp_path): +async def test_a_probe_answers_only_about_the_member_who_asks(tmp_path): """Same keying as the upload itself. Otherwise one member could measure another's progress on a file they never sent — and worse, resume it.""" ctx = _group_ctx(tmp_path) alice = _peer(ctx, "alice") - alice._do_file_upload(sealed_upload(alice, filename="film.mkv", + await alice._do_file_upload(sealed_upload(alice, filename="film.mkv", data=b"xxxx", chunk_index=0, total_chunks=5)) bob = _peer(ctx, "bob") - bob._do_file_upload(_probe(bob, "film.mkv")) + await bob._do_file_upload(_probe(bob, "film.mkv")) assert _acks(bob, ctx)[0]["resume_from"] == 0 -def test_an_ordinary_ack_carries_no_resume_field(tmp_path): +async def test_an_ordinary_ack_carries_no_resume_field(tmp_path): """So a client can tell a probe's answer from a chunk's without looking at the index it echoed.""" ctx = _group_ctx(tmp_path) peer = _peer(ctx) - peer._do_file_upload(sealed_upload(peer, filename="a.bin", data=b"x", + await peer._do_file_upload(sealed_upload(peer, filename="a.bin", data=b"x", chunk_index=0, total_chunks=2)) assert "resume_from" not in _acks(peer, ctx)[0] -def test_a_probe_is_refused_where_an_upload_would_be(tmp_path): +async def test_a_probe_is_refused_where_an_upload_would_be(tmp_path): """Every check the write path makes has already run when the probe is answered, so it cannot be used to ask questions about somewhere the caller may not write.""" ctx = _group_ctx(tmp_path) peer = _peer(ctx) - peer._do_file_upload(sealed_upload(peer, filename="../escape", + await peer._do_file_upload(sealed_upload(peer, filename="../escape", data=b"", chunk_index=UPLOAD_PROBE_INDEX, total_chunks=1)) assert [m.get("code") for m in _errors(peer)] == ["invalid_filename"] @@ -456,7 +456,7 @@ def test_a_probe_is_refused_where_an_upload_would_be(tmp_path): # ── the slot an upload holds ──────────────────────────────────────────────── -def test_an_upload_chunk_says_its_slot_is_in_use(tmp_path): +async def test_an_upload_chunk_says_its_slot_is_in_use(tmp_path): """A grant nobody takes up is reclaimed after thirty seconds and abandoned on the third miss. Uploads are not gated by the lease, so the file arrived anyway — but the widget follows the lease, and a 3.5 GB upload therefore @@ -482,7 +482,7 @@ def test_an_upload_chunk_says_its_slot_is_in_use(tmp_path): msg = sealed_upload(peer, filename="film.mkv", data=b"xxxx", chunk_index=0, total_chunks=2) msg["tr"] = "up-1" - peer._do_file_upload(msg) + await peer._do_file_upload(msg) assert _errors(peer) == [] assert slots.leases["up-1"].used is True, ( -- cgit v1.2.3