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, 190 insertions, 0 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/uploads.py b/packages/meshbay-node/src/meshbay_node/uploads.py
new file mode 100644
index 0000000..f8ae7f9
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/uploads.py
@@ -0,0 +1,190 @@
+"""
+Partial uploads: the state that must outlive a connection, and the files that
+must not outlive their upload.
+
+Two defects live here, and they are the same defect seen from two sides.
+
+An upload's progress was kept on the **session** — `WebRTCSession._uploads`,
+keyed by `rel_dir/filename`. A dropped connection therefore lost it, and the
+client's next chunk was refused with `not_started`: an upload interrupted at
+99% could only be started again from zero. The state belongs to the group, not
+to the connection that happened to carry it, and it is keyed by member as well,
+because a shared directory means two people can be sending `IMG_1234.jpg` at
+the same time and neither may inherit the other's position.
+
+And what the lost state left behind was a `.part` file that nothing would ever
+finish, delete or even look at again. One abandoned upload of a film is a
+gigabyte of somebody else's disk, kept for ever, invisible in the index because
+`.part` is not an index entry. That is the leak this module reaps.
+
+Pure logic, no asyncio and no transport — the same shape as `transfers.py`, and
+for the same reason: the rules are worth testing without a WebRTC connection to
+build first.
+"""
+
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Iterable
+
+# What an unfinished upload is called on disk while it is being written. The
+# node has always used this; it is named here because the reaper below has to
+# recognise one, and a second spelling of it would be a bug nobody could see.
+PART_SUFFIX = ".part"
+
+# How long a `.part` with no upload behind it is kept before it is deleted.
+#
+# Generous on purpose. The cost of waiting is disk; the cost of being wrong is
+# deleting an upload somebody is still making, which is unrecoverable and looks
+# to them like a transfer that failed for no reason. A day covers a laptop
+# closed overnight, a phone that lost signal in a tunnel, and a client that
+# reconnects on the next launch — all of which are resumable and none of which
+# should be swept.
+ORPHAN_AFTER_SECS = 24 * 3600
+
+
+@dataclass
+class Partial:
+    """One upload in progress, as the node knows it between two chunks."""
+
+    stored_name: str
+    # The `.part` this upload is writing. Recorded rather than recomputed: the
+    # reaper compares paths, and a path rebuilt from a root name and a relative
+    # directory is a second implementation of something that must agree exactly
+    # with the first, for ever, or a live upload gets deleted.
+    part_path: Path | None = None
+    next_index: int = 0
+    bytes: int = 0
+    updated_at: float = field(default_factory=time.time)
+
+
+class PartialUploads:
+    """
+    Every upload this group has in flight, keyed by member.
+
+    Held in the group context rather than on a session, so that a reconnecting
+    client finds its own upload exactly where it left it. The key is
+    `(user_id, rel_dir, filename)`: the directory and name alone would let one
+    member resume — or clobber the position of — another member's upload of the
+    same name, which a shared folder makes an ordinary occurrence rather than an
+    attack.
+    """
+
+    def __init__(self) -> None:
+        self._by_key: dict[tuple[str, str, str], Partial] = {}
+
+    # ── the state itself ────────────────────────────────────────────────────
+
+    def start(self, user_id: str, rel_dir: str, filename: str,
+              stored_name: str, part_path: Path | None = None,
+              now: float | None = None) -> Partial:
+        """Begin (or begin again) an upload, discarding any earlier position."""
+        state = Partial(stored_name=stored_name, part_path=part_path,
+                        updated_at=time.time() if now is None else now)
+        self._by_key[(user_id, rel_dir, filename)] = state
+        return state
+
+    def get(self, user_id: str, rel_dir: str, filename: str) -> Partial | None:
+        return self._by_key.get((user_id, rel_dir, filename))
+
+    def advance(self, user_id: str, rel_dir: str, filename: str,
+                chunk_index: int, nbytes: int,
+                now: float | None = None) -> Partial | None:
+        """Record one accepted chunk. Returns None if there is no such upload."""
+        state = self._by_key.get((user_id, rel_dir, filename))
+        if state is None:
+            return None
+        state.next_index = chunk_index + 1
+        state.bytes += nbytes
+        # Touched on every chunk, because the reaper measures *silence*, not
+        # age: an upload that has been running for two days is not an orphan,
+        # and one that stopped two days ago is, whatever it started as.
+        state.updated_at = time.time() if now is None else now
+        return state
+
+    def drop(self, user_id: str, rel_dir: str, filename: str) -> Partial | None:
+        return self._by_key.pop((user_id, rel_dir, filename), None)
+
+    def __len__(self) -> int:
+        return len(self._by_key)
+
+    # ── what the reaper must not touch ──────────────────────────────────────
+
+    def live_paths(self) -> set[Path]:
+        """The `.part` files that still have an upload behind them.
+
+        Deliberately without the member's identity: a file on disk has no owner,
+        and the only question the reaper asks is whether anyone is writing it.
+        """
+        return {state.part_path for state in self._by_key.values()
+                if state.part_path is not None}
+
+
+def orphaned_parts(candidates: Iterable[tuple[Path, float]],
+                   live: set[Path],
+                   now: float,
+                   older_than: float = ORPHAN_AFTER_SECS) -> list[Path]:
+    """
+    Which `.part` files may be deleted.
+
+    `candidates` is `(path, mtime)` for every `.part` found under the group's
+    writable roots. A file is an orphan when **both** are true: no upload in
+    `live` is writing it, and nothing has been written to it for `older_than`
+    seconds.
+
+    Both conditions are load-bearing. The first alone would delete an upload
+    that is mid-flight but whose state is held elsewhere; the second alone would
+    keep a file for a day after the upload that owned it was abandoned, which is
+    correct but is also the entire reason this is bounded rather than immediate.
+
+    A file with a future mtime — a clock that went backwards, a filesystem with
+    a different idea of now — is left alone rather than treated as infinitely
+    old, because deleting is not reversible and a wrong clock is not evidence.
+    """
+    doomed: list[Path] = []
+    for path, mtime in candidates:
+        if path.suffix != PART_SUFFIX:
+            continue
+        if path in live:
+            continue
+        age = now - mtime
+        if age < older_than:
+            continue
+        doomed.append(path)
+    return doomed
+
+
+def find_parts(roots: Iterable) -> list[tuple[Path, float]]:
+    """
+    Every `.part` under these roots, with its modification time.
+
+    Only writable, available roots are walked: a read-only root cannot have
+    received an upload, and an unavailable one is a disk that is not mounted —
+    walking it would find nothing and reporting nothing found there is how a
+    reaper deletes an entire library the day a drive is unplugged. (It cannot
+    here, since it only ever deletes what it finds, but the shape of that
+    mistake is worth refusing at the source.)
+
+    Errors are swallowed per entry rather than per walk: one unreadable
+    subdirectory must not stop the rest from being tidied.
+    """
+    found: list[tuple[Path, float]] = []
+    for root in roots:
+        if not getattr(root, "writable", False):
+            continue
+        if not getattr(root, "available", False):
+            continue
+        try:
+            candidates = root.path.rglob(f"*{PART_SUFFIX}")
+        except OSError:
+            continue
+        for path in candidates:
+            try:
+                if not path.is_file():
+                    continue
+                found.append((path, path.stat().st_mtime))
+            except OSError:
+                continue
+    return found