Fix public snapshot scheduler job failures

ai/tasks/done/TASK-243-fix-public-snapshot-scheduler-job-failures.md

@@ -0,0 +1,77 @@
+---
+id: TASK-243
+title: Fix public snapshot scheduler job failures
+status: done
+type: backend
+team: Backend Senior
+supporting_teams:
+  - Arquitecto Python
+roadmap_item: foundation
+priority: high
+---
+
+# TASK-243 - Fix public snapshot scheduler job failures
+
+## Goal
+
+Fix the real failures found when running manual public snapshot jobs through `python -m app.historical_runner --public-job ...`.
+
+## Context
+
+After TASK-240 the public HTTP audit was healthy, but manual runner jobs still exposed four operational problems:
+
+- CLI JSON serialization could crash on `datetime`
+- ranking snapshot item persistence could fail with duplicate `player_id` rows inside one snapshot
+- repeated PostgreSQL schema initialization could trigger deadlocks during heavy jobs
+- historical monthly generation could abort if `player_event_raw_ledger` was missing
+
+## Steps
+
+1. Reviewed the runner CLI path, ranking snapshot persistence path and historical monthly MVP V2 path.
+2. Added safe JSON serialization for manual job output.
+3. Deduplicated ranking snapshot rows by `player_id` before item insertion.
+4. Pre-initialized PostgreSQL snapshot storage once per heavy job and reused read/write connections without repeated DDL initialization.
+5. Degraded monthly MVP V2 to a controlled empty payload when `player_event_raw_ledger` is missing instead of aborting the whole job.
+
+## Files to Read First
+
+- `AGENTS.md`
+- `ai/architecture-index.md`
+- `ai/repo-context.md`
+- `backend/app/historical_runner.py`
+- `backend/app/rcon_historical_leaderboards.py`
+- `backend/app/historical_snapshots.py`
+
+## Expected Files to Modify
+
+- `backend/app/historical_runner.py`
+- `backend/app/rcon_historical_leaderboards.py`
+- `backend/app/rcon_historical_storage.py`
+- `backend/app/historical_snapshots.py`
+- `backend/app/historical_storage.py`
+- `backend/tests/test_historical_snapshot_refresh.py`
+- `docs/public-snapshot-refresh-schedule.md`
+
+## Constraints
+
+- Keep public GET endpoints snapshot-only.
+- No RCON host or server configuration changes.
+- No Elo/MMR reactivation.
+- No Comunidad Hispana #03 reintroduction.
+
+## 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`
+
+## Outcome
+
+- Manual public jobs now print JSON safely with datetime serialization.
+- Ranking snapshot item insertion no longer tries to persist the same `player_id` twice inside one snapshot.
+- Heavy scheduler jobs initialize PostgreSQL snapshot storage once and then reuse non-DDL connections for the rest of the job.
+- Missing `player_event_raw_ledger` now yields a controlled empty monthly MVP V2 result instead of aborting `historical-monthly`.
+
+## Change Budget
+
+- This task exceeded the preferred tiny scope because the fixes crossed runner orchestration, persistence and regression tests.

backend/app/historical_runner.py

@@ -51,11 +51,13 @@ from .rcon_annual_rankings import (
 )
 from .rcon_historical_leaderboards import refresh_ranking_snapshots
 from .rcon_historical_leaderboards import SNAPSHOT_GENERATOR_SERVER_KEYS
+from .rcon_historical_leaderboards import initialize_ranking_snapshot_storage
 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_storage import initialize_rcon_historical_storage
 from .rcon_historical_worker import run_rcon_historical_capture
 from .writer_lock import backend_writer_lock, build_writer_lock_holder
 
@@ -747,10 +749,12 @@ def refresh_public_weekly_ranking_snapshots(
         }
     )
     try:
+        initialize_ranking_snapshot_storage(ensure_storage=True)
         result = refresh_ranking_snapshots(
             limit=30,
             replace_existing=True,
             timeframes=("weekly",),
+            ensure_storage=False,
             now=anchor,
         )
     except Exception as exc:  # noqa: BLE001 - scheduler must keep the runner alive
@@ -803,10 +807,12 @@ def refresh_public_monthly_ranking_snapshots(
         }
     )
     try:
+        initialize_ranking_snapshot_storage(ensure_storage=True)
         result = refresh_ranking_snapshots(
             limit=30,
             replace_existing=True,
             timeframes=("monthly",),
+            ensure_storage=False,
             now=anchor,
         )
     except Exception as exc:  # noqa: BLE001 - scheduler must keep the runner alive
@@ -978,12 +984,14 @@ def refresh_public_weekly_historical_snapshots(
         }
     )
     try:
+        initialize_rcon_historical_storage(ensure_storage=True)
         result = {
             "status": "ok",
             **generate_and_persist_historical_leaderboard_snapshots(
                 timeframe="weekly",
                 server_keys=PREWARM_SNAPSHOT_SERVER_KEYS,
                 generated_at=anchor,
+                ensure_storage=False,
             ),
         }
     except Exception as exc:  # noqa: BLE001 - scheduler must keep the runner alive
@@ -1037,11 +1045,13 @@ def refresh_public_monthly_historical_snapshots(
         }
     )
     try:
+        initialize_rcon_historical_storage(ensure_storage=True)
         result = {
             "status": "ok",
             **generate_and_persist_historical_monthly_ui_snapshots(
                 server_keys=PREWARM_SNAPSHOT_SERVER_KEYS,
                 generated_at=anchor,
+                ensure_storage=False,
             ),
         }
     except Exception as exc:  # noqa: BLE001 - scheduler must keep the runner alive
@@ -1340,6 +1350,14 @@ def _to_iso(value: datetime) -> str:
     return _as_utc(value).isoformat().replace("+00:00", "Z")
 
 
+def _json_default(value: object) -> str:
+    if isinstance(value, datetime):
+        return _to_iso(value)
+    if isinstance(value, date):
+        return value.isoformat()
+    raise TypeError(f"Object of type {type(value).__name__} is not JSON serializable")
+
+
 def _resolve_refresh_cycle_status(**results: dict[str, Any]) -> str:
     statuses = [
         str(result.get("status") or "").strip().lower()
@@ -1658,7 +1676,7 @@ def main() -> None:
         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))
+        print(json.dumps({"status": "ok", "data": payload}, indent=2, default=_json_default))
         return
 
     run_periodic_historical_refresh(

backend/app/historical_snapshots.py

@@ -398,6 +398,7 @@ def generate_and_persist_historical_leaderboard_snapshots(
     server_keys: tuple[str, ...] = PREWARM_SNAPSHOT_SERVER_KEYS,
     generated_at: datetime | None = None,
     leaderboard_limit: int = DEFAULT_WEEKLY_LEADERBOARD_LIMIT,
+    ensure_storage: bool = True,
     db_path: Path | None = None,
 ) -> dict[str, object]:
     """Build and persist only the weekly or monthly leaderboard snapshot set."""
@@ -424,6 +425,7 @@ def generate_and_persist_historical_leaderboard_snapshots(
                         metric,
                         generated_at_value,
                         limit=leaderboard_limit,
+                        ensure_storage=ensure_storage if not snapshots else False,
                         db_path=db_path,
                     )
                 )
@@ -440,6 +442,7 @@ def generate_and_persist_historical_leaderboard_snapshots(
                     metric,
                     generated_at_value,
                     limit=leaderboard_limit,
+                    ensure_storage=ensure_storage if not snapshots else False,
                     db_path=db_path,
                 )
             )
@@ -467,6 +470,7 @@ 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,
+    ensure_storage: bool = True,
     db_path: Path | None = None,
 ) -> dict[str, object]:
     """Build and persist the monthly historical subset rendered by the public UI."""
@@ -489,6 +493,7 @@ def generate_and_persist_historical_monthly_ui_snapshots(
                     metric,
                     generated_at_value,
                     limit=leaderboard_limit,
+                    ensure_storage=ensure_storage if not snapshots else False,
                     db_path=db_path,
                 )
             )
@@ -507,6 +512,7 @@ def generate_and_persist_historical_monthly_ui_snapshots(
                 server_key,
                 generated_at_value,
                 limit=leaderboard_limit,
+                tolerate_missing_player_event_ledger=True,
                 db_path=db_path,
             )
         )
@@ -613,6 +619,7 @@ def _build_weekly_leaderboard_snapshot(
     generated_at: datetime,
     *,
     limit: int,
+    ensure_storage: bool = True,
     db_path: Path | None = None,
 ) -> dict[str, object]:
     if get_historical_data_source_kind() == SOURCE_KIND_RCON:
@@ -623,6 +630,7 @@ def _build_weekly_leaderboard_snapshot(
             server_key=server_key,
             metric=metric,
             timeframe="weekly",
+            ensure_storage=ensure_storage,
             db_path=db_path,
             now=generated_at,
         )
@@ -658,6 +666,7 @@ def _build_monthly_leaderboard_snapshot(
     generated_at: datetime,
     *,
     limit: int,
+    ensure_storage: bool = True,
     db_path: Path | None = None,
 ) -> dict[str, object]:
     if get_historical_data_source_kind() == SOURCE_KIND_RCON:
@@ -668,6 +677,7 @@ def _build_monthly_leaderboard_snapshot(
             server_key=server_key,
             metric=metric,
             timeframe="monthly",
+            ensure_storage=ensure_storage,
             db_path=db_path,
             now=generated_at,
         )
@@ -829,13 +839,49 @@ def _build_monthly_mvp_v2_snapshot(
     generated_at: datetime,
     *,
     limit: int,
+    tolerate_missing_player_event_ledger: bool = False,
     db_path: Path | None = None,
 ) -> dict[str, object]:
+    try:
         ranking_result = list_monthly_mvp_v2_ranking(
             limit=limit,
             server_id=server_key,
             db_path=db_path,
         )
+    except sqlite3.OperationalError as error:
+        if not tolerate_missing_player_event_ledger or "player_event_raw_ledger" not in str(error):
+            raise
+        ranking_result = {
+            "timeframe": "monthly",
+            "metric": "mvp-v2",
+            "ranking_version": "v2",
+            "window_start": None,
+            "window_end": None,
+            "window_days": None,
+            "window_kind": "missing-player-event-ledger",
+            "window_label": "Sin ledger de eventos",
+            "uses_fallback": True,
+            "selection_reason": "player-event-raw-ledger-missing",
+            "current_month_start": None,
+            "current_month_closed_matches": 0,
+            "previous_month_closed_matches": 0,
+            "sufficient_sample": {
+                "minimum_closed_matches": 0,
+                "current_month_closed_matches": 0,
+                "current_month_has_sufficient_sample": False,
+                "is_early_month": False,
+            },
+            "event_coverage": {
+                "ready": False,
+                "reason": "player-event-raw-ledger-missing",
+                "source_range_start": None,
+                "source_range_end": None,
+            },
+            "eligibility": None,
+            "eligible_players_count": 0,
+            "items": [],
+            "error": str(error),
+        }
     month_key = str(ranking_result.get("window_start") or "")[:7] or None
     event_coverage = ranking_result.get("event_coverage")
     source_range_start = None

backend/app/historical_storage.py

@@ -1761,11 +1761,22 @@ def list_monthly_mvp_v2_ranking(
     window_start = monthly_window["window_start"]
     window_end = monthly_window["window_end"]
     month_key = window_start.strftime("%Y-%m")
+    try:
         event_coverage = _get_monthly_player_event_coverage(
             server_id=server_id,
             month_key=month_key,
             db_path=resolved_path,
         )
+    except sqlite3.OperationalError as error:
+        if "player_event_raw_ledger" not in str(error):
+            raise
+        event_coverage = {
+            "ready": False,
+            "reason": "player-event-raw-ledger-missing",
+            "source_range_start": None,
+            "source_range_end": None,
+            "error": str(error),
+        }
     window_days = _calculate_window_days(window_start=window_start, window_end=window_end)
 
     empty_result = {

backend/app/rcon_historical_leaderboards.py

@@ -101,10 +101,18 @@ ON ranking_snapshot_items(snapshot_id, player_id);
 """
 
 
-def initialize_ranking_snapshot_storage(*, db_path: Path | None = None) -> Path:
+def initialize_ranking_snapshot_storage(
+    *,
+    db_path: Path | None = None,
+    ensure_storage: bool = True,
+) -> Path:
     """Create ranking snapshot tables used by weekly/monthly public ranking reads."""
-    resolved_path = initialize_rcon_materialized_storage(db_path=db_path)
+    resolved_path = initialize_rcon_materialized_storage(
+        db_path=db_path,
+        ensure_storage=ensure_storage,
+    )
     if use_postgres_rcon_storage(explicit_sqlite_path=db_path):
+        if ensure_storage:
             from .postgres_rcon_storage import initialize_postgres_rcon_storage
 
             initialize_postgres_rcon_storage()
@@ -132,6 +140,7 @@ def generate_ranking_snapshot(
     metric: str,
     limit: int,
     replace_existing: bool = True,
+    ensure_storage: bool = True,
     now: datetime | None = None,
     db_path: Path | None = None,
 ) -> dict[str, object]:
@@ -142,8 +151,15 @@ def generate_ranking_snapshot(
     normalized_limit = _normalize_limit(limit)
     anchor = _as_utc(now or datetime.now(timezone.utc))
 
-    resolved_path = initialize_ranking_snapshot_storage(db_path=db_path)
+    resolved_path = initialize_ranking_snapshot_storage(
-    connection_scope = _connect_write_scope(resolved_path, db_path=db_path)
+        db_path=db_path,
+        ensure_storage=ensure_storage,
+    )
+    connection_scope = _connect_write_scope(
+        resolved_path,
+        db_path=db_path,
+        initialize=ensure_storage,
+    )
     with connection_scope as connection:
         window = select_leaderboard_window(
             connection=connection,
@@ -231,6 +247,7 @@ def refresh_ranking_snapshots(
     limit: int = DEFAULT_RANKING_SNAPSHOT_REFRESH_LIMIT,
     replace_existing: bool = True,
     timeframes: tuple[str, ...] | None = None,
+    ensure_storage: bool = True,
     now: datetime | None = None,
     db_path: Path | None = None,
 ) -> dict[str, object]:
@@ -260,6 +277,7 @@ def refresh_ranking_snapshots(
                 metric=metric,
                 limit=normalized_limit,
                 replace_existing=replace_existing,
+                ensure_storage=ensure_storage if not results else False,
                 now=anchor,
                 db_path=db_path,
             )
@@ -532,6 +550,7 @@ def list_rcon_materialized_leaderboard(
     timeframe: str = "weekly",
     metric: str = "kills",
     limit: int = 10,
+    ensure_storage: bool = True,
     db_path: Path | None = None,
     now: datetime | None = None,
 ) -> dict[str, object]:
@@ -547,8 +566,15 @@ def list_rcon_materialized_leaderboard(
     normalized_limit = max(1, int(limit or 10))
     anchor = _as_utc(now or datetime.now(timezone.utc))
 
-    resolved_path = initialize_rcon_materialized_storage(db_path=db_path)
+    resolved_path = initialize_rcon_materialized_storage(
-    connection_scope = _connect_scope(resolved_path, db_path=db_path)
+        db_path=db_path,
+        ensure_storage=ensure_storage,
+    )
+    connection_scope = _connect_scope(
+        resolved_path,
+        db_path=db_path,
+        initialize=ensure_storage,
+    )
     with connection_scope as connection:
         window = select_leaderboard_window(
             connection=connection,
@@ -809,7 +835,8 @@ def _insert_snapshot_items(
     rows: list[dict[str, object]],
     limit: int,
 ) -> None:
-    for index, row in enumerate(rows[:limit], start=1):
+    deduplicated_rows = _dedupe_snapshot_rows(rows)
+    for index, row in enumerate(deduplicated_rows[:limit], start=1):
         item = _build_item(row, index=index)
         connection.execute(
             """
@@ -843,6 +870,19 @@ def _insert_snapshot_items(
         )
 
 
+def _dedupe_snapshot_rows(rows: list[dict[str, object]]) -> list[dict[str, object]]:
+    """Keep the first ranked row for each player id to preserve deterministic snapshot order."""
+    deduplicated_rows: list[dict[str, object]] = []
+    seen_player_ids: set[str] = set()
+    for row in rows:
+        player_id = str(row.get("player_id") or "").strip()
+        if not player_id or player_id in seen_player_ids:
+            continue
+        seen_player_ids.add(player_id)
+        deduplicated_rows.append(row)
+    return deduplicated_rows
+
+
 def _fetch_leaderboard_rows(
     connection: object,
     *,
@@ -1124,19 +1164,24 @@ def _normalize_snapshot_server_key(server_key: str | None) -> str:
     return normalized
 
 
-def _connect_scope(resolved_path: Path, *, db_path: Path | None):
+def _connect_scope(resolved_path: Path, *, db_path: Path | None, initialize: bool = True):
     if use_postgres_rcon_storage(explicit_sqlite_path=db_path):
         from .postgres_rcon_storage import connect_postgres_compat
 
-        return connect_postgres_compat()
+        return connect_postgres_compat(initialize=initialize)
     return closing(connect_sqlite_readonly(resolved_path))
 
 
-def _connect_write_scope(resolved_path: Path, *, db_path: Path | None):
+def _connect_write_scope(
+    resolved_path: Path,
+    *,
+    db_path: Path | None,
+    initialize: bool = True,
+):
     if use_postgres_rcon_storage(explicit_sqlite_path=db_path):
         from .postgres_rcon_storage import connect_postgres_compat
 
-        return connect_postgres_compat()
+        return connect_postgres_compat(initialize=initialize)
     return connect_sqlite_writer(resolved_path)
 
 

backend/app/rcon_historical_storage.py

@@ -20,9 +20,14 @@ COMPETITIVE_MODE_APPROXIMATE = "approximate"
 COMPETITIVE_MODE_EXACT = "exact"
 
 
-def initialize_rcon_historical_storage(*, db_path: Path | None = None) -> Path:
+def initialize_rcon_historical_storage(
+    *,
+    db_path: Path | None = None,
+    ensure_storage: bool = True,
+) -> Path:
     """Create the SQLite structures used by prospective RCON capture."""
     if use_postgres_rcon_storage(explicit_sqlite_path=db_path):
+        if ensure_storage:
             from .postgres_rcon_storage import initialize_postgres_rcon_storage
 
             initialize_postgres_rcon_storage()

backend/tests/test_historical_snapshot_refresh.py

@@ -5,9 +5,11 @@ from __future__ import annotations
 import io
 import json
 import os
+import sqlite3
 import unittest
 from contextlib import nullcontext, redirect_stdout
 from datetime import datetime, timezone
+from pathlib import Path
 from unittest.mock import patch
 
 import app.historical_runner as historical_runner_module
@@ -34,17 +36,20 @@ from app.payloads import (
 )
 from app.historical_runner import (
     _maybe_run_public_read_model_refreshes,
+    _json_default,
     _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
+from app.historical_storage import list_monthly_mvp_v2_ranking
 from app.postgres_display_storage import _json_payload_default
 from app.rcon_historical_read_model import (
     _calculate_coverage_hours,
     _calculate_duration_seconds,
 )
+from app.rcon_historical_leaderboards import _dedupe_snapshot_rows
 
 
 class HistoricalSnapshotRefreshTests(unittest.TestCase):
@@ -249,6 +254,58 @@ class HistoricalSnapshotRefreshTests(unittest.TestCase):
         self.assertEqual(result["job_name"], "public-historical-weekly")
         weekly_historical_refresh.assert_called_once()
 
+    def test_historical_runner_json_default_serializes_datetime_payloads(self) -> None:
+        payload = {
+            "status": "ok",
+            "data": {
+                "generated_at": datetime(2026, 6, 11, 8, 0, tzinfo=timezone.utc),
+            },
+        }
+
+        encoded = json.loads(json.dumps(payload, default=_json_default))
+
+        self.assertEqual(encoded["data"]["generated_at"], "2026-06-11T08:00:00Z")
+
+    def test_ranking_snapshot_dedupes_duplicate_player_rows(self) -> None:
+        rows = [
+            {"player_id": "player-1", "player_name": "Alpha", "metric_value": 20},
+            {"player_id": "player-1", "player_name": "Alpha duplicate", "metric_value": 19},
+            {"player_id": "player-2", "player_name": "Bravo", "metric_value": 18},
+        ]
+
+        deduplicated_rows = _dedupe_snapshot_rows(rows)
+
+        self.assertEqual([row["player_id"] for row in deduplicated_rows], ["player-1", "player-2"])
+        self.assertEqual(deduplicated_rows[0]["player_name"], "Alpha")
+
+    def test_monthly_mvp_v2_missing_player_event_ledger_returns_empty_payload(self) -> None:
+        monthly_window = {
+            "window_start": datetime(2026, 6, 1, tzinfo=timezone.utc),
+            "window_end": datetime(2026, 6, 11, tzinfo=timezone.utc),
+            "window_kind": "current-month",
+            "window_label": "Mes activo",
+            "uses_fallback": False,
+            "selection_reason": "test",
+            "current_month_closed_matches": 4,
+            "previous_month_closed_matches": 8,
+            "minimum_closed_matches": 3,
+            "current_month_has_sufficient_sample": True,
+            "is_early_month": False,
+        }
+        with (
+            patch("app.historical_storage.initialize_historical_storage", return_value=Path("dummy.sqlite3")),
+            patch("app.historical_storage._select_monthly_window", return_value=monthly_window),
+            patch(
+                "app.historical_storage._get_monthly_player_event_coverage",
+                side_effect=sqlite3.OperationalError("no such table: player_event_raw_ledger"),
+            ),
+        ):
+            payload = list_monthly_mvp_v2_ranking(server_id="all-servers", limit=10)
+
+        self.assertFalse(payload["event_coverage"]["ready"])
+        self.assertEqual(payload["event_coverage"]["reason"], "player-event-raw-ledger-missing")
+        self.assertEqual(payload["items"], [])
+
     def test_historical_leaderboard_snapshot_does_not_runtime_enrich_public_request(self) -> None:
         snapshot = {
             "generated_at": "2026-06-10T04:00:00Z",

docs/public-snapshot-refresh-schedule.md

@@ -94,6 +94,17 @@ Policy:
 
 This keeps the implementation simple and avoids CPU spikes without reintroducing request-time fallback work.
 
+## Manual Job Hardening
+
+Manual public jobs executed with `python -m app.historical_runner --public-job ...` now follow the same storage and error-handling rules as scheduled runs:
+
+- CLI JSON output serializes `datetime` and `date` values safely
+- ranking snapshot rows are deduplicated by `player_id` before inserting `ranking_snapshot_items`
+- PostgreSQL snapshot storage initialization runs once at the start of the heavy job, then substeps reuse non-DDL connections
+- missing `player_event_raw_ledger` no longer aborts `historical-monthly`; the monthly MVP V2 slice degrades to an empty payload with `event_coverage.ready = false`
+
+This keeps manual validation commands useful without hiding partial failures inside the job payload.
+
 ## Last Update Exposure
 
 Historical and ranking payloads continue to expose persisted generation metadata from the snapshot records: