v2.1.3

CHANGELOG.md

@@ -2,6 +2,18 @@
 
 All notable changes to F1 Replay Timing will be documented in this file.
 
+## 2.1.3
+
+### Improvements
+
+- **Storage management** — a **Storage** button in the header shows total disk use and sets a retention policy: sessions older than a chosen number of weeks or months are deleted, immediately and then daily. Individual sessions can be deleted by right-clicking them in the session picker. (requested by [@db-can](https://github.com/db-can))
+
+### Fixes
+
+- **Replay froze after a dropped connection** — an idle replay sent no data, so proxies could close the socket and the client never reconnected. The server now sends a keep-alive every 20s, and the client reconnects and resumes where it left off. ([#106](https://github.com/adn8naiagent/F1ReplayTiming/issues/106), reported by [@nbolland](https://github.com/nbolland))
+
+- **App unreachable behind a reverse proxy when `PORT` was set** — 2.1.2 passed `PORT` into the container, moving the port proxies point at (Traefik, nginx and cloudflared all target 8000). The container now always listens on 8000; `PORT` sets the published host port only. (found while investigating [#106](https://github.com/adn8naiagent/F1ReplayTiming/issues/106), reported by [@nbolland](https://github.com/nbolland))
+
 ## 2.1.2
 
 ### Fixes

backend/main.py

@@ -13,6 +13,7 @@
 from routers import sessions, track, laps, results, replay, telemetry, sync, live, live_status
 from routers import auth_routes
 from services.auto_precompute import auto_precompute_loop, get_allowed_session_types
+from services.retention import retention_loop
 
 load_dotenv()
 
@@ -29,13 +30,19 @@ async def lifespan(app: FastAPI):
         logger.info(f"Auto-precompute scheduled for session types: {sorted(allowed)}")
     else:
         logger.info("Auto-precompute disabled (AUTO_PRECOMPUTE=off)")
+    # Runs regardless of AUTO_PRECOMPUTE: someone who turned auto-download off
+    # is exactly who is most likely to be managing disk space.
+    retention_task = asyncio.create_task(retention_loop())
+
     yield
-    if task is not None:
-        task.cancel()
-        try:
-            await task
-        except asyncio.CancelledError:
-            pass
+
+    for t in (task, retention_task):
+        if t is not None:
+            t.cancel()
+            try:
+                await t
+            except asyncio.CancelledError:
+                pass
 
 
 app = FastAPI(

backend/routers/replay.py

@@ -37,6 +37,15 @@ def _log_memory():
 
 CACHE_EVICTION_SECONDS = 300  # 5 minutes after last client disconnects
 
+# Idle gap after which the server sends a keep-alive frame. Proxies and tunnels
+# that only count data frames towards their idle timeout (commonly 60s or 300s)
+# will close the socket otherwise — a paused replay sends nothing at all.
+HEARTBEAT_SECONDS = 20
+
+# Queued by the reader task when the client goes away, so the playback loop can
+# raise WebSocketDisconnect from wherever it happens to be waiting.
+_DISCONNECTED = object()
+
 # In-memory cache for pit loss data
 _pit_loss_cache: dict | None = None
 
@@ -206,6 +215,20 @@ async def _client_disconnect(key: str):
             _eviction_tasks[key] = task
 
 
+def evict_cached_session(year: int, round_num: int, session_type: str) -> None:
+    """Drop a session's frames from memory immediately.
+
+    Called when stored data is deleted; without it the frames stay resident
+    until the normal eviction delay elapses.
+    """
+    key = f"{year}_{round_num}_{session_type}"
+    task = _eviction_tasks.pop(key, None)
+    if task:
+        task.cancel()
+    if _replay_cache.pop(key, None) is not None:
+        logger.info(f"[memory] Evicted {key} after delete — {_log_memory()}")
+
+
 async def _evict_after_delay(key: str):
     """Wait, then evict a cached session if no new clients have connected."""
     try:
@@ -232,6 +255,8 @@ async def replay_websocket(
         return
     await websocket.accept()
 
+    receiver_task: asyncio.Task | None = None
+
     try:
         async def send_status(msg: str):
             await websocket.send_json({"type": "status", "message": msg})
@@ -311,8 +336,20 @@ def prepare_frame(f: dict) -> dict:
                 _add_pit_predictions(f, pit_loss_green, pit_loss_sc, pit_loss_vsc)
             return f
 
+        last_send = time.monotonic()
+
+        async def send_json(payload: dict) -> None:
+            """Send, recording the time so the heartbeat only fires when idle."""
+            nonlocal last_send
+            await websocket.send_json(payload)
+            last_send = time.monotonic()
+
+        async def beat_if_idle() -> None:
+            if time.monotonic() - last_send >= HEARTBEAT_SECONDS:
+                await send_json({"type": "ping"})
+
         # Send first frame immediately so cars are visible before play
-        await websocket.send_json({"type": "frame", **prepare_frame(frames[0])})
+        await send_json({"type": "frame", **prepare_frame(frames[0])})
 
         # Playback state
         playing = False
@@ -338,7 +375,7 @@ async def send_seek_frame(target_time: float):
                     frame_index = i
                     break
             if frame_index < len(frames):
-                await websocket.send_json({"type": "frame", **prepare_frame(frames[frame_index])})
+                await send_json({"type": "frame", **prepare_frame(frames[frame_index])})
 
         async def handle_command(cmd: str):
             nonlocal playing, speed, frame_index
@@ -370,32 +407,48 @@ async def handle_command(cmd: str):
                             frame_index = i
                             break
                     if frame_index < len(frames):
-                        await websocket.send_json({"type": "frame", **prepare_frame(frames[frame_index])})
+                        await send_json({"type": "frame", **prepare_frame(frames[frame_index])})
                     reset_anchor()
                 except ValueError:
                     pass
             elif cmd == "reset":
                 frame_index = 0
                 playing = False
-                await websocket.send_json({"type": "frame", **prepare_frame(frames[0])})
+                await send_json({"type": "frame", **prepare_frame(frames[0])})
                 reset_anchor()
 
+        # One long-lived reader feeding a queue. The playback loop polls the
+        # queue on a 50ms tick; cancelling a queue.get() is safe, whereas
+        # cancelling websocket.receive_text() that often can drop commands.
+        commands: asyncio.Queue = asyncio.Queue()
+
+        async def receive_commands():
+            try:
+                while True:
+                    await commands.put(await websocket.receive_text())
+            except Exception:
+                await commands.put(_DISCONNECTED)
+
+        receiver_task = asyncio.create_task(receive_commands())
+
         async def check_command(timeout: float) -> bool:
             try:
-                msg = await asyncio.wait_for(websocket.receive_text(), timeout=timeout)
-                await handle_command(msg.strip().lower())
-                return True
+                msg = await asyncio.wait_for(commands.get(), timeout=timeout)
             except asyncio.TimeoutError:
                 return False
+            if msg is _DISCONNECTED:
+                raise WebSocketDisconnect(1006)
+            await handle_command(msg.strip().lower())
+            return True
 
         while True:
             if playing and frame_index < len(frames):
-                await websocket.send_json({"type": "frame", **prepare_frame(frames[frame_index])})
+                await send_json({"type": "frame", **prepare_frame(frames[frame_index])})
                 frame_index += 1
 
                 if frame_index >= len(frames):
                     playing = False
-                    await websocket.send_json({"type": "finished"})
+                    await send_json({"type": "finished"})
                     continue
 
                 # Sleep until the next frame is due per wall clock.
@@ -408,9 +461,13 @@ async def check_command(timeout: float) -> bool:
                 while sleep_remaining > 0 and playing:
                     chunk = min(sleep_remaining, 0.05)
                     await check_command(chunk)
+                    # Covers long gaps between frames: red flags, and slow speeds.
+                    await beat_if_idle()
                     sleep_remaining = target_wall - time.monotonic()
             else:
+                # Paused, or waiting to start — nothing else is sent here.
                 await check_command(1.0)
+                await beat_if_idle()
 
     except WebSocketDisconnect:
         cache_key = f"{year}_{round_num}_{type}"
@@ -424,3 +481,6 @@ async def check_command(timeout: float) -> bool:
             await websocket.close()
         except Exception:
             pass
+    finally:
+        if receiver_task is not None:
+            receiver_task.cancel()

backend/routers/sessions.py

@@ -1,11 +1,26 @@
+import asyncio
 import logging
 import time
 from copy import deepcopy
 from datetime import datetime, timezone
 
 from fastapi import APIRouter, Query, HTTPException
 from services.storage import get_json, put_json, list_sizes
-from services.process import ensure_session_data, start_reprocess, get_reprocess_status
+from services.process import (
+    ensure_session_data,
+    start_reprocess,
+    get_reprocess_status,
+    delete_session,
+)
+from services.retention import (
+    RetentionError,
+    get_retention,
+    get_sweep_status,
+    save_retention,
+    session_index,
+    start_sweep,
+    sweep,
+)
 
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api", tags=["sessions"])
@@ -192,3 +207,62 @@ async def reprocess_session(year: int, round_num: int, type: str = Query("R")):
 @router.get("/sessions/{year}/{round_num}/reprocess/status")
 async def reprocess_session_status(year: int, round_num: int, type: str = Query("R")):
     return get_reprocess_status(year, round_num, type)
+
+
+@router.get("/storage")
+async def storage_usage():
+    """Stored size of every downloaded session, plus the retention setting."""
+    sessions, total = session_index()
+    return {
+        "total_bytes": total,
+        "session_count": len(sessions),
+        "sessions": sessions,
+        "retention": get_retention(),
+    }
+
+
+@router.delete("/sessions/{year}/{round_num}")
+async def delete_session_data(year: int, round_num: int, type: str = Query("R")):
+    """Delete one session's stored data. Auth is enforced by the /api middleware."""
+    if type not in SESSION_NAME_TO_TYPE.values():
+        raise HTTPException(status_code=400, detail=f"Unknown session type: {type}")
+    freed = delete_session(year, round_num, type)
+    _events_cache.clear()
+    return {"deleted": 1, "freed_bytes": freed}
+
+
+@router.put("/storage/retention")
+async def update_retention(
+    enabled: bool = Query(...),
+    amount: int = Query(...),
+    unit: str = Query("months"),
+    dry_run: bool = Query(False, description="Report what would be deleted, change nothing"),
+):
+    """Set the retention policy, applying it immediately.
+
+    Call with dry_run=true first to show the user what saving would remove;
+    saving then deletes everything already older than the threshold, and the
+    daily sweep keeps it that way.
+    """
+    try:
+        if dry_run:
+            return sweep(amount, unit, dry_run=True)
+
+        config = save_retention(enabled, amount, unit)
+        if not enabled:
+            return {"retention": config, "started": False}
+
+        # 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": 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/auto_precompute.py

@@ -62,7 +62,7 @@ async def _check_and_process():
     from services.f1_data import _fetch_schedule_sync, SESSION_NAME_TO_TYPE
     from services import storage
 
-    from services.process import process_session_sync as process_session
+    from services.process import process_session_sync as process_session, is_deleted
 
     now = datetime.now(timezone.utc)
     year = now.year
@@ -108,6 +108,11 @@ async def _check_and_process():
             if storage.exists(f"{base}/replay.json"):
                 continue
 
+            # Respect an explicit delete — don't silently re-download data the
+            # user removed. Cleared automatically if they reprocess it.
+            if is_deleted(year, round_num, session_type):
+                continue
+
             # New session data might be available - try to process it
             prefix = f"{year} R{round_num} {session_type}"
             logger.info(f"[auto] New session detected: {prefix}, attempting precompute...")

backend/services/process.py

@@ -9,6 +9,7 @@
 import asyncio
 import logging
 import traceback
+from datetime import datetime, timezone
 
 from services import storage
 from services.f1_data import (
@@ -29,6 +30,11 @@
 # state is "running" | "done" | "error"
 _reprocess_status: dict[str, dict] = {}
 
+# Written when a user deletes a session. auto_precompute skips sessions carrying
+# one, so an explicit delete isn't undone by the background downloader; any
+# successful reprocess clears it.
+TOMBSTONE = "deleted.json"
+
 
 def get_reprocess_status(year: int, round_num: int, session_type: str) -> dict:
     """Return the current reprocess state + latest progress message for a session."""
@@ -169,6 +175,20 @@ def status(msg: str):
     except Exception as e:
         logger.warning(f"[{prefix}] Telemetry upload issue: {e}")
 
+    # Session is back — drop any delete tombstone so auto-precompute resumes.
+    try:
+        storage.delete(f"{base}/{TOMBSTONE}")
+    except Exception:
+        pass
+
+    # Invalidate the cached events listing so the picker shows the downloaded
+    # marker straight away, rather than after the 5 minute cache expires.
+    try:
+        from routers.sessions import _events_cache
+        _events_cache.clear()
+    except Exception:
+        pass
+
     status("Processing complete")
     logger.info(f"[{prefix}] Done")
     return True
@@ -288,3 +308,39 @@ def sync_status(msg: str):
         except Exception as e:
             logger.error(f"On-demand processing failed for {key}: {e}")
             return False
+
+
+def is_deleted(year: int, round_num: int, session_type: str) -> bool:
+    """True if the user explicitly deleted this session's data."""
+    base = f"sessions/{year}/{round_num}/{session_type}"
+    return storage.exists(f"{base}/{TOMBSTONE}")
+
+
+def delete_session(year: int, round_num: int, session_type: str) -> int:
+    """Delete a session's stored data. Returns the number of bytes freed.
+
+    replay.json is removed first, because every availability check in the app
+    keys off it. If the delete is interrupted the session then reads as "not
+    downloaded" and reprocesses cleanly, rather than being left present but
+    incomplete — the failure mode people hit deleting these files by hand.
+    """
+    base = f"sessions/{year}/{round_num}/{session_type}"
+
+    freed = storage.delete(f"{base}/replay.json")
+    freed += storage.delete_prefix(base)
+
+    storage.put_json(
+        f"{base}/{TOMBSTONE}",
+        {"deleted_at": datetime.now(timezone.utc).isoformat()},
+    )
+
+    # Frames stay resident until the cache evicts them, so drop them now —
+    # otherwise deleting from disk frees no memory for up to 5 minutes.
+    try:
+        from routers.replay import evict_cached_session
+        evict_cached_session(year, round_num, session_type)
+    except Exception as e:
+        logger.warning(f"Could not evict replay cache for {base}: {e}")
+
+    logger.info(f"Deleted {base} — freed {freed} bytes")
+    return freed