From 00e85d470856704a5f78ea9357235383dca6c44e Mon Sep 17 00:00:00 2001 From: devRaGonSa Date: Thu, 11 Jun 2026 10:00:05 +0200 Subject: [PATCH] Implement public snapshot scheduler --- ...nking-and-historical-snapshot-scheduler.md | 72 +++ backend/app/config.py | 87 ++- backend/app/historical_runner.py | 509 +++++++++++++++++- backend/app/historical_snapshots.py | 138 +++++ backend/app/rcon_historical_leaderboards.py | 16 +- .../tests/test_historical_snapshot_refresh.py | 147 +++++ docker-compose.yml | 5 + docs/public-snapshot-refresh-schedule.md | 213 +++++--- 8 files changed, 1077 insertions(+), 110 deletions(-) create mode 100644 ai/tasks/done/TASK-240-implement-ranking-and-historical-snapshot-scheduler.md diff --git a/ai/tasks/done/TASK-240-implement-ranking-and-historical-snapshot-scheduler.md b/ai/tasks/done/TASK-240-implement-ranking-and-historical-snapshot-scheduler.md new file mode 100644 index 0000000..9639e22 --- /dev/null +++ b/ai/tasks/done/TASK-240-implement-ranking-and-historical-snapshot-scheduler.md @@ -0,0 +1,72 @@ +--- +id: TASK-240 +title: Implement ranking and historical snapshot scheduler +status: done +type: backend +team: Backend Senior +supporting_teams: + - Arquitecto Python +roadmap_item: foundation +priority: high +--- + +# TASK-240 - Implement ranking and historical snapshot scheduler + +## Goal + +Implement the real out-of-band scheduler for ranking and historical public snapshots so `ranking.html` and `historico.html` no longer depend on the daily full refresh to populate weekly and monthly data. + +## Context + +The previous audit in TASK-237 documented the gap but did not change runtime behavior. The runner refreshed ranking snapshots and recent matches, but the broader historical weekly/monthly snapshot subset used by the public historical page could remain missing for hours. + +## Steps + +1. Reviewed scheduler, snapshot generators, config and Compose wiring. +2. Added slot-based public jobs with explicit cadence and non-overlap rules. +3. Extended docs and tests for the implemented cadence. + +## Files to Read First + +- `AGENTS.md` +- `ai/architecture-index.md` +- `ai/repo-context.md` +- `backend/app/historical_runner.py` +- `backend/app/historical_snapshots.py` +- `docs/public-snapshot-refresh-schedule.md` + +## Expected Files to Modify + +- `backend/app/config.py` +- `backend/app/historical_runner.py` +- `backend/app/historical_snapshots.py` +- `backend/app/rcon_historical_leaderboards.py` +- `backend/tests/test_historical_snapshot_refresh.py` +- `docker-compose.yml` +- `docs/public-snapshot-refresh-schedule.md` + +## Constraints + +- No public request-time regeneration. +- No RCON host, port or server configuration changes. +- No Elo/MMR reactivation. +- No Comunidad Hispana #03 reintroduction. +- Keep the scheduler simple: locks and controlled skips instead of external dependencies. + +## Validation + +- `python -m compileall backend/app` +- `cd backend; python -m unittest tests.test_historical_snapshot_refresh` +- `cd backend; python -m unittest tests.test_current_match_payload tests.test_rcon_admin_log_storage tests.test_historical_snapshot_refresh` +- `git diff --name-only` + +## Outcome + +- Implemented wall-clock public jobs for ranking weekly/monthly and historical weekly/monthly subsets. +- Added manual runner entrypoints through `python -m app.historical_runner --public-job ...`. +- Kept recent matches on the existing fast path. +- Left request contracts compatible; `generated_at` remains the last-update field used by the public historical UI. + +## Change Budget + +- Scope exceeded the ideal small-task budget because scheduler, snapshot generation, tests, Compose wiring and docs had to move together to avoid partial behavior. diff --git a/backend/app/config.py b/backend/app/config.py index bb3ed04..51ef311 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -26,6 +26,11 @@ DEFAULT_PUBLIC_FULL_REFRESH_TIME = "06:00" DEFAULT_PUBLIC_FULL_REFRESH_TIMEZONE = "Europe/Madrid" DEFAULT_PUBLIC_RANKING_REFRESH_INTERVAL_SECONDS = 900 DEFAULT_PUBLIC_RECENT_MATCHES_REFRESH_INTERVAL_SECONDS = 60 +DEFAULT_PUBLIC_RANKING_WEEKLY_REFRESH_MINUTE = 10 +DEFAULT_PUBLIC_RANKING_MONTHLY_REFRESH_TIMES = ("07:00", "19:00") +DEFAULT_PUBLIC_HISTORICAL_WEEKLY_REFRESH_MINUTE = 25 +DEFAULT_PUBLIC_HISTORICAL_MONTHLY_REFRESH_MINUTE = 40 +DEFAULT_PUBLIC_HISTORICAL_MONTHLY_REFRESH_HOUR_INTERVAL = 2 DEFAULT_HISTORICAL_REFRESH_MAX_RETRIES = 2 DEFAULT_HISTORICAL_REFRESH_RETRY_DELAY_SECONDS = 30 DEFAULT_HISTORICAL_FULL_SNAPSHOT_EVERY_RUNS = 4 @@ -300,6 +305,29 @@ def _read_bool_env(name: str, *, default: bool) -> bool: raise ValueError(f"{name} must be a boolean value.") +def _normalize_hh_mm_value(name: str, configured_value: str) -> str: + """Normalize one HH:MM value used by public refresh scheduler settings.""" + normalized_value = configured_value.strip() + parts = normalized_value.split(":") + if len(parts) != 2: + raise ValueError(f"{name} must use HH:MM format.") + try: + hour, minute = (int(part) for part in parts) + except (TypeError, ValueError) as error: + raise ValueError(f"{name} must use HH:MM format.") from error + if hour < 0 or hour > 23 or minute < 0 or minute > 59: + raise ValueError(f"{name} must use HH:MM format.") + return f"{hour:02d}:{minute:02d}" + + +def _read_minute_env(name: str, default_value: int) -> int: + """Read one scheduler minute offset constrained to one wall-clock hour.""" + minute = _read_int_env(name, str(default_value), minimum=0) + if minute > 59: + raise ValueError(f"{name} must be at most 59.") + return minute + + def get_historical_refresh_overlap_hours() -> int: """Return the overlap window used by incremental historical refreshes.""" configured_value = os.getenv( @@ -528,14 +556,8 @@ def get_public_full_refresh_time() -> str: configured_value = os.getenv( "HLL_PUBLIC_FULL_REFRESH_TIME", DEFAULT_PUBLIC_FULL_REFRESH_TIME, - ).strip() - parts = configured_value.split(":") - if len(parts) != 2: - raise ValueError("HLL_PUBLIC_FULL_REFRESH_TIME must use HH:MM format.") - hour, minute = (int(part) for part in parts) - if hour < 0 or hour > 23 or minute < 0 or minute > 59: - raise ValueError("HLL_PUBLIC_FULL_REFRESH_TIME must use HH:MM format.") - return f"{hour:02d}:{minute:02d}" + ) + return _normalize_hh_mm_value("HLL_PUBLIC_FULL_REFRESH_TIME", configured_value) def get_public_full_refresh_timezone() -> str: @@ -567,6 +589,55 @@ def get_public_recent_matches_refresh_interval_seconds() -> int: ) +def get_public_ranking_weekly_refresh_minute() -> int: + """Return the minute offset used by hourly weekly ranking snapshot refreshes.""" + return _read_minute_env( + "HLL_PUBLIC_RANKING_WEEKLY_REFRESH_MINUTE", + DEFAULT_PUBLIC_RANKING_WEEKLY_REFRESH_MINUTE, + ) + + +def get_public_ranking_monthly_refresh_times() -> tuple[str, ...]: + """Return the local HH:MM slots used by monthly ranking refreshes.""" + configured_value = os.getenv( + "HLL_PUBLIC_RANKING_MONTHLY_REFRESH_TIMES", + ",".join(DEFAULT_PUBLIC_RANKING_MONTHLY_REFRESH_TIMES), + ) + values = tuple( + _normalize_hh_mm_value("HLL_PUBLIC_RANKING_MONTHLY_REFRESH_TIMES", item) + for item in configured_value.split(",") + if item.strip() + ) + if not values: + raise ValueError("HLL_PUBLIC_RANKING_MONTHLY_REFRESH_TIMES cannot be empty.") + return values + + +def get_public_historical_weekly_refresh_minute() -> int: + """Return the minute offset used by hourly historical weekly refreshes.""" + return _read_minute_env( + "HLL_PUBLIC_HISTORICAL_WEEKLY_REFRESH_MINUTE", + DEFAULT_PUBLIC_HISTORICAL_WEEKLY_REFRESH_MINUTE, + ) + + +def get_public_historical_monthly_refresh_minute() -> int: + """Return the minute offset used by periodic historical monthly refreshes.""" + return _read_minute_env( + "HLL_PUBLIC_HISTORICAL_MONTHLY_REFRESH_MINUTE", + DEFAULT_PUBLIC_HISTORICAL_MONTHLY_REFRESH_MINUTE, + ) + + +def get_public_historical_monthly_refresh_hour_interval() -> int: + """Return the hour interval used by periodic historical monthly refreshes.""" + return _read_int_env( + "HLL_PUBLIC_HISTORICAL_MONTHLY_REFRESH_HOUR_INTERVAL", + str(DEFAULT_PUBLIC_HISTORICAL_MONTHLY_REFRESH_HOUR_INTERVAL), + minimum=1, + ) + + def get_rcon_capture_mode() -> str: """Return whether the worker runs the normal historical pipeline or live capture only.""" configured_mode = os.getenv("HLL_RCON_CAPTURE_MODE") diff --git a/backend/app/historical_runner.py b/backend/app/historical_runner.py index b0bef23..371fbaf 100644 --- a/backend/app/historical_runner.py +++ b/backend/app/historical_runner.py @@ -24,7 +24,12 @@ from .config import ( get_public_full_refresh_enabled, get_public_full_refresh_time, get_public_full_refresh_timezone, + get_public_historical_monthly_refresh_hour_interval, + get_public_historical_monthly_refresh_minute, + get_public_historical_weekly_refresh_minute, + get_public_ranking_monthly_refresh_times, get_public_ranking_refresh_interval_seconds, + get_public_ranking_weekly_refresh_minute, get_public_recent_matches_refresh_interval_seconds, ) from .database_maintenance import run_database_maintenance_cleanup @@ -32,7 +37,10 @@ from .elo_mmr_engine import rebuild_elo_mmr_models from .elo_mmr_storage import get_latest_elo_mmr_generated_at from .historical_ingestion import run_incremental_refresh from .historical_snapshots import ( + PREWARM_SNAPSHOT_SERVER_KEYS, generate_and_persist_historical_snapshots, + generate_and_persist_historical_leaderboard_snapshots, + generate_and_persist_historical_monthly_ui_snapshots, generate_and_persist_priority_historical_snapshots, generate_and_persist_recent_historical_snapshots, ) @@ -58,13 +66,27 @@ DEFAULT_HISTORICAL_SERVER_SCOPE = ( ) _LAST_DATABASE_MAINTENANCE_RUN_AT: datetime | None = None _LAST_PUBLIC_FULL_REFRESH_LOCAL_DATE: date | None = None -_LAST_PUBLIC_RANKING_REFRESH_AT: datetime | None = None _LAST_PUBLIC_RECENT_MATCHES_REFRESH_AT: datetime | None = None +_LAST_PUBLIC_RANKING_WEEKLY_REFRESH_SLOT: str | None = None +_LAST_PUBLIC_RANKING_MONTHLY_REFRESH_SLOT: str | None = None +_LAST_PUBLIC_HISTORICAL_WEEKLY_REFRESH_SLOT: str | None = None +_LAST_PUBLIC_HISTORICAL_MONTHLY_REFRESH_SLOT: str | None = None _PUBLIC_REFRESH_IN_PROGRESS: set[str] = set() PUBLIC_FULL_REFRESH_YEAR = 2026 PUBLIC_REFRESH_LOCK_FULL = "public-full-refresh" -PUBLIC_REFRESH_LOCK_RANKING = "public-ranking-refresh" +PUBLIC_REFRESH_LOCK_RANKING_ANNUAL = "public-ranking-annual-refresh" +PUBLIC_REFRESH_LOCK_RANKING_MONTHLY = "public-ranking-monthly-refresh" +PUBLIC_REFRESH_LOCK_RANKING_WEEKLY = "public-ranking-weekly-refresh" +PUBLIC_REFRESH_LOCK_HISTORICAL_WEEKLY = "public-historical-weekly-refresh" +PUBLIC_REFRESH_LOCK_HISTORICAL_MONTHLY = "public-historical-monthly-refresh" PUBLIC_REFRESH_LOCK_RECENT_MATCHES = "public-recent-matches-refresh" +PUBLIC_HEAVY_REFRESH_KEYS = frozenset( + { + PUBLIC_REFRESH_LOCK_FULL, + PUBLIC_REFRESH_LOCK_RANKING_ANNUAL, + PUBLIC_REFRESH_LOCK_RANKING_MONTHLY, + } +) def run_periodic_historical_refresh( @@ -469,22 +491,118 @@ def _maybe_run_public_read_model_refreshes( """Run due public read-model refreshes without doing the heavy RCON capture cycle.""" anchor = _as_utc(now or datetime.now(timezone.utc)) results: dict[str, Any] = {} + heavy_job_ran = False if _is_public_full_refresh_due(anchor): results["public_full_refresh_result"] = _run_non_overlapping_public_refresh( PUBLIC_REFRESH_LOCK_FULL, lambda: refresh_public_full_read_models(run_number=run_number, now=anchor), + blocked_by=PUBLIC_HEAVY_REFRESH_KEYS - {PUBLIC_REFRESH_LOCK_FULL}, ) + heavy_job_ran = _refresh_result_counts_as_executed(results["public_full_refresh_result"]) - if _is_public_interval_refresh_due( - last_run_at=_LAST_PUBLIC_RANKING_REFRESH_AT, - interval_seconds=get_public_ranking_refresh_interval_seconds(), + ranking_monthly_slot = _resolve_latest_daily_schedule_slot( now=anchor, + times=get_public_ranking_monthly_refresh_times(), + ) + if _is_public_slot_refresh_due( + last_completed_slot=_LAST_PUBLIC_RANKING_MONTHLY_REFRESH_SLOT, + slot=ranking_monthly_slot, ): - results["ranking_snapshot_result"] = _run_non_overlapping_public_refresh( - PUBLIC_REFRESH_LOCK_RANKING, - lambda: refresh_public_ranking_snapshots(run_number=run_number, now=anchor), - ) + if heavy_job_ran: + results["ranking_monthly_snapshot_result"] = _build_public_scheduler_skip_result( + refresh_key=PUBLIC_REFRESH_LOCK_RANKING_MONTHLY, + reason="heavy-public-refresh-ran-this-cycle", + scheduled_slot=ranking_monthly_slot, + ) + else: + results["ranking_monthly_snapshot_result"] = _run_non_overlapping_public_refresh( + PUBLIC_REFRESH_LOCK_RANKING_MONTHLY, + lambda: refresh_public_monthly_ranking_snapshots( + run_number=run_number, + now=anchor, + scheduled_slot=ranking_monthly_slot, + ), + blocked_by=PUBLIC_HEAVY_REFRESH_KEYS - {PUBLIC_REFRESH_LOCK_RANKING_MONTHLY}, + ) + if _refresh_result_counts_as_executed(results["ranking_monthly_snapshot_result"]): + heavy_job_ran = True + + ranking_weekly_slot = _resolve_latest_hourly_schedule_slot( + now=anchor, + minute=get_public_ranking_weekly_refresh_minute(), + ) + if _is_public_slot_refresh_due( + last_completed_slot=_LAST_PUBLIC_RANKING_WEEKLY_REFRESH_SLOT, + slot=ranking_weekly_slot, + ): + if heavy_job_ran: + results["ranking_weekly_snapshot_result"] = _build_public_scheduler_skip_result( + refresh_key=PUBLIC_REFRESH_LOCK_RANKING_WEEKLY, + reason="heavy-public-refresh-ran-this-cycle", + scheduled_slot=ranking_weekly_slot, + ) + else: + results["ranking_weekly_snapshot_result"] = _run_non_overlapping_public_refresh( + PUBLIC_REFRESH_LOCK_RANKING_WEEKLY, + lambda: refresh_public_weekly_ranking_snapshots( + run_number=run_number, + now=anchor, + scheduled_slot=ranking_weekly_slot, + ), + blocked_by=PUBLIC_HEAVY_REFRESH_KEYS, + ) + + historical_weekly_slot = _resolve_latest_hourly_schedule_slot( + now=anchor, + minute=get_public_historical_weekly_refresh_minute(), + ) + if _is_public_slot_refresh_due( + last_completed_slot=_LAST_PUBLIC_HISTORICAL_WEEKLY_REFRESH_SLOT, + slot=historical_weekly_slot, + ): + if heavy_job_ran: + results["historical_weekly_snapshot_result"] = _build_public_scheduler_skip_result( + refresh_key=PUBLIC_REFRESH_LOCK_HISTORICAL_WEEKLY, + reason="heavy-public-refresh-ran-this-cycle", + scheduled_slot=historical_weekly_slot, + ) + else: + results["historical_weekly_snapshot_result"] = _run_non_overlapping_public_refresh( + PUBLIC_REFRESH_LOCK_HISTORICAL_WEEKLY, + lambda: refresh_public_weekly_historical_snapshots( + run_number=run_number, + now=anchor, + scheduled_slot=historical_weekly_slot, + ), + blocked_by=PUBLIC_HEAVY_REFRESH_KEYS | {PUBLIC_REFRESH_LOCK_HISTORICAL_MONTHLY}, + ) + + historical_monthly_slot = _resolve_latest_hourly_schedule_slot( + now=anchor, + minute=get_public_historical_monthly_refresh_minute(), + hour_interval=get_public_historical_monthly_refresh_hour_interval(), + ) + if _is_public_slot_refresh_due( + last_completed_slot=_LAST_PUBLIC_HISTORICAL_MONTHLY_REFRESH_SLOT, + slot=historical_monthly_slot, + ): + if heavy_job_ran: + results["historical_monthly_snapshot_result"] = _build_public_scheduler_skip_result( + refresh_key=PUBLIC_REFRESH_LOCK_HISTORICAL_MONTHLY, + reason="heavy-public-refresh-ran-this-cycle", + scheduled_slot=historical_monthly_slot, + ) + else: + results["historical_monthly_snapshot_result"] = _run_non_overlapping_public_refresh( + PUBLIC_REFRESH_LOCK_HISTORICAL_MONTHLY, + lambda: refresh_public_monthly_historical_snapshots( + run_number=run_number, + now=anchor, + scheduled_slot=historical_monthly_slot, + ), + blocked_by=PUBLIC_HEAVY_REFRESH_KEYS | {PUBLIC_REFRESH_LOCK_HISTORICAL_WEEKLY}, + ) if _is_public_interval_refresh_due( last_run_at=_LAST_PUBLIC_RECENT_MATCHES_REFRESH_AT, @@ -546,7 +664,10 @@ def refresh_public_full_read_models( ) steps["annual_ranking_snapshots"] = _run_public_refresh_step( "annual-ranking-snapshots-2026", - lambda: refresh_public_annual_ranking_snapshots(year=PUBLIC_FULL_REFRESH_YEAR), + lambda: refresh_public_annual_ranking_snapshots( + year=PUBLIC_FULL_REFRESH_YEAR, + now=anchor, + ), ) steps["player_search_index"] = _run_public_refresh_step( "player-search-index", @@ -579,8 +700,6 @@ def refresh_public_ranking_snapshots( now: datetime | None = None, ) -> dict[str, Any]: """Refresh weekly/monthly public ranking snapshots on the short cadence.""" - global _LAST_PUBLIC_RANKING_REFRESH_AT - anchor = _as_utc(now or datetime.now(timezone.utc)) started = time.perf_counter() _emit_json_log( @@ -591,9 +710,6 @@ def refresh_public_ranking_snapshots( } ) result = refresh_periodic_ranking_snapshots(run_number=run_number) - status = str(result.get("status") or "ok") - if status != "error": - _LAST_PUBLIC_RANKING_REFRESH_AT = anchor completed = { **result, "duration_ms": _elapsed_ms(started), @@ -605,6 +721,118 @@ def refresh_public_ranking_snapshots( return completed +def refresh_public_weekly_ranking_snapshots( + *, + run_number: int, + now: datetime | None = None, + scheduled_slot: datetime | None = None, +) -> dict[str, Any]: + """Refresh the hourly weekly public ranking matrix.""" + global _LAST_PUBLIC_RANKING_WEEKLY_REFRESH_SLOT + + anchor = _as_utc(now or datetime.now(timezone.utc)) + slot = scheduled_slot or _resolve_latest_hourly_schedule_slot( + now=anchor, + minute=get_public_ranking_weekly_refresh_minute(), + ) + started = time.perf_counter() + _emit_json_log( + { + "event": "public-ranking-weekly-refresh-started", + "job_name": "public-ranking-weekly", + "run_number": run_number, + "timeframe": "weekly", + "scheduled_slot": _to_iso(slot), + "scheduled_slot_local": _format_local_schedule_slot(slot), + } + ) + try: + result = refresh_ranking_snapshots( + limit=30, + replace_existing=True, + timeframes=("weekly",), + now=anchor, + ) + except Exception as exc: # noqa: BLE001 - scheduler must keep the runner alive + result = { + "status": "error", + "error_type": type(exc).__name__, + "error": str(exc), + "traceback": traceback.format_exc(), + } + if str(result.get("status") or "ok") != "error": + _LAST_PUBLIC_RANKING_WEEKLY_REFRESH_SLOT = _schedule_slot_key(slot) + completed = { + **result, + "job_name": "public-ranking-weekly", + "timeframe": "weekly", + "duration_ms": _elapsed_ms(started), + "generated_at": _to_iso(anchor), + "scheduled_slot": _to_iso(slot), + "scheduled_slot_local": _format_local_schedule_slot(slot), + "generation_policy": "scheduled-hourly-weekly-ranking-refresh", + "server_scope": "all-public-ranking-scopes", + } + _emit_json_log({"event": "public-ranking-weekly-refresh-completed", **completed}) + return completed + + +def refresh_public_monthly_ranking_snapshots( + *, + run_number: int, + now: datetime | None = None, + scheduled_slot: datetime | None = None, +) -> dict[str, Any]: + """Refresh the twice-daily monthly public ranking matrix.""" + global _LAST_PUBLIC_RANKING_MONTHLY_REFRESH_SLOT + + anchor = _as_utc(now or datetime.now(timezone.utc)) + slot = scheduled_slot or _resolve_latest_daily_schedule_slot( + now=anchor, + times=get_public_ranking_monthly_refresh_times(), + ) + started = time.perf_counter() + _emit_json_log( + { + "event": "public-ranking-monthly-refresh-started", + "job_name": "public-ranking-monthly", + "run_number": run_number, + "timeframe": "monthly", + "scheduled_slot": _to_iso(slot), + "scheduled_slot_local": _format_local_schedule_slot(slot), + } + ) + try: + result = refresh_ranking_snapshots( + limit=30, + replace_existing=True, + timeframes=("monthly",), + now=anchor, + ) + except Exception as exc: # noqa: BLE001 - scheduler must keep the runner alive + result = { + "status": "error", + "error_type": type(exc).__name__, + "error": str(exc), + "traceback": traceback.format_exc(), + } + if str(result.get("status") or "ok") != "error": + _LAST_PUBLIC_RANKING_MONTHLY_REFRESH_SLOT = _schedule_slot_key(slot) + completed = { + **result, + "job_name": "public-ranking-monthly", + "timeframe": "monthly", + "duration_ms": _elapsed_ms(started), + "generated_at": _to_iso(anchor), + "scheduled_slot": _to_iso(slot), + "scheduled_slot_local": _format_local_schedule_slot(slot), + "generation_policy": "scheduled-twice-daily-monthly-ranking-refresh", + "server_scope": "all-public-ranking-scopes", + } + _emit_json_log({"event": "public-ranking-monthly-refresh-completed", **completed}) + return completed + + def refresh_public_recent_matches_snapshots( *, run_number: int, @@ -657,8 +885,10 @@ def refresh_public_annual_ranking_snapshots( *, year: int = PUBLIC_FULL_REFRESH_YEAR, limit: int = 20, + now: datetime | None = None, ) -> dict[str, Any]: """Refresh supported annual ranking snapshots for all public scopes.""" + anchor = _as_utc(now or datetime.now(timezone.utc)) combinations = [ (server_key, metric) for server_key in SNAPSHOT_GENERATOR_SERVER_KEYS @@ -709,6 +939,7 @@ def refresh_public_annual_ranking_snapshots( status = "error" return { "status": status, + "generated_at": _to_iso(anchor), "year": year, "limit": limit, "combinations_expected": len(combinations), @@ -720,6 +951,122 @@ def refresh_public_annual_ranking_snapshots( } +def refresh_public_weekly_historical_snapshots( + *, + run_number: int, + now: datetime | None = None, + scheduled_slot: datetime | None = None, +) -> dict[str, Any]: + """Refresh the hourly weekly historical leaderboard subset used by historico.html.""" + global _LAST_PUBLIC_HISTORICAL_WEEKLY_REFRESH_SLOT + + anchor = _as_utc(now or datetime.now(timezone.utc)) + slot = scheduled_slot or _resolve_latest_hourly_schedule_slot( + now=anchor, + minute=get_public_historical_weekly_refresh_minute(), + ) + started = time.perf_counter() + _emit_json_log( + { + "event": "public-historical-weekly-refresh-started", + "job_name": "public-historical-weekly", + "run_number": run_number, + "timeframe": "weekly", + "scheduled_slot": _to_iso(slot), + "scheduled_slot_local": _format_local_schedule_slot(slot), + "server_keys": list(PREWARM_SNAPSHOT_SERVER_KEYS), + } + ) + try: + result = { + "status": "ok", + **generate_and_persist_historical_leaderboard_snapshots( + timeframe="weekly", + server_keys=PREWARM_SNAPSHOT_SERVER_KEYS, + generated_at=anchor, + ), + } + except Exception as exc: # noqa: BLE001 - scheduler must keep the runner alive + result = { + "status": "error", + "error_type": type(exc).__name__, + "error": str(exc), + "traceback": traceback.format_exc(), + } + if str(result.get("status") or "ok") != "error": + _LAST_PUBLIC_HISTORICAL_WEEKLY_REFRESH_SLOT = _schedule_slot_key(slot) + completed = { + **result, + "job_name": "public-historical-weekly", + "timeframe": "weekly", + "duration_ms": _elapsed_ms(started), + "generated_at": result.get("generated_at") or _to_iso(anchor), + "scheduled_slot": _to_iso(slot), + "scheduled_slot_local": _format_local_schedule_slot(slot), + "generation_policy": "scheduled-hourly-historical-weekly-refresh", + } + _emit_json_log({"event": "public-historical-weekly-refresh-completed", **completed}) + return completed + + +def refresh_public_monthly_historical_snapshots( + *, + run_number: int, + now: datetime | None = None, + scheduled_slot: datetime | None = None, +) -> dict[str, Any]: + """Refresh the periodic monthly historical subset used by historico.html.""" + global _LAST_PUBLIC_HISTORICAL_MONTHLY_REFRESH_SLOT + + anchor = _as_utc(now or datetime.now(timezone.utc)) + slot = scheduled_slot or _resolve_latest_hourly_schedule_slot( + now=anchor, + minute=get_public_historical_monthly_refresh_minute(), + hour_interval=get_public_historical_monthly_refresh_hour_interval(), + ) + started = time.perf_counter() + _emit_json_log( + { + "event": "public-historical-monthly-refresh-started", + "job_name": "public-historical-monthly", + "run_number": run_number, + "timeframe": "monthly", + "scheduled_slot": _to_iso(slot), + "scheduled_slot_local": _format_local_schedule_slot(slot), + "server_keys": list(PREWARM_SNAPSHOT_SERVER_KEYS), + } + ) + try: + result = { + "status": "ok", + **generate_and_persist_historical_monthly_ui_snapshots( + server_keys=PREWARM_SNAPSHOT_SERVER_KEYS, + generated_at=anchor, + ), + } + except Exception as exc: # noqa: BLE001 - scheduler must keep the runner alive + result = { + "status": "error", + "error_type": type(exc).__name__, + "error": str(exc), + "traceback": traceback.format_exc(), + } + if str(result.get("status") or "ok") != "error": + _LAST_PUBLIC_HISTORICAL_MONTHLY_REFRESH_SLOT = _schedule_slot_key(slot) + completed = { + **result, + "job_name": "public-historical-monthly", + "timeframe": "monthly", + "duration_ms": _elapsed_ms(started), + "generated_at": result.get("generated_at") or _to_iso(anchor), + "scheduled_slot": _to_iso(slot), + "scheduled_slot_local": _format_local_schedule_slot(slot), + "generation_policy": "scheduled-bi-hourly-historical-monthly-refresh", + } + _emit_json_log({"event": "public-historical-monthly-refresh-completed", **completed}) + return completed + + def get_next_public_full_refresh_at( *, now: datetime | None = None, @@ -788,6 +1135,7 @@ def _run_public_refresh_step( def _run_non_overlapping_public_refresh( refresh_key: str, callback: Any, + blocked_by: set[str] | frozenset[str] | None = None, ) -> dict[str, Any]: if refresh_key in _PUBLIC_REFRESH_IN_PROGRESS: return { @@ -795,6 +1143,14 @@ def _run_non_overlapping_public_refresh( "reason": "refresh-already-in-progress", "refresh_key": refresh_key, } + blocking_keys = sorted((blocked_by or set()) & _PUBLIC_REFRESH_IN_PROGRESS) + if blocking_keys: + return { + "status": "skipped", + "reason": "conflicting-refresh-in-progress", + "refresh_key": refresh_key, + "blocked_by": blocking_keys, + } _PUBLIC_REFRESH_IN_PROGRESS.add(refresh_key) try: return callback() @@ -807,11 +1163,7 @@ def _record_public_refreshes_from_cycle( *, now: datetime, ) -> None: - global _LAST_PUBLIC_RANKING_REFRESH_AT, _LAST_PUBLIC_RECENT_MATCHES_REFRESH_AT - - ranking_result = payload.get("ranking_snapshot_result") - if isinstance(ranking_result, dict) and ranking_result.get("status") != "error": - _LAST_PUBLIC_RANKING_REFRESH_AT = now + global _LAST_PUBLIC_RECENT_MATCHES_REFRESH_AT recent_result = payload.get("recent_matches_event_result") if isinstance(recent_result, dict) and recent_result.get("status") not in { "error", @@ -866,10 +1218,87 @@ def _resolve_runner_tick_seconds(interval_seconds: int) -> int: interval_seconds, get_public_ranking_refresh_interval_seconds(), get_public_recent_matches_refresh_interval_seconds(), + 60, ] return max(1, min(intervals)) +def _resolve_latest_hourly_schedule_slot( + *, + now: datetime, + minute: int, + hour_interval: int = 1, +) -> datetime: + local_now = now.astimezone(_get_public_refresh_zone()) + candidate = local_now.replace(minute=minute, second=0, microsecond=0) + if local_now < candidate: + candidate -= timedelta(hours=1) + candidate = candidate.replace(minute=minute, second=0, microsecond=0) + while candidate.hour % hour_interval != 0: + candidate -= timedelta(hours=1) + candidate = candidate.replace(minute=minute, second=0, microsecond=0) + return candidate.astimezone(timezone.utc) + + +def _resolve_latest_daily_schedule_slot( + *, + now: datetime, + times: tuple[str, ...], +) -> datetime: + local_now = now.astimezone(_get_public_refresh_zone()) + candidates: list[datetime] = [] + for time_value in times: + hour, minute = (int(part) for part in time_value.split(":")) + candidate = datetime.combine( + local_now.date(), + datetime_time(hour=hour, minute=minute), + tzinfo=local_now.tzinfo, + ) + if candidate > local_now: + candidate -= timedelta(days=1) + candidates.append(candidate) + return max(candidates).astimezone(timezone.utc) + + +def _schedule_slot_key(slot: datetime) -> str: + return _as_utc(slot).isoformat().replace("+00:00", "Z") + + +def _format_local_schedule_slot(slot: datetime) -> str: + return slot.astimezone(_get_public_refresh_zone()).isoformat(timespec="minutes") + + +def _is_public_slot_refresh_due( + *, + last_completed_slot: str | None, + slot: datetime, +) -> bool: + return last_completed_slot != _schedule_slot_key(slot) + + +def _build_public_scheduler_skip_result( + *, + refresh_key: str, + reason: str, + scheduled_slot: datetime | None = None, +) -> dict[str, Any]: + result: dict[str, Any] = { + "status": "skipped", + "reason": reason, + "refresh_key": refresh_key, + } + if scheduled_slot is not None: + result["scheduled_slot"] = _to_iso(scheduled_slot) + result["scheduled_slot_local"] = _format_local_schedule_slot(scheduled_slot) + return result + + +def _refresh_result_counts_as_executed(result: dict[str, Any]) -> bool: + if not isinstance(result, dict): + return False + return str(result.get("status") or "").lower() in {"ok", "partial"} + + def _get_public_refresh_zone() -> ZoneInfo: timezone_name = get_public_full_refresh_timezone() try: @@ -929,6 +1358,29 @@ def _resolve_refresh_cycle_status(**results: dict[str, Any]) -> str: return "ok" +def run_public_refresh_job_once(job_name: str, *, now: datetime | None = None) -> dict[str, Any]: + """Run one public snapshot job immediately for manual validation.""" + anchor = _as_utc(now or datetime.now(timezone.utc)) + normalized_job = str(job_name or "").strip().lower() + if normalized_job == "public-full": + return refresh_public_full_read_models(run_number=1, now=anchor) + if normalized_job == "ranking-weekly": + return refresh_public_weekly_ranking_snapshots(run_number=1, now=anchor) + if normalized_job == "ranking-monthly": + return refresh_public_monthly_ranking_snapshots(run_number=1, now=anchor) + if normalized_job == "historical-weekly": + return refresh_public_weekly_historical_snapshots(run_number=1, now=anchor) + if normalized_job == "historical-monthly": + return refresh_public_monthly_historical_snapshots(run_number=1, now=anchor) + if normalized_job == "recent-matches": + return refresh_public_recent_matches_snapshots( + run_number=1, + now=anchor, + trigger="manual-public-job", + ) + raise ValueError(f"Unsupported public job: {job_name}") + + def _emit_json_log(payload: dict[str, Any]) -> None: """Print JSON logs that remain safe for Compose and log collectors.""" print(json.dumps(payload, ensure_ascii=True, default=str), flush=True) @@ -1178,6 +1630,19 @@ def main() -> None: default=None, help="Optional safety limit for the number of refresh cycles to execute.", ) + parser.add_argument( + "--public-job", + choices=( + "public-full", + "ranking-weekly", + "ranking-monthly", + "historical-weekly", + "historical-monthly", + "recent-matches", + ), + default=None, + help="Run one public snapshot scheduler job immediately and exit.", + ) args = parser.parse_args() if args.hourly: @@ -1191,6 +1656,10 @@ def main() -> None: raise ValueError("--retry-delay must be zero or positive.") if args.max_runs is not None and args.max_runs <= 0: raise ValueError("--max-runs must be positive when provided.") + if args.public_job is not None: + payload = run_public_refresh_job_once(args.public_job) + print(json.dumps({"status": "ok", "data": payload}, indent=2)) + return run_periodic_historical_refresh( interval_seconds=args.interval, diff --git a/backend/app/historical_snapshots.py b/backend/app/historical_snapshots.py index 98d8dab..c818895 100644 --- a/backend/app/historical_snapshots.py +++ b/backend/app/historical_snapshots.py @@ -392,6 +392,144 @@ def generate_and_persist_priority_historical_snapshots( } +def generate_and_persist_historical_leaderboard_snapshots( + *, + timeframe: str, + server_keys: tuple[str, ...] = PREWARM_SNAPSHOT_SERVER_KEYS, + generated_at: datetime | None = None, + leaderboard_limit: int = DEFAULT_WEEKLY_LEADERBOARD_LIMIT, + db_path: Path | None = None, +) -> dict[str, object]: + """Build and persist only the weekly or monthly leaderboard snapshot set.""" + from .historical_snapshot_storage import persist_historical_snapshot_batch + + generated_at_value = _as_utc(generated_at or datetime.now(timezone.utc)) + leaderboard_limit = _normalize_snapshot_limit("leaderboard_limit", leaderboard_limit) + normalized_timeframe = str(timeframe or "").strip().lower() + if normalized_timeframe not in {"weekly", "monthly"}: + raise ValueError("timeframe must be 'weekly' or 'monthly'.") + + snapshots: list[dict[str, object]] = [] + for server_key in server_keys: + for metric in SNAPSHOT_LEADERBOARD_METRICS: + if normalized_timeframe == "weekly": + _log_snapshot_build_started( + server_key, + SNAPSHOT_TYPE_WEEKLY_LEADERBOARD, + metric=metric, + ) + snapshots.append( + _build_weekly_leaderboard_snapshot( + server_key, + metric, + generated_at_value, + limit=leaderboard_limit, + db_path=db_path, + ) + ) + continue + + _log_snapshot_build_started( + server_key, + SNAPSHOT_TYPE_MONTHLY_LEADERBOARD, + metric=metric, + ) + snapshots.append( + _build_monthly_leaderboard_snapshot( + server_key, + metric, + generated_at_value, + limit=leaderboard_limit, + db_path=db_path, + ) + ) + + persisted_records = persist_historical_snapshot_batch(snapshots, db_path=db_path) + snapshots_by_server: dict[str, int] = {} + for record in persisted_records: + snapshots_by_server.setdefault(record.server_key, 0) + snapshots_by_server[record.server_key] += 1 + + return { + "generated_at": _to_iso(generated_at_value), + "timeframe": normalized_timeframe, + "snapshot_policy": "leaderboard-subset", + "server_keys": list(server_keys), + "metrics": list(SNAPSHOT_LEADERBOARD_METRICS), + "snapshot_count": len(persisted_records), + "servers_processed": len(snapshots_by_server), + "snapshots_by_server": snapshots_by_server, + } + + +def generate_and_persist_historical_monthly_ui_snapshots( + *, + server_keys: tuple[str, ...] = PREWARM_SNAPSHOT_SERVER_KEYS, + generated_at: datetime | None = None, + leaderboard_limit: int = DEFAULT_WEEKLY_LEADERBOARD_LIMIT, + db_path: Path | None = None, +) -> dict[str, object]: + """Build and persist the monthly historical subset rendered by the public UI.""" + from .historical_snapshot_storage import persist_historical_snapshot_batch + + generated_at_value = _as_utc(generated_at or datetime.now(timezone.utc)) + leaderboard_limit = _normalize_snapshot_limit("leaderboard_limit", leaderboard_limit) + + snapshots: list[dict[str, object]] = [] + for server_key in server_keys: + for metric in SNAPSHOT_LEADERBOARD_METRICS: + _log_snapshot_build_started( + server_key, + SNAPSHOT_TYPE_MONTHLY_LEADERBOARD, + metric=metric, + ) + snapshots.append( + _build_monthly_leaderboard_snapshot( + server_key, + metric, + generated_at_value, + limit=leaderboard_limit, + db_path=db_path, + ) + ) + _log_snapshot_build_started(server_key, SNAPSHOT_TYPE_MONTHLY_MVP) + snapshots.append( + _build_monthly_mvp_snapshot( + server_key, + generated_at_value, + limit=leaderboard_limit, + db_path=db_path, + ) + ) + _log_snapshot_build_started(server_key, SNAPSHOT_TYPE_MONTHLY_MVP_V2) + snapshots.append( + _build_monthly_mvp_v2_snapshot( + server_key, + generated_at_value, + limit=leaderboard_limit, + db_path=db_path, + ) + ) + + persisted_records = persist_historical_snapshot_batch(snapshots, db_path=db_path) + snapshots_by_server: dict[str, int] = {} + for record in persisted_records: + snapshots_by_server.setdefault(record.server_key, 0) + snapshots_by_server[record.server_key] += 1 + + return { + "generated_at": _to_iso(generated_at_value), + "timeframe": "monthly", + "snapshot_policy": "monthly-ui-subset", + "server_keys": list(server_keys), + "metrics": list(SNAPSHOT_LEADERBOARD_METRICS), + "includes_monthly_mvp": True, + "snapshot_count": len(persisted_records), + "servers_processed": len(snapshots_by_server), + "snapshots_by_server": snapshots_by_server, + } + + def generate_and_persist_recent_historical_snapshots( *, server_key: str | None = None, diff --git a/backend/app/rcon_historical_leaderboards.py b/backend/app/rcon_historical_leaderboards.py index 41633cd..f6382ce 100644 --- a/backend/app/rcon_historical_leaderboards.py +++ b/backend/app/rcon_historical_leaderboards.py @@ -230,15 +230,20 @@ def refresh_ranking_snapshots( *, limit: int = DEFAULT_RANKING_SNAPSHOT_REFRESH_LIMIT, replace_existing: bool = True, + timeframes: tuple[str, ...] | None = None, now: datetime | None = None, db_path: Path | None = None, ) -> dict[str, object]: """Generate the full weekly/monthly ranking snapshot matrix with partial-failure reporting.""" normalized_limit = _normalize_limit(limit) anchor = _as_utc(now or datetime.now(timezone.utc)) + normalized_timeframes = tuple( + _normalize_generator_timeframe(timeframe) + for timeframe in (timeframes or SNAPSHOT_GENERATOR_TIMEFRAMES) + ) combinations = [ (timeframe, server_key, metric) - for timeframe in SNAPSHOT_GENERATOR_TIMEFRAMES + for timeframe in normalized_timeframes for server_key in SNAPSHOT_GENERATOR_SERVER_KEYS for metric in SNAPSHOT_GENERATOR_METRICS ] @@ -305,6 +310,7 @@ def refresh_ranking_snapshots( return { "status": overall_status, "generated_at": _to_iso(anchor), + "timeframes": list(normalized_timeframes), "limit": normalized_limit, "replace_existing": replace_existing, "combinations_expected": len(combinations), @@ -1421,6 +1427,13 @@ def _main(argv: list[str] | None = None) -> int: type=int, default=DEFAULT_RANKING_SNAPSHOT_REFRESH_LIMIT, ) + refresh_parser.add_argument( + "--timeframe", + choices=SNAPSHOT_GENERATOR_TIMEFRAMES, + action="append", + dest="timeframes", + help="Optional timeframe filter. Repeat to refresh more than one timeframe.", + ) refresh_parser.add_argument( "--sqlite-path", type=Path, @@ -1459,6 +1472,7 @@ def _main(argv: list[str] | None = None) -> int: payload = refresh_ranking_snapshots( limit=args.limit, replace_existing=args.replace_existing, + timeframes=tuple(args.timeframes) if args.timeframes else None, db_path=args.sqlite_path, ) print( diff --git a/backend/tests/test_historical_snapshot_refresh.py b/backend/tests/test_historical_snapshot_refresh.py index 4f0a1c1..d2ed1d2 100644 --- a/backend/tests/test_historical_snapshot_refresh.py +++ b/backend/tests/test_historical_snapshot_refresh.py @@ -10,6 +10,7 @@ from contextlib import nullcontext, redirect_stdout from datetime import datetime, timezone from unittest.mock import patch +import app.historical_runner as historical_runner_module from app.config import ( get_historical_refresh_interval_seconds, get_historical_refresh_max_retries, @@ -17,7 +18,12 @@ from app.config import ( get_public_full_refresh_enabled, get_public_full_refresh_time, get_public_full_refresh_timezone, + get_public_historical_monthly_refresh_hour_interval, + get_public_historical_monthly_refresh_minute, + get_public_historical_weekly_refresh_minute, + get_public_ranking_monthly_refresh_times, get_public_ranking_refresh_interval_seconds, + get_public_ranking_weekly_refresh_minute, get_public_recent_matches_refresh_interval_seconds, ) from app.payloads import ( @@ -27,8 +33,10 @@ from app.payloads import ( build_recent_historical_matches_snapshot_payload, ) from app.historical_runner import ( + _maybe_run_public_read_model_refreshes, _run_refresh_with_retries, get_next_public_full_refresh_at, + run_public_refresh_job_once, run_periodic_historical_refresh, ) from app.historical_snapshots import _normalize_snapshot_limit @@ -40,6 +48,15 @@ from app.rcon_historical_read_model import ( class HistoricalSnapshotRefreshTests(unittest.TestCase): + def setUp(self) -> None: + historical_runner_module._LAST_PUBLIC_FULL_REFRESH_LOCAL_DATE = None + historical_runner_module._LAST_PUBLIC_RECENT_MATCHES_REFRESH_AT = None + historical_runner_module._LAST_PUBLIC_RANKING_WEEKLY_REFRESH_SLOT = None + historical_runner_module._LAST_PUBLIC_RANKING_MONTHLY_REFRESH_SLOT = None + historical_runner_module._LAST_PUBLIC_HISTORICAL_WEEKLY_REFRESH_SLOT = None + historical_runner_module._LAST_PUBLIC_HISTORICAL_MONTHLY_REFRESH_SLOT = None + historical_runner_module._PUBLIC_REFRESH_IN_PROGRESS.clear() + def test_runner_numeric_env_values_are_parsed_before_use(self) -> None: with patch.dict( os.environ, @@ -75,6 +92,11 @@ class HistoricalSnapshotRefreshTests(unittest.TestCase): "HLL_PUBLIC_FULL_REFRESH_TIMEZONE": "Europe/Madrid", "HLL_PUBLIC_RANKING_REFRESH_INTERVAL_SECONDS": "900", "HLL_PUBLIC_RECENT_MATCHES_REFRESH_INTERVAL_SECONDS": "60", + "HLL_PUBLIC_RANKING_WEEKLY_REFRESH_MINUTE": "10", + "HLL_PUBLIC_RANKING_MONTHLY_REFRESH_TIMES": "07:00,19:00", + "HLL_PUBLIC_HISTORICAL_WEEKLY_REFRESH_MINUTE": "25", + "HLL_PUBLIC_HISTORICAL_MONTHLY_REFRESH_MINUTE": "40", + "HLL_PUBLIC_HISTORICAL_MONTHLY_REFRESH_HOUR_INTERVAL": "2", }, clear=False, ): @@ -83,6 +105,11 @@ class HistoricalSnapshotRefreshTests(unittest.TestCase): self.assertEqual(get_public_full_refresh_timezone(), "Europe/Madrid") self.assertEqual(get_public_ranking_refresh_interval_seconds(), 900) self.assertEqual(get_public_recent_matches_refresh_interval_seconds(), 60) + self.assertEqual(get_public_ranking_weekly_refresh_minute(), 10) + self.assertEqual(get_public_ranking_monthly_refresh_times(), ("07:00", "19:00")) + self.assertEqual(get_public_historical_weekly_refresh_minute(), 25) + self.assertEqual(get_public_historical_monthly_refresh_minute(), 40) + self.assertEqual(get_public_historical_monthly_refresh_hour_interval(), 2) def test_next_public_full_refresh_uses_madrid_six_am(self) -> None: with patch.dict( @@ -102,6 +129,126 @@ class HistoricalSnapshotRefreshTests(unittest.TestCase): datetime(2026, 6, 10, 4, 0, tzinfo=timezone.utc), ) + def test_public_scheduler_runs_hourly_weekly_ranking_job(self) -> None: + historical_runner_module._LAST_PUBLIC_RECENT_MATCHES_REFRESH_AT = datetime( + 2026, 6, 10, 6, 10, tzinfo=timezone.utc + ) + historical_runner_module._LAST_PUBLIC_RANKING_MONTHLY_REFRESH_SLOT = "2026-06-10T05:00:00Z" + historical_runner_module._LAST_PUBLIC_HISTORICAL_WEEKLY_REFRESH_SLOT = "2026-06-10T05:25:00Z" + historical_runner_module._LAST_PUBLIC_HISTORICAL_MONTHLY_REFRESH_SLOT = "2026-06-10T04:40:00Z" + with ( + patch.dict( + os.environ, + { + "HLL_PUBLIC_FULL_REFRESH_ENABLED": "false", + "HLL_PUBLIC_RANKING_WEEKLY_REFRESH_MINUTE": "10", + "HLL_PUBLIC_RANKING_MONTHLY_REFRESH_TIMES": "07:00,19:00", + "HLL_PUBLIC_HISTORICAL_WEEKLY_REFRESH_MINUTE": "25", + "HLL_PUBLIC_HISTORICAL_MONTHLY_REFRESH_MINUTE": "40", + "HLL_PUBLIC_HISTORICAL_MONTHLY_REFRESH_HOUR_INTERVAL": "2", + "HLL_PUBLIC_RECENT_MATCHES_REFRESH_INTERVAL_SECONDS": "3600", + }, + clear=False, + ), + patch( + "app.historical_runner.refresh_public_weekly_ranking_snapshots", + return_value={"status": "ok"}, + ) as weekly_ranking_refresh, + ): + payload = _maybe_run_public_read_model_refreshes( + run_number=3, + now=datetime(2026, 6, 10, 6, 10, tzinfo=timezone.utc), + ) + + self.assertEqual(payload["status"], "ok") + self.assertIn("ranking_weekly_snapshot_result", payload) + weekly_ranking_refresh.assert_called_once() + + def test_public_scheduler_runs_bi_hourly_monthly_historical_job(self) -> None: + historical_runner_module._LAST_PUBLIC_RECENT_MATCHES_REFRESH_AT = datetime( + 2026, 6, 10, 6, 40, tzinfo=timezone.utc + ) + historical_runner_module._LAST_PUBLIC_RANKING_MONTHLY_REFRESH_SLOT = "2026-06-10T05:00:00Z" + historical_runner_module._LAST_PUBLIC_RANKING_WEEKLY_REFRESH_SLOT = "2026-06-10T06:10:00Z" + historical_runner_module._LAST_PUBLIC_HISTORICAL_WEEKLY_REFRESH_SLOT = "2026-06-10T06:25:00Z" + with ( + patch.dict( + os.environ, + { + "HLL_PUBLIC_FULL_REFRESH_ENABLED": "false", + "HLL_PUBLIC_RANKING_WEEKLY_REFRESH_MINUTE": "10", + "HLL_PUBLIC_RANKING_MONTHLY_REFRESH_TIMES": "07:00,19:00", + "HLL_PUBLIC_HISTORICAL_WEEKLY_REFRESH_MINUTE": "25", + "HLL_PUBLIC_HISTORICAL_MONTHLY_REFRESH_MINUTE": "40", + "HLL_PUBLIC_HISTORICAL_MONTHLY_REFRESH_HOUR_INTERVAL": "2", + "HLL_PUBLIC_RECENT_MATCHES_REFRESH_INTERVAL_SECONDS": "3600", + }, + clear=False, + ), + patch( + "app.historical_runner.refresh_public_monthly_historical_snapshots", + return_value={"status": "ok"}, + ) as historical_monthly_refresh, + ): + payload = _maybe_run_public_read_model_refreshes( + run_number=4, + now=datetime(2026, 6, 10, 6, 40, tzinfo=timezone.utc), + ) + + self.assertEqual(payload["status"], "ok") + self.assertIn("historical_monthly_snapshot_result", payload) + historical_monthly_refresh.assert_called_once() + + def test_public_scheduler_skips_lower_priority_jobs_when_heavy_job_runs(self) -> None: + historical_runner_module._LAST_PUBLIC_RECENT_MATCHES_REFRESH_AT = datetime( + 2026, 6, 10, 5, 30, tzinfo=timezone.utc + ) + with ( + patch.dict( + os.environ, + { + "HLL_PUBLIC_FULL_REFRESH_ENABLED": "false", + "HLL_PUBLIC_RANKING_WEEKLY_REFRESH_MINUTE": "10", + "HLL_PUBLIC_RANKING_MONTHLY_REFRESH_TIMES": "07:00,19:00", + "HLL_PUBLIC_HISTORICAL_WEEKLY_REFRESH_MINUTE": "25", + "HLL_PUBLIC_HISTORICAL_MONTHLY_REFRESH_MINUTE": "40", + "HLL_PUBLIC_HISTORICAL_MONTHLY_REFRESH_HOUR_INTERVAL": "2", + "HLL_PUBLIC_RECENT_MATCHES_REFRESH_INTERVAL_SECONDS": "3600", + }, + clear=False, + ), + patch( + "app.historical_runner.refresh_public_monthly_ranking_snapshots", + return_value={"status": "ok"}, + ) as monthly_ranking_refresh, + patch("app.historical_runner.refresh_public_weekly_ranking_snapshots") as weekly_ranking_refresh, + patch("app.historical_runner.refresh_public_weekly_historical_snapshots") as weekly_historical_refresh, + ): + payload = _maybe_run_public_read_model_refreshes( + run_number=5, + now=datetime(2026, 6, 10, 5, 30, tzinfo=timezone.utc), + ) + + self.assertEqual(payload["ranking_monthly_snapshot_result"]["status"], "ok") + self.assertEqual(payload["ranking_weekly_snapshot_result"]["status"], "skipped") + self.assertEqual(payload["historical_weekly_snapshot_result"]["status"], "skipped") + monthly_ranking_refresh.assert_called_once() + weekly_ranking_refresh.assert_not_called() + weekly_historical_refresh.assert_not_called() + + def test_manual_public_job_runner_routes_to_supported_job(self) -> None: + with patch( + "app.historical_runner.refresh_public_weekly_historical_snapshots", + return_value={"status": "ok", "job_name": "public-historical-weekly"}, + ) as weekly_historical_refresh: + result = run_public_refresh_job_once( + "historical-weekly", + now=datetime(2026, 6, 10, 6, 25, tzinfo=timezone.utc), + ) + + self.assertEqual(result["job_name"], "public-historical-weekly") + weekly_historical_refresh.assert_called_once() + def test_historical_leaderboard_snapshot_does_not_runtime_enrich_public_request(self) -> None: snapshot = { "generated_at": "2026-06-10T04:00:00Z", diff --git a/docker-compose.yml b/docker-compose.yml index 737cd0b..fabb556 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -57,6 +57,11 @@ services: HLL_PUBLIC_FULL_REFRESH_TIME: ${HLL_PUBLIC_FULL_REFRESH_TIME:-06:00} HLL_PUBLIC_FULL_REFRESH_TIMEZONE: ${HLL_PUBLIC_FULL_REFRESH_TIMEZONE:-Europe/Madrid} HLL_PUBLIC_RANKING_REFRESH_INTERVAL_SECONDS: ${HLL_PUBLIC_RANKING_REFRESH_INTERVAL_SECONDS:-900} + HLL_PUBLIC_RANKING_WEEKLY_REFRESH_MINUTE: ${HLL_PUBLIC_RANKING_WEEKLY_REFRESH_MINUTE:-10} + HLL_PUBLIC_RANKING_MONTHLY_REFRESH_TIMES: ${HLL_PUBLIC_RANKING_MONTHLY_REFRESH_TIMES:-07:00,19:00} + HLL_PUBLIC_HISTORICAL_WEEKLY_REFRESH_MINUTE: ${HLL_PUBLIC_HISTORICAL_WEEKLY_REFRESH_MINUTE:-25} + HLL_PUBLIC_HISTORICAL_MONTHLY_REFRESH_MINUTE: ${HLL_PUBLIC_HISTORICAL_MONTHLY_REFRESH_MINUTE:-40} + HLL_PUBLIC_HISTORICAL_MONTHLY_REFRESH_HOUR_INTERVAL: ${HLL_PUBLIC_HISTORICAL_MONTHLY_REFRESH_HOUR_INTERVAL:-2} HLL_PUBLIC_RECENT_MATCHES_REFRESH_INTERVAL_SECONDS: ${HLL_PUBLIC_RECENT_MATCHES_REFRESH_INTERVAL_SECONDS:-60} HLL_BACKEND_RCON_TARGETS: >- ${HLL_BACKEND_RCON_TARGETS:-[{"name":"Comunidad Hispana #01","slug":"comunidad-hispana-01","external_server_id":"comunidad-hispana-01","host":"152.114.195.174","port":7779,"password":"replace-me-01","source_name":"community-hispana-rcon","region":"ES","game_port":null,"query_port":null},{"name":"Comunidad Hispana #02","slug":"comunidad-hispana-02","external_server_id":"comunidad-hispana-02","host":"152.114.195.150","port":7879,"password":"replace-me-02","source_name":"community-hispana-rcon","region":"ES","game_port":null,"query_port":null}]} diff --git a/docs/public-snapshot-refresh-schedule.md b/docs/public-snapshot-refresh-schedule.md index e9b15f2..3f79ab4 100644 --- a/docs/public-snapshot-refresh-schedule.md +++ b/docs/public-snapshot-refresh-schedule.md @@ -2,117 +2,168 @@ ## Goal -Public pages should read precomputed PostgreSQL read models or persisted public snapshots. Ranking requests must not regenerate snapshots and must not query RCON directly. +Public pages must read persisted snapshots and read models only. No public GET endpoint should regenerate ranking or historical snapshots at request time. -## Runner Scheduling +## Current Scheduler Owner The internal `historical-runner` owns public refresh scheduling. Host cron is not required. -Daily full refresh: +## Implemented Cadence -- Controlled by `HLL_PUBLIC_FULL_REFRESH_ENABLED`. -- Runs once per local day after `HLL_PUBLIC_FULL_REFRESH_TIME` in `HLL_PUBLIC_FULL_REFRESH_TIMEZONE`. -- Default: `06:00 Europe/Madrid`. -- Intended for the lowest RCON and database load window. +Local timezone for scheduled jobs: `Europe/Madrid`. -The daily full refresh rebuilds: +Heavy daily window: -- full historical public snapshots/read models from `generate_and_persist_historical_snapshots()`; -- weekly/monthly `ranking_snapshots`; -- annual 2026 ranking snapshots for `kills`, `deaths`, `teamkills`, `matches_considered`, `kd_ratio` and `kills_per_match`; -- `player_search_index`; -- `player_period_stats`. +- full public refresh: + - `06:00` + - rebuilds full historical snapshots, weekly/monthly ranking snapshots, annual ranking snapshots, player search index and player period stats + - runs under `public-full-refresh` -Frequent refreshes: +Ranking page: -- `HLL_PUBLIC_RANKING_REFRESH_INTERVAL_SECONDS`, default `900`, refreshes weekly/monthly public ranking snapshots. -- `HLL_PUBLIC_RECENT_MATCHES_REFRESH_INTERVAL_SECONDS`, default `60`, refreshes recent-match snapshots when no direct finished-match hook fires. -- When the RCON capture cycle reports newly materialized finished matches, the runner refreshes recent-match snapshots immediately. +- annual ranking: + - rebuilt inside the daily full refresh at `06:00` + - generated sequentially per scope and metric +- monthly ranking: + - `07:00` and `19:00` + - generated sequentially per scope and metric + - runs under `public-ranking-monthly-refresh` +- weekly ranking: + - every hour at minute `10` + - generated sequentially per scope and metric + - runs under `public-ranking-weekly-refresh` -## Gap Found In Scheduler Audit +Historical page: -Current short cadence refresh only covers: +- weekly historical leaderboard subset: + - every hour at minute `25` + - scopes: + - `all-servers` + - `comunidad-hispana-01` + - `comunidad-hispana-02` + - metrics: + - `kills` + - `deaths` + - `matches_over_100_kills` + - `support` + - runs under `public-historical-weekly-refresh` +- monthly historical UI subset: + - every `2` hours at minute `40` + - same scopes as weekly historical + - leaderboard metrics: + - `kills` + - `deaths` + - `matches_over_100_kills` + - `support` + - also refreshes: + - `monthly-mvp` + - `monthly-mvp-v2` + - runs under `public-historical-monthly-refresh` -- `refresh_public_ranking_snapshots()` for `/api/ranking` -- `refresh_public_recent_matches_snapshots()` for recent matches +Recent matches: -It does not refresh the broader historical snapshot matrix consumed by `historico.html` every hour. That means weekly/monthly historical leaderboard or summary data can remain `snapshot_status=missing` until the next daily full refresh at `06:00`, even while ranking snapshots are already updating every 15 minutes. +- interval polling fallback remains enabled through `HLL_PUBLIC_RECENT_MATCHES_REFRESH_INTERVAL_SECONDS` +- default: `60` seconds +- immediate refresh still happens when the RCON capture loop materializes a finished match -## Recommended Future Cadence +## Why Historical Weekly And Monthly Were Missing -Target cadence identified by the audit: +The previous short cadence only refreshed: -- annual ranking snapshots: - - daily at `06:00 Europe/Madrid` -- monthly ranking snapshots: - - `07:00` and `19:00 Europe/Madrid` -- weekly ranking snapshots: - - every hour -- historical weekly leaderboard snapshots used by `historico.html`: - - every hour -- historical monthly leaderboard snapshots used by `historico.html`: - - every 2 hours +- `ranking_snapshots` for `/api/ranking` +- `recent-matches` snapshots -Operational constraints for that future backend change: +It did not refresh the broader historical snapshot subset consumed by `historico.html`. That left weekly and monthly historical leaderboard payloads in `snapshot_status=missing` until the next daily full refresh. -- large ranking matrix refreshes should run sequentially -- annual/full refresh should not overlap with shorter historical refreshes -- per-job start/end/duration/scope/result logs should remain explicit -- public GET endpoints must stay snapshot-only and must not regenerate heavy data on request +## Locking And Overlap Policy -Refreshes are idempotent: existing ranking snapshots are replaced for the same window/scope, player read models are rebuilt per scope, and persisted historical snapshots are replaced/upserted by snapshot identity. +The scheduler uses in-process locks and controlled skips. It does not enqueue backlog jobs. -## Portainer +Heavy jobs: -Run the `historical-runner` service from the Compose stack with the advanced profile. The service runs: +- `public-full-refresh` +- `public-ranking-annual-refresh` +- `public-ranking-monthly-refresh` + +Policy: + +- heavy jobs do not overlap with other heavy jobs +- hourly and bi-hourly jobs skip when a conflicting heavy job is in progress +- if a heavy job already ran in the same scheduler tick, lower-priority jobs skip and retry on the next tick +- weekly and monthly historical subset jobs also avoid overlapping with each other + +This keeps the implementation simple and avoids CPU spikes without reintroducing request-time fallback work. + +## Last Update Exposure + +Historical and ranking payloads continue to expose persisted generation metadata from the snapshot records: + +- `snapshot_status` +- `generated_at` +- `source_range_start` +- `source_range_end` +- `is_stale` + +`historico.js` already renders `generated_at` as the visible "Actualizado" label, so no frontend contract change was required for this task. + +## Environment Variables + +Existing variables kept: + +- `HLL_PUBLIC_FULL_REFRESH_ENABLED` +- `HLL_PUBLIC_FULL_REFRESH_TIME` +- `HLL_PUBLIC_FULL_REFRESH_TIMEZONE` +- `HLL_PUBLIC_RECENT_MATCHES_REFRESH_INTERVAL_SECONDS` + +Retained for compatibility: + +- `HLL_PUBLIC_RANKING_REFRESH_INTERVAL_SECONDS` + - no longer drives the hourly/slot-based ranking scheduler + - still participates in runner tick resolution and legacy compatibility paths + +New scheduler variables: + +- `HLL_PUBLIC_RANKING_WEEKLY_REFRESH_MINUTE` + - default `10` +- `HLL_PUBLIC_RANKING_MONTHLY_REFRESH_TIMES` + - default `07:00,19:00` +- `HLL_PUBLIC_HISTORICAL_WEEKLY_REFRESH_MINUTE` + - default `25` +- `HLL_PUBLIC_HISTORICAL_MONTHLY_REFRESH_MINUTE` + - default `40` +- `HLL_PUBLIC_HISTORICAL_MONTHLY_REFRESH_HOUR_INTERVAL` + - default `2` + +## Manual Validation Commands + +One-off public jobs through the runner: ```powershell -python -m app.historical_runner +docker compose exec historical-runner python -m app.historical_runner --public-job ranking-weekly +docker compose exec historical-runner python -m app.historical_runner --public-job ranking-monthly +docker compose exec historical-runner python -m app.historical_runner --public-job historical-weekly +docker compose exec historical-runner python -m app.historical_runner --public-job historical-monthly +docker compose exec historical-runner python -m app.historical_runner --public-job public-full ``` -Recommended Portainer environment values: - -```text -HLL_PUBLIC_FULL_REFRESH_ENABLED=true -HLL_PUBLIC_FULL_REFRESH_TIME=06:00 -HLL_PUBLIC_FULL_REFRESH_TIMEZONE=Europe/Madrid -HLL_PUBLIC_RANKING_REFRESH_INTERVAL_SECONDS=900 -HLL_PUBLIC_RECENT_MATCHES_REFRESH_INTERVAL_SECONDS=60 -``` - -Do not add host cron unless the internal runner cannot be used in the deployment. If a sidecar is ever required, it should call the same Python modules, not duplicate SQL logic. - -## Manual Emergency Commands - -One-off full runner cycle: +Direct ranking matrix commands: ```powershell -docker compose exec historical-runner python -m app.historical_runner --max-runs 1 -``` - -Weekly/monthly ranking snapshots: - -```powershell -docker compose exec historical-runner python -m app.rcon_historical_leaderboards refresh-ranking-snapshots --limit 30 -``` - -Annual ranking snapshot, one metric/scope: - -```powershell -docker compose exec historical-runner python -m app.rcon_annual_rankings generate --year 2026 --metric kills --server-key all -``` - -Player read models: - -```powershell -docker compose exec historical-runner python -m app.rcon_historical_player_stats refresh-player-search-index -docker compose exec historical-runner python -m app.rcon_historical_player_stats refresh-player-period-stats +docker compose exec historical-runner python -m app.rcon_historical_leaderboards refresh-ranking-snapshots --timeframe weekly --limit 30 +docker compose exec historical-runner python -m app.rcon_historical_leaderboards refresh-ranking-snapshots --timeframe monthly --limit 30 ``` Operational checks: ```powershell -docker compose ps historical-runner -docker compose logs --tail=200 historical-runner -docker compose exec backend python -m app.storage_diagnostics +docker logs --tail=300 hll-vietnam-historical-runner-1 +docker exec hll-vietnam-historical-runner-1 sh -lc 'env | sort | grep HLL_PUBLIC' +python .\scripts\audit_public_requests.py --base-url https://comunidadhll.devzamode.es --timeout 30 --filter servers --output tmp\task240_servers_after.json +python .\scripts\audit_public_requests.py --base-url https://comunidadhll.devzamode.es --timeout 30 --output tmp\task240_full_audit_after.json ``` + +UI checks: + +- verify `historico.html` weekly rankings render without `snapshot_status=missing` +- verify `historico.html` monthly rankings render without `snapshot_status=missing` +- verify the visible "Actualizado" label changes after the corresponding runner job