diff --git a/backend/routers/sessions.py b/backend/routers/sessions.py index d0bdf8c..8c0fa3b 100644 --- a/backend/routers/sessions.py +++ b/backend/routers/sessions.py @@ -15,8 +15,10 @@ from services.retention import ( RetentionError, get_retention, + get_sweep_status, save_retention, session_index, + start_sweep, sweep, ) @@ -248,11 +250,19 @@ async def update_retention( config = save_retention(enabled, amount, unit) if not enabled: - return {"retention": config, "count": 0, "freed_bytes": 0} + return {"retention": config, "started": False} - result = await asyncio.to_thread(sweep, amount, unit) + # Deleting a large backlog takes a while, so run it in the background + # and let the client poll for progress rather than holding the request. + status = await start_sweep(amount, unit) except RetentionError as e: raise HTTPException(status_code=400, detail=str(e)) _events_cache.clear() - return {"retention": get_retention(), **result} + return {"retention": config, "started": True, "status": status} + + +@router.get("/storage/retention/status") +async def retention_status(): + """Progress of the running or most recent sweep.""" + return get_sweep_status() diff --git a/backend/services/retention.py b/backend/services/retention.py index 645f3e4..c838adf 100644 --- a/backend/services/retention.py +++ b/backend/services/retention.py @@ -42,6 +42,21 @@ class RetentionError(ValueError): """Invalid retention settings.""" +# State of the running/last sweep, polled by the UI so a large delete shows +# progress instead of a request that just hangs. +# state is "idle" | "running" | "done" | "error" +_sweep_status: dict = {"state": "idle", "message": "", "done": 0, "total": 0, + "count": 0, "freed_bytes": 0} + + +def get_sweep_status() -> dict: + return dict(_sweep_status) + + +def _set_status(**fields) -> None: + _sweep_status.update(fields) + + def cutoff_for(amount: int, unit: str, now: datetime | None = None) -> datetime: """The date before which sessions are considered expired.""" now = now or datetime.now(timezone.utc) @@ -161,8 +176,6 @@ def expired_sessions(amount: int, unit: str) -> list[dict]: def sweep(amount: int, unit: str, dry_run: bool = False) -> dict: """Delete every stored session older than the given age.""" - from services.process import delete_session - stale = expired_sessions(amount, unit) if dry_run: return { @@ -172,16 +185,42 @@ def sweep(amount: int, unit: str, dry_run: bool = False) -> dict: "sessions": stale, } - freed = 0 - deleted = 0 + # Bulk delete rather than one call per session: on R2 a per-session loop is + # several sequential round trips each, which for a large first sweep means + # minutes with the request held open. + # + # No tombstones here. They exist to stop auto_precompute re-downloading a + # session the user removed, and it only ever looks at the last 7 days — + # retention never deletes anything newer than a week. + prefixes = [f"sessions/{s['year']}/{s['round']}/{s['type']}" for s in stale] + freed, deleted = 0, 0 + try: + _set_status(state="running", message="Finding files to delete…") + sizes = storage.keys_under(prefixes) + keys = list(sizes) + + def progress(done: int, total: int): + _set_status( + state="running", + message=f"Deleting {len(stale)} session{'' if len(stale) == 1 else 's'}…", + done=done, + total=total, + ) + + progress(0, len(keys)) + storage.delete_keys(keys, on_progress=progress) + freed = sum(sizes.values()) + deleted = len(stale) + except Exception as e: + logger.error(f"Retention sweep failed: {e}") + _set_status(state="error", message="Delete failed.") + for s in stale: try: - freed += delete_session(s["year"], s["round"], s["type"]) - deleted += 1 - except Exception as e: - logger.warning( - f"Retention: failed to delete {s['year']}/{s['round']}/{s['type']}: {e}" - ) + from routers.replay import evict_cached_session + evict_cached_session(s["year"], s["round"], s["type"]) + except Exception: + pass if deleted: logger.info(f"Retention: deleted {deleted} session(s), freed {freed} bytes") @@ -194,9 +233,34 @@ def sweep(amount: int, unit: str, dry_run: bool = False) -> dict: }) storage.put_json(CONFIG_PATH, config) + if _sweep_status.get("state") != "error": + _set_status( + state="done", + message=f"Deleted {deleted} session{'' if deleted == 1 else 's'}.", + count=deleted, + freed_bytes=freed, + ) + return {"dry_run": False, "count": deleted, "freed_bytes": freed, "sessions": stale} +async def start_sweep(amount: int, unit: str) -> dict: + """Run a sweep in the background so the UI can poll for progress.""" + _sweep_status.clear() + _sweep_status.update({"state": "running", "message": "Starting…", + "done": 0, "total": 0, "count": 0, "freed_bytes": 0}) + + async def run(): + try: + await asyncio.to_thread(sweep, amount, unit) + except Exception as e: + logger.error(f"Sweep task failed: {e}") + _set_status(state="error", message="Delete failed.") + + asyncio.create_task(run()) + return get_sweep_status() + + async def retention_loop(): """Apply the retention policy shortly after startup, then once a day.""" logger.info("Retention sweep task started") diff --git a/backend/services/storage.py b/backend/services/storage.py index bdef87d..d877855 100644 --- a/backend/services/storage.py +++ b/backend/services/storage.py @@ -205,6 +205,32 @@ def _r2_delete(path: str) -> int: return size +def _r2_delete_prefixes(prefixes: list[str]) -> int: + """Delete many prefixes with one listing and batched deletes. + + Deleting a hundred sessions one at a time is hundreds of sequential + round trips; this keeps it to a single listing plus one call per 1000 keys. + """ + client = _get_r2_client() + bucket = _r2_bucket() + + wanted = tuple(p if p.endswith("/") else p + "/" for p in prefixes) + # List once from the deepest shared root rather than per prefix. + root = os.path.commonprefix(list(wanted)) + root = root[: root.rfind("/") + 1] if "/" in root else "" + sizes = {k: v for k, v in _r2_list_sizes(root).items() if k.startswith(wanted)} + if not sizes: + return 0 + + keys = list(sizes) + for i in range(0, len(keys), 1000): + client.delete_objects( + Bucket=bucket, + Delete={"Objects": [{"Key": k} for k in keys[i:i + 1000]]}, + ) + return sum(sizes.values()) + + def _r2_delete_prefix(prefix: str) -> int: client = _get_r2_client() bucket = _r2_bucket() @@ -282,3 +308,65 @@ def delete_prefix(prefix: str) -> int: if _mode() == "r2": return _r2_delete_prefix(cleaned) return _local_delete_prefix(cleaned) + + +def delete_prefixes(prefixes: list[str]) -> int: + """Delete every file under each of *prefixes*. Returns bytes freed.""" + cleaned = [] + for prefix in prefixes: + c = (prefix or "").strip("/") + if not c or ".." in c.split("/"): + raise ValueError(f"refusing to delete unsafe prefix: {prefix!r}") + cleaned.append(c) + if not cleaned: + return 0 + if _mode() == "r2": + return _r2_delete_prefixes(cleaned) + return sum(_local_delete_prefix(c) for c in cleaned) + + +def keys_under(prefixes: list[str]) -> dict[str, int]: + """Every stored key beneath any of *prefixes*, with sizes. + + Does a single listing so a caller can delete in progress-reporting batches + without paying for a fresh listing per batch. + """ + cleaned = [] + for prefix in prefixes: + c = (prefix or "").strip("/") + if not c or ".." in c.split("/"): + raise ValueError(f"refusing to list unsafe prefix: {prefix!r}") + cleaned.append(c if c.endswith("/") else c + "/") + if not cleaned: + return {} + + root = os.path.commonprefix(cleaned) + root = root[: root.rfind("/") + 1] if "/" in root else "" + wanted = tuple(cleaned) + return {k: v for k, v in list_sizes(root).items() if k.startswith(wanted)} + + +def delete_keys(keys: list[str], on_progress=None) -> None: + """Delete exactly these keys. on_progress(done, total) fires per batch.""" + total = len(keys) + if not total: + return + r2 = _mode() == "r2" + client = _get_r2_client() if r2 else None + bucket = _r2_bucket() if r2 else None + + step = 1000 if r2 else 200 + for i in range(0, total, step): + batch = keys[i:i + step] + if r2: + client.delete_objects( + Bucket=bucket, Delete={"Objects": [{"Key": k} for k in batch]} + ) + else: + for k in batch: + try: + (_data_dir() / k).unlink() + except FileNotFoundError: + pass + if on_progress: + on_progress(min(i + step, total), total) diff --git a/frontend/src/components/StorageManager.tsx b/frontend/src/components/StorageManager.tsx index 0c905d3..1f426b6 100644 --- a/frontend/src/components/StorageManager.tsx +++ b/frontend/src/components/StorageManager.tsx @@ -29,6 +29,15 @@ interface StorageResponse { retention: Retention; } +interface SweepStatus { + state: "idle" | "running" | "done" | "error"; + message: string; + done: number; + total: number; + count: number; + freed_bytes: number; +} + interface SweepResult { count: number; freed_bytes: number; @@ -93,6 +102,7 @@ export default function StorageManager({ onClose }: { onClose: () => void }) { const unitLabel = amountNum === 1 ? unit.slice(0, -1) : unit; const [confirm, setConfirm] = useState(null); + const [progress, setProgress] = useState(null); const [busy, setBusy] = useState(false); const [result, setResult] = useState(null); const [error, setError] = useState(null); @@ -175,16 +185,30 @@ export default function StorageManager({ onClose }: { onClose: () => void }) { async function applySave() { setBusy(true); setError(null); + setConfirm(null); + setProgress({ state: "running", message: "Starting…", done: 0, total: 0, count: 0, freed_bytes: 0 }); try { const res = await authFetch(`/api/storage/retention?${query}`, { method: "PUT" }); if (!res.ok) throw new Error(); - const body: SweepResult = await res.json(); - setConfirm(null); - setResult( - `Deleted ${body.count} session${body.count === 1 ? "" : "s"}, freeing ${formatSize(body.freed_bytes)}.`, - ); + + // The sweep runs server-side; poll so a large delete shows progress. + for (;;) { + await new Promise((r) => setTimeout(r, 400)); + const s: SweepStatus = await authFetch("/api/storage/retention/status").then((r) => r.json()); + setProgress(s); + if (s.state === "done") { + setResult(`Deleted ${s.count} session${s.count === 1 ? "" : "s"}, freeing ${formatSize(s.freed_bytes)}.`); + break; + } + if (s.state === "error") { + setError(s.message || "Delete failed."); + break; + } + } + setProgress(null); await load(); } catch { + setProgress(null); setError("Failed to apply the retention setting."); } finally { setBusy(false); @@ -209,7 +233,14 @@ export default function StorageManager({ onClose }: { onClose: () => void }) { )} - {/* Retention policy */} + {/* Retention policy — hidden until usage is known, so the threshold + can't be set against a total that hasn't loaded yet. */} + {loading ? ( +
+ + Calculating current storage… +
+ ) : (
)} + {progress && ( +
+
+ + {progress.message} +
+ {progress.total > 0 && ( + <> +
+
+
+

+ {progress.done} of {progress.total} files +

+ + )} +
+ )} + {result &&

{result}

} {error &&

{error}

}
+ )}