Throttle elo rebuilds on rcon cycles

backend/app/config.py

@@ -24,6 +24,8 @@ DEFAULT_HISTORICAL_SNAPSHOT_REFRESH_INTERVAL_SECONDS = 900
 DEFAULT_HISTORICAL_REFRESH_MAX_RETRIES = 2
 DEFAULT_HISTORICAL_REFRESH_RETRY_DELAY_SECONDS = 30
 DEFAULT_HISTORICAL_FULL_SNAPSHOT_EVERY_RUNS = 4
+DEFAULT_HISTORICAL_ELO_MMR_REBUILD_INTERVAL_MINUTES = 1440
+DEFAULT_HISTORICAL_ELO_MMR_MIN_NEW_SAMPLES = 12
 DEFAULT_HISTORICAL_WEEKLY_FALLBACK_MIN_MATCHES = 3
 DEFAULT_HISTORICAL_WEEKLY_FALLBACK_MAX_WEEKDAY = 2
 DEFAULT_PLAYER_EVENT_REFRESH_INTERVAL_SECONDS = 1800
@@ -318,6 +320,30 @@ def get_historical_full_snapshot_every_runs() -> int:
     return run_count
 
 
+def get_historical_elo_mmr_rebuild_interval_minutes() -> int:
+    """Return the minimum minutes between automatic Elo/MMR rebuilds."""
+    configured_value = os.getenv(
+        "HLL_HISTORICAL_ELO_MMR_REBUILD_INTERVAL_MINUTES",
+        str(DEFAULT_HISTORICAL_ELO_MMR_REBUILD_INTERVAL_MINUTES),
+    )
+    interval_minutes = int(configured_value)
+    if interval_minutes <= 0:
+        raise ValueError("HLL_HISTORICAL_ELO_MMR_REBUILD_INTERVAL_MINUTES must be positive.")
+    return interval_minutes
+
+
+def get_historical_elo_mmr_min_new_samples() -> int:
+    """Return the minimum new RCON samples required for an automatic Elo/MMR rebuild."""
+    configured_value = os.getenv(
+        "HLL_HISTORICAL_ELO_MMR_MIN_NEW_SAMPLES",
+        str(DEFAULT_HISTORICAL_ELO_MMR_MIN_NEW_SAMPLES),
+    )
+    min_samples = int(configured_value)
+    if min_samples <= 0:
+        raise ValueError("HLL_HISTORICAL_ELO_MMR_MIN_NEW_SAMPLES must be positive.")
+    return min_samples
+
+
 def get_historical_weekly_fallback_min_matches() -> int:
     """Return the minimum closed matches required to trust the current week."""
     configured_value = os.getenv(

backend/app/elo_mmr_storage.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import json
 import sqlite3
+from datetime import datetime
 from pathlib import Path
 
 from .config import get_storage_path
@@ -546,6 +547,25 @@ def get_latest_elo_mmr_month_key(
     return str(row["latest_month_key"]) if row and row["latest_month_key"] else None
 
 
+def get_latest_elo_mmr_generated_at(*, db_path: Path | None = None) -> datetime | None:
+    """Return the latest persisted Elo/MMR checkpoint generation time, if any."""
+    resolved_path = _resolve_db_path(db_path)
+    try:
+        with _connect_readonly(resolved_path) as connection:
+            row = connection.execute(
+                """
+                SELECT MAX(generated_at) AS latest_generated_at
+                FROM elo_mmr_monthly_checkpoints
+                """
+            ).fetchone()
+    except sqlite3.OperationalError:
+        return None
+    latest_generated_at = str(row["latest_generated_at"] or "").strip() if row else ""
+    if not latest_generated_at:
+        return None
+    return datetime.fromisoformat(latest_generated_at.replace("Z", "+00:00"))
+
+
 def _connect_writer(db_path: Path):
     return connect_sqlite_writer(db_path)
 

backend/app/historical_runner.py

@@ -10,17 +10,21 @@ from typing import Any
 
 from .config import (
     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,
 )
 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 (
     generate_and_persist_historical_snapshots,
     generate_and_persist_priority_historical_snapshots,
 )
+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
 
@@ -136,10 +140,22 @@ def _run_refresh_with_retries(
                             "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-snapshots-only-elo-rebuild-deferred",
+                                "reason": "rcon-primary-useful-cycle-elo-rebuild-throttled",
-                            "generation_policy": "rcon-primary-useful-cycle-snapshots-only",
+                                "generation_policy": "rcon-primary-useful-cycle-elo-rebuild-throttled",
+                                **elo_policy,
                             }
                     else:
                         snapshot_result = {
@@ -151,6 +167,9 @@ def _run_refresh_with_retries(
                             "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
+                            ),
                         }
             return {
                 "status": "ok",
@@ -264,6 +283,49 @@ def _rcon_capture_has_new_useful_data(rcon_capture_result: dict[str, Any]) -> bo
     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(

backend/app/rcon_historical_storage.py

@@ -10,6 +10,7 @@ from pathlib import Path
 
 from .config import get_storage_path
 from .normalizers import normalize_map_name
+from .rcon_client import load_rcon_targets
 from .sqlite_utils import connect_sqlite_readonly, connect_sqlite_writer
 
 
@@ -356,8 +357,13 @@ def list_recent_rcon_historical_samples(
     where_clause = ""
     params: list[object] = [limit]
     if target_key:
-        where_clause = "WHERE targets.target_key = ? OR targets.external_server_id = ?"
+        aliases = _expand_target_key_aliases(target_key)
-        params = [target_key, target_key, limit]
+        alias_placeholders = ", ".join("?" for _ in aliases)
+        where_clause = (
+            "WHERE targets.target_key IN "
+            f"({alias_placeholders}) OR targets.external_server_id IN ({alias_placeholders})"
+        )
+        params = [*aliases, *aliases, limit]
 
     try:
         with _connect_readonly(resolved_path) as connection:
@@ -411,8 +417,13 @@ def list_rcon_historical_competitive_windows(
     where_clause = ""
     params: list[object] = [limit]
     if target_key:
-        where_clause = "WHERE targets.target_key = ? OR targets.external_server_id = ?"
+        aliases = _expand_target_key_aliases(target_key)
-        params = [target_key, target_key, limit]
+        alias_placeholders = ", ".join("?" for _ in aliases)
+        where_clause = (
+            "WHERE targets.target_key IN "
+            f"({alias_placeholders}) OR targets.external_server_id IN ({alias_placeholders})"
+        )
+        params = [*aliases, *aliases, limit]
 
     try:
         with _connect_readonly(resolved_path) as connection:
@@ -479,6 +490,30 @@ def list_rcon_historical_competitive_windows(
     return items
 
 
+def count_rcon_historical_samples_since(
+    since: str | None,
+    *,
+    db_path: Path | None = None,
+) -> int:
+    """Return how many RCON samples were captured after one timestamp."""
+    if not since:
+        return 0
+    resolved_path = _resolve_db_path(db_path)
+    try:
+        with _connect_readonly(resolved_path) as connection:
+            row = connection.execute(
+                """
+                SELECT COUNT(*) AS sample_count
+                FROM rcon_historical_samples
+                WHERE captured_at > ?
+                """,
+                (since,),
+            ).fetchone()
+    except sqlite3.OperationalError:
+        return 0
+    return int(row["sample_count"] or 0) if row else 0
+
+
 def list_rcon_historical_competitive_summary_rows(
     *,
     target_key: str | None = None,
@@ -489,8 +524,13 @@ def list_rcon_historical_competitive_summary_rows(
     where_clause = ""
     params: list[object] = []
     if target_key:
-        where_clause = "WHERE targets.target_key = ? OR targets.external_server_id = ?"
+        aliases = _expand_target_key_aliases(target_key)
-        params = [target_key, target_key]
+        alias_placeholders = ", ".join("?" for _ in aliases)
+        where_clause = (
+            "WHERE targets.target_key IN "
+            f"({alias_placeholders}) OR targets.external_server_id IN ({alias_placeholders})"
+        )
+        params = [*aliases, *aliases]
 
     try:
         with _connect_readonly(resolved_path) as connection:
@@ -555,10 +595,12 @@ def find_rcon_historical_competitive_window(
         return None
     resolved_path = _resolve_db_path(db_path)
     normalized_map_name = normalize_map_name(map_name)
+    aliases = _expand_target_key_aliases(server_key)
+    alias_placeholders = ", ".join("?" for _ in aliases)
     try:
         with _connect_readonly(resolved_path) as connection:
             candidates = connection.execute(
-                """
+                f"""
             SELECT
                 windows.session_key,
                 windows.first_seen_at,
@@ -573,11 +615,14 @@ def find_rcon_historical_competitive_window(
             FROM rcon_historical_competitive_windows AS windows
             INNER JOIN rcon_historical_targets AS targets
                 ON targets.id = windows.target_id
-            WHERE (targets.target_key = ? OR targets.external_server_id = ?)
+            WHERE (
+                targets.target_key IN ({alias_placeholders})
+                OR targets.external_server_id IN ({alias_placeholders})
+            )
             ORDER BY windows.last_seen_at DESC
             LIMIT 12
             """,
-                (server_key, server_key),
+                [*aliases, *aliases],
             ).fetchall()
     except sqlite3.OperationalError:
         return None
@@ -629,6 +674,31 @@ def _resolve_db_path(db_path: Path | None) -> Path:
     return db_path or get_storage_path()
 
 
+def _expand_target_key_aliases(target_key: str) -> list[str]:
+    normalized_target_key = str(target_key or "").strip()
+    if not normalized_target_key:
+        return [normalized_target_key]
+
+    aliases = {normalized_target_key}
+    try:
+        configured_targets = load_rcon_targets()
+    except Exception:
+        configured_targets = ()
+
+    for target in configured_targets:
+        external_server_id = str(target.external_server_id or "").strip()
+        legacy_target_key = f"rcon:{target.host}:{target.port}"
+        if external_server_id and external_server_id == normalized_target_key:
+            aliases.add(legacy_target_key)
+            aliases.add(external_server_id)
+        elif legacy_target_key == normalized_target_key:
+            aliases.add(legacy_target_key)
+            if external_server_id:
+                aliases.add(external_server_id)
+
+    return sorted(alias for alias in aliases if alias)
+
+
 def _upsert_target(connection: sqlite3.Connection, *, target: Mapping[str, object]) -> int:
     target_key = str(target.get("target_key") or "").strip()
     if not target_key: