Files
ProgressionMods/progression_backend/refresh.py
T
HRiggs 208ae62ca6
Build and Publish Container / Publish Linux container (push) Successful in 3m53s
fix: download cases
2026-10-06 01:04:23 -04:00

661 lines
28 KiB
Python

from __future__ import annotations
import asyncio
import logging
import shutil
import time
import uuid
from collections.abc import Callable
from datetime import datetime, timedelta, timezone
from typing import Any
from .config import Settings
from .database import Database
from .steam_api import SteamWorkshopClient, WorkshopDetails
from .workshop import SteamCmdRunner, directory_size, parse_about_xml, utc_now
logger = logging.getLogger("uvicorn.error")
ProgressCallback = Callable[[dict[str, Any]], None]
class InitialDownloadTimeout(RuntimeError):
pass
def _batches(values: list[str], size: int) -> list[list[str]]:
return [values[index : index + size] for index in range(0, len(values), size)]
class RefreshService:
def __init__(
self,
settings: Settings,
database: Database,
steam: SteamWorkshopClient | None = None,
runner: SteamCmdRunner | None = None,
):
self.settings = settings
self.database = database
self.steam = steam or SteamWorkshopClient()
self.runner = runner or SteamCmdRunner(
settings.steamcmd_path,
settings.workshop_content_dir,
settings.steamcmd_timeout_seconds,
)
def needs_initial_download(self) -> bool:
rows = self.database.current_mods()
if not rows:
return True
for row in rows:
if row.get("status") == "skipped":
continue
if row.get("status") != "ready" or not (
self.settings.workshop_content_dir / row["workshop_id"]
).is_dir():
return True
return False
def _store_metadata(
self,
detail: WorkshopDetails,
previous: dict[str, Any] | None,
stale: bool,
skip_reason: str | None = None,
) -> dict[str, Any]:
invalid_reason = skip_reason
if invalid_reason is None and detail.result != 1:
invalid_reason = f"Steam returned result {detail.result}"
elif invalid_reason is None and detail.consumer_app_id not in (0, 294100):
invalid_reason = f"Workshop item belongs to app {detail.consumer_app_id}"
return {
"workshop_id": detail.workshop_id,
"title": detail.title,
"package_id": None,
"steam_app_id": None,
"display_name": None,
"time_updated": detail.time_updated,
"metadata_file_size": detail.file_size,
"disk_size": None,
"local_path": None,
"status": "skipped"
if invalid_reason
else ("pending" if stale else (previous or {}).get("status", "pending")),
"error": invalid_reason,
"last_checked_at": utc_now(),
"last_downloaded_at": None,
}
def _inspection_values(
self,
workshop_id: str,
downloaded: bool,
forced_error: str | None = None,
) -> tuple[bool, dict[str, Any]]:
mod_path = self.settings.workshop_content_dir / workshop_id
if forced_error:
return False, {
"workshop_id": workshop_id,
"status": "failed",
"error": forced_error,
"disk_size": directory_size(mod_path),
"local_path": str(mod_path) if mod_path.exists() else None,
"last_checked_at": utc_now(),
"last_downloaded_at": utc_now() if downloaded else None,
}
try:
parsed = parse_about_xml(mod_path)
return True, {
"workshop_id": workshop_id,
"package_id": parsed.package_id,
"steam_app_id": parsed.declared_steam_app_id,
"display_name": parsed.display_name,
"disk_size": directory_size(mod_path),
"local_path": str(mod_path),
"status": "ready",
"error": "",
"last_checked_at": utc_now(),
"last_downloaded_at": utc_now() if downloaded else None,
}
except (OSError, ValueError) as error:
return False, {
"workshop_id": workshop_id,
"status": "failed",
"error": str(error),
"disk_size": directory_size(mod_path),
"local_path": str(mod_path) if mod_path.exists() else None,
"last_checked_at": utc_now(),
"last_downloaded_at": utc_now() if downloaded else None,
}
def refresh(
self,
run_id: str,
trigger: str,
*,
initial_download: bool = False,
overall_timeout_seconds: int | None = None,
progress: ProgressCallback | None = None,
) -> None:
started_at = utc_now()
started = time.monotonic()
deadline = (
started + overall_timeout_seconds
if initial_download and overall_timeout_seconds is not None
else None
)
self.database.create_run(run_id, trigger, started_at)
metrics: dict[str, Any] = {
"collection_items": 0,
"unique_items": 0,
"ready_items": 0,
"downloads_requested": 0,
"downloads_succeeded": 0,
"downloads_failed": 0,
"bytes_before": directory_size(self.settings.workshop_content_dir),
"bytes_after": 0,
"bytes_downloaded": 0,
"cache_bytes": 0,
"status": "failed",
"error": None,
}
def report(**values: Any) -> None:
if progress:
progress(values)
def remaining_time() -> float | None:
if deadline is None:
return None
remaining = deadline - time.monotonic()
if remaining <= 0:
raise InitialDownloadTimeout(
"The initial Workshop download reached its "
f"{overall_timeout_seconds}-second limit"
)
return remaining
try:
report(stage="collections", completed_mods=0, total_mods=0, percent=0.0)
remaining_time()
collection_ids = list(self.settings.collections.values())
response, collection_reference_ids = self.steam.get_collections(collection_ids)
named_collections: dict[str, tuple[str, list[str]]] = {}
all_ids: list[str] = []
for name, collection_id in self.settings.collections.items():
workshop_ids = response.get(collection_id)
if workshop_ids is None:
raise RuntimeError(
f"Steam did not return collection {collection_id}"
)
named_collections[name] = (collection_id, workshop_ids)
all_ids.extend(workshop_ids)
self.database.replace_collection_members(named_collections)
unique_ids = sorted(set(all_ids), key=int)
metrics["collection_items"] = len(all_ids)
metrics["unique_items"] = len(unique_ids)
details = self.steam.get_item_details(unique_ids)
previous = self.database.mod_map(unique_ids)
targets: list[str] = []
metadata_rows: list[dict[str, Any]] = []
for workshop_id in unique_ids:
detail = details.get(workshop_id)
if not detail:
self.database.upsert_mod(
{
"workshop_id": workshop_id,
"status": "skipped",
"error": "Steam returned no item metadata",
"last_checked_at": utc_now(),
}
)
continue
old = previous.get(workshop_id)
local_path = self.settings.workshop_content_dir / workshop_id
stale = (
old is None
or int(old.get("time_updated") or 0) != detail.time_updated
or old.get("status") != "ready"
or not local_path.is_dir()
)
skip_reason = (
"Workshop entry is a collection reference"
if workshop_id in collection_reference_ids
else None
)
metadata_rows.append(
self._store_metadata(detail, old, stale, skip_reason=skip_reason)
)
if (
detail.result == 1
and detail.consumer_app_id in (0, 294100)
and workshop_id not in collection_reference_ids
and stale
):
targets.append(workshop_id)
self.database.upsert_mods(metadata_rows)
metrics["downloads_requested"] = len(targets)
total_targets = len(targets)
completed_targets = 0
current_after_metadata = self.database.mod_map(unique_ids)
resolved_before = sum(
1
for row in current_after_metadata.values()
if row.get("status") in {"ready", "skipped"}
)
progress_total = len(unique_ids) if initial_download else total_targets
progress_completed = resolved_before if initial_download else 0
title_by_id = {
workshop_id: details[workshop_id].title
for workshop_id in targets
if workshop_id in details
}
report(
stage="downloading" if targets else "inspecting",
completed_mods=progress_completed,
total_mods=progress_total,
percent=round(
(progress_completed / max(progress_total, 1)) * 100, 1
)
if progress_total
else 100.0,
)
if initial_download:
logger.info(
"Initial Workshop download started: %d mods need download, "
"%d/%d collection entries already resolved, %.1f-hour hard limit",
total_targets,
resolved_before,
len(unique_ids),
(overall_timeout_seconds or 0) / 3600,
)
if targets:
remaining_bytes = sum(
max(int(details[workshop_id].file_size or 0), 0)
for workshop_id in targets
if workshop_id in details
)
workshop_root = self.settings.workshop_content_dir.parent.parent
workshop_root.mkdir(parents=True, exist_ok=True)
free_bytes = shutil.disk_usage(workshop_root).free
required_bytes = int(remaining_bytes * 1.1)
logger.info(
"Workshop capacity check: %.2f GiB expected, %.2f GiB free",
remaining_bytes / (1024**3),
free_bytes / (1024**3),
)
if required_bytes and free_bytes < required_bytes:
raise RuntimeError(
"Insufficient Workshop volume space: "
f"{free_bytes / (1024**3):.2f} GiB free, "
f"approximately {required_bytes / (1024**3):.2f} GiB required"
)
if targets and not self.runner.available():
reason = (
f"SteamCMD was not found at '{self.settings.steamcmd_path}'. "
"Set STEAMCMD_PATH to enable anonymous downloads."
)
inspection_rows: list[dict[str, Any]] = []
for workshop_id in targets:
mod_path = self.settings.workshop_content_dir / workshop_id
success, values = self._inspection_values(
workshop_id,
downloaded=False,
forced_error=None if mod_path.is_dir() else reason,
)
inspection_rows.append(values)
if not success:
metrics["downloads_failed"] += 1
self.database.upsert_mods(inspection_rows)
else:
download_queue = list(targets)
ready_target_ids: set[str] = set()
batch_number = 0
pass_number = 1
while download_queue:
# SteamCMD stops processing the rest of a command when one
# Workshop item times out. Keep the first pass efficient,
# then isolate every remaining item so one bad mod cannot
# permanently block the items that followed it in a batch.
pass_batch_size = (
self.settings.steamcmd_batch_size if pass_number == 1 else 1
)
batches = _batches(download_queue, pass_batch_size)
if pass_number > 1:
logger.info(
"Initial Workshop download retry pass %d is "
"isolating %d remaining mods individually",
pass_number,
len(download_queue),
)
for pass_batch_number, batch in enumerate(batches, start=1):
remaining = remaining_time()
batch_number += 1
logger.info(
"%s pass %d batch %d/%d starting (%d mods): %s",
"Initial Workshop download"
if initial_download
else "Workshop refresh",
pass_number,
pass_batch_number,
len(batches),
len(batch),
", ".join(
f"{title_by_id.get(workshop_id, 'Unknown mod')} "
f"[{workshop_id}]"
for workshop_id in batch
),
)
result = self.runner.run_batch(
batch, timeout_seconds=remaining
)
if result.status != "completed":
logger.warning(
"SteamCMD pass %d batch %d/%d finished with "
"status=%s exit_code=%s after %.1fs. Output:\n%s",
pass_number,
pass_batch_number,
len(batches),
result.status,
result.exit_code,
result.elapsed_seconds,
result.output_tail or "(no SteamCMD output)",
)
if result.has_io_failure:
recover = getattr(self.runner, "recover_io_failure", None)
moved = recover() if callable(recover) else []
if moved:
logger.warning(
"SteamCMD I/O recovery quarantined transient "
"Workshop state: %s",
", ".join(str(path) for path in moved),
)
self.database.add_download_attempt(
{
"run_id": run_id,
"batch_number": batch_number,
"item_count": len(batch),
"workshop_ids": batch,
"started_at": result.started_at,
"finished_at": result.finished_at,
"elapsed_seconds": result.elapsed_seconds,
"exit_code": result.exit_code,
"status": result.status,
"output_tail": result.output_tail,
}
)
inspection_rows = []
for workshop_id in batch:
failure = (
(
"SteamCMD did not complete this item; "
"queued for retry"
if initial_download
else "SteamCMD reported a download failure"
)
if workshop_id in result.failed_ids
else None
)
success, values = self._inspection_values(
workshop_id, downloaded=True, forced_error=failure
)
inspection_rows.append(values)
if success:
metrics["downloads_succeeded"] += 1
ready_target_ids.add(workshop_id)
else:
metrics["downloads_failed"] += 1
completed_targets = len(ready_target_ids)
progress_completed = (
resolved_before + completed_targets
if initial_download
else completed_targets
)
percent = round(
(progress_completed / max(progress_total, 1)) * 100,
1,
)
title = title_by_id.get(workshop_id, "Unknown mod")
outcome = (
"ready"
if success
else ("retrying" if initial_download else "failed")
)
report(
stage="downloading",
completed_mods=progress_completed,
total_mods=progress_total,
percent=percent,
current_workshop_id=workshop_id,
current_title=title,
current_status=outcome,
)
logger.info(
"%s %d/%d (%.1f%%) %s: %s [%s]",
"Initial mod download"
if initial_download
else "Mod refresh",
progress_completed,
progress_total,
percent,
outcome,
title,
workshop_id,
)
self.database.upsert_mods(inspection_rows)
remaining_time()
if not initial_download:
break
download_queue = [
workshop_id
for workshop_id in download_queue
if workshop_id not in ready_target_ids
]
if download_queue:
remaining = remaining_time()
retry_delay = min(30.0, remaining or 30.0)
logger.warning(
"Initial Workshop download pass %d left %d mods; "
"retrying in %.0f seconds",
pass_number,
len(download_queue),
retry_delay,
)
time.sleep(retry_delay)
remaining_time()
pass_number += 1
# Re-inspect ready cached items whose metadata has never been parsed by this backend.
current = self.database.mod_map(unique_ids)
pending_rows = []
for workshop_id in unique_ids:
row = current.get(workshop_id)
if row and row.get("status") == "pending":
_, values = self._inspection_values(workshop_id, downloaded=False)
pending_rows.append(values)
if pending_rows:
self.database.upsert_mods(pending_rows)
metrics["bytes_after"] = directory_size(self.settings.workshop_content_dir)
metrics["cache_bytes"] = metrics["bytes_after"]
metrics["bytes_downloaded"] = max(
metrics["bytes_after"] - metrics["bytes_before"], 0
)
final_rows = self.database.mod_map(unique_ids)
metrics["ready_items"] = sum(
1 for row in final_rows.values() if row.get("status") == "ready"
)
resolved_items = sum(
1
for row in final_rows.values()
if row.get("status") in {"ready", "skipped"}
)
metrics["status"] = (
"completed"
if resolved_items == metrics["unique_items"]
else "partial"
)
if metrics["status"] == "partial" and not metrics["error"]:
metrics["error"] = (
f"{metrics['unique_items'] - resolved_items} items are not resolved"
)
report(
stage=metrics["status"],
completed_mods=resolved_items,
total_mods=metrics["unique_items"],
percent=round(
(resolved_items / max(metrics["unique_items"], 1)) * 100, 1
)
if metrics["unique_items"]
else 100.0,
)
except Exception as error: # noqa: BLE001 - the run must always persist a terminal state
metrics["status"] = "failed"
metrics["error"] = str(error)
metrics["bytes_after"] = directory_size(self.settings.workshop_content_dir)
metrics["cache_bytes"] = metrics["bytes_after"]
metrics["bytes_downloaded"] = max(
metrics["bytes_after"] - metrics["bytes_before"], 0
)
report(stage="failed", error=str(error))
logger.error(
"%s failed: %s",
"Initial Workshop download" if initial_download else "Workshop refresh",
error,
)
finally:
if metrics["unique_items"]:
metrics["ready_items"] = sum(
1
for row in self.database.current_mods()
if row.get("status") == "ready"
)
elapsed = max(time.monotonic() - started, 0.001)
metrics["elapsed_seconds"] = round(elapsed, 3)
metrics["average_mbps"] = round(
(metrics["bytes_downloaded"] * 8 / 1_000_000) / elapsed, 3
)
metrics["finished_at"] = utc_now()
self.database.finish_run(run_id, metrics)
logger.info(
"%s finished: status=%s, ready=%d/%d, elapsed=%.1fs, "
"cache=%d bytes",
"Initial Workshop download" if initial_download else "Workshop refresh",
metrics["status"],
metrics["ready_items"],
metrics["unique_items"],
elapsed,
metrics["cache_bytes"],
)
class RefreshCoordinator:
def __init__(
self,
service: RefreshService,
interval_seconds: int,
refresh_on_startup: bool,
initial_download_timeout_seconds: int = 43200,
):
self.service = service
self.interval_seconds = interval_seconds
self.refresh_on_startup = refresh_on_startup
self.initial_download_timeout_seconds = initial_download_timeout_seconds
self.current_run_id: str | None = None
self.next_refresh_at: str | None = None
self._refresh_task: asyncio.Task[None] | None = None
self._scheduler_task: asyncio.Task[None] | None = None
self._stop = asyncio.Event()
self._active_initial_download = False
self._progress: dict[str, Any] | None = None
async def start(self) -> None:
if self._scheduler_task:
return
self._scheduler_task = asyncio.create_task(self._scheduler_loop())
async def stop(self) -> None:
self._stop.set()
if self._scheduler_task:
self._scheduler_task.cancel()
await asyncio.gather(self._scheduler_task, return_exceptions=True)
async def trigger(self, trigger: str) -> tuple[str, bool]:
if self._refresh_task and not self._refresh_task.done():
return self.current_run_id or "", False
run_id = uuid.uuid4().hex
self.current_run_id = run_id
initial_download = self.service.needs_initial_download()
self._active_initial_download = initial_download
self._progress = {
"stage": "starting",
"completed_mods": 0,
"total_mods": 0,
"percent": 0.0,
}
def update_progress(values: dict[str, Any]) -> None:
self._progress = {**(self._progress or {}), **values}
async def run() -> None:
try:
await asyncio.to_thread(
self.service.refresh,
run_id,
trigger,
initial_download=initial_download,
overall_timeout_seconds=(
self.initial_download_timeout_seconds
if initial_download
else None
),
progress=update_progress,
)
finally:
self.current_run_id = None
self._active_initial_download = False
self._refresh_task = asyncio.create_task(run())
return run_id, True
async def _wait_for_initial_download(self, started: bool) -> None:
if started and self._active_initial_download and self._refresh_task:
self.next_refresh_at = None
await asyncio.shield(self._refresh_task)
async def _scheduler_loop(self) -> None:
if self.refresh_on_startup:
_, started = await self.trigger("startup")
await self._wait_for_initial_download(started)
while not self._stop.is_set():
next_time = datetime.now(timezone.utc) + timedelta(
seconds=self.interval_seconds
)
self.next_refresh_at = next_time.isoformat()
try:
await asyncio.wait_for(self._stop.wait(), timeout=self.interval_seconds)
except asyncio.TimeoutError:
_, started = await self.trigger("scheduled")
await self._wait_for_initial_download(started)
def state(self) -> dict[str, Any]:
return {
"running": bool(self._refresh_task and not self._refresh_task.done()),
"current_run_id": self.current_run_id,
"next_refresh_at": self.next_refresh_at,
"interval_seconds": self.interval_seconds,
"initial_download": bool(
self._active_initial_download
and self._refresh_task
and not self._refresh_task.done()
),
"initial_download_timeout_seconds": self.initial_download_timeout_seconds,
"progress": dict(self._progress) if self._progress else None,
}