Make retention delete fast and show progress

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()

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")

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)

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<SweepResult | null>(null);
+  const [progress, setProgress] = useState<SweepStatus | null>(null);
   const [busy, setBusy] = useState(false);
   const [result, setResult] = useState<string | null>(null);
   const [error, setError] = useState<string | null>(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 }) {
           )}
         </div>
 
-        {/* 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 ? (
+          <div className="p-6 py-8 border-b border-f1-border flex items-center justify-center gap-3">
+            <span className="w-5 h-5 border-2 border-f1-muted border-t-f1-red rounded-full animate-spin flex-shrink-0" />
+            <span className="text-f1-muted text-sm">Calculating current storage…</span>
+          </div>
+        ) : (
         <div className="p-6 py-4 border-b border-f1-border overflow-y-auto min-h-0">
           <button
             onClick={() => setEnabled(!enabled)}
@@ -316,9 +347,32 @@ export default function StorageManager({ onClose }: { onClose: () => void }) {
             </div>
           )}
 
+          {progress && (
+            <div className="mt-3 p-3 bg-f1-dark border border-f1-border rounded">
+              <div className="flex items-center gap-3 mb-2">
+                <span className="w-4 h-4 border-2 border-f1-muted border-t-f1-red rounded-full animate-spin flex-shrink-0" />
+                <span className="text-white text-sm">{progress.message}</span>
+              </div>
+              {progress.total > 0 && (
+                <>
+                  <div className="h-1.5 bg-f1-border rounded overflow-hidden">
+                    <div
+                      className="h-full bg-f1-red transition-all duration-200"
+                      style={{ width: `${Math.round((progress.done / progress.total) * 100)}%` }}
+                    />
+                  </div>
+                  <p className="text-f1-muted text-xs mt-1">
+                    {progress.done} of {progress.total} files
+                  </p>
+                </>
+              )}
+            </div>
+          )}
+
           {result && <p className="text-green-400 text-sm mt-2">{result}</p>}
           {error && <p className="text-red-400 text-sm mt-2">{error}</p>}
         </div>
+        )}
 
         <div className="p-6 pt-4 flex justify-end">
           <button