1677 lines
61 KiB
Python
1677 lines
61 KiB
Python
"""Local development loop for periodic historical CRCON refreshes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import time
|
|
import traceback
|
|
from datetime import date, datetime, time as datetime_time, timedelta, timezone
|
|
from typing import Any
|
|
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
|
|
|
from .config import (
|
|
DEFAULT_DB_MAINTENANCE_INTERVAL_SECONDS,
|
|
get_db_maintenance_enabled,
|
|
get_db_maintenance_interval_seconds,
|
|
get_historical_full_snapshot_every_runs,
|
|
get_historical_elo_mmr_min_new_samples,
|
|
get_historical_elo_mmr_rebuild_interval_minutes,
|
|
get_historical_refresh_interval_seconds,
|
|
get_historical_refresh_max_retries,
|
|
get_historical_refresh_retry_delay_seconds,
|
|
get_historical_data_source_kind,
|
|
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
|
|
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,
|
|
)
|
|
from .historical_storage import ALL_SERVERS_SLUG
|
|
from .rcon_annual_rankings import (
|
|
SUPPORTED_ANNUAL_RANKING_METRICS,
|
|
generate_annual_ranking_snapshot,
|
|
)
|
|
from .rcon_historical_leaderboards import refresh_ranking_snapshots
|
|
from .rcon_historical_leaderboards import SNAPSHOT_GENERATOR_SERVER_KEYS
|
|
from .rcon_historical_player_stats import (
|
|
refresh_player_period_stats,
|
|
refresh_player_search_index,
|
|
)
|
|
from .rcon_historical_storage import count_rcon_historical_samples_since
|
|
from .rcon_historical_worker import run_rcon_historical_capture
|
|
from .writer_lock import backend_writer_lock, build_writer_lock_holder
|
|
|
|
HOURLY_INTERVAL_SECONDS = 3600
|
|
DEFAULT_HISTORICAL_SERVER_SCOPE = (
|
|
"comunidad-hispana-01",
|
|
"comunidad-hispana-02",
|
|
)
|
|
_LAST_DATABASE_MAINTENANCE_RUN_AT: datetime | None = None
|
|
_LAST_PUBLIC_FULL_REFRESH_LOCAL_DATE: date | 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_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(
|
|
*,
|
|
interval_seconds: int,
|
|
max_retries: int,
|
|
retry_delay_seconds: float,
|
|
server_slug: str | None = None,
|
|
max_pages: int | None = None,
|
|
page_size: int | None = None,
|
|
max_runs: int | None = None,
|
|
) -> None:
|
|
"""Run periodic historical refreshes and rebuild persisted snapshots."""
|
|
completed_runs = 0
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"event": "historical-refresh-loop-started",
|
|
"interval_seconds": interval_seconds,
|
|
"max_retries": max_retries,
|
|
"retry_delay_seconds": retry_delay_seconds,
|
|
"server_scope": _describe_refresh_scope(server_slug),
|
|
"snapshot_scope": _describe_snapshot_scope(server_slug),
|
|
},
|
|
indent=2,
|
|
)
|
|
)
|
|
print("Press Ctrl+C to stop.")
|
|
|
|
last_historical_refresh_started_at: datetime | None = None
|
|
loop_tick_seconds = _resolve_runner_tick_seconds(interval_seconds)
|
|
try:
|
|
while max_runs is None or completed_runs < max_runs:
|
|
now = datetime.now(timezone.utc)
|
|
historical_refresh_due = _is_interval_due(
|
|
last_run_at=last_historical_refresh_started_at,
|
|
interval_seconds=interval_seconds,
|
|
now=now,
|
|
)
|
|
if historical_refresh_due:
|
|
completed_runs += 1
|
|
last_historical_refresh_started_at = now
|
|
payload = _run_refresh_with_retries(
|
|
max_retries=max_retries,
|
|
retry_delay_seconds=retry_delay_seconds,
|
|
server_slug=server_slug,
|
|
max_pages=max_pages,
|
|
page_size=page_size,
|
|
run_number=completed_runs,
|
|
)
|
|
_record_public_refreshes_from_cycle(payload, now=now)
|
|
_emit_json_log({"run": completed_runs, **payload})
|
|
else:
|
|
payload = _maybe_run_public_read_model_refreshes(
|
|
run_number=completed_runs,
|
|
now=now,
|
|
)
|
|
if payload["status"] != "skipped":
|
|
_emit_json_log({"run": completed_runs, **payload})
|
|
|
|
if max_runs is not None and completed_runs >= max_runs:
|
|
break
|
|
|
|
time.sleep(loop_tick_seconds)
|
|
except KeyboardInterrupt:
|
|
print("\nHistorical refresh loop stopped by user.")
|
|
|
|
|
|
def _run_refresh_with_retries(
|
|
*,
|
|
max_retries: int,
|
|
retry_delay_seconds: float,
|
|
server_slug: str | None,
|
|
max_pages: int | None,
|
|
page_size: int | None,
|
|
run_number: int,
|
|
) -> dict[str, Any]:
|
|
attempt = 0
|
|
while True:
|
|
attempt += 1
|
|
try:
|
|
with backend_writer_lock(
|
|
holder=build_writer_lock_holder(
|
|
f"app.historical_runner refresh:{server_slug or 'all-servers'}"
|
|
)
|
|
):
|
|
rcon_capture_result = _run_primary_rcon_capture()
|
|
should_run_classic_fallback, classic_fallback_reason = (
|
|
_resolve_classic_fallback_policy(
|
|
server_slug=server_slug,
|
|
run_number=run_number,
|
|
rcon_capture_result=rcon_capture_result,
|
|
)
|
|
)
|
|
if should_run_classic_fallback:
|
|
refresh_result = run_incremental_refresh(
|
|
server_slug=server_slug,
|
|
max_pages=max_pages,
|
|
page_size=page_size,
|
|
rebuild_snapshots=False,
|
|
)
|
|
historical_snapshot_result = refresh_periodic_historical_snapshots(
|
|
server_slug=server_slug,
|
|
run_number=run_number,
|
|
)
|
|
elo_mmr_result = rebuild_elo_mmr_models()
|
|
else:
|
|
should_generate_snapshots = _rcon_capture_has_new_useful_data(
|
|
rcon_capture_result
|
|
)
|
|
refresh_result = {
|
|
"status": "skipped",
|
|
"reason": "rcon-primary-cycle-no-classic-fallback-needed",
|
|
}
|
|
if should_generate_snapshots:
|
|
historical_snapshot_result = refresh_periodic_historical_snapshots(
|
|
server_slug=server_slug,
|
|
run_number=run_number,
|
|
)
|
|
historical_snapshot_result = {
|
|
**historical_snapshot_result,
|
|
"generation_policy": "rcon-primary-useful-cycle",
|
|
"reason": "rcon-primary-cycle-produced-new-useful-coverage",
|
|
}
|
|
elo_policy = _build_elo_mmr_rebuild_policy(
|
|
rcon_capture_result=rcon_capture_result
|
|
)
|
|
if bool(elo_policy["due"]):
|
|
elo_mmr_result = {
|
|
**rebuild_elo_mmr_models(),
|
|
"generation_policy": "rcon-primary-useful-cycle-elo-rebuild-due",
|
|
"reason": "rcon-primary-useful-cycle-met-elo-rebuild-threshold",
|
|
**elo_policy,
|
|
}
|
|
else:
|
|
elo_mmr_result = {
|
|
"status": "skipped",
|
|
"reason": "rcon-primary-useful-cycle-elo-rebuild-throttled",
|
|
"generation_policy": "rcon-primary-useful-cycle-elo-rebuild-throttled",
|
|
**elo_policy,
|
|
}
|
|
else:
|
|
historical_snapshot_result = {
|
|
"status": "skipped",
|
|
"reason": "rcon-primary-cycle-had-no-new-useful-data",
|
|
"generation_policy": "rcon-primary-no-new-useful-data",
|
|
}
|
|
elo_mmr_result = {
|
|
"status": "skipped",
|
|
"reason": "rcon-primary-cycle-had-no-new-useful-data",
|
|
"generation_policy": "rcon-primary-no-new-useful-data",
|
|
**_build_elo_mmr_rebuild_policy(
|
|
rcon_capture_result=rcon_capture_result
|
|
),
|
|
}
|
|
player_search_index_result = refresh_periodic_player_search_index(
|
|
server_slug=server_slug,
|
|
run_number=run_number,
|
|
)
|
|
player_period_stats_result = refresh_periodic_player_period_stats(
|
|
server_slug=server_slug,
|
|
run_number=run_number,
|
|
)
|
|
ranking_snapshot_result = refresh_periodic_ranking_snapshots(
|
|
server_slug=server_slug,
|
|
run_number=run_number,
|
|
)
|
|
recent_matches_event_result = _maybe_refresh_recent_matches_after_capture(
|
|
rcon_capture_result=rcon_capture_result,
|
|
server_slug=server_slug,
|
|
run_number=run_number,
|
|
)
|
|
maintenance_result = _maybe_run_database_maintenance()
|
|
return {
|
|
"status": _resolve_refresh_cycle_status(
|
|
refresh_result=refresh_result,
|
|
historical_snapshot_result=historical_snapshot_result,
|
|
player_search_index_result=player_search_index_result,
|
|
player_period_stats_result=player_period_stats_result,
|
|
ranking_snapshot_result=ranking_snapshot_result,
|
|
recent_matches_event_result=recent_matches_event_result,
|
|
elo_mmr_result=elo_mmr_result,
|
|
database_maintenance_result=maintenance_result,
|
|
),
|
|
"attempts_used": attempt,
|
|
"max_retries": max_retries,
|
|
"rcon_capture_result": rcon_capture_result,
|
|
"classic_fallback_used": should_run_classic_fallback,
|
|
"classic_fallback_reason": classic_fallback_reason,
|
|
"refresh_result": refresh_result,
|
|
"historical_snapshot_result": historical_snapshot_result,
|
|
"snapshot_result": historical_snapshot_result,
|
|
"player_search_index_result": player_search_index_result,
|
|
"player_period_stats_result": player_period_stats_result,
|
|
"ranking_snapshot_result": ranking_snapshot_result,
|
|
"recent_matches_event_result": recent_matches_event_result,
|
|
"elo_mmr_result": elo_mmr_result,
|
|
"database_maintenance_result": maintenance_result,
|
|
}
|
|
except Exception as exc:
|
|
failure_payload = {
|
|
"event": "historical-refresh-attempt-failed",
|
|
"attempt": attempt,
|
|
"max_retries": max_retries,
|
|
"server_scope": _describe_refresh_scope(server_slug),
|
|
"snapshot_scope": _describe_snapshot_scope(server_slug),
|
|
"error_type": type(exc).__name__,
|
|
"error": str(exc),
|
|
"traceback": traceback.format_exc(),
|
|
}
|
|
_emit_json_log(failure_payload)
|
|
if attempt > max_retries:
|
|
return {
|
|
"status": "error",
|
|
"attempts_used": attempt,
|
|
"max_retries": max_retries,
|
|
"error_type": type(exc).__name__,
|
|
"error": str(exc),
|
|
"traceback": failure_payload["traceback"],
|
|
}
|
|
if retry_delay_seconds > 0:
|
|
time.sleep(retry_delay_seconds)
|
|
|
|
|
|
def generate_historical_snapshots(
|
|
*,
|
|
server_slug: str | None = None,
|
|
run_number: int = 1,
|
|
) -> dict[str, Any]:
|
|
"""Build priority prewarm snapshots on every run and the full matrix on cadence."""
|
|
generated_at = datetime.now(timezone.utc)
|
|
full_snapshot_every_runs = get_historical_full_snapshot_every_runs()
|
|
should_run_full_refresh = bool(server_slug) or run_number % full_snapshot_every_runs == 0
|
|
_emit_json_log(
|
|
{
|
|
"event": "historical-snapshot-refresh-started",
|
|
"run_number": run_number,
|
|
"snapshot_step": "full-matrix" if should_run_full_refresh else "priority-prewarm",
|
|
"server_slug": server_slug,
|
|
"snapshot_scope": _describe_snapshot_scope(server_slug),
|
|
}
|
|
)
|
|
if should_run_full_refresh:
|
|
result = generate_and_persist_historical_snapshots(
|
|
server_key=server_slug,
|
|
generated_at=generated_at,
|
|
)
|
|
else:
|
|
result = generate_and_persist_priority_historical_snapshots(
|
|
generated_at=generated_at,
|
|
)
|
|
return {
|
|
**result,
|
|
"run_number": run_number,
|
|
"full_snapshot_every_runs": full_snapshot_every_runs,
|
|
"prewarm_only": not should_run_full_refresh,
|
|
"refresh_interval_seconds": get_historical_refresh_interval_seconds(),
|
|
"includes_monthly_mvp_v2": True,
|
|
}
|
|
|
|
|
|
def refresh_periodic_historical_snapshots(
|
|
*,
|
|
server_slug: str | None = None,
|
|
run_number: int = 1,
|
|
) -> dict[str, Any]:
|
|
"""Refresh legacy historical snapshots without aborting operational read-model refreshes."""
|
|
try:
|
|
return generate_historical_snapshots(
|
|
server_slug=server_slug,
|
|
run_number=run_number,
|
|
)
|
|
except Exception as exc: # noqa: BLE001 - legacy failures must remain visible but isolated
|
|
failure_result = {
|
|
"status": "error",
|
|
"error_type": type(exc).__name__,
|
|
"error": str(exc),
|
|
"traceback": traceback.format_exc(),
|
|
"run_number": run_number,
|
|
"refresh_interval_seconds": get_historical_refresh_interval_seconds(),
|
|
"server_slug": server_slug,
|
|
"generation_policy": "periodic-historical-refresh-cycle",
|
|
"snapshot_scope": _describe_snapshot_scope(server_slug),
|
|
"legacy_block": "historical-snapshots",
|
|
}
|
|
_emit_json_log({"event": "historical-snapshot-refresh-failed", **failure_result})
|
|
return failure_result
|
|
|
|
|
|
def refresh_periodic_ranking_snapshots(
|
|
*,
|
|
server_slug: str | None = None,
|
|
run_number: int = 1,
|
|
) -> dict[str, Any]:
|
|
"""Refresh the public weekly/monthly ranking snapshot matrix for the current cycle."""
|
|
_emit_json_log(
|
|
{
|
|
"event": "ranking-snapshot-refresh-started",
|
|
"run_number": run_number,
|
|
"server_slug": server_slug,
|
|
"snapshot_scope": _describe_snapshot_scope(server_slug),
|
|
"ranking_snapshot_limit": 30,
|
|
}
|
|
)
|
|
result = refresh_ranking_snapshots(limit=30)
|
|
return {
|
|
**result,
|
|
"run_number": run_number,
|
|
"refresh_interval_seconds": get_historical_refresh_interval_seconds(),
|
|
"server_slug": server_slug,
|
|
"generation_policy": "periodic-historical-refresh-cycle",
|
|
"recommended_frequency": {
|
|
"weekly": "5-15-minutes",
|
|
"monthly": "15-30-minutes",
|
|
},
|
|
}
|
|
|
|
|
|
def refresh_periodic_player_search_index(
|
|
*,
|
|
server_slug: str | None = None,
|
|
run_number: int = 1,
|
|
) -> dict[str, Any]:
|
|
"""Refresh the player search read model without aborting the remaining cycle on failure."""
|
|
_emit_json_log(
|
|
{
|
|
"event": "player-search-index-refresh-started",
|
|
"run_number": run_number,
|
|
"server_slug": server_slug,
|
|
"refresh_scope": "supported-public-player-search-scopes",
|
|
}
|
|
)
|
|
try:
|
|
result = refresh_player_search_index()
|
|
except Exception as exc: # noqa: BLE001 - one read-model failure must stay visible
|
|
failure_result = {
|
|
"status": "error",
|
|
"error_type": type(exc).__name__,
|
|
"error": str(exc),
|
|
"run_number": run_number,
|
|
"refresh_interval_seconds": get_historical_refresh_interval_seconds(),
|
|
"server_slug": server_slug,
|
|
"generation_policy": "periodic-historical-refresh-cycle",
|
|
"scope_policy": "always-refresh-supported-public-player-search-scopes",
|
|
}
|
|
_emit_json_log({"event": "player-search-index-refresh-failed", **failure_result})
|
|
return failure_result
|
|
return {
|
|
**result,
|
|
"run_number": run_number,
|
|
"refresh_interval_seconds": get_historical_refresh_interval_seconds(),
|
|
"server_slug": server_slug,
|
|
"generation_policy": "periodic-historical-refresh-cycle",
|
|
"scope_policy": "always-refresh-supported-public-player-search-scopes",
|
|
}
|
|
|
|
|
|
def refresh_periodic_player_period_stats(
|
|
*,
|
|
server_slug: str | None = None,
|
|
run_number: int = 1,
|
|
) -> dict[str, Any]:
|
|
"""Refresh the player period stats read model without aborting the remaining cycle on failure."""
|
|
_emit_json_log(
|
|
{
|
|
"event": "player-period-stats-refresh-started",
|
|
"run_number": run_number,
|
|
"server_slug": server_slug,
|
|
"refresh_scope": "supported-public-player-period-scopes",
|
|
}
|
|
)
|
|
try:
|
|
result = refresh_player_period_stats()
|
|
except Exception as exc: # noqa: BLE001 - one read-model failure must stay visible
|
|
failure_result = {
|
|
"status": "error",
|
|
"error_type": type(exc).__name__,
|
|
"error": str(exc),
|
|
"run_number": run_number,
|
|
"refresh_interval_seconds": get_historical_refresh_interval_seconds(),
|
|
"server_slug": server_slug,
|
|
"generation_policy": "periodic-historical-refresh-cycle",
|
|
"scope_policy": "always-refresh-supported-public-player-period-scopes",
|
|
}
|
|
_emit_json_log({"event": "player-period-stats-refresh-failed", **failure_result})
|
|
return failure_result
|
|
return {
|
|
**result,
|
|
"run_number": run_number,
|
|
"refresh_interval_seconds": get_historical_refresh_interval_seconds(),
|
|
"server_slug": server_slug,
|
|
"generation_policy": "periodic-historical-refresh-cycle",
|
|
"scope_policy": "always-refresh-supported-public-player-period-scopes",
|
|
}
|
|
|
|
|
|
def _maybe_run_public_read_model_refreshes(
|
|
*,
|
|
run_number: int,
|
|
now: datetime | None = None,
|
|
) -> dict[str, Any]:
|
|
"""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"])
|
|
|
|
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,
|
|
):
|
|
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,
|
|
interval_seconds=get_public_recent_matches_refresh_interval_seconds(),
|
|
now=anchor,
|
|
):
|
|
results["recent_matches_snapshot_result"] = _run_non_overlapping_public_refresh(
|
|
PUBLIC_REFRESH_LOCK_RECENT_MATCHES,
|
|
lambda: refresh_public_recent_matches_snapshots(
|
|
run_number=run_number,
|
|
now=anchor,
|
|
trigger="polling-interval",
|
|
),
|
|
)
|
|
|
|
if not results:
|
|
return {
|
|
"event": "public-read-model-refresh-scheduler-skipped",
|
|
"status": "skipped",
|
|
"reason": "no-public-refresh-due",
|
|
}
|
|
|
|
return {
|
|
"event": "public-read-model-refresh-scheduler-completed",
|
|
"status": _resolve_refresh_cycle_status(**results),
|
|
**results,
|
|
}
|
|
|
|
|
|
def refresh_public_full_read_models(
|
|
*,
|
|
run_number: int,
|
|
now: datetime | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Refresh the full public read-model set intended for the low-load nightly window."""
|
|
global _LAST_PUBLIC_FULL_REFRESH_LOCAL_DATE
|
|
|
|
anchor = _as_utc(now or datetime.now(timezone.utc))
|
|
local_anchor = anchor.astimezone(_get_public_refresh_zone())
|
|
steps: dict[str, Any] = {}
|
|
started = time.perf_counter()
|
|
_emit_json_log(
|
|
{
|
|
"event": "public-full-refresh-started",
|
|
"run_number": run_number,
|
|
"scheduled_local_date": local_anchor.date().isoformat(),
|
|
"scheduled_time": get_public_full_refresh_time(),
|
|
"timezone": get_public_full_refresh_timezone(),
|
|
}
|
|
)
|
|
|
|
steps["historical_snapshots"] = _run_public_refresh_step(
|
|
"historical-snapshots-full",
|
|
lambda: generate_and_persist_historical_snapshots(generated_at=anchor),
|
|
)
|
|
steps["ranking_snapshots"] = _run_public_refresh_step(
|
|
"ranking-snapshots-weekly-monthly",
|
|
lambda: refresh_periodic_ranking_snapshots(run_number=run_number),
|
|
)
|
|
steps["annual_ranking_snapshots"] = _run_public_refresh_step(
|
|
"annual-ranking-snapshots-2026",
|
|
lambda: refresh_public_annual_ranking_snapshots(
|
|
year=PUBLIC_FULL_REFRESH_YEAR,
|
|
now=anchor,
|
|
),
|
|
)
|
|
steps["player_search_index"] = _run_public_refresh_step(
|
|
"player-search-index",
|
|
refresh_player_search_index,
|
|
)
|
|
steps["player_period_stats"] = _run_public_refresh_step(
|
|
"player-period-stats",
|
|
refresh_player_period_stats,
|
|
)
|
|
|
|
status = _resolve_refresh_cycle_status(**steps)
|
|
if status in {"ok", "partial"}:
|
|
_LAST_PUBLIC_FULL_REFRESH_LOCAL_DATE = local_anchor.date()
|
|
result = {
|
|
"status": status,
|
|
"run_number": run_number,
|
|
"generated_at": _to_iso(anchor),
|
|
"duration_ms": _elapsed_ms(started),
|
|
"timezone": get_public_full_refresh_timezone(),
|
|
"scheduled_time": get_public_full_refresh_time(),
|
|
"steps": steps,
|
|
}
|
|
_emit_json_log({"event": "public-full-refresh-completed", **result})
|
|
return result
|
|
|
|
|
|
def refresh_public_ranking_snapshots(
|
|
*,
|
|
run_number: int,
|
|
now: datetime | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Refresh weekly/monthly public ranking snapshots on the short cadence."""
|
|
anchor = _as_utc(now or datetime.now(timezone.utc))
|
|
started = time.perf_counter()
|
|
_emit_json_log(
|
|
{
|
|
"event": "public-ranking-refresh-started",
|
|
"run_number": run_number,
|
|
"interval_seconds": get_public_ranking_refresh_interval_seconds(),
|
|
}
|
|
)
|
|
result = refresh_periodic_ranking_snapshots(run_number=run_number)
|
|
completed = {
|
|
**result,
|
|
"duration_ms": _elapsed_ms(started),
|
|
"generated_at": _to_iso(anchor),
|
|
"interval_seconds": get_public_ranking_refresh_interval_seconds(),
|
|
"generation_policy": "public-ranking-short-cadence",
|
|
}
|
|
_emit_json_log({"event": "public-ranking-refresh-completed", **completed})
|
|
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,
|
|
now: datetime | None = None,
|
|
trigger: str,
|
|
server_slug: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Refresh recent-match snapshots after match-finalization evidence or short polling."""
|
|
global _LAST_PUBLIC_RECENT_MATCHES_REFRESH_AT
|
|
|
|
anchor = _as_utc(now or datetime.now(timezone.utc))
|
|
started = time.perf_counter()
|
|
_emit_json_log(
|
|
{
|
|
"event": "public-recent-matches-refresh-started",
|
|
"run_number": run_number,
|
|
"trigger": trigger,
|
|
"server_slug": server_slug,
|
|
"interval_seconds": get_public_recent_matches_refresh_interval_seconds(),
|
|
}
|
|
)
|
|
try:
|
|
result = generate_and_persist_recent_historical_snapshots(
|
|
server_key=server_slug,
|
|
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(),
|
|
}
|
|
status = str(result.get("status") or "ok")
|
|
if status != "error":
|
|
_LAST_PUBLIC_RECENT_MATCHES_REFRESH_AT = anchor
|
|
completed = {
|
|
**result,
|
|
"duration_ms": _elapsed_ms(started),
|
|
"generated_at": _to_iso(anchor),
|
|
"trigger": trigger,
|
|
"server_slug": server_slug,
|
|
"interval_seconds": get_public_recent_matches_refresh_interval_seconds(),
|
|
}
|
|
_emit_json_log({"event": "public-recent-matches-refresh-completed", **completed})
|
|
return completed
|
|
|
|
|
|
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
|
|
for metric in SUPPORTED_ANNUAL_RANKING_METRICS
|
|
]
|
|
results: list[dict[str, Any]] = []
|
|
succeeded = 0
|
|
failed = 0
|
|
for server_key, metric in combinations:
|
|
try:
|
|
payload = generate_annual_ranking_snapshot(
|
|
year=year,
|
|
server_key=None if server_key == ALL_SERVERS_SLUG else server_key,
|
|
metric=metric,
|
|
limit=limit,
|
|
replace_existing=True,
|
|
)
|
|
snapshot = payload.get("snapshot") if isinstance(payload, dict) else {}
|
|
succeeded += 1
|
|
results.append(
|
|
{
|
|
"status": "ok",
|
|
"year": year,
|
|
"server_key": server_key,
|
|
"metric": metric,
|
|
"snapshot_id": snapshot.get("id") if isinstance(snapshot, dict) else None,
|
|
"ranked_players": int(payload.get("ranked_players") or 0),
|
|
"source_matches_count": int(payload.get("source_matches_count") or 0),
|
|
}
|
|
)
|
|
except Exception as exc: # noqa: BLE001 - report per-snapshot failures
|
|
failed += 1
|
|
results.append(
|
|
{
|
|
"status": "error",
|
|
"year": year,
|
|
"server_key": server_key,
|
|
"metric": metric,
|
|
"error_type": type(exc).__name__,
|
|
"error": str(exc),
|
|
}
|
|
)
|
|
|
|
status = "ok"
|
|
if failed and succeeded:
|
|
status = "partial"
|
|
elif failed:
|
|
status = "error"
|
|
return {
|
|
"status": status,
|
|
"generated_at": _to_iso(anchor),
|
|
"year": year,
|
|
"limit": limit,
|
|
"combinations_expected": len(combinations),
|
|
"totals": {
|
|
"succeeded": succeeded,
|
|
"failed": failed,
|
|
},
|
|
"results": results,
|
|
}
|
|
|
|
|
|
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,
|
|
) -> datetime:
|
|
"""Return the next configured daily public full refresh in UTC."""
|
|
anchor = _as_utc(now or datetime.now(timezone.utc))
|
|
zone = _get_public_refresh_zone()
|
|
local_anchor = anchor.astimezone(zone)
|
|
hour, minute = (int(part) for part in get_public_full_refresh_time().split(":"))
|
|
candidate = datetime.combine(
|
|
local_anchor.date(),
|
|
datetime_time(hour=hour, minute=minute),
|
|
tzinfo=zone,
|
|
)
|
|
if local_anchor >= candidate:
|
|
candidate += timedelta(days=1)
|
|
return candidate.astimezone(timezone.utc)
|
|
|
|
|
|
def _maybe_refresh_recent_matches_after_capture(
|
|
*,
|
|
rcon_capture_result: dict[str, Any],
|
|
server_slug: str | None,
|
|
run_number: int,
|
|
) -> dict[str, Any]:
|
|
if not _rcon_capture_materialized_finished_match(rcon_capture_result):
|
|
return {
|
|
"status": "skipped",
|
|
"reason": "no-materialized-finished-match-detected",
|
|
"trigger": "rcon-capture",
|
|
}
|
|
return _run_non_overlapping_public_refresh(
|
|
PUBLIC_REFRESH_LOCK_RECENT_MATCHES,
|
|
lambda: refresh_public_recent_matches_snapshots(
|
|
run_number=run_number,
|
|
trigger="rcon-capture-materialized-match",
|
|
server_slug=server_slug,
|
|
),
|
|
)
|
|
|
|
|
|
def _run_public_refresh_step(
|
|
step_name: str,
|
|
callback: Any,
|
|
) -> dict[str, Any]:
|
|
started = time.perf_counter()
|
|
_emit_json_log({"event": "public-refresh-step-started", "step": step_name})
|
|
try:
|
|
result = callback()
|
|
except Exception as exc: # noqa: BLE001 - keep neighboring refreshes running
|
|
result = {
|
|
"status": "error",
|
|
"error_type": type(exc).__name__,
|
|
"error": str(exc),
|
|
"traceback": traceback.format_exc(),
|
|
}
|
|
completed = {
|
|
**result,
|
|
"step": step_name,
|
|
"duration_ms": _elapsed_ms(started),
|
|
}
|
|
_emit_json_log({"event": "public-refresh-step-completed", **completed})
|
|
return completed
|
|
|
|
|
|
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 {
|
|
"status": "skipped",
|
|
"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()
|
|
finally:
|
|
_PUBLIC_REFRESH_IN_PROGRESS.discard(refresh_key)
|
|
|
|
|
|
def _record_public_refreshes_from_cycle(
|
|
payload: dict[str, Any],
|
|
*,
|
|
now: datetime,
|
|
) -> None:
|
|
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",
|
|
"skipped",
|
|
}:
|
|
_LAST_PUBLIC_RECENT_MATCHES_REFRESH_AT = now
|
|
|
|
|
|
def _is_public_full_refresh_due(now: datetime) -> bool:
|
|
if not get_public_full_refresh_enabled():
|
|
return False
|
|
local_now = now.astimezone(_get_public_refresh_zone())
|
|
hour, minute = (int(part) for part in get_public_full_refresh_time().split(":"))
|
|
scheduled_today = datetime.combine(
|
|
local_now.date(),
|
|
datetime_time(hour=hour, minute=minute),
|
|
tzinfo=local_now.tzinfo,
|
|
)
|
|
return (
|
|
local_now >= scheduled_today
|
|
and _LAST_PUBLIC_FULL_REFRESH_LOCAL_DATE != local_now.date()
|
|
)
|
|
|
|
|
|
def _is_public_interval_refresh_due(
|
|
*,
|
|
last_run_at: datetime | None,
|
|
interval_seconds: int,
|
|
now: datetime,
|
|
) -> bool:
|
|
return _is_interval_due(
|
|
last_run_at=last_run_at,
|
|
interval_seconds=interval_seconds,
|
|
now=now,
|
|
)
|
|
|
|
|
|
def _is_interval_due(
|
|
*,
|
|
last_run_at: datetime | None,
|
|
interval_seconds: int,
|
|
now: datetime,
|
|
) -> bool:
|
|
if last_run_at is None:
|
|
return True
|
|
elapsed_seconds = (now - _as_utc(last_run_at)).total_seconds()
|
|
return elapsed_seconds >= interval_seconds
|
|
|
|
|
|
def _resolve_runner_tick_seconds(interval_seconds: int) -> int:
|
|
intervals = [
|
|
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:
|
|
return ZoneInfo(timezone_name)
|
|
except ZoneInfoNotFoundError as error:
|
|
raise ValueError(
|
|
f"HLL_PUBLIC_FULL_REFRESH_TIMEZONE is not a valid IANA timezone: {timezone_name}"
|
|
) from error
|
|
|
|
|
|
def _rcon_capture_materialized_finished_match(
|
|
rcon_capture_result: dict[str, Any],
|
|
) -> bool:
|
|
totals = rcon_capture_result.get("totals")
|
|
if isinstance(totals, dict) and int(totals.get("materialized_matches_inserted") or 0) > 0:
|
|
return True
|
|
targets = rcon_capture_result.get("targets")
|
|
if not isinstance(targets, list):
|
|
return False
|
|
for target in targets:
|
|
if not isinstance(target, dict):
|
|
continue
|
|
if int(target.get("materialized_matches_inserted") or 0) > 0:
|
|
return True
|
|
return False
|
|
|
|
|
|
def _elapsed_ms(started: float) -> int:
|
|
return int((time.perf_counter() - started) * 1000)
|
|
|
|
|
|
def _as_utc(value: datetime) -> datetime:
|
|
if value.tzinfo is None:
|
|
return value.replace(tzinfo=timezone.utc)
|
|
return value.astimezone(timezone.utc)
|
|
|
|
|
|
def _to_iso(value: datetime) -> str:
|
|
return _as_utc(value).isoformat().replace("+00:00", "Z")
|
|
|
|
|
|
def _resolve_refresh_cycle_status(**results: dict[str, Any]) -> str:
|
|
statuses = [
|
|
str(result.get("status") or "").strip().lower()
|
|
for result in results.values()
|
|
if isinstance(result, dict) and result.get("status") is not None
|
|
]
|
|
if not statuses:
|
|
return "ok"
|
|
if any(status == "error" for status in statuses):
|
|
ok_like_statuses = {"ok", "skipped"}
|
|
if all(status not in ok_like_statuses for status in statuses):
|
|
return "error"
|
|
return "partial"
|
|
if any(status == "partial" for status in statuses):
|
|
return "partial"
|
|
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)
|
|
|
|
|
|
def _maybe_run_database_maintenance(*, now: datetime | None = None) -> dict[str, Any]:
|
|
"""Optionally run scheduled database maintenance without crashing the runner."""
|
|
global _LAST_DATABASE_MAINTENANCE_RUN_AT
|
|
|
|
anchor = now.astimezone(timezone.utc) if now else datetime.now(timezone.utc)
|
|
if not get_db_maintenance_enabled():
|
|
result = {"status": "skipped", "reason": "disabled", "enabled": False}
|
|
_emit_json_log({"event": "database-maintenance-scheduler-skipped-disabled", **result})
|
|
return result
|
|
|
|
interval_seconds, interval_source = _resolve_db_maintenance_interval_seconds()
|
|
if _LAST_DATABASE_MAINTENANCE_RUN_AT is not None:
|
|
elapsed_seconds = max(
|
|
0,
|
|
int((anchor - _LAST_DATABASE_MAINTENANCE_RUN_AT).total_seconds()),
|
|
)
|
|
if elapsed_seconds < interval_seconds:
|
|
result = {
|
|
"status": "skipped",
|
|
"reason": "not-due",
|
|
"enabled": True,
|
|
"interval_seconds": interval_seconds,
|
|
"interval_source": interval_source,
|
|
"elapsed_seconds": elapsed_seconds,
|
|
"last_run_at": _LAST_DATABASE_MAINTENANCE_RUN_AT.isoformat().replace(
|
|
"+00:00", "Z"
|
|
),
|
|
}
|
|
_emit_json_log({"event": "database-maintenance-scheduler-skipped-not-due", **result})
|
|
return result
|
|
|
|
_emit_json_log(
|
|
{
|
|
"event": "database-maintenance-scheduler-started",
|
|
"enabled": True,
|
|
"interval_seconds": interval_seconds,
|
|
"interval_source": interval_source,
|
|
"scheduled_at": anchor.isoformat().replace("+00:00", "Z"),
|
|
}
|
|
)
|
|
try:
|
|
result = run_database_maintenance_cleanup(apply=True, now=anchor)
|
|
except Exception as exc: # noqa: BLE001 - scheduler must not crash the runner
|
|
result = {
|
|
"status": "error",
|
|
"error_type": type(exc).__name__,
|
|
"error": str(exc),
|
|
"enabled": True,
|
|
"interval_seconds": interval_seconds,
|
|
"interval_source": interval_source,
|
|
}
|
|
_emit_json_log({"event": "database-maintenance-scheduler-failed", **result})
|
|
return result
|
|
|
|
if result.get("status") == "ok":
|
|
_LAST_DATABASE_MAINTENANCE_RUN_AT = anchor
|
|
_emit_json_log(
|
|
{
|
|
"event": "database-maintenance-scheduler-completed",
|
|
"enabled": True,
|
|
"interval_seconds": interval_seconds,
|
|
"interval_source": interval_source,
|
|
"result": result,
|
|
}
|
|
)
|
|
return result
|
|
|
|
failed_result = {
|
|
"enabled": True,
|
|
"interval_seconds": interval_seconds,
|
|
"interval_source": interval_source,
|
|
"result": result,
|
|
}
|
|
_emit_json_log({"event": "database-maintenance-scheduler-failed", **failed_result})
|
|
return result
|
|
|
|
|
|
def _resolve_db_maintenance_interval_seconds() -> tuple[int, str]:
|
|
"""Return a safe maintenance interval even if env configuration is invalid."""
|
|
try:
|
|
return get_db_maintenance_interval_seconds(), "env"
|
|
except ValueError:
|
|
return DEFAULT_DB_MAINTENANCE_INTERVAL_SECONDS, "default-invalid-env-fallback"
|
|
|
|
|
|
def _describe_refresh_scope(server_slug: str | None) -> list[str]:
|
|
if server_slug:
|
|
return [server_slug]
|
|
return list(DEFAULT_HISTORICAL_SERVER_SCOPE)
|
|
|
|
|
|
def _describe_snapshot_scope(server_slug: str | None) -> list[str]:
|
|
if server_slug:
|
|
return [server_slug, "all-servers"]
|
|
return [*DEFAULT_HISTORICAL_SERVER_SCOPE, "all-servers"]
|
|
|
|
|
|
def _run_primary_rcon_capture() -> dict[str, Any]:
|
|
if get_historical_data_source_kind() != "rcon":
|
|
return {
|
|
"status": "skipped",
|
|
"reason": "historical-data-source-configured-without-rcon-primary",
|
|
}
|
|
return run_rcon_historical_capture()
|
|
|
|
|
|
def _resolve_classic_fallback_policy(
|
|
*,
|
|
server_slug: str | None,
|
|
run_number: int,
|
|
rcon_capture_result: dict[str, Any],
|
|
) -> tuple[bool, str]:
|
|
if get_historical_data_source_kind() != "rcon":
|
|
return True, "public-scoreboard-configured-as-primary-historical-source"
|
|
|
|
if not _rcon_capture_has_usable_results(rcon_capture_result):
|
|
return True, "rcon-historical-capture-failed-or-returned-no-usable-targets"
|
|
|
|
if server_slug:
|
|
return True, "manual-server-scope-still-needs-classic-historical-fallback"
|
|
|
|
if run_number % get_historical_full_snapshot_every_runs() == 0:
|
|
return True, "periodic-classic-fallback-for-competitive-historical-coverage"
|
|
|
|
return False, "rcon-primary-cycle-succeeded-without-needing-classic-fallback"
|
|
|
|
|
|
def _rcon_capture_has_usable_results(rcon_capture_result: dict[str, Any]) -> bool:
|
|
if rcon_capture_result.get("status") != "ok":
|
|
return False
|
|
targets = rcon_capture_result.get("targets")
|
|
return isinstance(targets, list) and len(targets) > 0
|
|
|
|
|
|
def _rcon_capture_has_new_useful_data(rcon_capture_result: dict[str, Any]) -> bool:
|
|
if rcon_capture_result.get("status") != "ok":
|
|
return False
|
|
totals = rcon_capture_result.get("totals")
|
|
if isinstance(totals, dict) and int(totals.get("samples_inserted") or 0) > 0:
|
|
return True
|
|
if isinstance(totals, dict) and int(totals.get("admin_log_events_inserted") or 0) > 0:
|
|
return True
|
|
if isinstance(totals, dict) and int(totals.get("materialized_matches_inserted") or 0) > 0:
|
|
return True
|
|
targets = rcon_capture_result.get("targets")
|
|
if not isinstance(targets, list):
|
|
return False
|
|
return any(bool(target.get("sample_inserted")) for target in targets if isinstance(target, dict))
|
|
|
|
|
|
def _build_elo_mmr_rebuild_policy(
|
|
*,
|
|
rcon_capture_result: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
interval_minutes = get_historical_elo_mmr_rebuild_interval_minutes()
|
|
min_new_samples = get_historical_elo_mmr_min_new_samples()
|
|
last_generated_at = get_latest_elo_mmr_generated_at()
|
|
last_generated_at_iso = (
|
|
last_generated_at.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
|
|
if last_generated_at is not None
|
|
else None
|
|
)
|
|
minutes_since_last_rebuild = None
|
|
if last_generated_at is not None:
|
|
minutes_since_last_rebuild = int(
|
|
max(
|
|
0,
|
|
(
|
|
datetime.now(timezone.utc) - last_generated_at.astimezone(timezone.utc)
|
|
).total_seconds() // 60,
|
|
)
|
|
)
|
|
samples_since_last_rebuild = count_rcon_historical_samples_since(last_generated_at_iso)
|
|
due = (
|
|
_rcon_capture_has_new_useful_data(rcon_capture_result)
|
|
and samples_since_last_rebuild >= min_new_samples
|
|
and (
|
|
last_generated_at is None
|
|
or minutes_since_last_rebuild is None
|
|
or minutes_since_last_rebuild >= interval_minutes
|
|
)
|
|
)
|
|
return {
|
|
"policy": "min-new-rcon-samples-and-minutes-since-last-successful-rebuild",
|
|
"due": due,
|
|
"last_generated_at": last_generated_at_iso,
|
|
"samples_since_last_rebuild": samples_since_last_rebuild,
|
|
"minutes_since_last_rebuild": minutes_since_last_rebuild,
|
|
"rebuild_interval_minutes": interval_minutes,
|
|
"min_new_samples": min_new_samples,
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
"""Allow local scheduled historical refresh execution without external infra."""
|
|
parser = argparse.ArgumentParser(
|
|
description="Run periodic historical refreshes and regenerate snapshots for HLL Vietnam.",
|
|
)
|
|
parser.add_argument(
|
|
"--interval",
|
|
type=int,
|
|
default=get_historical_refresh_interval_seconds(),
|
|
help="Seconds to wait between refresh-plus-snapshot runs.",
|
|
)
|
|
parser.add_argument(
|
|
"--hourly",
|
|
action="store_true",
|
|
help="Shortcut for running the refresh loop every 3600 seconds.",
|
|
)
|
|
parser.add_argument(
|
|
"--retries",
|
|
type=int,
|
|
default=get_historical_refresh_max_retries(),
|
|
help="Retry attempts after a failed incremental refresh.",
|
|
)
|
|
parser.add_argument(
|
|
"--retry-delay",
|
|
type=float,
|
|
default=get_historical_refresh_retry_delay_seconds(),
|
|
help="Seconds to wait between failed attempts.",
|
|
)
|
|
parser.add_argument(
|
|
"--server",
|
|
dest="server_slug",
|
|
help="Optional historical server slug.",
|
|
)
|
|
parser.add_argument(
|
|
"--max-pages",
|
|
type=int,
|
|
default=None,
|
|
help="Optional page cap for local validation.",
|
|
)
|
|
parser.add_argument(
|
|
"--page-size",
|
|
type=int,
|
|
default=None,
|
|
help="Optional override for CRCON page size.",
|
|
)
|
|
parser.add_argument(
|
|
"--max-runs",
|
|
type=int,
|
|
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:
|
|
args.interval = HOURLY_INTERVAL_SECONDS
|
|
|
|
if args.interval <= 0:
|
|
raise ValueError("--interval must be a positive integer.")
|
|
if args.retries < 0:
|
|
raise ValueError("--retries must be zero or positive.")
|
|
if args.retry_delay < 0:
|
|
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,
|
|
max_retries=args.retries,
|
|
retry_delay_seconds=args.retry_delay,
|
|
server_slug=args.server_slug,
|
|
max_pages=args.max_pages,
|
|
page_size=args.page_size,
|
|
max_runs=args.max_runs,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|