Fix live AdminLog worker startup with safe historical guard

This commit is contained in:
devRaGonSa
2026-06-22 15:16:19 +02:00
parent 43ca4696e0
commit e18177ff63
8 changed files with 651 additions and 193 deletions

View File

@@ -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"])

View File

@@ -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)

View File

@@ -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),
)

View File

@@ -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: