feat(node): uploads outlive their connection, and their leftovers are reaped

Stage 8 of ~/next/improve-downloads.md, first half. Two defects that are the
same defect seen from two sides.

An upload's progress lived on the session, keyed by `rel_dir/filename`. A
dropped connection threw it away and the client's next chunk was refused with
`not_started`: an upload interrupted at 99% could only be started again from
zero, on a link flaky enough to have interrupted it once. It now lives in the
group context, keyed by member as well -- a shared directory means two people
can be sending IMG_1234.jpg at the same moment and neither may inherit, or
overwrite the position of, the other's.

What the lost state left behind was a `.part` nothing would ever finish, delete
or look at again. It is not an index entry, so it is invisible to every member
and to the operator's own file list: one abandoned film is a gigabyte of their
disk, kept for ever. That leak predates this branch.

A `.part` is deleted only when **both** hold: no upload is writing it, and
nothing has been written to it for 24 hours. Waiting costs disk; being wrong
costs somebody their upload, and is not reversible -- so a read-only root is
never walked (it cannot have received an upload), an unavailable one is never
walked (an unmounted drive reporting "nothing found" is how a careless janitor
deletes a library), and a file whose mtime is in the future is left alone (a
clock that went backwards is not evidence). The reaper matches whole paths and
the state records the path it is writing, rather than both sides rebuilding one
from a root name -- two implementations of one rule whose failure mode is
deleting a live upload.

The rules are in `uploads.py`, pure logic with no asyncio and no transport, the
same shape as `transfers.py` and for the same reason.

23 cases, four of them checked against the unfixed source. Node suite 1195
passed.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HCGdheDLxGReuKHga3BtST

1 files changed, 41 insertions, 14 deletions

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 ce599cc..6bae27c 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -129,6 +129,7 @@ from meshbay_node.indexer import GroupIndex
 from meshbay_node.indexer.indexer import DirectoryIndexer
 from meshbay_node import linkpreview, ops, platform
 from meshbay_node import transfers as transfers_mod
+from meshbay_node import uploads as uploads_mod
 from meshbay_node.transfers import TransferSlots
 # Re-imported under its original name: every call site and existing test in
 # this module still refers to it as `_probe_video`. The implementation lives
@@ -426,7 +427,8 @@ class WebRTCPeerSession:
         self._join_attempts = 0
         self._nonce_client: bytes = b""
         self._admin_ops: dict[str, dict] = {}   # op_id → pending admin operation
-        self._uploads: dict[str, dict] = {}   # filename → {next_index, bytes}
+        # Uploads in progress live in the group context, not here: see
+        # `_partial_uploads` and `uploads.py`.
         # Diagnostics only (_WEBRTC_TRACE): when the last DataChannel message
         # arrived, so the heartbeat can report silence duration.
         self._last_msg_at: float = 0.0
@@ -4757,6 +4759,19 @@ class WebRTCPeerSession:
                                       if k not in ("type", "v")})
         self._send(resp)
 
+    def _partial_uploads(self, ctx: dict) -> uploads_mod.PartialUploads:
+        """This group's uploads in progress, created on first use.
+
+        In the group context rather than on the session, so a client that
+        reconnects finds its own upload where it left it — and so the reaper has
+        something to ask "is anyone still writing this?".
+        """
+        store = ctx.get("partial_uploads")
+        if store is None:
+            store = uploads_mod.PartialUploads()
+            ctx["partial_uploads"] = store
+        return store
+
     def _do_file_upload(self, msg: dict) -> None:
         """
         One chunk of an upload, sealed under the group key (MNP 2.0).
@@ -4912,13 +4927,26 @@ class WebRTCPeerSession:
                         "root_unavailable")
                 return
 
-        upload_key = f"{rel_dir}/{filename}"
-        state = self._uploads.get(upload_key)
+        # Held by the group, not by this connection.
+        #
+        # This used to be `self._uploads`, on the session. A dropped link threw
+        # the position away and the next chunk was refused with `not_started`:
+        # an upload interrupted at 99% could only be started again from zero, on
+        # a connection flaky enough to have interrupted it once. And the state
+        # it lost was the only thing that knew about the `.part` file left
+        # behind — see `uploads.orphaned_parts`, which is the other half of this.
+        #
+        # Keyed by member as well as by name, because a shared directory means
+        # two people can be sending IMG_1234.jpg at the same moment and neither
+        # may inherit the other's position.
+        uploads = self._partial_uploads(ctx)
+        user_id = self._user_id or ""
+        state = uploads.get(user_id, rel_dir, filename)
         # 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)
-        tmp_path = target_dir / f"{stored_name}.part"
+        stored_name = state.stored_name if state else _free_name(target_dir, filename)
+        tmp_path = target_dir / f"{stored_name}{uploads_mod.PART_SUFFIX}"
         final_path = target_dir / stored_name
 
         if chunk_index == 0:
@@ -4927,28 +4955,27 @@ class WebRTCPeerSession:
             if final_path.exists():
                 _refuse("File already exists", "already_exists")
                 return
-            state = {"next_index": 0, "bytes": 0, "stored_name": stored_name}
-            self._uploads[upload_key] = state
+            state = uploads.start(user_id, rel_dir, filename, stored_name,
+                                  part_path=tmp_path)
         elif state is None:
             _refuse("Upload not started", "not_started")
             return
 
         # Reject out-of-order or replayed chunks — otherwise chunk_index>0 appends
         # blindly to whatever .part file is already on disk.
-        if chunk_index != state["next_index"]:
+        if chunk_index != state.next_index:
             _refuse("Unexpected chunk index", "bad_chunk_index")
             return
 
-        if state["bytes"] + len(chunk_bytes) > MAX_UPLOAD_BYTES:
-            self._uploads.pop(upload_key, None)
+        if state.bytes + len(chunk_bytes) > MAX_UPLOAD_BYTES:
+            uploads.drop(user_id, rel_dir, filename)
             tmp_path.unlink(missing_ok=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)
-        state["next_index"] = chunk_index + 1
-        state["bytes"] += len(chunk_bytes)
+        uploads.advance(user_id, rel_dir, filename, chunk_index, len(chunk_bytes))
 
         self._send(file_upload_ack_wire(
             gek, self._group_id or "",
@@ -4962,10 +4989,10 @@ class WebRTCPeerSession:
         ))
 
         if chunk_index + 1 >= total_chunks:
-            self._uploads.pop(upload_key, None)
+            uploads.drop(user_id, rel_dir, filename)
             tmp_path.rename(final_path)
             log.info("Upload complete: %s (%d chunks, %d bytes)",
-                     stored_name, total_chunks, state["bytes"])
+                     stored_name, total_chunks, state.bytes)
             self._audit("file_upload", f"{rel_dir}/{stored_name}")
             self._register_uploader(ctx, rel_dir, stored_name)