fix(node): take the availability poll and every upload write off the loop

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 <noreply@anthropic.com>

1 files changed, 21 insertions, 21 deletions

diff --git a/packages/meshbay-node/tests/test_upload_sealed.py b/packages/meshbay-node/tests/test_upload_sealed.py
index 7c1be96..7f78ba7 100644
--- a/packages/meshbay-node/tests/test_upload_sealed.py
+++ b/packages/meshbay-node/tests/test_upload_sealed.py
@@ -80,14 +80,14 @@ def test_the_wire_message_carries_no_filename_and_no_plaintext(tmp_path):
     assert b"JPEGDATA" not in blob, "the file content is on the wire in clear"
 
 
-def test_the_ack_carries_no_stored_name(tmp_path):
+async def test_the_ack_carries_no_stored_name(tmp_path):
     """
     `stored_as` is the name the node settled on — it finds a free one rather
     than replacing anything — and naming it in clear would hand back exactly
     what the request took the trouble to hide.
     """
     session = _session(tmp_path)
-    session._do_file_upload(sealed_upload(
+    await session._do_file_upload(sealed_upload(
         session, filename="holiday.jpg", data=b"x", dir=_root(session).name))
 
     ack = [m for m in session.sent if m.get("type") == MNP.FILE_UPLOAD_ACK][-1]
@@ -98,7 +98,7 @@ def test_the_ack_carries_no_stored_name(tmp_path):
         "dir": _root(session).name}
 
 
-def test_the_ack_names_the_upload_so_one_refusal_fails_one_upload(tmp_path):
+async def test_the_ack_names_the_upload_so_one_refusal_fails_one_upload(tmp_path):
     """
     `filename` used to be the correlation key on both sides. It cannot be one
     any more, and `upload_id` replaces it — a client-chosen label, opaque to
@@ -107,10 +107,10 @@ def test_the_ack_names_the_upload_so_one_refusal_fails_one_upload(tmp_path):
     for one file used to fail every upload in flight.
     """
     session = _session(tmp_path)
-    session._do_file_upload(sealed_upload(
+    await session._do_file_upload(sealed_upload(
         session, filename="a.txt", data=b"x", dir=_root(session).name,
         upload_id="upload-A"))
-    session._do_file_upload(sealed_upload(
+    await session._do_file_upload(sealed_upload(
         session, filename="../evil", data=b"x", dir=_root(session).name,
         upload_id="upload-B"))
 
@@ -121,14 +121,14 @@ def test_the_ack_names_the_upload_so_one_refusal_fails_one_upload(tmp_path):
 
 # ── What is refused ──────────────────────────────────────────────────────────
 
-def test_a_plaintext_upload_is_refused(tmp_path):
+async def test_a_plaintext_upload_is_refused(tmp_path):
     """
     The MNP 1.x shape, which is what an un-updated client sends. Refused with a
     code and a message saying which side is old — never accepted "just this
     once", because a path that still takes plaintext is not a sealed path.
     """
     session = _session(tmp_path)
-    session._do_file_upload({
+    await session._do_file_upload({
         "filename": "note.txt", "dir": _root(session).name,
         "chunk_index": 0, "total_chunks": 1, "data": b"x",
     })
@@ -137,7 +137,7 @@ def test_a_plaintext_upload_is_refused(tmp_path):
     assert not _wrote_anything(tmp_path)
 
 
-def test_a_tampered_chunk_is_refused(tmp_path):
+async def test_a_tampered_chunk_is_refused(tmp_path):
     """
     AES-GCM's tag, asserted where it matters: a flipped bit in the ciphertext
     must stop the upload, not produce a corrupt file with a plausible name.
@@ -146,13 +146,13 @@ def test_a_tampered_chunk_is_refused(tmp_path):
     msg = sealed_upload(session, filename="note.txt", data=b"x" * 64,
                         dir=_root(session).name)
     msg["ct"] = bytes([msg["ct"][0] ^ 0x01]) + msg["ct"][1:]
-    session._do_file_upload(msg)
+    await session._do_file_upload(msg)
 
     assert _errors(session)[0]["code"] == "upload_not_sealed"
     assert not _wrote_anything(tmp_path)
 
 
-def test_an_upload_sealed_for_another_group_is_refused(tmp_path):
+async def test_an_upload_sealed_for_another_group_is_refused(tmp_path):
     """
     The group is the AAD, so a node hosting two groups cannot have a chunk
     moved between them — and a member of one cannot write into the other by
@@ -163,20 +163,20 @@ def test_an_upload_sealed_for_another_group_is_refused(tmp_path):
         session._ctx["gek"], "some-other-group",
         upload_id="u1", chunk_index=0, total_chunks=1,
         filename="note.txt", data=b"x", dir=_root(session).name)
-    session._do_file_upload(msg)
+    await session._do_file_upload(msg)
 
     assert _errors(session)[0]["code"] == "upload_not_sealed"
     assert not _wrote_anything(tmp_path)
 
 
-def test_an_upload_under_another_key_is_refused(tmp_path):
+async def test_an_upload_under_another_key_is_refused(tmp_path):
     """A peer past the handshake with the wrong GEK still writes nothing."""
     session = _session(tmp_path)
     msg = file_upload_wire(
         generate_gek(), GROUP,
         upload_id="u1", chunk_index=0, total_chunks=1,
         filename="note.txt", data=b"x", dir=_root(session).name)
-    session._do_file_upload(msg)
+    await session._do_file_upload(msg)
 
     assert _errors(session)[0]["code"] == "upload_not_sealed"
     assert not _wrote_anything(tmp_path)
@@ -199,7 +199,7 @@ def test_an_ack_replayed_as_a_request_does_not_open(tmp_path):
         file_upload_payload(session._ctx["gek"], GROUP, ack)
 
 
-def test_a_group_with_no_key_refuses_rather_than_falling_back(tmp_path):
+async def test_a_group_with_no_key_refuses_rather_than_falling_back(tmp_path):
     """
     A node whose group has no GEK yet cannot open anything. It must say so, not
     read the message as though it were the old plaintext shape.
@@ -208,7 +208,7 @@ def test_a_group_with_no_key_refuses_rather_than_falling_back(tmp_path):
     msg = sealed_upload(session, filename="note.txt", data=b"x",
                         dir=_root(session).name)
     session._ctx["gek"] = b""
-    session._do_file_upload(msg)
+    await session._do_file_upload(msg)
 
     assert _errors(session)[0]["code"] == "no_group_key"
     assert not _wrote_anything(tmp_path)
@@ -216,7 +216,7 @@ def test_a_group_with_no_key_refuses_rather_than_falling_back(tmp_path):
 
 # ── What must still work ─────────────────────────────────────────────────────
 
-def test_a_multi_chunk_upload_reassembles(tmp_path):
+async def test_a_multi_chunk_upload_reassembles(tmp_path):
     """
     Every chunk is sealed under its own nonce, and the node appends in order.
     Nothing about the seal may change what lands on disk.
@@ -225,7 +225,7 @@ def test_a_multi_chunk_upload_reassembles(tmp_path):
     body = bytes(range(256)) * 40
     parts = [body[i:i + 1024] for i in range(0, len(body), 1024)]
     for i, part in enumerate(parts):
-        session._do_file_upload(sealed_upload(
+        await session._do_file_upload(sealed_upload(
             session, filename="blob.bin", data=part,
             chunk_index=i, total_chunks=len(parts), dir=_root(session).name))
 
@@ -247,7 +247,7 @@ def test_two_identical_chunks_do_not_reuse_a_nonce(tmp_path):
     assert a["ct"] != b["ct"]
 
 
-def test_a_sealed_payload_is_authenticated_not_validated(tmp_path):
+async def test_a_sealed_payload_is_authenticated_not_validated(tmp_path):
     """
     Opening a payload proves a member wrote it, not that they wrote something
     sensible. A member can seal anything, so the fields still need their types
@@ -261,7 +261,7 @@ def test_a_sealed_payload_is_authenticated_not_validated(tmp_path):
                     {"filename": "note.txt", "data": "not bytes"},
                     {"filename": "note.txt"}):
         session.sent.clear()
-        session._do_file_upload({
+        await session._do_file_upload({
             "type": MNP.FILE_UPLOAD, "v": "2.0", "upload_id": "u1",
             "chunk_index": 0, "total_chunks": 1,
             **seal(session._ctx["gek"], PURPOSE_UPLOAD, MNP.FILE_UPLOAD,
@@ -273,13 +273,13 @@ def test_a_sealed_payload_is_authenticated_not_validated(tmp_path):
     assert not _wrote_anything(tmp_path)
 
 
-def test_a_peer_controlled_chunk_index_cannot_crash_the_handler(tmp_path):
+async def test_a_peer_controlled_chunk_index_cannot_crash_the_handler(tmp_path):
     """`chunk_index` is outside the seal by necessity, so it is unchecked input."""
     session = _session(tmp_path)
     msg = sealed_upload(session, filename="note.txt", data=b"x",
                         dir=_root(session).name)
     msg["chunk_index"] = "zero"
-    session._do_file_upload(msg)
+    await session._do_file_upload(msg)
 
     assert _errors(session)[0]["code"] == "bad_chunk_index"
     assert not _wrote_anything(tmp_path)