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

diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index e1cab61..371c2f0 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -32,6 +32,7 @@ import logging
 import os
 import signal
 import sys
+import time
 from pathlib import Path
 
 import uvicorn
@@ -51,6 +52,7 @@ from meshbay_node.indexer.enrich_audio import AudioEnricher
 from meshbay_node.indexer.enrich_photo import PhotoEnricher
 from meshbay_node.media_cache import MediaCache
 from meshbay_node.tmdb import TmdbClient
+from meshbay_node import uploads as uploads_mod
 from meshbay_node.musicbrainz import MusicBrainzClient
 from meshbay_node.keystore import create_keystore, load_keystore, load_or_create_keystore
 from meshbay_node.platform import chmod_private, config_dir, data_dir, state_dir
@@ -249,6 +251,8 @@ class NodeDaemon:
         self._tasks.append(asyncio.create_task(ui_server.serve()))
         log.info("Control API on 127.0.0.1:%d", self._config.node.ui_port)
 
+        self._tasks.append(asyncio.create_task(self._reap_partial_uploads()))
+
         # 3. Hub connection (Ed25519 auth — retries until node key is linked)
         hub_cfg = HubConfig(
             hub_url=self._config.hub.url,
@@ -1057,6 +1061,60 @@ class NodeDaemon:
                       group_id[:8], e)
             return 0
 
+    async def _reap_partial_uploads(self, interval: float = 3600.0,
+                                    first_delay: float = 60.0) -> None:
+        """
+        Delete `.part` files that no upload will ever finish.
+
+        An upload interrupted for good leaves its partial file behind, and
+        nothing else ever looks at it: `.part` is not an index entry, so it is
+        invisible to every group member and to the operator's own file list. One
+        abandoned film is a gigabyte of their disk, kept for ever.
+
+        Two conditions, both required, and `uploads.orphaned_parts` is where
+        they are stated and tested. What this adds is the walk and the deletion,
+        and one rule of its own: it runs a minute after start rather than at
+        once, so a client reconnecting to finish an upload that outlived a node
+        restart is not raced by the janitor that would have deleted it — the age
+        threshold makes that impossible in practice, and doing it anyway costs a
+        minute.
+
+        `interval` and `first_delay` are parameters so a test can drive this
+        without waiting an hour.
+        """
+        await asyncio.sleep(first_delay)
+        while True:
+            try:
+                self._reap_once()
+            except Exception as exc:   # never let the janitor kill the node
+                log.warning("Reaping partial uploads failed: %s", exc)
+            await asyncio.sleep(interval)
+
+    def _reap_once(self, now: float | None = None) -> int:
+        """One pass over every group. Returns how many files were deleted."""
+        groups = (self._webrtc._ctx.get("groups") or {}) if self._webrtc else {}
+        when = time.time() if now is None else now
+        deleted = 0
+        for gid, ctx in groups.items():
+            roots = ctx.get("roots")
+            if roots is None:
+                continue
+            store = ctx.get("partial_uploads")
+            live = store.live_paths() if store is not None else set()
+            for path in uploads_mod.orphaned_parts(
+                    uploads_mod.find_parts(roots.roots), live, when):
+                try:
+                    size = path.stat().st_size
+                    path.unlink()
+                except OSError as exc:
+                    log.warning("Could not remove abandoned upload %s: %s",
+                                path.name, exc)
+                    continue
+                deleted += 1
+                log.info("Removed abandoned upload %s (%d bytes, group %s)",
+                         path.name, size, gid[:8])
+        return deleted
+
     async def _progress_pusher(self, indexer: DirectoryIndexer,
                                interval: float = 2.0) -> None:
         """