diff --git a/ai/tasks/done/TASK-271-split-rcon-live-ingestion-from-historical-materialization.md b/ai/tasks/done/TASK-271-split-rcon-live-ingestion-from-historical-materialization.md index b44e670..64d2f34 100644 --- a/ai/tasks/done/TASK-271-split-rcon-live-ingestion-from-historical-materialization.md +++ b/ai/tasks/done/TASK-271-split-rcon-live-ingestion-from-historical-materialization.md @@ -1,7 +1,7 @@ --- id: TASK-271 title: Split RCON live ingestion from historical materialization -status: in-progress +status: done type: backend team: Backend Senior supporting_teams: ["Arquitecto Python"] @@ -117,10 +117,64 @@ Validation completed: - `cd backend; python -m unittest tests.test_rcon_historical_worker` - `cd backend; python -m unittest tests.test_rcon_current_match_worker` +Post-deploy hotfix note: + +- live AdminLog PostgreSQL startup now initializes only AdminLog tables instead of the full historical/materialization schema path +- the old `idx_rcon_historical_single_running_historical` unique index is removed from schema bootstrap and dropped when present +- PostgreSQL historical single-running protection now uses a runtime advisory lock on the heavy historical worker path +- duplicate or stale historical `running` rows no longer crash the live worker startup path + Follow-up intentionally left out of central scope: - `TEAM KILL` parser correctness remains a separate follow-up task unless handled later with a tiny isolated patch +## Production Hotfix Root Cause + +The TASK-271 split originally added the PostgreSQL unique index +`idx_rcon_historical_single_running_historical` as a schema-level historical +single-running guard. Production already had more than one +`rcon_historical_capture_runs` row with `mode='historical'`, so full PostgreSQL +schema initialization attempted to create a unique index over non-unique +existing data and failed with `psycopg.errors.UniqueViolation`. + +The live AdminLog worker called `initialize_rcon_admin_log_storage()` on startup. +Before the hotfix, that PostgreSQL path delegated to full RCON schema bootstrap, +so live AdminLog ingestion could crash because of historical capture-run state. + +## Production Hotfix Fix + +- Removed creation of `idx_rcon_historical_single_running_historical` from + runtime PostgreSQL schema SQL. +- Added an idempotent `DROP INDEX IF EXISTS + idx_rcon_historical_single_running_historical` cleanup to full PostgreSQL + bootstrap. +- Added `initialize_postgres_admin_log_storage()` so the live AdminLog worker + initializes only `rcon_admin_log_events` and profile snapshot tables. +- Replaced PostgreSQL historical overlap protection with + `pg_try_advisory_lock()` on the heavy historical path only. +- Kept SQLite historical overlap protection query-based with stale `running` + rows marked after the existing runtime timeout. + +## Production Hotfix Validation + +- Live worker startup is independent from duplicate or stale + `mode='historical'` capture rows. +- The live worker persists AdminLog rows idempotently and does not call + `materialize_rcon_admin_log`. +- The historical worker returns `skipped` with reason `already-running` when the + PostgreSQL advisory lock is unavailable. +- Runtime schema code no longer creates + `idx_rcon_historical_single_running_historical`. + +## Remaining Risks + +- Existing stale PostgreSQL `running` rows remain historical audit data; they no + longer provide locking semantics and should be interpreted through worker + logs plus advisory-lock behavior. +- Live freshness still depends on RCON AdminLog availability and credentials per + configured trusted target. +- `TEAM KILL` AdminLog parser correctness remains a separate follow-up. + ## Change Budget - Prefer fewer than 5 modified files when feasible, but accept a slightly larger backend/deploy/docs scope if required to complete the split safely. diff --git a/backend/app/postgres_rcon_storage.py b/backend/app/postgres_rcon_storage.py index d129b94..640626a 100644 --- a/backend/app/postgres_rcon_storage.py +++ b/backend/app/postgres_rcon_storage.py @@ -3,7 +3,7 @@ from __future__ import annotations import json -from collections.abc import Iterable, Mapping +from collections.abc import Iterable, Iterator, Mapping from contextlib import contextmanager from datetime import datetime, timezone from typing import Any @@ -20,6 +20,10 @@ COMPETITIVE_MODE_EXACT = "exact" RUNNING_HISTORICAL_CAPTURE_CONFLICT_MESSAGE = ( "historical materialization capture already running" ) +HISTORICAL_CAPTURE_ADVISORY_LOCK_KEY = 2710001 +DROP_LEGACY_HISTORICAL_GUARD_INDEX_SQL = """ +DROP INDEX IF EXISTS idx_rcon_historical_single_running_historical; +""" RCON_SCHEMA_SQL = """ @@ -54,10 +58,6 @@ CREATE TABLE IF NOT EXISTS rcon_historical_capture_runs ( created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP ); -CREATE UNIQUE INDEX IF NOT EXISTS idx_rcon_historical_single_running_historical -ON rcon_historical_capture_runs(mode) -WHERE status = 'running' AND mode = 'historical'; - CREATE TABLE IF NOT EXISTS rcon_historical_samples ( id BIGSERIAL PRIMARY KEY, target_id BIGINT NOT NULL REFERENCES rcon_historical_targets(id), @@ -388,6 +388,59 @@ CREATE INDEX IF NOT EXISTS idx_rcon_scoreboard_candidates_server_end ON rcon_scoreboard_match_candidates(server_slug, ended_at DESC, started_at DESC); """ +POSTGRES_ADMIN_LOG_SCHEMA_SQL = """ +CREATE TABLE IF NOT EXISTS rcon_admin_log_events ( + id BIGSERIAL PRIMARY KEY, + target_key TEXT NOT NULL, + external_server_id TEXT, + event_timestamp TEXT, + server_time BIGINT, + relative_time TEXT, + event_type TEXT NOT NULL, + raw_message TEXT NOT NULL, + canonical_message TEXT NOT NULL, + parsed_payload_json TEXT NOT NULL, + raw_entry_json TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + UNIQUE NULLS NOT DISTINCT(target_key, server_time, canonical_message) +); + +CREATE TABLE IF NOT EXISTS rcon_player_profile_snapshots ( + id BIGSERIAL PRIMARY KEY, + target_key TEXT NOT NULL, + external_server_id TEXT, + player_id TEXT NOT NULL, + player_name TEXT NOT NULL, + source_server_time BIGINT NOT NULL, + event_timestamp TEXT, + first_seen TEXT, + sessions INTEGER, + matches_played INTEGER, + play_time TEXT, + total_kills INTEGER, + total_deaths INTEGER, + teamkills_done INTEGER, + teamkills_received INTEGER, + kd_ratio DOUBLE PRECISION, + favorite_weapons_json TEXT NOT NULL DEFAULT '{}', + victims_json TEXT NOT NULL DEFAULT '{}', + nemesis_json TEXT NOT NULL DEFAULT '{}', + averages_json TEXT NOT NULL DEFAULT '{}', + sanctions_json TEXT NOT NULL DEFAULT '{}', + raw_content TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + UNIQUE(target_key, player_id, source_server_time) +); + +CREATE INDEX IF NOT EXISTS idx_rcon_admin_log_events_target_time +ON rcon_admin_log_events(target_key, server_time DESC); +CREATE INDEX IF NOT EXISTS idx_rcon_admin_log_events_type +ON rcon_admin_log_events(event_type); +CREATE INDEX IF NOT EXISTS idx_rcon_player_profile_snapshots_player +ON rcon_player_profile_snapshots(target_key, player_id, source_server_time DESC); +""" + POSTGRES_ANNUAL_RANKING_SCHEMA_MIGRATION_SQL = """ ALTER TABLE rcon_annual_ranking_snapshot_items ALTER COLUMN metric_value TYPE DOUBLE PRECISION USING metric_value::double precision; @@ -437,11 +490,19 @@ def initialize_postgres_rcon_storage() -> None: """Create deterministic PostgreSQL schema for migrated RCON domains.""" with connect_postgres() as connection: with connection.cursor() as cursor: + cursor.execute(DROP_LEGACY_HISTORICAL_GUARD_INDEX_SQL) cursor.execute(RCON_SCHEMA_SQL) cursor.execute(POSTGRES_ANNUAL_RANKING_SCHEMA_MIGRATION_SQL) cursor.execute(POSTGRES_RCON_MATCH_PLAYER_STATS_ACTIVE_TIME_MIGRATION_SQL) +def initialize_postgres_admin_log_storage() -> None: + """Create only the PostgreSQL AdminLog structures used by the live worker.""" + with connect_postgres() as connection: + with connection.cursor() as cursor: + cursor.execute(POSTGRES_ADMIN_LOG_SCHEMA_SQL) + + @contextmanager def connect_postgres(): """Yield one PostgreSQL connection with dict-shaped rows.""" @@ -479,30 +540,38 @@ def connect_postgres_compat(*, initialize: bool = True): yield PostgresCompatConnection(connection) +@contextmanager +def postgres_historical_capture_advisory_guard() -> Iterator[bool]: + """Hold one PostgreSQL advisory lock for the heavy historical capture path.""" + with connect_postgres() as connection: + row = connection.execute( + "SELECT pg_try_advisory_lock(%s) AS acquired", + (HISTORICAL_CAPTURE_ADVISORY_LOCK_KEY,), + ).fetchone() + acquired = bool(row and row["acquired"]) + if not acquired: + yield False + return + try: + yield True + finally: + connection.execute( + "SELECT pg_advisory_unlock(%s)", + (HISTORICAL_CAPTURE_ADVISORY_LOCK_KEY,), + ) + + def start_capture_run(*, mode: str, target_scope: str) -> int: initialize_postgres_rcon_storage() with connect_postgres() as connection: - try: - row = connection.execute( - """ - INSERT INTO rcon_historical_capture_runs (mode, status, target_scope, started_at) - VALUES (%s, 'running', %s, %s) - RETURNING id - """, - (mode, target_scope, _utc_now_iso()), - ).fetchone() - except Exception as error: - try: - import psycopg - except ImportError: - psycopg = None # type: ignore[assignment] - if ( - mode == "historical" - and psycopg is not None - and isinstance(error, psycopg.errors.UniqueViolation) - ): - raise RuntimeError(RUNNING_HISTORICAL_CAPTURE_CONFLICT_MESSAGE) from error - raise + row = connection.execute( + """ + INSERT INTO rcon_historical_capture_runs (mode, status, target_scope, started_at) + VALUES (%s, 'running', %s, %s) + RETURNING id + """, + (mode, target_scope, _utc_now_iso()), + ).fetchone() return int(row["id"]) diff --git a/backend/app/rcon_admin_log_storage.py b/backend/app/rcon_admin_log_storage.py index fa59cdb..f2112d7 100644 --- a/backend/app/rcon_admin_log_storage.py +++ b/backend/app/rcon_admin_log_storage.py @@ -31,9 +31,9 @@ CURRENT_MATCH_PLAYER_EVENT_TYPES = ( def initialize_rcon_admin_log_storage(*, db_path: Path | None = None) -> Path: """Create SQLite structures for parsed RCON AdminLog events.""" if use_postgres_rcon_storage(explicit_sqlite_path=db_path): - from .postgres_rcon_storage import initialize_postgres_rcon_storage + from .postgres_rcon_storage import initialize_postgres_admin_log_storage - initialize_postgres_rcon_storage() + initialize_postgres_admin_log_storage() return get_storage_path() resolved_path = initialize_rcon_historical_storage(db_path=db_path) diff --git a/backend/app/rcon_historical_storage.py b/backend/app/rcon_historical_storage.py index aa886a7..29e3dd9 100644 --- a/backend/app/rcon_historical_storage.py +++ b/backend/app/rcon_historical_storage.py @@ -7,7 +7,7 @@ import sqlite3 from collections.abc import Mapping from collections.abc import Iterator from contextlib import closing, contextmanager -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from pathlib import Path from .config import get_storage_path, use_postgres_rcon_storage @@ -23,6 +23,7 @@ COMPETITIVE_MODE_EXACT = "exact" RUNNING_HISTORICAL_CAPTURE_CONFLICT_MESSAGE = ( "historical materialization capture already running" ) +HISTORICAL_RUNNING_STALE_TIMEOUT = timedelta(hours=6) def initialize_rcon_historical_storage( @@ -75,10 +76,6 @@ def initialize_rcon_historical_storage( created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP ); - CREATE UNIQUE INDEX IF NOT EXISTS idx_rcon_historical_single_running_historical - ON rcon_historical_capture_runs(mode) - WHERE status = 'running' AND mode = 'historical'; - CREATE TABLE IF NOT EXISTS rcon_historical_samples ( id INTEGER PRIMARY KEY AUTOINCREMENT, target_id INTEGER NOT NULL, @@ -140,6 +137,7 @@ def initialize_rcon_historical_storage( ON rcon_historical_competitive_windows(target_id, last_seen_at DESC); """ ) + connection.execute("DROP INDEX IF EXISTS idx_rcon_historical_single_running_historical") return resolved_path @@ -158,25 +156,62 @@ def start_rcon_historical_capture_run( resolved_path = initialize_rcon_historical_storage(db_path=db_path) with _connect(resolved_path) as connection: - try: - cursor = connection.execute( + if mode == "historical": + _mark_stale_historical_runs(connection) + active_running_row = connection.execute( """ - INSERT INTO rcon_historical_capture_runs ( - mode, - status, - target_scope, - started_at - ) VALUES (?, 'running', ?, ?) - """, - (mode, target_scope, _utc_now_iso()), - ) - except sqlite3.IntegrityError as error: - if mode == "historical": - raise RuntimeError(RUNNING_HISTORICAL_CAPTURE_CONFLICT_MESSAGE) from error - raise + SELECT id + FROM rcon_historical_capture_runs + WHERE mode = 'historical' AND status = 'running' + ORDER BY started_at DESC, id DESC + LIMIT 1 + """ + ).fetchone() + if active_running_row is not None: + raise RuntimeError(RUNNING_HISTORICAL_CAPTURE_CONFLICT_MESSAGE) + cursor = connection.execute( + """ + INSERT INTO rcon_historical_capture_runs ( + mode, + status, + target_scope, + started_at + ) VALUES (?, 'running', ?, ?) + """, + (mode, target_scope, _utc_now_iso()), + ) return int(cursor.lastrowid) +@contextmanager +def historical_capture_runtime_guard(*, capture_mode: str, db_path: Path | None = None): + """Guard the heavy historical path without relying on a schema-level unique index.""" + if capture_mode != "historical": + yield True + return + + if use_postgres_rcon_storage(explicit_sqlite_path=db_path): + from .postgres_rcon_storage import postgres_historical_capture_advisory_guard + + with postgres_historical_capture_advisory_guard() as acquired: + yield acquired + return + + resolved_path = initialize_rcon_historical_storage(db_path=db_path) + with _connect(resolved_path) as connection: + _mark_stale_historical_runs(connection) + active_running_row = connection.execute( + """ + SELECT id + FROM rcon_historical_capture_runs + WHERE mode = 'historical' AND status = 'running' + ORDER BY started_at DESC, id DESC + LIMIT 1 + """ + ).fetchone() + yield active_running_row is None + + def finalize_rcon_historical_capture_run( run_id: int, *, @@ -1135,3 +1170,25 @@ def _calculate_duration_seconds(first_seen_at: str | None, last_seen_at: str | N def _utc_now_iso() -> str: return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + + +def _mark_stale_historical_runs(connection: sqlite3.Connection) -> None: + stale_before = ( + datetime.now(timezone.utc) - HISTORICAL_RUNNING_STALE_TIMEOUT + ).isoformat().replace("+00:00", "Z") + connection.execute( + """ + UPDATE rcon_historical_capture_runs + SET status = 'stale', + completed_at = ?, + notes = CASE + WHEN notes IS NULL OR notes = '' + THEN 'auto-marked stale after runtime guard timeout' + ELSE notes + END + WHERE mode = 'historical' + AND status = 'running' + AND started_at < ? + """, + (_utc_now_iso(), stale_before), + ) diff --git a/backend/app/rcon_historical_worker.py b/backend/app/rcon_historical_worker.py index d7c6f89..bcba444 100644 --- a/backend/app/rcon_historical_worker.py +++ b/backend/app/rcon_historical_worker.py @@ -30,6 +30,7 @@ from .rcon_client import ( ) from .rcon_historical_storage import ( finalize_rcon_historical_capture_run, + historical_capture_runtime_guard, initialize_rcon_historical_storage, list_rcon_historical_target_statuses, mark_rcon_historical_capture_failure, @@ -99,27 +100,16 @@ def run_rcon_historical_capture_unlocked( capture_mode=capture_mode, skip_materialization=skip_materialization, ) - initialize_rcon_historical_storage() - selected_targets = _select_targets(target_key) - selected_target_keys = {build_rcon_target_key(target) for target in selected_targets} - admin_log_lookback_minutes = get_rcon_admin_log_lookback_minutes() - captured_at = utc_now().isoformat().replace("+00:00", "Z") - target_scope = target_key or "all-configured-rcon-targets" - try: - run_id = start_rcon_historical_capture_run( - mode=resolved_capture_mode, - target_scope=target_scope, - ) - except RuntimeError as error: - if _is_historical_run_conflict(error, capture_mode=resolved_capture_mode): + with historical_capture_runtime_guard(capture_mode=resolved_capture_mode) as guard_acquired: + if not guard_acquired: return { "status": "skipped", "run_status": "skipped", - "captured_at": captured_at, - "target_scope": target_scope, + "captured_at": utc_now().isoformat().replace("+00:00", "Z"), + "target_scope": target_key or "all-configured-rcon-targets", "capture_mode": resolved_capture_mode, "materialization_skipped": resolved_skip_materialization, - "admin_log_lookback_minutes": admin_log_lookback_minutes, + "admin_log_lookback_minutes": get_rcon_admin_log_lookback_minutes(), "admin_log_events_seen": 0, "admin_log_events_inserted": 0, "duplicate_events": 0, @@ -145,131 +135,140 @@ def run_rcon_historical_capture_unlocked( "materialized_matches_updated": 0, }, } - raise - stats = RconHistoricalCaptureStats() - items: list[dict[str, object]] = [] - errors: list[dict[str, object]] = [] - admin_log_errors: list[dict[str, object]] = [] - timeout_seconds = get_rcon_request_timeout_seconds() + initialize_rcon_historical_storage() + selected_targets = _select_targets(target_key) + selected_target_keys = {build_rcon_target_key(target) for target in selected_targets} + admin_log_lookback_minutes = get_rcon_admin_log_lookback_minutes() + captured_at = utc_now().isoformat().replace("+00:00", "Z") + target_scope = target_key or "all-configured-rcon-targets" + run_id = start_rcon_historical_capture_run( + mode=resolved_capture_mode, + target_scope=target_scope, + ) + stats = RconHistoricalCaptureStats() + items: list[dict[str, object]] = [] + errors: list[dict[str, object]] = [] + admin_log_errors: list[dict[str, object]] = [] + timeout_seconds = get_rcon_request_timeout_seconds() - try: - for target in selected_targets: - target_metadata = _serialize_target(target) - stats.targets_seen += 1 - try: - sample = query_live_server_sample( - target, - timeout_seconds=timeout_seconds, + try: + for target in selected_targets: + target_metadata = _serialize_target(target) + stats.targets_seen += 1 + try: + sample = query_live_server_sample( + target, + timeout_seconds=timeout_seconds, + ) + delta = persist_rcon_historical_sample( + run_id=run_id, + captured_at=captured_at, + target=target_metadata, + normalized_payload=sample["normalized"], + raw_payload=sample["raw_session"], + ) + stats.samples_inserted += int(delta["samples_inserted"]) + stats.duplicate_samples += int(delta["duplicate_samples"]) + items.append( + { + "target_key": target_metadata["target_key"], + "external_server_id": target.external_server_id, + "name": target.name, + "host": target.host, + "port": target.port, + "timeout_seconds": timeout_seconds, + "captured_at": captured_at, + "sample_inserted": bool(delta["samples_inserted"]), + "normalized": sample["normalized"], + } + ) + except Exception as exc: # noqa: BLE001 - controlled worker failures + stats.failed_targets += 1 + mark_rcon_historical_capture_failure( + run_id=run_id, + target=target_metadata, + error_message=_format_error_message(exc), + ) + errors.append(_serialize_capture_error(target, exc, timeout_seconds=timeout_seconds)) + + admin_log_result = _ingest_target_admin_log( + target_key=str(target_metadata["target_key"]), + minutes=admin_log_lookback_minutes, ) - delta = persist_rcon_historical_sample( - run_id=run_id, - captured_at=captured_at, + _merge_admin_log_result( + stats=stats, + admin_log_errors=admin_log_errors, target=target_metadata, - normalized_payload=sample["normalized"], - raw_payload=sample["raw_session"], + result=admin_log_result, ) - stats.samples_inserted += int(delta["samples_inserted"]) - stats.duplicate_samples += int(delta["duplicate_samples"]) - items.append( - { - "target_key": target_metadata["target_key"], - "external_server_id": target.external_server_id, - "name": target.name, - "host": target.host, - "port": target.port, - "timeout_seconds": timeout_seconds, - "captured_at": captured_at, - "sample_inserted": bool(delta["samples_inserted"]), - "normalized": sample["normalized"], - } + + materialization_result = _run_materialization_if_enabled( + skip_materialization=resolved_skip_materialization + ) + if not resolved_skip_materialization: + stats.materialized_matches_inserted = int( + materialization_result.get("matches_materialized") or 0 ) - except Exception as exc: # noqa: BLE001 - controlled worker failures - stats.failed_targets += 1 - mark_rcon_historical_capture_failure( - run_id=run_id, - target=target_metadata, - error_message=_format_error_message(exc), + stats.materialized_matches_updated = int( + materialization_result.get("matches_updated") or 0 ) - errors.append(_serialize_capture_error(target, exc, timeout_seconds=timeout_seconds)) - admin_log_result = _ingest_target_admin_log( - target_key=str(target_metadata["target_key"]), - minutes=admin_log_lookback_minutes, + status = "success" if not errors else ("partial" if items else "failed") + finalize_rcon_historical_capture_run( + run_id, + status=status, + targets_seen=stats.targets_seen, + samples_inserted=stats.samples_inserted, + duplicate_samples=stats.duplicate_samples, + failed_targets=stats.failed_targets, + notes=None if not errors else json.dumps(errors, separators=(",", ":")), ) - _merge_admin_log_result( - stats=stats, - admin_log_errors=admin_log_errors, - target=target_metadata, - result=admin_log_result, + except Exception as exc: + finalize_rcon_historical_capture_run( + run_id, + status="failed", + targets_seen=stats.targets_seen, + samples_inserted=stats.samples_inserted, + duplicate_samples=stats.duplicate_samples, + failed_targets=max(1, stats.failed_targets), + notes=str(exc), ) + raise - materialization_result = _run_materialization_if_enabled( - skip_materialization=resolved_skip_materialization - ) - if not resolved_skip_materialization: - stats.materialized_matches_inserted = int( - materialization_result.get("matches_materialized") or 0 - ) - stats.materialized_matches_updated = int( - materialization_result.get("matches_updated") or 0 - ) - - status = "success" if not errors else ("partial" if items else "failed") - finalize_rcon_historical_capture_run( - run_id, - status=status, - targets_seen=stats.targets_seen, - samples_inserted=stats.samples_inserted, - duplicate_samples=stats.duplicate_samples, - failed_targets=stats.failed_targets, - notes=None if not errors else json.dumps(errors, separators=(",", ":")), - ) - except Exception as exc: - finalize_rcon_historical_capture_run( - run_id, - status="failed", - targets_seen=stats.targets_seen, - samples_inserted=stats.samples_inserted, - duplicate_samples=stats.duplicate_samples, - failed_targets=max(1, stats.failed_targets), - notes=str(exc), - ) - raise - - return { - "status": "ok" if items else "error", - "run_status": status, - "captured_at": captured_at, - "target_scope": target_scope, - "capture_mode": resolved_capture_mode, - "materialization_skipped": resolved_skip_materialization, - "admin_log_lookback_minutes": admin_log_lookback_minutes, - "admin_log_events_seen": stats.admin_log_events_seen, - "admin_log_events_inserted": stats.admin_log_events_inserted, - "duplicate_events": stats.admin_log_duplicate_events, - "samples_inserted": stats.samples_inserted, - "targets": items, - "errors": errors, - "admin_log_errors": admin_log_errors, - "materialization_result": materialization_result, - "storage_status": [ - status - for status in list_rcon_historical_target_statuses() - if status.get("target_key") in selected_target_keys - ], - "totals": { - "targets_seen": stats.targets_seen, - "samples_inserted": stats.samples_inserted, - "duplicate_samples": stats.duplicate_samples, - "failed_targets": stats.failed_targets, + return { + "status": "ok" if items else "error", + "run_status": status, + "captured_at": captured_at, + "target_scope": target_scope, + "capture_mode": resolved_capture_mode, + "materialization_skipped": resolved_skip_materialization, + "admin_log_lookback_minutes": admin_log_lookback_minutes, "admin_log_events_seen": stats.admin_log_events_seen, "admin_log_events_inserted": stats.admin_log_events_inserted, - "admin_log_duplicate_events": stats.admin_log_duplicate_events, - "admin_log_failed_targets": stats.admin_log_failed_targets, - "materialized_matches_inserted": stats.materialized_matches_inserted, - "materialized_matches_updated": stats.materialized_matches_updated, - }, - } + "duplicate_events": stats.admin_log_duplicate_events, + "samples_inserted": stats.samples_inserted, + "targets": items, + "errors": errors, + "admin_log_errors": admin_log_errors, + "materialization_result": materialization_result, + "storage_status": [ + status + for status in list_rcon_historical_target_statuses() + if status.get("target_key") in selected_target_keys + ], + "totals": { + "targets_seen": stats.targets_seen, + "samples_inserted": stats.samples_inserted, + "duplicate_samples": stats.duplicate_samples, + "failed_targets": stats.failed_targets, + "admin_log_events_seen": stats.admin_log_events_seen, + "admin_log_events_inserted": stats.admin_log_events_inserted, + "admin_log_duplicate_events": stats.admin_log_duplicate_events, + "admin_log_failed_targets": stats.admin_log_failed_targets, + "materialized_matches_inserted": stats.materialized_matches_inserted, + "materialized_matches_updated": stats.materialized_matches_updated, + }, + } def run_periodic_rcon_historical_capture( @@ -392,12 +391,6 @@ def _run_capture_with_retries( time.sleep(retry_delay_seconds) -def _is_historical_run_conflict(error: RuntimeError, *, capture_mode: str) -> bool: - return capture_mode == CAPTURE_MODE_HISTORICAL and ( - "already running" in str(error).lower() - ) - - def _select_targets(target_key: str | None) -> list[object]: configured_targets = list(load_rcon_targets()) if not configured_targets: diff --git a/backend/tests/test_rcon_current_match_worker.py b/backend/tests/test_rcon_current_match_worker.py index 8f6485d..91da210 100644 --- a/backend/tests/test_rcon_current_match_worker.py +++ b/backend/tests/test_rcon_current_match_worker.py @@ -1,12 +1,16 @@ from __future__ import annotations import os +import sqlite3 import tempfile import unittest -from contextlib import contextmanager +from contextlib import closing, contextmanager +from pathlib import Path from types import SimpleNamespace from unittest.mock import patch +from app import postgres_rcon_storage +from app import rcon_admin_log_storage from app.config import ( get_current_match_adminlog_enabled, get_current_match_adminlog_interval_seconds, @@ -190,6 +194,101 @@ class RconCurrentMatchWorkerTests(unittest.TestCase): self.assertEqual(second["totals"]["events_inserted"], 0) self.assertEqual(second["totals"]["duplicate_events"], 1) + def test_live_worker_starts_when_duplicate_historical_runs_exist(self) -> None: + entry = { + "timestamp": "2026-06-18T18:00:00Z", + "message": ( + "[5:00 min (321)] KILL: Alpha(Allies/steam-alpha) -> " + "Victim(Axis/steam-victim) with Rifle" + ), + } + with tempfile.TemporaryDirectory() as temp_dir: + db_path = os.path.join(temp_dir, "current_match.sqlite3") + rcon_admin_log_storage.initialize_rcon_admin_log_storage(db_path=Path(db_path)) + with closing(sqlite3.connect(db_path)) as connection: + connection.execute( + """ + INSERT INTO rcon_historical_capture_runs ( + mode, status, target_scope, started_at + ) VALUES (?, ?, ?, ?) + """, + ("historical", "running", "all-configured-rcon-targets", "2099-01-01T00:00:00Z"), + ) + connection.execute( + """ + INSERT INTO rcon_historical_capture_runs ( + mode, status, target_scope, started_at + ) VALUES (?, ?, ?, ?) + """, + ("historical", "running", "all-configured-rcon-targets", "2099-01-01T00:05:00Z"), + ) + connection.commit() + + result = run_current_match_adminlog_refresh_once_unlocked( + lookback_seconds=180, + targets=[TARGET_01], + fetch_entries_fn=lambda *_args, **_kwargs: [entry], + db_path=db_path, + ) + + self.assertEqual(result["status"], "ok") + self.assertEqual(result["totals"]["events_inserted"], 1) + self.assertEqual(result["totals"]["failed_targets"], 0) + + def test_live_worker_does_not_trigger_admin_log_materialization(self) -> None: + def fake_fetch(target, *, lookback_seconds, timeout_seconds): + return [ + { + "timestamp": "2026-06-18T18:00:00Z", + "message": "[1:00 min (60)] CONNECTED Alpha (steam-alpha)", + } + ] + + with ( + tempfile.TemporaryDirectory() as temp_dir, + patch("app.rcon_admin_log_materialization.materialize_rcon_admin_log") as materialize, + ): + db_path = os.path.join(temp_dir, "current_match.sqlite3") + result = run_current_match_adminlog_refresh_once_unlocked( + lookback_seconds=180, + targets=[TARGET_01], + fetch_entries_fn=fake_fetch, + db_path=db_path, + ) + + materialize.assert_not_called() + self.assertEqual(result["status"], "ok") + + def test_live_worker_does_not_use_historical_runtime_guard(self) -> None: + def fake_fetch(target, *, lookback_seconds, timeout_seconds): + return [ + { + "timestamp": "2026-06-18T18:00:00Z", + "message": "[1:00 min (60)] CONNECTED Alpha (steam-alpha)", + } + ] + + def fake_persist(*, target, entries, db_path=None, ensure_storage=True): + return { + "events_seen": len(entries), + "events_inserted": len(entries), + "duplicate_events": 0, + } + + with ( + patch("app.rcon_current_match_worker.initialize_rcon_admin_log_storage"), + patch("app.rcon_historical_storage.historical_capture_runtime_guard") as guard, + ): + result = run_current_match_adminlog_refresh_once_unlocked( + lookback_seconds=180, + targets=[TARGET_01], + fetch_entries_fn=fake_fetch, + persist_entries_fn=fake_persist, + ) + + guard.assert_not_called() + self.assertEqual(result["status"], "ok") + def test_loop_honors_max_runs(self) -> None: results = [{"status": "ok"}, {"status": "ok"}] @@ -211,6 +310,51 @@ class RconCurrentMatchWorkerTests(unittest.TestCase): self.assertEqual(run_once.call_count, 2) self.assertEqual(sleep.call_count, 1) + def test_postgres_admin_log_storage_uses_admin_log_bootstrap_only(self) -> None: + with ( + patch.object(rcon_admin_log_storage, "use_postgres_rcon_storage", return_value=True), + patch("app.postgres_rcon_storage.initialize_postgres_admin_log_storage") as admin_log_init, + patch("app.postgres_rcon_storage.initialize_postgres_rcon_storage") as full_init, + ): + rcon_admin_log_storage.initialize_rcon_admin_log_storage() + + admin_log_init.assert_called_once_with() + full_init.assert_not_called() + + def test_postgres_schema_strings_no_longer_create_historical_running_unique_index(self) -> None: + self.assertNotIn( + "CREATE UNIQUE INDEX IF NOT EXISTS idx_rcon_historical_single_running_historical", + postgres_rcon_storage.RCON_SCHEMA_SQL, + ) + self.assertNotIn( + "idx_rcon_historical_single_running_historical", + postgres_rcon_storage.POSTGRES_ADMIN_LOG_SCHEMA_SQL, + ) + self.assertIn( + "DROP INDEX IF EXISTS idx_rcon_historical_single_running_historical", + postgres_rcon_storage.DROP_LEGACY_HISTORICAL_GUARD_INDEX_SQL, + ) + + def test_initialize_postgres_rcon_storage_does_not_execute_removed_unique_index(self) -> None: + executed_sql: list[str] = [] + + with patch.object( + postgres_rcon_storage, + "connect_postgres", + return_value=_FakePostgresConnectionScope(executed_sql), + ): + postgres_rcon_storage.initialize_postgres_rcon_storage() + + executed_text = "\n".join(executed_sql) + self.assertIn( + "DROP INDEX IF EXISTS idx_rcon_historical_single_running_historical", + executed_text, + ) + self.assertNotIn( + "CREATE UNIQUE INDEX IF NOT EXISTS idx_rcon_historical_single_running_historical", + executed_text, + ) + def test_compose_nas_runs_split_live_worker_and_safe_historical_interval(self) -> None: compose_path = os.path.join( os.path.dirname(os.path.dirname(__file__)), @@ -264,3 +408,31 @@ def _temporary_env(**values: str | None): os.environ.pop(name, None) else: os.environ[name] = value + + +class _FakePostgresConnectionScope: + def __init__(self, executed_sql: list[str]) -> None: + self.connection = _FakePostgresConnection(executed_sql) + + def __enter__(self): + return self.connection + + def __exit__(self, exc_type, exc, traceback) -> None: + return None + + +class _FakePostgresConnection: + def __init__(self, executed_sql: list[str]) -> None: + self.executed_sql = executed_sql + + @contextmanager + def cursor(self): + yield _FakePostgresCursor(self.executed_sql) + + +class _FakePostgresCursor: + def __init__(self, executed_sql: list[str]) -> None: + self.executed_sql = executed_sql + + def execute(self, sql: str) -> None: + self.executed_sql.append(sql) diff --git a/backend/tests/test_rcon_historical_worker.py b/backend/tests/test_rcon_historical_worker.py index bc4aa98..646ef8c 100644 --- a/backend/tests/test_rcon_historical_worker.py +++ b/backend/tests/test_rcon_historical_worker.py @@ -1,11 +1,15 @@ from __future__ import annotations import os +import sqlite3 +import tempfile import unittest -from contextlib import contextmanager +from contextlib import closing, contextmanager +from pathlib import Path from types import SimpleNamespace from unittest.mock import patch +from app import rcon_historical_storage from app.rcon_historical_worker import ( CAPTURE_MODE_CURRENT_LIVE, CAPTURE_MODE_HISTORICAL, @@ -157,19 +161,98 @@ class RconHistoricalWorkerTests(unittest.TestCase): def test_historical_capture_skips_when_previous_heavy_run_is_still_running(self) -> None: with ( - patch("app.rcon_historical_worker.initialize_rcon_historical_storage"), - patch("app.rcon_historical_worker._select_targets", return_value=[TARGET]), patch( - "app.rcon_historical_worker.start_rcon_historical_capture_run", - side_effect=RuntimeError("historical materialization capture already running"), + "app.rcon_historical_worker.historical_capture_runtime_guard", + side_effect=_yield_guard(False), ), + patch("app.rcon_historical_worker.start_rcon_historical_capture_run") as start_run, ): payload = run_rcon_historical_capture_unlocked(capture_mode=CAPTURE_MODE_HISTORICAL) + start_run.assert_not_called() self.assertEqual(payload["status"], "skipped") self.assertEqual(payload["run_status"], "skipped") self.assertEqual(payload["materialization_result"]["reason"], "already-running") + def test_historical_storage_can_start_with_duplicate_stale_running_rows(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + db_path = Path(temp_dir) / "historical.sqlite3" + rcon_historical_storage.initialize_rcon_historical_storage(db_path=db_path) + + with closing(sqlite3.connect(db_path)) as connection: + connection.execute( + """ + INSERT INTO rcon_historical_capture_runs ( + mode, status, target_scope, started_at + ) VALUES (?, ?, ?, ?) + """, + ("historical", "running", "all-configured-rcon-targets", "2026-01-01T00:00:00Z"), + ) + connection.execute( + """ + INSERT INTO rcon_historical_capture_runs ( + mode, status, target_scope, started_at + ) VALUES (?, ?, ?, ?) + """, + ("historical", "running", "all-configured-rcon-targets", "2026-01-01T00:05:00Z"), + ) + connection.commit() + + run_id = rcon_historical_storage.start_rcon_historical_capture_run( + mode="historical", + target_scope="all-configured-rcon-targets", + db_path=db_path, + ) + + self.assertGreater(run_id, 0) + with closing(sqlite3.connect(db_path)) as connection: + rows = connection.execute( + """ + SELECT status + FROM rcon_historical_capture_runs + WHERE mode = 'historical' + ORDER BY id ASC + """ + ).fetchall() + self.assertEqual([row[0] for row in rows[:2]], ["stale", "stale"]) + self.assertEqual(rows[-1][0], "running") + + def test_historical_runtime_guard_releases_non_stale_conflict_only(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + db_path = Path(temp_dir) / "historical.sqlite3" + rcon_historical_storage.initialize_rcon_historical_storage(db_path=db_path) + with closing(sqlite3.connect(db_path)) as connection: + connection.execute( + """ + INSERT INTO rcon_historical_capture_runs ( + mode, status, target_scope, started_at + ) VALUES (?, ?, ?, ?) + """, + ("historical", "running", "all-configured-rcon-targets", "2099-01-01T00:00:00Z"), + ) + connection.commit() + + with rcon_historical_storage.historical_capture_runtime_guard( + capture_mode="historical", + db_path=db_path, + ) as acquired: + self.assertIs(acquired, False) + + def test_postgres_historical_runtime_guard_uses_advisory_lock(self) -> None: + with ( + patch.object(rcon_historical_storage, "use_postgres_rcon_storage", return_value=True), + patch( + "app.postgres_rcon_storage.postgres_historical_capture_advisory_guard", + side_effect=_yield_guard(False), + ) as advisory_guard, + ): + with rcon_historical_storage.historical_capture_runtime_guard( + capture_mode="historical", + ) as acquired: + self.assertIs(acquired, False) + + advisory_guard.assert_called_once_with() + @contextmanager def _temporary_env(**values: str): @@ -184,3 +267,11 @@ def _temporary_env(**values: str): os.environ.pop(name, None) else: os.environ[name] = value + + +def _yield_guard(value: bool): + @contextmanager + def _guard(*_args, **_kwargs): + yield value + + return _guard diff --git a/docs/current-match-adminlog-freshness.md b/docs/current-match-adminlog-freshness.md index 172a6e1..e0cb66a 100644 --- a/docs/current-match-adminlog-freshness.md +++ b/docs/current-match-adminlog-freshness.md @@ -80,18 +80,28 @@ Operational behavior: - historical worker remains `python -m app.rcon_historical_worker loop` - it now runs on an explicit safe interval in Portainer instead of inheriting a too-fast runtime cadence - heavy materialization stays in the historical worker -- a single-running-historical guard was added through the capture-run storage layer +- the single-running-historical guard now uses a runtime lock instead of a schema-level unique index +- PostgreSQL uses an advisory lock only around the heavy historical worker path - if a historical run is already active, the next historical attempt returns a skipped result with reason `already-running` +- stale or duplicated old `mode='historical'` run rows no longer crash live worker startup and no longer block PostgreSQL schema bootstrap ### Schema and initialization behavior The hot live path was reduced in two places: - current-match API reads already use `ensure_storage=False`, so public `GET /api/current-match/kills` and `GET /api/current-match/players` do not request schema initialization on each read -- the live AdminLog worker now initializes storage once at worker startup or once-mode execution, then persists with `ensure_storage=False` +- the live AdminLog worker now initializes only the PostgreSQL AdminLog tables at startup or once-mode execution, then persists with `ensure_storage=False` +- the full PostgreSQL bootstrap now removes the obsolete `idx_rcon_historical_single_running_historical` index instead of attempting to recreate it This removes obvious repeated PostgreSQL DDL/init from the live ingestion loop. +TASK-271 production hotfix note: + +- the original split exposed a startup crash when production already contained duplicate historical capture rows +- the crash came from schema-level creation of `idx_rcon_historical_single_running_historical` +- historical single-running protection now uses a runtime PostgreSQL advisory lock on the heavy worker path +- the live AdminLog worker remains independent from historical materialization and historical capture-run state + ## Deployment Decision The chosen deployment split in `deploy/portainer/docker-compose.nas.yml` is: @@ -206,10 +216,23 @@ Expected: - short cycles - per-target AdminLog counts +- no traceback or `UniqueViolation` +- no attempt to create `idx_rcon_historical_single_running_historical` - no `player_stats_seen 838k` - no `materialized_matches_updated` output every few seconds - no repeated deadlocks +### 4a. Verify removed runtime index creation + +```powershell +docker exec hll-vietnam-backend-1 sh -lc "grep -R \"idx_rcon_historical_single_running_historical\" -n /app/app || true" +``` + +Expected: + +- no runtime schema code creates the removed unique index +- documentation or tests may still mention it as a removed regression string + ### 5. Historical worker/materializer logs ```powershell @@ -329,7 +352,6 @@ This was intentionally not expanded into the central scope of the ingestion/mate Primary residual risks: -- a stale historical `running` row can cause later heavy cycles to skip until an operator clears the stale state - if production has multiple unexpected worker replicas, logs should be reviewed after redeploy to confirm the split behaves as intended - live freshness still depends on RCON AdminLog availability per target