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, 22 insertions, 22 deletions

diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py
index fa55ff6..e203811 100644
--- a/packages/meshbay-node/tests/test_security_regressions.py
+++ b/packages/meshbay-node/tests/test_security_regressions.py
@@ -174,7 +174,7 @@ def _session(tmp_path: Path, user_id: str) -> WebRTCPeerSession:
     return session
 
 
-def test_upload_cannot_overwrite_another_members_file(tmp_path):
+async def test_upload_cannot_overwrite_another_members_file(tmp_path):
     """
     C5a: uploads used to land in the shared root under a client-chosen name and
     overwrite whatever was there. That let any member destroy the operator's
@@ -192,7 +192,7 @@ def test_upload_cannot_overwrite_another_members_file(tmp_path):
     original.write_bytes(b"operator's original content")
 
     attacker = _session(tmp_path, "attacker-user")
-    attacker._do_file_upload(sealed_upload(
+    await attacker._do_file_upload(sealed_upload(
         attacker, filename="important.mp4", data=b"attacker content"))
 
     assert original.read_bytes() == b"operator's original content", (
@@ -200,18 +200,18 @@ def test_upload_cannot_overwrite_another_members_file(tmp_path):
     assert (uploads / "important (2).mp4").read_bytes() == b"attacker content"
 
 
-def test_upload_second_attempt_cannot_replace_own_completed_file(tmp_path):
+async def test_upload_second_attempt_cannot_replace_own_completed_file(tmp_path):
     """C5a: even the original uploader does not get to overwrite."""
     session = _session(tmp_path, "user-1")
-    def _send_it():
+    async def _send_it():
         # Sealed afresh each time: a nonce is drawn per message, so re-sending
         # the same dict would be a replay rather than a second upload.
-        session._do_file_upload(sealed_upload(
+        await session._do_file_upload(sealed_upload(
             session, filename="movie.mp4", data=b"first"))
 
-    _send_it()
+    await _send_it()
     session.sent.clear()
-    _send_it()
+    await _send_it()
     uploads = _uploads_dir(session)
     assert (uploads / "movie.mp4").read_bytes() == b"first", (
         "the first upload was replaced")
@@ -236,7 +236,7 @@ def test_dir_create_cannot_escape_the_shared_root(tmp_path, bad):
     assert set(tmp_path.rglob("*")) == before, f"created something via {bad!r}"
 
 
-def test_the_client_names_a_folder_and_never_a_filesystem_path(tmp_path):
+async def test_the_client_names_a_folder_and_never_a_filesystem_path(tmp_path):
     """
     The destination is now the folder the sender is looking at, which means the
     client does choose it — and the whole of what keeps that safe is that the
@@ -255,7 +255,7 @@ def test_the_client_names_a_folder_and_never_a_filesystem_path(tmp_path):
     for bad in ("../../etc", "/etc", "shared/../..", "shared/../../etc",
                 "nope", "shared/missing"):
         session.sent.clear()
-        session._do_file_upload(sealed_upload(
+        await session._do_file_upload(sealed_upload(
             session, filename="note.txt", data=b"x", dir=bad))
         refusal = [m for m in session.sent if m.get("type") == "error"]
         assert refusal, f"{bad!r} was accepted"
@@ -264,7 +264,7 @@ def test_the_client_names_a_folder_and_never_a_filesystem_path(tmp_path):
     assert set(tmp_path.rglob("*")) == before, "a refused upload still wrote"
 
 
-def test_an_upload_lands_in_the_folder_it_names(tmp_path):
+async def test_an_upload_lands_in_the_folder_it_names(tmp_path):
     """
     And in that folder itself — the `uploads/` subdirectory the node used to
     create is gone. Somebody dropping a file into the folder they are looking
@@ -274,7 +274,7 @@ def test_an_upload_lands_in_the_folder_it_names(tmp_path):
     root = session._ctx["roots"].roots[0]
     (root.path / "Albums").mkdir()
 
-    session._do_file_upload(sealed_upload(
+    await session._do_file_upload(sealed_upload(
         session, filename="note.txt", data=b"x", dir=f"{root.name}/Albums"))
 
     assert (root.path / "Albums" / "note.txt").read_bytes() == b"x"
@@ -283,7 +283,7 @@ def test_an_upload_lands_in_the_folder_it_names(tmp_path):
     assert not (root.path / "uploads").exists()
 
 
-def test_an_upload_goes_to_the_root_it_names(tmp_path):
+async def test_an_upload_goes_to_the_root_it_names(tmp_path):
     """
     With two writable roots there is no defensible default, and the client is
     the only party that knows which directory the person is looking at. The
@@ -301,14 +301,14 @@ def test_an_upload_goes_to_the_root_it_names(tmp_path):
         {"path": str(incoming), "writable": True},
     ])
 
-    session._do_file_upload(sealed_upload(
+    await session._do_file_upload(sealed_upload(
         session, filename="note.txt", data=b"x", dir="Incoming"))
 
     assert (incoming / "note.txt").read_bytes() == b"x"
     assert not (media / "note.txt").exists(), "it went to the first root instead"
 
 
-def test_a_read_only_root_refuses_an_upload(tmp_path):
+async def test_a_read_only_root_refuses_an_upload(tmp_path):
     """
     RO is the mechanism now, not a hidden button. It binds the operator too:
     "read-only for everyone" is what makes a published library one, and an
@@ -321,7 +321,7 @@ def test_a_read_only_root_refuses_an_upload(tmp_path):
     session._ctx["roots"] = RootSet.build([{"path": str(published)}])
     session._is_node_admin = lambda: True
 
-    session._do_file_upload(sealed_upload(
+    await session._do_file_upload(sealed_upload(
         session, filename="note.txt", data=b"x", dir="Published"))
 
     refusal = [m for m in session.sent if m.get("type") == "error"]
@@ -329,7 +329,7 @@ def test_a_read_only_root_refuses_an_upload(tmp_path):
     assert not (published / "note.txt").exists()
 
 
-def test_a_fully_read_only_group_refuses_an_unaddressed_upload(tmp_path):
+async def test_a_fully_read_only_group_refuses_an_unaddressed_upload(tmp_path):
     """
     An MNP 1.0 client names no root, so the node falls back to the first
     writable one. There isn't one here, and the fallback must refuse rather
@@ -340,7 +340,7 @@ def test_a_fully_read_only_group_refuses_an_unaddressed_upload(tmp_path):
     session = _session(tmp_path, "user-1")
     session._ctx["roots"] = RootSet.build([{"path": str(published)}])
 
-    session._do_file_upload(sealed_upload(
+    await session._do_file_upload(sealed_upload(
         session, filename="note.txt", data=b"x"))
 
     refusal = [m for m in session.sent if m.get("type") == "error"]
@@ -348,7 +348,7 @@ def test_a_fully_read_only_group_refuses_an_unaddressed_upload(tmp_path):
     assert not (published / "note.txt").exists()
 
 
-def test_an_ejected_root_refuses_an_upload(tmp_path):
+async def test_an_ejected_root_refuses_an_upload(tmp_path):
     """
     Writing to a drive somebody has their hand on is the thing eject exists to
     stop. `writable` is still true — that is configuration — so availability
@@ -363,7 +363,7 @@ def test_an_ejected_root_refuses_an_upload(tmp_path):
     roots.roots[0].available = False
     session._ctx["roots"] = roots
 
-    session._do_file_upload(sealed_upload(
+    await session._do_file_upload(sealed_upload(
         session, filename="note.txt", data=b"x", dir="USB"))
 
     refusal = [m for m in session.sent if m.get("type") == "error"]
@@ -371,19 +371,19 @@ def test_an_ejected_root_refuses_an_upload(tmp_path):
     assert not (usb / "note.txt").exists()
 
 
-def test_two_members_can_send_the_same_filename(tmp_path):
+async def test_two_members_can_send_the_same_filename(tmp_path):
     """
     One shared uploads/ means collisions are ordinary — every camera produces
     IMG_1234.jpg. The second gets a free name; neither replaces the other.
     """
     first = _session(tmp_path, "user-1")
-    first._do_file_upload(sealed_upload(
+    await first._do_file_upload(sealed_upload(
         first, filename="IMG_1234.jpg", data=b"first"))
     second = _session(tmp_path, "user-2")
     # Same group, so the same key: `_session` builds one per call, and two
     # members of one group do not have two.
     second._ctx["gek"] = first._ctx["gek"]
-    second._do_file_upload(sealed_upload(
+    await second._do_file_upload(sealed_upload(
         second, filename="IMG_1234.jpg", data=b"second"))
 
     uploads = _uploads_dir(first)