From 915d074b46e540e82c16bf092ad67aacbdcd135f Mon Sep 17 00:00:00 2001 From: devRaGonSa Date: Thu, 11 Jun 2026 10:22:37 +0200 Subject: [PATCH] Fix public snapshot scheduler job failures --- ...-public-snapshot-scheduler-job-failures.md | 77 +++++++++++++++++++ backend/app/historical_runner.py | 20 ++++- backend/app/historical_snapshots.py | 56 ++++++++++++-- backend/app/historical_storage.py | 21 +++-- backend/app/rcon_historical_leaderboards.py | 71 +++++++++++++---- backend/app/rcon_historical_storage.py | 11 ++- .../tests/test_historical_snapshot_refresh.py | 57 ++++++++++++++ docs/public-snapshot-refresh-schedule.md | 11 +++ 8 files changed, 297 insertions(+), 27 deletions(-) create mode 100644 ai/tasks/done/TASK-243-fix-public-snapshot-scheduler-job-failures.md diff --git a/ai/tasks/done/TASK-243-fix-public-snapshot-scheduler-job-failures.md b/ai/tasks/done/TASK-243-fix-public-snapshot-scheduler-job-failures.md new file mode 100644 index 0000000..c94969d --- /dev/null +++ b/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. diff --git a/backend/app/historical_runner.py b/backend/app/historical_runner.py index 371fbaf..1526901 100644 --- a/backend/app/historical_runner.py +++ b/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( diff --git a/backend/app/historical_snapshots.py b/backend/app/historical_snapshots.py index c818895..300bcf3 100644 --- a/backend/app/historical_snapshots.py +++ b/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]: - ranking_result = list_monthly_mvp_v2_ranking( - limit=limit, - server_id=server_key, - db_path=db_path, - ) + 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 diff --git a/backend/app/historical_storage.py b/backend/app/historical_storage.py index db7dd57..24cd35d 100644 --- a/backend/app/historical_storage.py +++ b/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") - event_coverage = _get_monthly_player_event_coverage( - server_id=server_id, - month_key=month_key, - db_path=resolved_path, - ) + 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 = { diff --git a/backend/app/rcon_historical_leaderboards.py b/backend/app/rcon_historical_leaderboards.py index f6382ce..54d0309 100644 --- a/backend/app/rcon_historical_leaderboards.py +++ b/backend/app/rcon_historical_leaderboards.py @@ -101,13 +101,21 @@ 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): - from .postgres_rcon_storage import initialize_postgres_rcon_storage + if ensure_storage: + from .postgres_rcon_storage import initialize_postgres_rcon_storage - initialize_postgres_rcon_storage() + initialize_postgres_rcon_storage() return resolved_path with closing(connect_sqlite_writer(resolved_path)) as connection: @@ -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) - connection_scope = _connect_write_scope(resolved_path, db_path=db_path) + resolved_path = initialize_ranking_snapshot_storage( + 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) - connection_scope = _connect_scope(resolved_path, db_path=db_path) + resolved_path = initialize_rcon_materialized_storage( + 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) diff --git a/backend/app/rcon_historical_storage.py b/backend/app/rcon_historical_storage.py index 30d75b4..bf97289 100644 --- a/backend/app/rcon_historical_storage.py +++ b/backend/app/rcon_historical_storage.py @@ -20,12 +20,17 @@ 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): - from .postgres_rcon_storage import initialize_postgres_rcon_storage + if ensure_storage: + from .postgres_rcon_storage import initialize_postgres_rcon_storage - initialize_postgres_rcon_storage() + initialize_postgres_rcon_storage() return get_storage_path() resolved_path = db_path or get_storage_path() diff --git a/backend/tests/test_historical_snapshot_refresh.py b/backend/tests/test_historical_snapshot_refresh.py index d2ed1d2..e8a03fc 100644 --- a/backend/tests/test_historical_snapshot_refresh.py +++ b/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", diff --git a/docs/public-snapshot-refresh-schedule.md b/docs/public-snapshot-refresh-schedule.md index 3f79ab4..ca45ef8 100644 --- a/docs/public-snapshot-refresh-schedule.md +++ b/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: