diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/daemon.py')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/daemon.py | 58 |
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: """ |