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>

4 files changed, 68 insertions, 25 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index 69c7e66..26f73fc 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -45,7 +45,7 @@ from meshbay_node.audit import RETENTION_DAYS as AUDIT_RETENTION_DAYS, AuditStor
 from meshbay_node.bundle_store import BundleStore
 from meshbay_node.chat.store import ChatStore
 from meshbay_node.config import Config, DEFAULT_CONFIG_PATH, load_config, write_example_config
-from meshbay_node.roots import RootSet, RootError, entry_abs_path
+from meshbay_node.roots import RootSet, RootError, entry_abs_path, off_disk
 from meshbay_node.hub_client import HubClient, HubConfig
 from meshbay_node.indexer import DirectoryIndexer, IndexCache, GroupIndex
 from meshbay_node.indexer.enrich import Enricher
@@ -331,7 +331,7 @@ class NodeDaemon:
                     log.error("Group %r: %s — skipping", group_cfg.name, e)
                     continue
 
-                roots.refresh_availability()
+                await off_disk(roots, roots.refresh_availability)
                 if not any(r.available for r in roots):
                     # Not skipped for being empty: a group whose only drive is
                     # unplugged still exists, and its index is frozen rather
@@ -826,7 +826,7 @@ class NodeDaemon:
             # reports "nothing changed" for exactly that edit.
             if _root_shape(ctx["roots"]) == _root_shape(roots):
                 continue
-            roots.refresh_availability()
+            await off_disk(roots, roots.refresh_availability)
             indexer = next((i for i in self._indexers
                             if i.group_id == group_cfg.id), None)
             if indexer is None:
@@ -869,7 +869,7 @@ class NodeDaemon:
             except RootError as e:
                 log.error("New group %r: %s — skipping", group_cfg.name, e)
                 continue
-            roots.refresh_availability()
+            await off_disk(roots, roots.refresh_availability)
 
             gek = None
             if sk_x_raw and pk_x_raw:
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
index c1b151a..887d435 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
@@ -40,7 +40,7 @@ from meshbay_common.paths import fold, find_fold_collisions, long_path
 from meshbay_common.protocol import IndexEntry
 from meshbay_node.indexer.cache import IndexCache
 from meshbay_node.indexer.group_index import GroupIndex
-from meshbay_node.roots import Root, RootSet
+from meshbay_node.roots import Root, RootSet, off_disk
 
 log = logging.getLogger(__name__)
 
@@ -387,7 +387,7 @@ class DirectoryIndexer:
             await self._initial_scan()
 
     async def _initial_scan(self) -> None:
-        self.roots.refresh_availability()
+        await off_disk(self.roots, self.roots.refresh_availability)
         total = 0
         waiting = [r.name for r in self.roots if r.available]
         self._queue(waiting)
@@ -695,7 +695,7 @@ class DirectoryIndexer:
                     self._index.remove_entry(entry.id)
 
         self.roots = roots
-        roots.refresh_availability()
+        await off_disk(roots, roots.refresh_availability)
         added = [r for r in roots if r.folded not in old_names and r.available]
 
         self._index.roots = roots.describe()
@@ -794,7 +794,7 @@ class DirectoryIndexer:
             return await self._reconcile()
 
     async def _reconcile(self) -> bool:
-        changed = self.roots.refresh_availability()
+        changed = await off_disk(self.roots, self.roots.refresh_availability)
         touched = False
 
         # Drained before the loop below, because persisting the flag is what
@@ -1042,7 +1042,7 @@ class DirectoryIndexer:
         if root is None:
             return
         root.ejected = False
-        root.available = root.is_live()
+        root.available = await off_disk(self.roots, root.is_live)
         if not root.available:
             await self._finish_plug(None)
             return
@@ -1071,7 +1071,8 @@ class DirectoryIndexer:
                     # Ejected or removed again while it waited. `_rescan_root`
                     # drops the entries before it walks, so going ahead would
                     # empty a root that is not there to be read.
-                    if self._holds(root) and not root.ejected and root.is_live():
+                    live = await off_disk(self.roots, root.is_live)
+                    if self._holds(root) and not root.ejected and live:
                         await self._rescan_root(root)
             finally:
                 if waiting:
@@ -1143,7 +1144,7 @@ class DirectoryIndexer:
             if root is None:
                 return
 
-            if deleted and not root.is_live():
+            if deleted and not await off_disk(self.roots, root.is_live):
                 # The volume went away rather than the file. Freeze: mark the
                 # root and touch nothing. Every other event for this root
                 # will arrive here too and be dropped the same way, which is
diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py
index 8cd893c..2221bea 100644
--- a/packages/meshbay-node/src/meshbay_node/ops.py
+++ b/packages/meshbay-node/src/meshbay_node/ops.py
@@ -39,7 +39,7 @@ from meshbay_common.crypto import (
 )
 from meshbay_node.config import DEFAULT_CONFIG_PATH
 from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR
-from meshbay_node.roots import RootError, RootSet
+from meshbay_node.roots import RootError, RootSet, off_disk
 
 log = logging.getLogger(__name__)
 
@@ -1130,7 +1130,7 @@ async def plug_root(state: dict, group_id: str, root_name: str) -> dict:
     if not root.ejected:
         return {"status": "already_plugged", "name": root_name,
                 "group_id": group_id, "roots": roots.describe()}
-    if not root.is_live():
+    if not await off_disk(roots, root.is_live):
         raise OpError(
             f"Directory not found: {root.path}. Is the device connected?",
             status=409)
@@ -1145,7 +1145,7 @@ async def plug_root(state: dict, group_id: str, root_name: str) -> dict:
     if indexer:
         await indexer.plug_root(root_name)
     root.ejected = False
-    root.available = root.is_live()
+    root.available = await off_disk(roots, root.is_live)
 
     log.info("Root plugged: %s in group %s", root_name, group_id[:8])
     return {"status": "plugged", "name": root_name, "group_id": group_id,
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 e748be9..92f2951 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -644,7 +644,7 @@ class WebRTCPeerSession:
             elif mtype == MNP.TRANSFER_CLOSE:
                 self._do_transfer_close(msg)
             elif mtype == MNP.FILE_UPLOAD:
-                self._do_file_upload(msg)
+                self._spawn(self._do_file_upload(msg))
             elif mtype == MNP.DIR_CREATE:
                 self._do_dir_create(msg)
             elif mtype == MNP.DIR_DELETE:
@@ -5205,7 +5205,43 @@ class WebRTCPeerSession:
             ctx["partial_uploads"] = store
         return store
 
-    def _do_file_upload(self, msg: dict) -> None:
+    @staticmethod
+    def _upload_lock(ctx: dict) -> asyncio.Lock:
+        """
+        One lock per group, beside the state it protects.
+
+        Not per session: `partial_uploads` lives in the group context so a
+        client that reconnects finds its upload where it left it, which means
+        two sessions of the same member share the position of one `.part` file.
+        A lock on the session would let them interleave — and the loop no longer
+        serializes them for free now that a chunk write is awaited.
+        """
+        lock = ctx.get("upload_lock")
+        if lock is None:
+            lock = asyncio.Lock()
+            ctx["upload_lock"] = lock
+        return lock
+
+    async def _do_file_upload(self, msg: dict) -> None:
+        """
+        One chunk of an upload, in the order it arrived.
+
+        The chunk ordering rule — `chunk_index != state.next_index` is refused —
+        used to hold for free: the handler was synchronous, so nothing could run
+        between the check and the `advance` that answers it. Awaiting the write
+        opens that gap, and two chunks of one upload racing through it is a
+        `.part` file with a hole in it or a chunk refused for arriving on time.
+        So the check, the write and the advance are one critical section again.
+
+        The order is the arrival order: `_dispatch_message` runs per message as
+        it arrives and creates these tasks in that order, tasks start in
+        creation order, and this lock is the first thing each one waits on, so
+        its queue of waiters is in arrival order too.
+        """
+        async with self._upload_lock(self._group_ctx()):
+            await self._upload_chunk(msg)
+
+    async def _upload_chunk(self, msg: dict) -> None:
         """
         One chunk of an upload, sealed under the group key (MNP 2.0).
 
@@ -5386,8 +5422,8 @@ class WebRTCPeerSession:
         # client names *where among the group's own folders*, never a path on
         # the operator's filesystem.
         if target_rel:
-            target_dir = roots.resolve(target_rel)
-            if target_dir is None or not target_dir.is_dir():
+            target_dir = await off_disk(roots, roots.resolve, target_rel)
+            if target_dir is None or not await off_disk(roots, target_dir.is_dir):
                 _refuse("Not a directory in this group", "no_such_directory")
                 return
             rel_dir = target_rel
@@ -5396,7 +5432,7 @@ class WebRTCPeerSession:
             # one destination is.
             target_dir = upload_root.path
             rel_dir = upload_root.name
-            if not target_dir.is_dir():
+            if not await off_disk(roots, target_dir.is_dir):
                 _refuse("That directory is currently unavailable",
                         "root_unavailable")
                 return
@@ -5419,7 +5455,8 @@ class WebRTCPeerSession:
         # A shared directory means two people can send the same name. Refusing the
         # second is safe but silly — everyone's camera produces IMG_1234.jpg — so
         # a free name is found instead. Never a replacement.
-        stored_name = state.stored_name if state else _free_name(target_dir, filename)
+        stored_name = (state.stored_name if state
+                       else await off_disk(roots, _free_name, target_dir, filename))
         tmp_path = target_dir / f"{stored_name}{uploads_mod.PART_SUFFIX}"
         final_path = target_dir / stored_name
 
@@ -5449,7 +5486,7 @@ class WebRTCPeerSession:
         if chunk_index == 0:
             # Backstop: _free_name already guarantees this, and it stays because
             # it asserts the invariant where the write happens.
-            if final_path.exists():
+            if await off_disk(roots, final_path.exists):
                 _refuse("File already exists", "already_exists")
                 return
             state = uploads.start(user_id, rel_dir, filename, stored_name,
@@ -5466,12 +5503,11 @@ class WebRTCPeerSession:
 
         if state.bytes + len(chunk_bytes) > MAX_UPLOAD_BYTES:
             uploads.drop(user_id, rel_dir, filename)
-            tmp_path.unlink(missing_ok=True)
+            await off_disk(roots, tmp_path.unlink, True)
             _refuse("Upload exceeds size limit", "too_large")
             return
 
-        with open(tmp_path, "wb" if chunk_index == 0 else "ab") as f:
-            f.write(chunk_bytes)
+        await off_disk(roots, _append_chunk, tmp_path, chunk_bytes, chunk_index == 0)
         uploads.advance(user_id, rel_dir, filename, chunk_index, len(chunk_bytes))
 
         self._send(file_upload_ack_wire(
@@ -5487,7 +5523,7 @@ class WebRTCPeerSession:
 
         if chunk_index + 1 >= total_chunks:
             uploads.drop(user_id, rel_dir, filename)
-            tmp_path.rename(final_path)
+            await off_disk(roots, tmp_path.rename, final_path)
             log.info("Upload complete: %s (%d chunks, %d bytes)",
                      stored_name, total_chunks, state.bytes)
             self._audit("file_upload", f"{rel_dir}/{stored_name}")
@@ -6589,6 +6625,12 @@ def _locate(roots: RootSet, entry) -> tuple[Path | None, str | None]:
     return path, None
 
 
+def _append_chunk(tmp_path: Path, chunk_bytes: bytes, first: bool) -> None:
+    """Add one chunk to a partial upload. Blocking; called through `off_disk`."""
+    with open(tmp_path, "wb" if first else "ab") as f:
+        f.write(chunk_bytes)
+
+
 def _read_and_encrypt(
     gek: bytes,
     file_path: Path,