feat: add database maintenance and update home copy
This commit is contained in:
devRaGonSa
@@ -38,6 +38,13 @@ DEFAULT_RCON_HISTORICAL_CAPTURE_RETRY_DELAY_SECONDS = 15
|
||||
DEFAULT_RCON_BACKFILL_CHUNK_HOURS = 6
|
||||
DEFAULT_RCON_BACKFILL_SLEEP_SECONDS = 1.0
|
||||
DEFAULT_RCON_BACKFILL_MAX_DAYS_BACK = 45
|
||||
DEFAULT_RECENT_MATCHES_KEEP = 100
|
||||
DEFAULT_ADMIN_LOG_NONCRITICAL_RETENTION_DAYS = 30
|
||||
DEFAULT_ADMIN_LOG_CRITICAL_RETENTION_DAYS = 90
|
||||
DEFAULT_SERVER_SNAPSHOT_RETENTION_DAYS = 14
|
||||
DEFAULT_DB_MAINTENANCE_BATCH_SIZE = 5000
|
||||
DEFAULT_DB_MAINTENANCE_ENABLED = False
|
||||
DEFAULT_DB_MAINTENANCE_INTERVAL_SECONDS = 43200
|
||||
DEFAULT_SQLITE_WRITER_TIMEOUT_SECONDS = 30.0
|
||||
DEFAULT_SQLITE_BUSY_TIMEOUT_MS = 30000
|
||||
DEFAULT_WRITER_LOCK_TIMEOUT_SECONDS = 120.0
|
||||
@@ -514,6 +521,69 @@ def get_rcon_backfill_max_days_back() -> int:
|
||||
)
|
||||
|
||||
|
||||
def get_recent_matches_keep() -> int:
|
||||
"""Return how many recent closed materialized matches maintenance must protect."""
|
||||
return _read_int_env(
|
||||
"HLL_RECENT_MATCHES_KEEP",
|
||||
str(DEFAULT_RECENT_MATCHES_KEEP),
|
||||
minimum=1,
|
||||
)
|
||||
|
||||
|
||||
def get_admin_log_noncritical_retention_days() -> int:
|
||||
"""Return retention days for non-critical AdminLog events."""
|
||||
return _read_int_env(
|
||||
"HLL_ADMIN_LOG_NONCRITICAL_RETENTION_DAYS",
|
||||
str(DEFAULT_ADMIN_LOG_NONCRITICAL_RETENTION_DAYS),
|
||||
minimum=1,
|
||||
)
|
||||
|
||||
|
||||
def get_admin_log_critical_retention_days() -> int:
|
||||
"""Return retention days for critical AdminLog events."""
|
||||
return _read_int_env(
|
||||
"HLL_ADMIN_LOG_CRITICAL_RETENTION_DAYS",
|
||||
str(DEFAULT_ADMIN_LOG_CRITICAL_RETENTION_DAYS),
|
||||
minimum=1,
|
||||
)
|
||||
|
||||
|
||||
def get_server_snapshot_retention_days() -> int:
|
||||
"""Return retention days for live server snapshots."""
|
||||
return _read_int_env(
|
||||
"HLL_SERVER_SNAPSHOT_RETENTION_DAYS",
|
||||
str(DEFAULT_SERVER_SNAPSHOT_RETENTION_DAYS),
|
||||
minimum=1,
|
||||
)
|
||||
|
||||
|
||||
def get_db_maintenance_batch_size() -> int:
|
||||
"""Return the delete batch size used by database maintenance."""
|
||||
return _read_int_env(
|
||||
"HLL_DB_MAINTENANCE_BATCH_SIZE",
|
||||
str(DEFAULT_DB_MAINTENANCE_BATCH_SIZE),
|
||||
minimum=1,
|
||||
)
|
||||
|
||||
|
||||
def get_db_maintenance_enabled() -> bool:
|
||||
"""Return whether scheduled database maintenance is enabled."""
|
||||
normalized = os.getenv(
|
||||
"HLL_DB_MAINTENANCE_ENABLED",
|
||||
"true" if DEFAULT_DB_MAINTENANCE_ENABLED else "false",
|
||||
).strip().lower()
|
||||
return normalized in {"1", "true", "yes", "on"}
|
||||
|
||||
|
||||
def get_db_maintenance_interval_seconds() -> int:
|
||||
"""Return the scheduled database maintenance interval in seconds."""
|
||||
return _read_int_env(
|
||||
"HLL_DB_MAINTENANCE_INTERVAL_SECONDS",
|
||||
str(DEFAULT_DB_MAINTENANCE_INTERVAL_SECONDS),
|
||||
minimum=1,
|
||||
)
|
||||
|
||||
|
||||
def get_a2s_targets_payload() -> str | None:
|
||||
"""Return the optional JSON payload that overrides local A2S targets."""
|
||||
raw_payload = os.getenv(DEFAULT_A2S_TARGETS_ENV_VAR)
|
||||
|
||||
638
backend/app/database_maintenance.py
Normal file
638
backend/app/database_maintenance.py
Normal file
@@ -0,0 +1,638 @@
|
||||
"""Application-level database maintenance for bounded historical storage."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sqlite3
|
||||
from contextlib import closing
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any, Iterable, Sequence
|
||||
|
||||
from .config import (
|
||||
get_admin_log_critical_retention_days,
|
||||
get_admin_log_noncritical_retention_days,
|
||||
get_database_url,
|
||||
get_db_maintenance_batch_size,
|
||||
get_historical_weekly_fallback_min_matches,
|
||||
get_recent_matches_keep,
|
||||
get_server_snapshot_retention_days,
|
||||
)
|
||||
from .rcon_admin_log_materialization import MATCH_RESULT_SOURCE
|
||||
from .sqlite_utils import connect_sqlite_writer
|
||||
from .writer_lock import backend_writer_lock, build_writer_lock_holder
|
||||
|
||||
CRITICAL_ADMIN_LOG_EVENT_TYPES = frozenset({"kill", "match_start", "match_end"})
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class MaintenanceOptions:
|
||||
apply: bool
|
||||
recent_matches_keep: int
|
||||
admin_log_noncritical_retention_days: int
|
||||
admin_log_critical_retention_days: int
|
||||
server_snapshot_retention_days: int
|
||||
batch_size: int
|
||||
vacuum_analyze: bool
|
||||
now: datetime
|
||||
|
||||
|
||||
def run_database_maintenance_cleanup(
|
||||
*,
|
||||
apply: bool = False,
|
||||
recent_matches_keep: int | None = None,
|
||||
admin_log_noncritical_retention_days: int | None = None,
|
||||
admin_log_critical_retention_days: int | None = None,
|
||||
server_snapshot_retention_days: int | None = None,
|
||||
batch_size: int | None = None,
|
||||
vacuum_analyze: bool = False,
|
||||
now: str | datetime | None = None,
|
||||
db_path: Path | None = None,
|
||||
) -> dict[str, object]:
|
||||
"""Plan or apply safe bounded cleanup for supported storage tables."""
|
||||
options = MaintenanceOptions(
|
||||
apply=apply,
|
||||
recent_matches_keep=recent_matches_keep or get_recent_matches_keep(),
|
||||
admin_log_noncritical_retention_days=(
|
||||
admin_log_noncritical_retention_days or get_admin_log_noncritical_retention_days()
|
||||
),
|
||||
admin_log_critical_retention_days=(
|
||||
admin_log_critical_retention_days or get_admin_log_critical_retention_days()
|
||||
),
|
||||
server_snapshot_retention_days=(
|
||||
server_snapshot_retention_days or get_server_snapshot_retention_days()
|
||||
),
|
||||
batch_size=batch_size or get_db_maintenance_batch_size(),
|
||||
vacuum_analyze=vacuum_analyze,
|
||||
now=_resolve_now(now),
|
||||
)
|
||||
_emit_json_log(
|
||||
{
|
||||
"event": "database-maintenance-started",
|
||||
"mode": "apply" if options.apply else "dry-run",
|
||||
"database_backend": _database_backend_name(db_path=db_path),
|
||||
"database_url_configured": bool(get_database_url()) and db_path is None,
|
||||
"db_path": str(db_path) if db_path is not None else None,
|
||||
"recent_matches_keep": options.recent_matches_keep,
|
||||
"admin_log_noncritical_retention_days": options.admin_log_noncritical_retention_days,
|
||||
"admin_log_critical_retention_days": options.admin_log_critical_retention_days,
|
||||
"server_snapshot_retention_days": options.server_snapshot_retention_days,
|
||||
"batch_size": options.batch_size,
|
||||
"vacuum_analyze": options.vacuum_analyze,
|
||||
"now": _to_iso(options.now),
|
||||
}
|
||||
)
|
||||
|
||||
try:
|
||||
if options.apply:
|
||||
with backend_writer_lock(
|
||||
holder=build_writer_lock_holder("app.database_maintenance cleanup"),
|
||||
storage_path=db_path,
|
||||
):
|
||||
payload = _run_cleanup(options=options, db_path=db_path)
|
||||
else:
|
||||
payload = _run_cleanup(options=options, db_path=db_path)
|
||||
_emit_json_log(
|
||||
{
|
||||
"event": "database-maintenance-completed",
|
||||
**payload,
|
||||
}
|
||||
)
|
||||
return payload
|
||||
except Exception as exc: # noqa: BLE001 - CLI reports structured diagnostics
|
||||
error_payload = {
|
||||
"status": "error",
|
||||
"mode": "apply" if options.apply else "dry-run",
|
||||
"error_type": type(exc).__name__,
|
||||
"error": str(exc),
|
||||
}
|
||||
_emit_json_log({"event": "database-maintenance-error", **error_payload})
|
||||
return error_payload
|
||||
|
||||
|
||||
def _run_cleanup(*, options: MaintenanceOptions, db_path: Path | None) -> dict[str, object]:
|
||||
with _connect_maintenance(db_path=db_path) as connection:
|
||||
existing_tables = _existing_table_names(connection)
|
||||
plan = _build_cleanup_plan(connection, existing_tables=existing_tables, options=options)
|
||||
_emit_json_log(
|
||||
{
|
||||
"event": "database-maintenance-plan",
|
||||
**plan["summary"],
|
||||
}
|
||||
)
|
||||
|
||||
deleted_counts = {
|
||||
"rcon_match_player_stats": 0,
|
||||
"rcon_materialized_matches": 0,
|
||||
"rcon_admin_log_events": 0,
|
||||
"server_snapshots": 0,
|
||||
}
|
||||
if options.apply:
|
||||
deleted_counts["rcon_match_player_stats"] = _delete_match_player_stats(
|
||||
connection,
|
||||
matches=plan["candidate_matches"],
|
||||
batch_size=options.batch_size,
|
||||
)
|
||||
deleted_counts["rcon_materialized_matches"] = _delete_ids_in_batches(
|
||||
connection,
|
||||
table_name="rcon_materialized_matches",
|
||||
ids=[int(row["id"]) for row in plan["candidate_matches"]],
|
||||
batch_size=options.batch_size,
|
||||
)
|
||||
deleted_counts["rcon_admin_log_events"] = _delete_ids_in_batches(
|
||||
connection,
|
||||
table_name="rcon_admin_log_events",
|
||||
ids=plan["candidate_admin_log_ids"],
|
||||
batch_size=options.batch_size,
|
||||
)
|
||||
deleted_counts["server_snapshots"] = _delete_ids_in_batches(
|
||||
connection,
|
||||
table_name="server_snapshots",
|
||||
ids=plan["candidate_server_snapshot_ids"],
|
||||
batch_size=options.batch_size,
|
||||
)
|
||||
if options.vacuum_analyze:
|
||||
_run_vacuum_analyze(connection)
|
||||
|
||||
return {
|
||||
"status": "ok",
|
||||
"mode": "apply" if options.apply else "dry-run",
|
||||
"deleted_counts": deleted_counts,
|
||||
"plan": plan["summary"],
|
||||
}
|
||||
|
||||
|
||||
def _build_cleanup_plan(
|
||||
connection: sqlite3.Connection | Any,
|
||||
*,
|
||||
existing_tables: set[str],
|
||||
options: MaintenanceOptions,
|
||||
) -> dict[str, object]:
|
||||
candidate_server_snapshot_ids: list[int] = []
|
||||
candidate_admin_log_ids: list[int] = []
|
||||
candidate_matches: list[dict[str, object]] = []
|
||||
protected_match_keys: list[str] = []
|
||||
skipped_tables: list[str] = []
|
||||
|
||||
if "server_snapshots" not in existing_tables:
|
||||
skipped_tables.append("server_snapshots")
|
||||
_emit_skip("server_snapshots", "table-missing")
|
||||
else:
|
||||
cutoff = options.now - timedelta(days=options.server_snapshot_retention_days)
|
||||
for row in connection.execute(
|
||||
"SELECT id, captured_at FROM server_snapshots ORDER BY id ASC"
|
||||
).fetchall():
|
||||
captured_at = _parse_datetime(row["captured_at"])
|
||||
if captured_at is None:
|
||||
continue
|
||||
if captured_at < cutoff:
|
||||
candidate_server_snapshot_ids.append(int(row["id"]))
|
||||
|
||||
protected_ranges: dict[str, list[tuple[int, int]]] = {}
|
||||
if "rcon_materialized_matches" not in existing_tables:
|
||||
skipped_tables.append("rcon_materialized_matches")
|
||||
_emit_skip("rcon_materialized_matches", "table-missing")
|
||||
else:
|
||||
(
|
||||
candidate_matches,
|
||||
protected_matches,
|
||||
protected_ranges,
|
||||
protection_summary,
|
||||
) = _plan_materialized_match_cleanup(connection, options=options)
|
||||
protected_match_keys = [str(row["match_key"]) for row in protected_matches]
|
||||
if "rcon_match_player_stats" not in existing_tables:
|
||||
skipped_tables.append("rcon_match_player_stats")
|
||||
_emit_skip("rcon_match_player_stats", "table-missing")
|
||||
|
||||
if "rcon_admin_log_events" not in existing_tables:
|
||||
skipped_tables.append("rcon_admin_log_events")
|
||||
_emit_skip("rcon_admin_log_events", "table-missing")
|
||||
else:
|
||||
candidate_admin_log_ids = _plan_admin_log_cleanup(
|
||||
connection,
|
||||
options=options,
|
||||
protected_ranges=protected_ranges,
|
||||
)
|
||||
|
||||
candidate_player_stat_rows = 0
|
||||
if candidate_matches and "rcon_match_player_stats" in existing_tables:
|
||||
candidate_player_stat_rows = _count_candidate_player_stats(connection, candidate_matches)
|
||||
|
||||
summary = {
|
||||
"status": "ok",
|
||||
"protected_match_count": len(protected_match_keys),
|
||||
"candidate_match_count": len(candidate_matches),
|
||||
"candidate_match_player_stat_count": candidate_player_stat_rows,
|
||||
"candidate_admin_log_event_count": len(candidate_admin_log_ids),
|
||||
"candidate_server_snapshot_count": len(candidate_server_snapshot_ids),
|
||||
"skipped_tables": skipped_tables,
|
||||
"protected_match_keys_preview": protected_match_keys[:10],
|
||||
}
|
||||
if "protection_summary" in locals():
|
||||
summary["protection_summary"] = protection_summary
|
||||
|
||||
return {
|
||||
"candidate_server_snapshot_ids": candidate_server_snapshot_ids,
|
||||
"candidate_admin_log_ids": candidate_admin_log_ids,
|
||||
"candidate_matches": candidate_matches,
|
||||
"summary": summary,
|
||||
}
|
||||
|
||||
|
||||
def _plan_materialized_match_cleanup(
|
||||
connection: sqlite3.Connection | Any,
|
||||
*,
|
||||
options: MaintenanceOptions,
|
||||
) -> tuple[list[dict[str, object]], list[dict[str, object]], dict[str, list[tuple[int, int]]], dict[str, object]]:
|
||||
rows = [
|
||||
dict(row)
|
||||
for row in connection.execute(
|
||||
"""
|
||||
SELECT id, target_key, match_key, started_at, ended_at,
|
||||
started_server_time, ended_server_time, source_basis
|
||||
FROM rcon_materialized_matches
|
||||
WHERE source_basis = ?
|
||||
""",
|
||||
(MATCH_RESULT_SOURCE,),
|
||||
).fetchall()
|
||||
]
|
||||
closed_rows: list[dict[str, object]] = []
|
||||
protected_rows: list[dict[str, object]] = []
|
||||
current_month_start = options.now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
|
||||
previous_month_start = (current_month_start - timedelta(days=1)).replace(day=1)
|
||||
current_week_start = (options.now - timedelta(days=options.now.weekday())).replace(
|
||||
hour=0,
|
||||
minute=0,
|
||||
second=0,
|
||||
microsecond=0,
|
||||
)
|
||||
previous_week_start = current_week_start - timedelta(days=7)
|
||||
|
||||
for row in rows:
|
||||
closed_at = _parse_datetime(row.get("ended_at") or row.get("started_at"))
|
||||
if closed_at is None:
|
||||
row["_protect_reason"] = "unparseable-closed-at"
|
||||
protected_rows.append(row)
|
||||
continue
|
||||
row["_closed_at"] = closed_at
|
||||
closed_rows.append(row)
|
||||
|
||||
closed_rows.sort(
|
||||
key=lambda row: (
|
||||
row["_closed_at"],
|
||||
_coerce_int(row.get("ended_server_time")) or _coerce_int(row.get("started_server_time")) or 0,
|
||||
_coerce_int(row.get("id")) or 0,
|
||||
),
|
||||
reverse=True,
|
||||
)
|
||||
latest_ids = {int(row["id"]) for row in closed_rows[: options.recent_matches_keep]}
|
||||
current_week_count = sum(
|
||||
1 for row in closed_rows if current_week_start <= row["_closed_at"] < options.now
|
||||
)
|
||||
previous_week_count = sum(
|
||||
1 for row in closed_rows if previous_week_start <= row["_closed_at"] < current_week_start
|
||||
)
|
||||
protect_previous_week = (
|
||||
current_week_count < get_historical_weekly_fallback_min_matches()
|
||||
and previous_week_count > 0
|
||||
)
|
||||
protect_previous_month = options.now.day <= 7
|
||||
|
||||
candidate_rows: list[dict[str, object]] = []
|
||||
protected_ranges: dict[str, list[tuple[int, int]]] = {}
|
||||
for row in closed_rows:
|
||||
closed_at = row["_closed_at"]
|
||||
should_protect = False
|
||||
if int(row["id"]) in latest_ids:
|
||||
should_protect = True
|
||||
elif closed_at >= current_month_start:
|
||||
should_protect = True
|
||||
elif protect_previous_month and previous_month_start <= closed_at < current_month_start:
|
||||
should_protect = True
|
||||
elif closed_at >= current_week_start:
|
||||
should_protect = True
|
||||
elif protect_previous_week and previous_week_start <= closed_at < current_week_start:
|
||||
should_protect = True
|
||||
|
||||
if should_protect:
|
||||
protected_rows.append(row)
|
||||
lower = _coerce_int(row.get("started_server_time"))
|
||||
upper = _coerce_int(row.get("ended_server_time"))
|
||||
if lower is not None and upper is not None:
|
||||
protected_ranges.setdefault(str(row["target_key"]), []).append((lower, upper))
|
||||
else:
|
||||
candidate_rows.append(row)
|
||||
|
||||
return (
|
||||
candidate_rows,
|
||||
protected_rows,
|
||||
protected_ranges,
|
||||
{
|
||||
"recent_matches_keep": options.recent_matches_keep,
|
||||
"current_week_closed_matches": current_week_count,
|
||||
"previous_week_closed_matches": previous_week_count,
|
||||
"protect_previous_week": protect_previous_week,
|
||||
"protect_previous_month": protect_previous_month,
|
||||
"current_week_start": _to_iso(current_week_start),
|
||||
"previous_week_start": _to_iso(previous_week_start),
|
||||
"current_month_start": _to_iso(current_month_start),
|
||||
"previous_month_start": _to_iso(previous_month_start),
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def _plan_admin_log_cleanup(
|
||||
connection: sqlite3.Connection | Any,
|
||||
*,
|
||||
options: MaintenanceOptions,
|
||||
protected_ranges: dict[str, list[tuple[int, int]]],
|
||||
) -> list[int]:
|
||||
noncritical_cutoff = options.now - timedelta(days=options.admin_log_noncritical_retention_days)
|
||||
critical_cutoff = options.now - timedelta(days=options.admin_log_critical_retention_days)
|
||||
candidate_ids: list[int] = []
|
||||
rows = connection.execute(
|
||||
"""
|
||||
SELECT id, target_key, event_type, event_timestamp, server_time
|
||||
FROM rcon_admin_log_events
|
||||
ORDER BY id ASC
|
||||
"""
|
||||
).fetchall()
|
||||
for row in rows:
|
||||
event_type = str(row["event_type"] or "").strip()
|
||||
event_time = _parse_datetime(row["event_timestamp"])
|
||||
if event_time is None:
|
||||
continue
|
||||
if event_type in CRITICAL_ADMIN_LOG_EVENT_TYPES:
|
||||
if event_time >= critical_cutoff:
|
||||
continue
|
||||
server_time = _coerce_int(row["server_time"])
|
||||
if server_time is None:
|
||||
continue
|
||||
if _server_time_is_protected(
|
||||
target_key=str(row["target_key"] or ""),
|
||||
server_time=server_time,
|
||||
protected_ranges=protected_ranges,
|
||||
):
|
||||
continue
|
||||
candidate_ids.append(int(row["id"]))
|
||||
continue
|
||||
if event_time < noncritical_cutoff:
|
||||
candidate_ids.append(int(row["id"]))
|
||||
return candidate_ids
|
||||
|
||||
|
||||
def _count_candidate_player_stats(
|
||||
connection: sqlite3.Connection | Any,
|
||||
matches: Sequence[dict[str, object]],
|
||||
) -> int:
|
||||
count = 0
|
||||
for batch in _chunked(list(matches), 250):
|
||||
clause, params = _match_pair_clause(batch)
|
||||
row = connection.execute(
|
||||
f"SELECT COUNT(*) AS count FROM rcon_match_player_stats WHERE {clause}",
|
||||
params,
|
||||
).fetchone()
|
||||
count += int(row["count"] or 0)
|
||||
return count
|
||||
|
||||
|
||||
def _delete_match_player_stats(
|
||||
connection: sqlite3.Connection | Any,
|
||||
*,
|
||||
matches: Sequence[dict[str, object]],
|
||||
batch_size: int,
|
||||
) -> int:
|
||||
deleted = 0
|
||||
for batch in _chunked(list(matches), max(1, min(batch_size, 250))):
|
||||
clause, params = _match_pair_clause(batch)
|
||||
deleted_in_batch = int(
|
||||
connection.execute(
|
||||
f"DELETE FROM rcon_match_player_stats WHERE {clause}",
|
||||
params,
|
||||
).rowcount
|
||||
or 0
|
||||
)
|
||||
_commit(connection)
|
||||
deleted += deleted_in_batch
|
||||
_emit_json_log(
|
||||
{
|
||||
"event": "database-maintenance-delete-batch",
|
||||
"table": "rcon_match_player_stats",
|
||||
"deleted_rows": deleted_in_batch,
|
||||
"batch_size": len(batch),
|
||||
}
|
||||
)
|
||||
return deleted
|
||||
|
||||
|
||||
def _delete_ids_in_batches(
|
||||
connection: sqlite3.Connection | Any,
|
||||
*,
|
||||
table_name: str,
|
||||
ids: Sequence[int],
|
||||
batch_size: int,
|
||||
) -> int:
|
||||
deleted = 0
|
||||
for batch in _chunked(list(ids), batch_size):
|
||||
placeholders = ",".join("?" for _ in batch)
|
||||
deleted_in_batch = int(
|
||||
connection.execute(
|
||||
f"DELETE FROM {table_name} WHERE id IN ({placeholders})",
|
||||
batch,
|
||||
).rowcount
|
||||
or 0
|
||||
)
|
||||
_commit(connection)
|
||||
deleted += deleted_in_batch
|
||||
_emit_json_log(
|
||||
{
|
||||
"event": "database-maintenance-delete-batch",
|
||||
"table": table_name,
|
||||
"deleted_rows": deleted_in_batch,
|
||||
"batch_size": len(batch),
|
||||
}
|
||||
)
|
||||
return deleted
|
||||
|
||||
|
||||
def _run_vacuum_analyze(connection: sqlite3.Connection | Any) -> None:
|
||||
raw_connection = _raw_connection(connection)
|
||||
if isinstance(raw_connection, sqlite3.Connection):
|
||||
raw_connection.execute("VACUUM")
|
||||
raw_connection.execute("ANALYZE")
|
||||
raw_connection.commit()
|
||||
return
|
||||
raw_connection.commit()
|
||||
raw_connection.autocommit = True
|
||||
try:
|
||||
raw_connection.execute("VACUUM ANALYZE")
|
||||
finally:
|
||||
raw_connection.autocommit = False
|
||||
|
||||
|
||||
def _match_pair_clause(matches: Sequence[dict[str, object]]) -> tuple[str, list[object]]:
|
||||
clauses: list[str] = []
|
||||
params: list[object] = []
|
||||
for row in matches:
|
||||
clauses.append("(target_key = ? AND match_key = ?)")
|
||||
params.extend([row["target_key"], row["match_key"]])
|
||||
return " OR ".join(clauses), params
|
||||
|
||||
|
||||
def _existing_table_names(connection: sqlite3.Connection | Any) -> set[str]:
|
||||
raw_connection = _raw_connection(connection)
|
||||
if isinstance(raw_connection, sqlite3.Connection):
|
||||
rows = connection.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type = 'table'"
|
||||
).fetchall()
|
||||
return {str(row["name"]) for row in rows}
|
||||
rows = raw_connection.execute(
|
||||
"""
|
||||
SELECT table_name
|
||||
FROM information_schema.tables
|
||||
WHERE table_schema = 'public'
|
||||
"""
|
||||
).fetchall()
|
||||
return {str(row["table_name"]) for row in rows}
|
||||
|
||||
|
||||
def _emit_skip(table_name: str, reason: str) -> None:
|
||||
_emit_json_log(
|
||||
{
|
||||
"event": "database-maintenance-table-skipped",
|
||||
"table": table_name,
|
||||
"reason": reason,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _server_time_is_protected(
|
||||
*,
|
||||
target_key: str,
|
||||
server_time: int,
|
||||
protected_ranges: dict[str, list[tuple[int, int]]],
|
||||
) -> bool:
|
||||
for lower, upper in protected_ranges.get(target_key, []):
|
||||
if lower <= server_time <= upper:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _connect_maintenance(*, db_path: Path | None):
|
||||
if get_database_url() and db_path is None:
|
||||
from .postgres_rcon_storage import connect_postgres_compat
|
||||
|
||||
return connect_postgres_compat()
|
||||
resolved_path = db_path or Path.cwd() / "backend" / "data" / "hll_vietnam_dev.sqlite3"
|
||||
resolved_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
return closing(connect_sqlite_writer(resolved_path))
|
||||
|
||||
|
||||
def _commit(connection: sqlite3.Connection | Any) -> None:
|
||||
_raw_connection(connection).commit()
|
||||
|
||||
|
||||
def _raw_connection(connection: sqlite3.Connection | Any) -> sqlite3.Connection | Any:
|
||||
return connection.connection if hasattr(connection, "connection") else connection
|
||||
|
||||
|
||||
def _database_backend_name(*, db_path: Path | None) -> str:
|
||||
return "postgres" if get_database_url() and db_path is None else "sqlite"
|
||||
|
||||
|
||||
def _resolve_now(value: str | datetime | None) -> datetime:
|
||||
if value is None:
|
||||
return datetime.now(timezone.utc)
|
||||
if isinstance(value, datetime):
|
||||
return value.astimezone(timezone.utc) if value.tzinfo else value.replace(tzinfo=timezone.utc)
|
||||
parsed = _parse_datetime(value)
|
||||
if parsed is None:
|
||||
raise ValueError("--now must be an ISO 8601 timestamp or date.")
|
||||
return parsed
|
||||
|
||||
|
||||
def _parse_datetime(value: object) -> datetime | None:
|
||||
text = str(value or "").strip()
|
||||
if not text:
|
||||
return None
|
||||
if len(text) == 10:
|
||||
text = f"{text}T00:00:00+00:00"
|
||||
try:
|
||||
parsed = datetime.fromisoformat(text.replace("Z", "+00:00"))
|
||||
except ValueError:
|
||||
return None
|
||||
return parsed.astimezone(timezone.utc) if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def _to_iso(value: datetime) -> str:
|
||||
return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
|
||||
|
||||
|
||||
def _coerce_int(value: object) -> int | None:
|
||||
try:
|
||||
return None if value is None else int(value)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def _chunked(values: Sequence[Any], size: int) -> Iterable[list[Any]]:
|
||||
for index in range(0, len(values), size):
|
||||
yield list(values[index : index + size])
|
||||
|
||||
|
||||
def _emit_json_log(payload: dict[str, object]) -> None:
|
||||
print(json.dumps(payload, ensure_ascii=True, default=str), flush=True)
|
||||
|
||||
|
||||
def build_arg_parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(description="Database maintenance for HLL Vietnam.")
|
||||
subparsers = parser.add_subparsers(dest="command", required=True)
|
||||
cleanup_parser = subparsers.add_parser("cleanup")
|
||||
cleanup_parser.add_argument("--dry-run", action="store_true")
|
||||
cleanup_parser.add_argument("--apply", action="store_true")
|
||||
cleanup_parser.add_argument("--recent-matches-keep", type=int, default=get_recent_matches_keep())
|
||||
cleanup_parser.add_argument(
|
||||
"--admin-log-noncritical-retention-days",
|
||||
type=int,
|
||||
default=get_admin_log_noncritical_retention_days(),
|
||||
)
|
||||
cleanup_parser.add_argument(
|
||||
"--admin-log-critical-retention-days",
|
||||
type=int,
|
||||
default=get_admin_log_critical_retention_days(),
|
||||
)
|
||||
cleanup_parser.add_argument(
|
||||
"--server-snapshot-retention-days",
|
||||
type=int,
|
||||
default=get_server_snapshot_retention_days(),
|
||||
)
|
||||
cleanup_parser.add_argument("--batch-size", type=int, default=get_db_maintenance_batch_size())
|
||||
cleanup_parser.add_argument("--vacuum-analyze", action="store_true")
|
||||
cleanup_parser.add_argument("--now", default=None)
|
||||
return parser
|
||||
|
||||
|
||||
def main(argv: Sequence[str] | None = None) -> int:
|
||||
parser = build_arg_parser()
|
||||
args = parser.parse_args(list(argv) if argv is not None else None)
|
||||
if args.command != "cleanup":
|
||||
raise ValueError("Unsupported command.")
|
||||
if args.apply and args.dry_run:
|
||||
raise ValueError("--apply and --dry-run are mutually exclusive.")
|
||||
payload = run_database_maintenance_cleanup(
|
||||
apply=bool(args.apply),
|
||||
recent_matches_keep=args.recent_matches_keep,
|
||||
admin_log_noncritical_retention_days=args.admin_log_noncritical_retention_days,
|
||||
admin_log_critical_retention_days=args.admin_log_critical_retention_days,
|
||||
server_snapshot_retention_days=args.server_snapshot_retention_days,
|
||||
batch_size=args.batch_size,
|
||||
vacuum_analyze=bool(args.vacuum_analyze),
|
||||
now=args.now,
|
||||
)
|
||||
return 0 if payload.get("status") == "ok" else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -10,6 +10,9 @@ from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
from .config import (
|
||||
DEFAULT_DB_MAINTENANCE_INTERVAL_SECONDS,
|
||||
get_db_maintenance_enabled,
|
||||
get_db_maintenance_interval_seconds,
|
||||
get_historical_full_snapshot_every_runs,
|
||||
get_historical_elo_mmr_min_new_samples,
|
||||
get_historical_elo_mmr_rebuild_interval_minutes,
|
||||
@@ -18,6 +21,7 @@ from .config import (
|
||||
get_historical_refresh_retry_delay_seconds,
|
||||
get_historical_data_source_kind,
|
||||
)
|
||||
from .database_maintenance import run_database_maintenance_cleanup
|
||||
from .elo_mmr_engine import rebuild_elo_mmr_models
|
||||
from .elo_mmr_storage import get_latest_elo_mmr_generated_at
|
||||
from .historical_ingestion import run_incremental_refresh
|
||||
@@ -34,6 +38,7 @@ DEFAULT_HISTORICAL_SERVER_SCOPE = (
|
||||
"comunidad-hispana-01",
|
||||
"comunidad-hispana-02",
|
||||
)
|
||||
_LAST_DATABASE_MAINTENANCE_RUN_AT: datetime | None = None
|
||||
|
||||
|
||||
def run_periodic_historical_refresh(
|
||||
@@ -171,6 +176,7 @@ def _run_refresh_with_retries(
|
||||
rcon_capture_result=rcon_capture_result
|
||||
),
|
||||
}
|
||||
maintenance_result = _maybe_run_database_maintenance()
|
||||
return {
|
||||
"status": "ok",
|
||||
"attempts_used": attempt,
|
||||
@@ -181,6 +187,7 @@ def _run_refresh_with_retries(
|
||||
"refresh_result": refresh_result,
|
||||
"snapshot_result": snapshot_result,
|
||||
"elo_mmr_result": elo_mmr_result,
|
||||
"database_maintenance_result": maintenance_result,
|
||||
}
|
||||
except Exception as exc:
|
||||
failure_payload = {
|
||||
@@ -249,6 +256,91 @@ def _emit_json_log(payload: dict[str, Any]) -> None:
|
||||
print(json.dumps(payload, ensure_ascii=True, default=str), flush=True)
|
||||
|
||||
|
||||
def _maybe_run_database_maintenance(*, now: datetime | None = None) -> dict[str, Any]:
|
||||
"""Optionally run scheduled database maintenance without crashing the runner."""
|
||||
global _LAST_DATABASE_MAINTENANCE_RUN_AT
|
||||
|
||||
anchor = now.astimezone(timezone.utc) if now else datetime.now(timezone.utc)
|
||||
if not get_db_maintenance_enabled():
|
||||
result = {"status": "skipped", "reason": "disabled", "enabled": False}
|
||||
_emit_json_log({"event": "database-maintenance-scheduler-skipped-disabled", **result})
|
||||
return result
|
||||
|
||||
interval_seconds, interval_source = _resolve_db_maintenance_interval_seconds()
|
||||
if _LAST_DATABASE_MAINTENANCE_RUN_AT is not None:
|
||||
elapsed_seconds = max(
|
||||
0,
|
||||
int((anchor - _LAST_DATABASE_MAINTENANCE_RUN_AT).total_seconds()),
|
||||
)
|
||||
if elapsed_seconds < interval_seconds:
|
||||
result = {
|
||||
"status": "skipped",
|
||||
"reason": "not-due",
|
||||
"enabled": True,
|
||||
"interval_seconds": interval_seconds,
|
||||
"interval_source": interval_source,
|
||||
"elapsed_seconds": elapsed_seconds,
|
||||
"last_run_at": _LAST_DATABASE_MAINTENANCE_RUN_AT.isoformat().replace(
|
||||
"+00:00", "Z"
|
||||
),
|
||||
}
|
||||
_emit_json_log({"event": "database-maintenance-scheduler-skipped-not-due", **result})
|
||||
return result
|
||||
|
||||
_emit_json_log(
|
||||
{
|
||||
"event": "database-maintenance-scheduler-started",
|
||||
"enabled": True,
|
||||
"interval_seconds": interval_seconds,
|
||||
"interval_source": interval_source,
|
||||
"scheduled_at": anchor.isoformat().replace("+00:00", "Z"),
|
||||
}
|
||||
)
|
||||
try:
|
||||
result = run_database_maintenance_cleanup(apply=True, now=anchor)
|
||||
except Exception as exc: # noqa: BLE001 - scheduler must not crash the runner
|
||||
result = {
|
||||
"status": "error",
|
||||
"error_type": type(exc).__name__,
|
||||
"error": str(exc),
|
||||
"enabled": True,
|
||||
"interval_seconds": interval_seconds,
|
||||
"interval_source": interval_source,
|
||||
}
|
||||
_emit_json_log({"event": "database-maintenance-scheduler-failed", **result})
|
||||
return result
|
||||
|
||||
if result.get("status") == "ok":
|
||||
_LAST_DATABASE_MAINTENANCE_RUN_AT = anchor
|
||||
_emit_json_log(
|
||||
{
|
||||
"event": "database-maintenance-scheduler-completed",
|
||||
"enabled": True,
|
||||
"interval_seconds": interval_seconds,
|
||||
"interval_source": interval_source,
|
||||
"result": result,
|
||||
}
|
||||
)
|
||||
return result
|
||||
|
||||
failed_result = {
|
||||
"enabled": True,
|
||||
"interval_seconds": interval_seconds,
|
||||
"interval_source": interval_source,
|
||||
"result": result,
|
||||
}
|
||||
_emit_json_log({"event": "database-maintenance-scheduler-failed", **failed_result})
|
||||
return result
|
||||
|
||||
|
||||
def _resolve_db_maintenance_interval_seconds() -> tuple[int, str]:
|
||||
"""Return a safe maintenance interval even if env configuration is invalid."""
|
||||
try:
|
||||
return get_db_maintenance_interval_seconds(), "env"
|
||||
except ValueError:
|
||||
return DEFAULT_DB_MAINTENANCE_INTERVAL_SECONDS, "default-invalid-env-fallback"
|
||||
|
||||
|
||||
def _describe_refresh_scope(server_slug: str | None) -> list[str]:
|
||||
if server_slug:
|
||||
return [server_slug]
|
||||
|
||||
448
backend/tests/test_database_maintenance.py
Normal file
448
backend/tests/test_database_maintenance.py
Normal file
@@ -0,0 +1,448 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import json
|
||||
import sqlite3
|
||||
import tempfile
|
||||
import unittest
|
||||
from contextlib import closing, redirect_stdout
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
|
||||
from app.database_maintenance import run_database_maintenance_cleanup
|
||||
from app.rcon_admin_log_materialization import MATCH_RESULT_SOURCE, initialize_rcon_materialized_storage
|
||||
from app.rcon_admin_log_storage import initialize_rcon_admin_log_storage
|
||||
from app.storage import initialize_storage
|
||||
|
||||
|
||||
class DatabaseMaintenanceTests(unittest.TestCase):
|
||||
def test_dry_run_does_not_delete(self) -> None:
|
||||
with _temp_db() as db_path:
|
||||
_insert_server_snapshot(db_path, snapshot_id=1, captured_at="2026-05-01T00:00:00Z")
|
||||
|
||||
payload = run_database_maintenance_cleanup(
|
||||
db_path=db_path,
|
||||
now="2026-06-20T12:00:00Z",
|
||||
)
|
||||
|
||||
self.assertEqual(payload["status"], "ok")
|
||||
self.assertEqual(payload["mode"], "dry-run")
|
||||
with closing(sqlite3.connect(db_path)) as connection:
|
||||
self.assertEqual(
|
||||
connection.execute("SELECT COUNT(*) FROM server_snapshots").fetchone()[0],
|
||||
1,
|
||||
)
|
||||
|
||||
def test_apply_deletes_old_server_snapshots(self) -> None:
|
||||
with _temp_db() as db_path:
|
||||
_insert_server_snapshot(db_path, snapshot_id=1, captured_at="2026-05-01T00:00:00Z")
|
||||
_insert_server_snapshot(db_path, snapshot_id=2, captured_at="2026-06-18T00:00:00Z")
|
||||
|
||||
run_database_maintenance_cleanup(
|
||||
apply=True,
|
||||
db_path=db_path,
|
||||
now="2026-06-20T12:00:00Z",
|
||||
recent_matches_keep=1,
|
||||
)
|
||||
|
||||
with closing(sqlite3.connect(db_path)) as connection:
|
||||
ids = [row[0] for row in connection.execute("SELECT id FROM server_snapshots ORDER BY id")]
|
||||
self.assertEqual(ids, [2])
|
||||
|
||||
def test_apply_deletes_old_noncritical_admin_log_events(self) -> None:
|
||||
with _temp_db() as db_path:
|
||||
_insert_admin_log_event(
|
||||
db_path,
|
||||
event_id=1,
|
||||
event_type="chat",
|
||||
event_timestamp="2026-04-01T00:00:00Z",
|
||||
server_time=100,
|
||||
)
|
||||
_insert_admin_log_event(
|
||||
db_path,
|
||||
event_id=2,
|
||||
event_type="chat",
|
||||
event_timestamp="2026-06-15T00:00:00Z",
|
||||
server_time=200,
|
||||
)
|
||||
|
||||
run_database_maintenance_cleanup(
|
||||
apply=True,
|
||||
db_path=db_path,
|
||||
now="2026-06-20T12:00:00Z",
|
||||
)
|
||||
|
||||
with closing(sqlite3.connect(db_path)) as connection:
|
||||
remaining = [
|
||||
tuple(row)
|
||||
for row in connection.execute(
|
||||
"SELECT id, event_type FROM rcon_admin_log_events ORDER BY id"
|
||||
)
|
||||
]
|
||||
self.assertEqual(remaining, [(2, "chat")])
|
||||
|
||||
def test_apply_preserves_critical_events_within_retention(self) -> None:
|
||||
with _temp_db() as db_path:
|
||||
_insert_admin_log_event(
|
||||
db_path,
|
||||
event_id=1,
|
||||
event_type="kill",
|
||||
event_timestamp="2026-06-10T00:00:00Z",
|
||||
server_time=100,
|
||||
)
|
||||
|
||||
run_database_maintenance_cleanup(
|
||||
apply=True,
|
||||
db_path=db_path,
|
||||
now="2026-06-20T12:00:00Z",
|
||||
)
|
||||
|
||||
with closing(sqlite3.connect(db_path)) as connection:
|
||||
count = connection.execute(
|
||||
"SELECT COUNT(*) FROM rcon_admin_log_events WHERE event_type = 'kill'"
|
||||
).fetchone()[0]
|
||||
self.assertEqual(count, 1)
|
||||
|
||||
def test_apply_preserves_latest_100_materialized_matches(self) -> None:
|
||||
with _temp_db() as db_path:
|
||||
for index in range(101):
|
||||
ended_at = (
|
||||
datetime(2026, 1, 1, 12, tzinfo=timezone.utc) + timedelta(days=index)
|
||||
).isoformat().replace("+00:00", "Z")
|
||||
_insert_materialized_match(
|
||||
db_path,
|
||||
match_id=index + 1,
|
||||
match_key=f"match-{index + 1}",
|
||||
ended_at=ended_at,
|
||||
server_time_start=(index + 1) * 10,
|
||||
server_time_end=(index + 1) * 10 + 5,
|
||||
)
|
||||
|
||||
run_database_maintenance_cleanup(
|
||||
apply=True,
|
||||
db_path=db_path,
|
||||
now="2026-06-20T12:00:00Z",
|
||||
)
|
||||
|
||||
with closing(sqlite3.connect(db_path)) as connection:
|
||||
remaining = connection.execute(
|
||||
"SELECT COUNT(*) FROM rcon_materialized_matches"
|
||||
).fetchone()[0]
|
||||
oldest = connection.execute(
|
||||
"SELECT COUNT(*) FROM rcon_materialized_matches WHERE match_key = 'match-1'"
|
||||
).fetchone()[0]
|
||||
self.assertEqual(remaining, 100)
|
||||
self.assertEqual(oldest, 0)
|
||||
|
||||
def test_apply_preserves_current_month_matches(self) -> None:
|
||||
with _temp_db() as db_path:
|
||||
_insert_materialized_match(
|
||||
db_path,
|
||||
match_id=1,
|
||||
match_key="old",
|
||||
ended_at="2026-01-10T12:00:00Z",
|
||||
server_time_start=10,
|
||||
server_time_end=20,
|
||||
)
|
||||
_insert_materialized_match(
|
||||
db_path,
|
||||
match_id=2,
|
||||
match_key="current-month",
|
||||
ended_at="2026-06-03T12:00:00Z",
|
||||
server_time_start=30,
|
||||
server_time_end=40,
|
||||
)
|
||||
|
||||
run_database_maintenance_cleanup(
|
||||
apply=True,
|
||||
db_path=db_path,
|
||||
now="2026-06-20T12:00:00Z",
|
||||
recent_matches_keep=1,
|
||||
)
|
||||
|
||||
with closing(sqlite3.connect(db_path)) as connection:
|
||||
keys = [row[0] for row in connection.execute("SELECT match_key FROM rcon_materialized_matches")]
|
||||
self.assertEqual(keys, ["current-month"])
|
||||
|
||||
def test_apply_preserves_previous_month_when_now_day_is_early(self) -> None:
|
||||
with _temp_db() as db_path:
|
||||
_insert_materialized_match(
|
||||
db_path,
|
||||
match_id=1,
|
||||
match_key="previous-month",
|
||||
ended_at="2026-05-15T12:00:00Z",
|
||||
server_time_start=10,
|
||||
server_time_end=20,
|
||||
)
|
||||
_insert_materialized_match(
|
||||
db_path,
|
||||
match_id=2,
|
||||
match_key="older",
|
||||
ended_at="2026-04-15T12:00:00Z",
|
||||
server_time_start=30,
|
||||
server_time_end=40,
|
||||
)
|
||||
|
||||
run_database_maintenance_cleanup(
|
||||
apply=True,
|
||||
db_path=db_path,
|
||||
now="2026-06-05T12:00:00Z",
|
||||
recent_matches_keep=1,
|
||||
)
|
||||
|
||||
with closing(sqlite3.connect(db_path)) as connection:
|
||||
keys = [row[0] for row in connection.execute("SELECT match_key FROM rcon_materialized_matches")]
|
||||
self.assertEqual(keys, ["previous-month"])
|
||||
|
||||
def test_apply_preserves_current_week(self) -> None:
|
||||
with _temp_db() as db_path:
|
||||
_insert_materialized_match(
|
||||
db_path,
|
||||
match_id=1,
|
||||
match_key="current-week",
|
||||
ended_at="2026-06-10T12:00:00Z",
|
||||
server_time_start=10,
|
||||
server_time_end=20,
|
||||
)
|
||||
_insert_materialized_match(
|
||||
db_path,
|
||||
match_id=2,
|
||||
match_key="older",
|
||||
ended_at="2026-05-01T12:00:00Z",
|
||||
server_time_start=30,
|
||||
server_time_end=40,
|
||||
)
|
||||
|
||||
run_database_maintenance_cleanup(
|
||||
apply=True,
|
||||
db_path=db_path,
|
||||
now="2026-06-10T13:00:00Z",
|
||||
recent_matches_keep=1,
|
||||
)
|
||||
|
||||
with closing(sqlite3.connect(db_path)) as connection:
|
||||
keys = [row[0] for row in connection.execute("SELECT match_key FROM rcon_materialized_matches")]
|
||||
self.assertEqual(keys, ["current-week"])
|
||||
|
||||
def test_apply_preserves_previous_week_when_fallback_may_need_it(self) -> None:
|
||||
with _temp_db() as db_path:
|
||||
_insert_materialized_match(
|
||||
db_path,
|
||||
match_id=1,
|
||||
match_key="previous-week",
|
||||
ended_at="2026-06-03T12:00:00Z",
|
||||
server_time_start=10,
|
||||
server_time_end=20,
|
||||
)
|
||||
_insert_materialized_match(
|
||||
db_path,
|
||||
match_id=2,
|
||||
match_key="current-week-sample",
|
||||
ended_at="2026-06-09T12:00:00Z",
|
||||
server_time_start=30,
|
||||
server_time_end=40,
|
||||
)
|
||||
_insert_materialized_match(
|
||||
db_path,
|
||||
match_id=3,
|
||||
match_key="older",
|
||||
ended_at="2026-05-01T12:00:00Z",
|
||||
server_time_start=50,
|
||||
server_time_end=60,
|
||||
)
|
||||
|
||||
run_database_maintenance_cleanup(
|
||||
apply=True,
|
||||
db_path=db_path,
|
||||
now="2026-06-10T13:00:00Z",
|
||||
recent_matches_keep=1,
|
||||
)
|
||||
|
||||
with closing(sqlite3.connect(db_path)) as connection:
|
||||
keys = {
|
||||
row[0]
|
||||
for row in connection.execute("SELECT match_key FROM rcon_materialized_matches")
|
||||
}
|
||||
self.assertEqual(keys, {"previous-week", "current-week-sample"})
|
||||
|
||||
def test_apply_deletes_old_non_protected_match_and_child_stats(self) -> None:
|
||||
with _temp_db() as db_path:
|
||||
_insert_materialized_match(
|
||||
db_path,
|
||||
match_id=1,
|
||||
match_key="delete-me",
|
||||
ended_at="2026-01-10T12:00:00Z",
|
||||
server_time_start=10,
|
||||
server_time_end=20,
|
||||
)
|
||||
_insert_materialized_match(
|
||||
db_path,
|
||||
match_id=2,
|
||||
match_key="keep-me",
|
||||
ended_at="2026-06-18T12:00:00Z",
|
||||
server_time_start=30,
|
||||
server_time_end=40,
|
||||
)
|
||||
_insert_player_stat(db_path, match_key="delete-me", player_id="player-1")
|
||||
_insert_player_stat(db_path, match_key="keep-me", player_id="player-2")
|
||||
|
||||
run_database_maintenance_cleanup(
|
||||
apply=True,
|
||||
db_path=db_path,
|
||||
now="2026-06-20T12:00:00Z",
|
||||
recent_matches_keep=1,
|
||||
)
|
||||
|
||||
with closing(sqlite3.connect(db_path)) as connection:
|
||||
deleted_match_count = connection.execute(
|
||||
"SELECT COUNT(*) FROM rcon_materialized_matches WHERE match_key = 'delete-me'"
|
||||
).fetchone()[0]
|
||||
deleted_stat_count = connection.execute(
|
||||
"SELECT COUNT(*) FROM rcon_match_player_stats WHERE match_key = 'delete-me'"
|
||||
).fetchone()[0]
|
||||
kept_stat_count = connection.execute(
|
||||
"SELECT COUNT(*) FROM rcon_match_player_stats WHERE match_key = 'keep-me'"
|
||||
).fetchone()[0]
|
||||
self.assertEqual(deleted_match_count, 0)
|
||||
self.assertEqual(deleted_stat_count, 0)
|
||||
self.assertEqual(kept_stat_count, 1)
|
||||
|
||||
def test_missing_optional_tables_are_logged_and_do_not_crash(self) -> None:
|
||||
with _temp_db(create_schema=False) as db_path:
|
||||
stream = io.StringIO()
|
||||
with redirect_stdout(stream):
|
||||
payload = run_database_maintenance_cleanup(
|
||||
db_path=db_path,
|
||||
now="2026-06-20T12:00:00Z",
|
||||
)
|
||||
|
||||
self.assertEqual(payload["status"], "ok")
|
||||
self.assertIn("database-maintenance-table-skipped", stream.getvalue())
|
||||
|
||||
|
||||
def _temp_db(*, create_schema: bool = True):
|
||||
class _TempDbContext:
|
||||
def __enter__(self) -> Path:
|
||||
self._tmpdir = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
|
||||
self.db_path = Path(self._tmpdir.name) / "maintenance.sqlite3"
|
||||
if create_schema:
|
||||
initialize_storage(db_path=self.db_path)
|
||||
initialize_rcon_admin_log_storage(db_path=self.db_path)
|
||||
initialize_rcon_materialized_storage(db_path=self.db_path)
|
||||
return self.db_path
|
||||
|
||||
def __exit__(self, exc_type, exc, tb) -> None:
|
||||
self._tmpdir.cleanup()
|
||||
|
||||
return _TempDbContext()
|
||||
|
||||
|
||||
def _insert_server_snapshot(db_path: Path, *, snapshot_id: int, captured_at: str) -> None:
|
||||
with closing(sqlite3.connect(db_path)) as connection:
|
||||
connection.execute(
|
||||
"""
|
||||
INSERT OR IGNORE INTO game_sources (
|
||||
id, slug, display_name, provider_kind, is_active, created_at, updated_at
|
||||
) VALUES (1, 'current-hll', 'Current Hell Let Loose', 'development', 1, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
|
||||
"""
|
||||
)
|
||||
connection.execute(
|
||||
"""
|
||||
INSERT OR IGNORE INTO servers (
|
||||
id, game_source_id, external_server_id, server_name, region, first_seen_at, last_seen_at
|
||||
) VALUES (1, 1, 'server-1', 'Server 1', 'ES', ?, ?)
|
||||
""",
|
||||
(captured_at, captured_at),
|
||||
)
|
||||
connection.execute(
|
||||
"""
|
||||
INSERT INTO server_snapshots (
|
||||
id, server_id, captured_at, status, players, max_players, current_map, source_name
|
||||
) VALUES (?, 1, ?, 'online', 10, 100, 'hurtgen', 'test')
|
||||
""",
|
||||
(snapshot_id, captured_at),
|
||||
)
|
||||
connection.commit()
|
||||
|
||||
|
||||
def _insert_admin_log_event(
|
||||
db_path: Path,
|
||||
*,
|
||||
event_id: int,
|
||||
event_type: str,
|
||||
event_timestamp: str,
|
||||
server_time: int,
|
||||
) -> None:
|
||||
with closing(sqlite3.connect(db_path)) as connection:
|
||||
connection.execute(
|
||||
"""
|
||||
INSERT INTO rcon_admin_log_events (
|
||||
id, target_key, external_server_id, event_timestamp, server_time,
|
||||
relative_time, event_type, raw_message, canonical_message,
|
||||
parsed_payload_json, raw_entry_json
|
||||
) VALUES (?, 'comunidad-hispana-01', 'comunidad-hispana-01', ?, ?, '', ?, '', '', '{}', '{}')
|
||||
""",
|
||||
(event_id, event_timestamp, server_time, event_type),
|
||||
)
|
||||
connection.commit()
|
||||
|
||||
|
||||
def _insert_materialized_match(
|
||||
db_path: Path,
|
||||
*,
|
||||
match_id: int,
|
||||
match_key: str,
|
||||
ended_at: str,
|
||||
server_time_start: int,
|
||||
server_time_end: int,
|
||||
) -> None:
|
||||
started_at = _shift_iso(ended_at, hours=-1)
|
||||
with closing(sqlite3.connect(db_path)) as connection:
|
||||
connection.execute(
|
||||
"""
|
||||
INSERT INTO rcon_materialized_matches (
|
||||
id, target_key, external_server_id, match_key, map_name, map_pretty_name,
|
||||
game_mode, started_server_time, ended_server_time, started_at, ended_at,
|
||||
allied_score, axis_score, winner, confidence_mode, source_basis
|
||||
) VALUES (?, 'comunidad-hispana-01', 'comunidad-hispana-01', ?, 'hurtgen', 'Hurtgen Forest',
|
||||
'warfare', ?, ?, ?, ?, 5, 3, 'allied', 'exact', ?)
|
||||
""",
|
||||
(
|
||||
match_id,
|
||||
match_key,
|
||||
server_time_start,
|
||||
server_time_end,
|
||||
started_at,
|
||||
ended_at,
|
||||
MATCH_RESULT_SOURCE,
|
||||
),
|
||||
)
|
||||
connection.commit()
|
||||
|
||||
|
||||
def _insert_player_stat(db_path: Path, *, match_key: str, player_id: str) -> None:
|
||||
with closing(sqlite3.connect(db_path)) as connection:
|
||||
connection.execute(
|
||||
"""
|
||||
INSERT INTO rcon_match_player_stats (
|
||||
target_key, match_key, player_id, player_name, team,
|
||||
kills, deaths, teamkills, deaths_by_teamkill,
|
||||
weapons_json, death_by_weapons_json, most_killed_json, death_by_json
|
||||
) VALUES (
|
||||
'comunidad-hispana-01', ?, ?, ?, 'Allies',
|
||||
1, 1, 0, 0, '{}', '{}', '{}', '{}'
|
||||
)
|
||||
""",
|
||||
(match_key, player_id, player_id),
|
||||
)
|
||||
connection.commit()
|
||||
|
||||
|
||||
def _shift_iso(value: str, *, hours: int) -> str:
|
||||
point = datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(timezone.utc)
|
||||
shifted = point + timedelta(hours=hours)
|
||||
return shifted.isoformat().replace("+00:00", "Z")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
153
backend/tests/test_historical_runner_maintenance.py
Normal file
153
backend/tests/test_historical_runner_maintenance.py
Normal file
@@ -0,0 +1,153 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import os
|
||||
import unittest
|
||||
from contextlib import nullcontext, redirect_stdout
|
||||
from datetime import datetime, timezone
|
||||
from unittest.mock import patch
|
||||
|
||||
import app.historical_runner as historical_runner
|
||||
from app.historical_runner import _maybe_run_database_maintenance, _run_refresh_with_retries
|
||||
|
||||
|
||||
class HistoricalRunnerMaintenanceTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
historical_runner._LAST_DATABASE_MAINTENANCE_RUN_AT = None
|
||||
|
||||
def tearDown(self) -> None:
|
||||
historical_runner._LAST_DATABASE_MAINTENANCE_RUN_AT = None
|
||||
|
||||
def test_scheduler_disabled_does_not_call_cleanup(self) -> None:
|
||||
with (
|
||||
patch.dict(os.environ, {"HLL_DB_MAINTENANCE_ENABLED": "false"}, clear=False),
|
||||
patch("app.historical_runner.run_database_maintenance_cleanup") as cleanup,
|
||||
):
|
||||
result = _maybe_run_database_maintenance(
|
||||
now=datetime(2026, 6, 20, 12, tzinfo=timezone.utc)
|
||||
)
|
||||
|
||||
cleanup.assert_not_called()
|
||||
self.assertEqual(result["status"], "skipped")
|
||||
self.assertEqual(result["reason"], "disabled")
|
||||
|
||||
def test_scheduler_enabled_but_not_due_does_not_call_cleanup(self) -> None:
|
||||
with (
|
||||
patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"HLL_DB_MAINTENANCE_ENABLED": "true",
|
||||
"HLL_DB_MAINTENANCE_INTERVAL_SECONDS": "43200",
|
||||
},
|
||||
clear=False,
|
||||
),
|
||||
patch(
|
||||
"app.historical_runner.run_database_maintenance_cleanup",
|
||||
return_value={"status": "ok"},
|
||||
) as cleanup,
|
||||
):
|
||||
first = _maybe_run_database_maintenance(
|
||||
now=datetime(2026, 6, 20, 0, tzinfo=timezone.utc)
|
||||
)
|
||||
second = _maybe_run_database_maintenance(
|
||||
now=datetime(2026, 6, 20, 1, tzinfo=timezone.utc)
|
||||
)
|
||||
|
||||
self.assertEqual(first["status"], "ok")
|
||||
self.assertEqual(second["status"], "skipped")
|
||||
self.assertEqual(second["reason"], "not-due")
|
||||
cleanup.assert_called_once()
|
||||
|
||||
def test_scheduler_enabled_and_due_calls_cleanup(self) -> None:
|
||||
with (
|
||||
patch.dict(os.environ, {"HLL_DB_MAINTENANCE_ENABLED": "true"}, clear=False),
|
||||
patch(
|
||||
"app.historical_runner.run_database_maintenance_cleanup",
|
||||
return_value={"status": "ok"},
|
||||
) as cleanup,
|
||||
):
|
||||
result = _maybe_run_database_maintenance(
|
||||
now=datetime(2026, 6, 20, 12, tzinfo=timezone.utc)
|
||||
)
|
||||
|
||||
cleanup.assert_called_once()
|
||||
self.assertEqual(result["status"], "ok")
|
||||
|
||||
def test_cleanup_exception_is_logged_and_runner_continues(self) -> None:
|
||||
stream = io.StringIO()
|
||||
with (
|
||||
patch.dict(os.environ, {"HLL_DB_MAINTENANCE_ENABLED": "true"}, clear=False),
|
||||
patch("app.historical_runner.backend_writer_lock", return_value=nullcontext()),
|
||||
patch(
|
||||
"app.historical_runner._run_primary_rcon_capture",
|
||||
return_value={"status": "ok", "targets": []},
|
||||
),
|
||||
patch(
|
||||
"app.historical_runner.run_incremental_refresh",
|
||||
return_value={"status": "ok"},
|
||||
),
|
||||
patch(
|
||||
"app.historical_runner.generate_historical_snapshots",
|
||||
return_value={"status": "ok"},
|
||||
),
|
||||
patch(
|
||||
"app.historical_runner.rebuild_elo_mmr_models",
|
||||
return_value={"status": "ok"},
|
||||
),
|
||||
patch(
|
||||
"app.historical_runner.run_database_maintenance_cleanup",
|
||||
side_effect=RuntimeError("maintenance failed"),
|
||||
),
|
||||
redirect_stdout(stream),
|
||||
):
|
||||
result = _run_refresh_with_retries(
|
||||
max_retries=0,
|
||||
retry_delay_seconds=0,
|
||||
server_slug="comunidad-hispana-01",
|
||||
max_pages=None,
|
||||
page_size=None,
|
||||
run_number=1,
|
||||
)
|
||||
|
||||
self.assertEqual(result["status"], "ok")
|
||||
self.assertEqual(result["database_maintenance_result"]["status"], "error")
|
||||
self.assertIn("database-maintenance-scheduler-failed", stream.getvalue())
|
||||
|
||||
def test_interval_parsing_handles_invalid_values_safely(self) -> None:
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"HLL_DB_MAINTENANCE_ENABLED": "true",
|
||||
"HLL_DB_MAINTENANCE_INTERVAL_SECONDS": "bad",
|
||||
},
|
||||
clear=False,
|
||||
):
|
||||
interval_seconds, source = historical_runner._resolve_db_maintenance_interval_seconds()
|
||||
|
||||
self.assertEqual(interval_seconds, 43200)
|
||||
self.assertEqual(source, "default-invalid-env-fallback")
|
||||
|
||||
def test_maintenance_state_is_tracked_in_process(self) -> None:
|
||||
with (
|
||||
patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"HLL_DB_MAINTENANCE_ENABLED": "true",
|
||||
"HLL_DB_MAINTENANCE_INTERVAL_SECONDS": "3600",
|
||||
},
|
||||
clear=False,
|
||||
),
|
||||
patch(
|
||||
"app.historical_runner.run_database_maintenance_cleanup",
|
||||
return_value={"status": "ok"},
|
||||
),
|
||||
):
|
||||
_maybe_run_database_maintenance(now=datetime(2026, 6, 20, 12, tzinfo=timezone.utc))
|
||||
self.assertEqual(
|
||||
historical_runner._LAST_DATABASE_MAINTENANCE_RUN_AT,
|
||||
datetime(2026, 6, 20, 12, tzinfo=timezone.utc),
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user