From 02b3bc9ad1923e169a327bf208eefc6538dbf9dc Mon Sep 17 00:00:00 2001 From: devRaGonSa Date: Wed, 25 Mar 2026 12:07:11 +0100 Subject: [PATCH] Harden sqlite writers and add shared lock --- .../done/TASK-091-sqlite-writer-hardening.md | 53 +++++ ...e-writer-locking-for-automation-workers.md | 81 +++++++ backend/README.md | 85 +++++++ backend/app/config.py | 54 +++++ backend/app/historical_ingestion.py | 54 +++-- backend/app/historical_runner.py | 26 +- backend/app/historical_storage.py | 7 +- backend/app/player_event_storage.py | 6 +- backend/app/player_event_worker.py | 198 ++++++++-------- backend/app/rcon_historical_storage.py | 6 +- backend/app/rcon_historical_worker.py | 176 +++++++------- backend/app/sqlite_utils.py | 41 ++++ backend/app/storage.py | 9 +- backend/app/writer_lock.py | 223 ++++++++++++++++++ 14 files changed, 788 insertions(+), 231 deletions(-) create mode 100644 ai/tasks/done/TASK-091-sqlite-writer-hardening.md create mode 100644 ai/tasks/done/TASK-092-single-writer-locking-for-automation-workers.md create mode 100644 backend/app/sqlite_utils.py create mode 100644 backend/app/writer_lock.py diff --git a/ai/tasks/done/TASK-091-sqlite-writer-hardening.md b/ai/tasks/done/TASK-091-sqlite-writer-hardening.md new file mode 100644 index 0000000..63fe59d --- /dev/null +++ b/ai/tasks/done/TASK-091-sqlite-writer-hardening.md @@ -0,0 +1,53 @@ +# TASK-091-sqlite-writer-hardening + +## Goal +Unificar y endurecer la politica de conexion SQLite para todas las rutas de escritura del backend que comparten la misma base de datos, reduciendo contencion evitable y haciendo el comportamiento consistente entre historico base, player-events y captura prospectiva RCON. + +## Context +La repo ya tiene varias rutas de escritura sobre el mismo SQLite: +- historico base CRCON +- player-event ledger V2 +- captura prospectiva RCON +- snapshots y runners asociados + +Parte del storage ya usa timeout/WAL/busy_timeout, pero no toda la capa de persistencia lo hace de forma uniforme. Antes de cualquier politica de locking entre procesos, hace falta que todas las conexiones writer-capable usen una configuracion SQLite coherente. + +## Scope +Backend solamente. Sin cambios en frontend. + +## Steps +1. Auditar todas las funciones `_connect()` o equivalentes que abren SQLite en: + - `backend/app/historical_storage.py` + - `backend/app/player_event_storage.py` + - `backend/app/rcon_historical_storage.py` + - cualquier otro storage writer-capable relacionado +2. Crear una utilidad compartida y pequena para abrir conexiones SQLite con politica comun de escritura. +3. La politica comun debe incluir como minimo: + - `timeout` explicito + - `PRAGMA foreign_keys = ON` + - `PRAGMA journal_mode = WAL` + - `PRAGMA busy_timeout` + - `row_factory = sqlite3.Row` +4. Reusar esa utilidad en todas las capas de persistencia con escritura que comparten el mismo DB. +5. Mantener compatibilidad con la ruta actual del storage y sin cambiar contratos HTTP. +6. Actualizar README/runbook backend con una nota breve sobre la politica SQLite usada por writers. + +## Constraints +- No tocar frontend. +- No cambiar la semantica funcional de los endpoints. +- No cambiar el proveedor historico por defecto. +- No meter dependencias nuevas salvo necesidad extrema. +- Mantener la utilidad pequeña y facil de leer. + +## Validation +- Todas las capas writer-capable relevantes comparten la misma politica SQLite. +- `python -m compileall app` pasa. +- Las CLIs principales siguen arrancando con `--help`. +- La repo queda consistente. + +## Expected Files +- `backend/app/historical_storage.py` +- `backend/app/player_event_storage.py` +- `backend/app/rcon_historical_storage.py` +- un util compartido nuevo bajo `backend/app/` si hace falta +- `backend/README.md` diff --git a/ai/tasks/done/TASK-092-single-writer-locking-for-automation-workers.md b/ai/tasks/done/TASK-092-single-writer-locking-for-automation-workers.md new file mode 100644 index 0000000..4df73e7 --- /dev/null +++ b/ai/tasks/done/TASK-092-single-writer-locking-for-automation-workers.md @@ -0,0 +1,81 @@ +# TASK-092-single-writer-locking-for-automation-workers + +## Goal +Imponer una coordinacion de single-writer entre todos los procesos del backend que escriben sobre el mismo SQLite compartido, evitando colisiones entre automatizaciones y ejecuciones manuales y sustituyendo errores opacos de `database is locked` por una coordinacion controlada y mensajes claros. + +## Context +Actualmente pueden coincidir sobre el mismo volumen `/app/data`: +- `historical-runner` +- `player_event_worker` +- `rcon-historical-worker` +- ejecuciones manuales via `docker compose exec backend ...` + +Aunque WAL y busy_timeout ayudan, no garantizan una operativa limpia si varios writers largos arrancan a la vez. Hace falta una politica de exclusion mutua a nivel de proceso/job, no solo a nivel SQLite. + +## Scope +Backend y orquestacion minima/documentacion. Sin cambios en frontend. + +## Desired Design +Implementar un unico lock compartido para todos los writers que tocan el mismo SQLite. + +Ese lock debe: +- vivir bajo el storage compartido en `/app/data` o una ruta derivada del `storage_path` +- ser comun para: + - `app.historical_runner` + - `app.historical_ingestion` + - `app.player_event_worker` + - `app.rcon_historical_worker` +- adquirirse al principio de cada run writer-oriented +- liberarse siempre aunque haya excepcion +- incluir metadata visible del holder: + - proceso/comando + - started_at + - host/container si es viable +- tener espera configurable con timeout y poll interval +- fallar con error claro cuando no pueda adquirir el lock dentro del timeout +- evitar dependencias nuevas si es posible + +## Steps +1. Crear una utilidad comun de writer lock compartido para el backend. +2. Aplicarla a: + - `historical_ingestion` manual + - `historical_runner` + - `player_event_worker` + - `rcon_historical_worker` +3. Asegurarte de que el lock cubre la ejecucion writer-oriented completa, no solo una sentencia aislada. +4. Exponer configuracion/env para: + - lock timeout + - poll interval +5. Añadir mensajes de error/estado legibles cuando el lock este ocupado. +6. Mantener las rutas HTTP read-only fuera de este lock. +7. Actualizar `docker-compose.yml` solo si es necesario para dejar la operativa alineada. +8. Actualizar `backend/README.md` con runbook claro: + - que writers comparten lock + - como hacer una pasada manual + - que pasa si el lock esta ocupado + - como convivir con automatizaciones sin parar contenedores salvo necesidad excepcional + +## Constraints +- No tocar frontend. +- No convertir esto en un rediseño total del scheduler. +- No romper los workers ya existentes. +- No bloquear las rutas HTTP de lectura. +- No ampliar el alcance funcional del read model RCON. +- Debe quedar claro que la solucion principal es single-writer coordination, no solo subir retries. + +## Validation +- Existe un lock compartido real para los writers. +- `historical_runner`, `historical_ingestion`, `player_event_worker` y `rcon_historical_worker` lo usan. +- Si dos writers coinciden, no se produce un fallo opaco de SQLite; hay espera controlada o error claro de lock ocupado. +- `python -m compileall app` pasa. +- Las CLIs principales siguen respondiendo con `--help`. +- La repo queda consistente. + +## Expected Files +- uno o varios utilitarios nuevos bajo `backend/app/` +- `backend/app/historical_runner.py` +- `backend/app/historical_ingestion.py` +- `backend/app/player_event_worker.py` +- `backend/app/rcon_historical_worker.py` +- `docker-compose.yml` si hace falta +- `backend/README.md` diff --git a/backend/README.md b/backend/README.md index 59edbd8..959cc5f 100644 --- a/backend/README.md +++ b/backend/README.md @@ -71,6 +71,10 @@ Variables opcionales: - `HLL_RCON_HISTORICAL_CAPTURE_INTERVAL_SECONDS` - `HLL_RCON_HISTORICAL_CAPTURE_MAX_RETRIES` - `HLL_RCON_HISTORICAL_CAPTURE_RETRY_DELAY_SECONDS` +- `HLL_BACKEND_SQLITE_WRITER_TIMEOUT_SECONDS` +- `HLL_BACKEND_SQLITE_BUSY_TIMEOUT_MS` +- `HLL_BACKEND_WRITER_LOCK_TIMEOUT_SECONDS` +- `HLL_BACKEND_WRITER_LOCK_POLL_INTERVAL_SECONDS` - `HLL_HISTORICAL_CRCON_PAGE_SIZE` - `HLL_HISTORICAL_CRCON_TIMEOUT_SECONDS` - `HLL_HISTORICAL_CRCON_DETAIL_WORKERS` @@ -82,6 +86,10 @@ Variables opcionales: - `HLL_HISTORICAL_FULL_SNAPSHOT_EVERY_RUNS` - `HLL_HISTORICAL_REFRESH_MAX_RETRIES` - `HLL_HISTORICAL_REFRESH_RETRY_DELAY_SECONDS` +- `HLL_BACKEND_SQLITE_WRITER_TIMEOUT_SECONDS` +- `HLL_BACKEND_SQLITE_BUSY_TIMEOUT_MS` +- `HLL_BACKEND_WRITER_LOCK_TIMEOUT_SECONDS` +- `HLL_BACKEND_WRITER_LOCK_POLL_INTERVAL_SECONDS` - `HLL_HISTORICAL_WEEKLY_FALLBACK_MIN_MATCHES` - `HLL_HISTORICAL_WEEKLY_FALLBACK_MAX_WEEKDAY` @@ -490,6 +498,22 @@ En Docker, ese mismo rol de persistencia debe montarse fuera del contenedor en: /app/data/hll_vietnam_dev.sqlite3 ``` +Politica comun SQLite para writers: + +- `timeout` explicito compartido +- `PRAGMA foreign_keys = ON` +- `PRAGMA journal_mode = WAL` +- `PRAGMA busy_timeout` +- `row_factory = sqlite3.Row` + +Esta politica se aplica de forma uniforme a las capas writer-capable que +comparten el mismo SQLite, incluyendo: + +- `historical_storage.py` +- `player_event_storage.py` +- `rcon_historical_storage.py` +- `storage.py` + Variable opcional: - `HLL_BACKEND_STORAGE_PATH` @@ -1119,6 +1143,67 @@ El servicio `historical-runner` usa el mismo volumen persistente `./backend/data y ejecuta `python -m app.historical_runner --hourly` como bucle operativo dedicado, sin mezclar el scheduler con el proceso HTTP principal. +## Coordinacion single-writer para automatizaciones y CLI + +Todos los procesos writer-oriented que comparten el mismo SQLite usan ahora un +lock comun derivado de `HLL_BACKEND_STORAGE_PATH` y persistido junto al volumen +de datos compartido. Ese lock coordina: + +- `app.historical_ingestion` +- `app.historical_runner` +- `app.player_event_worker` +- `app.rcon_historical_worker` + +Rutas HTTP read-only como `/api/historical/snapshots/*`, `/api/servers` en modo +cache local y el read model minimo RCON no adquieren este lock. + +Variables operativas: + +- `HLL_BACKEND_WRITER_LOCK_TIMEOUT_SECONDS` +- `HLL_BACKEND_WRITER_LOCK_POLL_INTERVAL_SECONDS` + +Comportamiento: + +- si un writer ya esta ejecutandose, el siguiente espera de forma controlada + hasta agotar el timeout configurado +- si no puede adquirir el lock, falla con un error claro indicando: + - lock path + - holder + - `started_at` + - host + - pid +- la coordinacion principal es este single-writer lock; WAL y `busy_timeout` + quedan como endurecimiento complementario, no como solucion unica + +Runbook minimo: + +- pasada manual del historico base mientras el runner automatico existe: + + ```powershell + docker compose exec backend python -m app.historical_ingestion refresh --overlap-hours 48 + ``` + + Si el lock esta ocupado, el comando esperara hasta el timeout configurado y, + si no se libera, terminara con un mensaje claro de lock ocupado. + +- pasada manual de player-events: + + ```powershell + docker compose exec backend python -m app.player_event_worker refresh --overlap-hours 48 + ``` + +- pasada manual de captura prospectiva RCON: + + ```powershell + docker compose exec backend python -m app.rcon_historical_worker capture + ``` + +- convivencia recomendada con automatizaciones: + - no hace falta parar contenedores por defecto + - dejar que el lock coordine la exclusión mutua + - usar `--max-runs 1` o comandos manuales puntuales cuando se quiera una + pasada controlada + Comprobaciones utiles con Compose: - `docker compose ps historical-runner` diff --git a/backend/app/config.py b/backend/app/config.py index 2cb0455..d2de8f6 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -33,6 +33,10 @@ DEFAULT_PLAYER_EVENT_REFRESH_RETRY_DELAY_SECONDS = 30 DEFAULT_RCON_HISTORICAL_CAPTURE_INTERVAL_SECONDS = 120 DEFAULT_RCON_HISTORICAL_CAPTURE_MAX_RETRIES = 2 DEFAULT_RCON_HISTORICAL_CAPTURE_RETRY_DELAY_SECONDS = 15 +DEFAULT_SQLITE_WRITER_TIMEOUT_SECONDS = 30.0 +DEFAULT_SQLITE_BUSY_TIMEOUT_MS = 30000 +DEFAULT_WRITER_LOCK_TIMEOUT_SECONDS = 120.0 +DEFAULT_WRITER_LOCK_POLL_INTERVAL_SECONDS = 1.0 DEFAULT_ALLOWED_ORIGINS = ( "null", "http://127.0.0.1:5500", @@ -79,6 +83,56 @@ def get_storage_path() -> Path: return Path(configured_path) if configured_path else default_path +def get_sqlite_writer_timeout_seconds() -> float: + """Return the SQLite connection timeout shared by writer-capable storage layers.""" + configured_value = os.getenv( + "HLL_BACKEND_SQLITE_WRITER_TIMEOUT_SECONDS", + str(DEFAULT_SQLITE_WRITER_TIMEOUT_SECONDS), + ) + timeout_seconds = float(configured_value) + if timeout_seconds <= 0: + raise ValueError("HLL_BACKEND_SQLITE_WRITER_TIMEOUT_SECONDS must be positive.") + return timeout_seconds + + +def get_sqlite_busy_timeout_ms() -> int: + """Return the SQLite busy_timeout shared by writer-capable storage layers.""" + configured_value = os.getenv( + "HLL_BACKEND_SQLITE_BUSY_TIMEOUT_MS", + str(DEFAULT_SQLITE_BUSY_TIMEOUT_MS), + ) + busy_timeout_ms = int(configured_value) + if busy_timeout_ms <= 0: + raise ValueError("HLL_BACKEND_SQLITE_BUSY_TIMEOUT_MS must be positive.") + return busy_timeout_ms + + +def get_writer_lock_timeout_seconds() -> float: + """Return how long writer jobs should wait for the shared backend writer lock.""" + configured_value = os.getenv( + "HLL_BACKEND_WRITER_LOCK_TIMEOUT_SECONDS", + str(DEFAULT_WRITER_LOCK_TIMEOUT_SECONDS), + ) + timeout_seconds = float(configured_value) + if timeout_seconds < 0: + raise ValueError("HLL_BACKEND_WRITER_LOCK_TIMEOUT_SECONDS must be zero or positive.") + return timeout_seconds + + +def get_writer_lock_poll_interval_seconds() -> float: + """Return how often writer jobs should poll the shared backend writer lock.""" + configured_value = os.getenv( + "HLL_BACKEND_WRITER_LOCK_POLL_INTERVAL_SECONDS", + str(DEFAULT_WRITER_LOCK_POLL_INTERVAL_SECONDS), + ) + poll_interval_seconds = float(configured_value) + if poll_interval_seconds <= 0: + raise ValueError( + "HLL_BACKEND_WRITER_LOCK_POLL_INTERVAL_SECONDS must be positive." + ) + return poll_interval_seconds + + def get_refresh_interval_seconds() -> int: """Return the default interval used by the local refresh loop.""" configured_value = os.getenv( diff --git a/backend/app/historical_ingestion.py b/backend/app/historical_ingestion.py index 9c841fc..2cbecfe 100644 --- a/backend/app/historical_ingestion.py +++ b/backend/app/historical_ingestion.py @@ -27,6 +27,7 @@ from .historical_storage import ( start_ingestion_run, upsert_historical_match, ) +from .writer_lock import backend_writer_lock, build_writer_lock_holder @dataclass(slots=True) @@ -57,16 +58,22 @@ def run_bootstrap( rebuild_snapshots: bool = True, ) -> dict[str, object]: """Run a first full historical import against one or all configured servers.""" - return _run_ingestion( - mode="bootstrap", - server_slug=server_slug, - max_pages=max_pages, - page_size=page_size, - start_page=start_page, - detail_workers=detail_workers, - incremental=False, - rebuild_snapshots=rebuild_snapshots, - ) + with backend_writer_lock( + holder=build_writer_lock_holder( + f"app.historical_ingestion bootstrap:{server_slug or 'all-servers'}" + ) + ): + return _run_ingestion( + mode="bootstrap", + server_slug=server_slug, + max_pages=max_pages, + page_size=page_size, + start_page=start_page, + detail_workers=detail_workers, + overlap_hours=None, + incremental=False, + rebuild_snapshots=rebuild_snapshots, + ) def run_incremental_refresh( @@ -80,17 +87,22 @@ def run_incremental_refresh( rebuild_snapshots: bool = True, ) -> dict[str, object]: """Refresh recent historical pages without replaying the whole archive.""" - return _run_ingestion( - mode="incremental", - server_slug=server_slug, - max_pages=max_pages, - page_size=page_size, - start_page=start_page, - detail_workers=detail_workers, - overlap_hours=overlap_hours, - incremental=True, - rebuild_snapshots=rebuild_snapshots, - ) + with backend_writer_lock( + holder=build_writer_lock_holder( + f"app.historical_ingestion refresh:{server_slug or 'all-servers'}" + ) + ): + return _run_ingestion( + mode="incremental", + server_slug=server_slug, + max_pages=max_pages, + page_size=page_size, + start_page=start_page, + detail_workers=detail_workers, + overlap_hours=overlap_hours, + incremental=True, + rebuild_snapshots=rebuild_snapshots, + ) def _run_ingestion( diff --git a/backend/app/historical_runner.py b/backend/app/historical_runner.py index f81c38e..b56943c 100644 --- a/backend/app/historical_runner.py +++ b/backend/app/historical_runner.py @@ -19,6 +19,7 @@ from .historical_snapshots import ( generate_and_persist_historical_snapshots, generate_and_persist_priority_historical_snapshots, ) +from .writer_lock import backend_writer_lock, build_writer_lock_holder HOURLY_INTERVAL_SECONDS = 3600 DEFAULT_HISTORICAL_SERVER_SCOPE = ( @@ -89,16 +90,21 @@ def _run_refresh_with_retries( while True: attempt += 1 try: - refresh_result = run_incremental_refresh( - server_slug=server_slug, - max_pages=max_pages, - page_size=page_size, - rebuild_snapshots=False, - ) - snapshot_result = generate_historical_snapshots( - server_slug=server_slug, - run_number=run_number, - ) + with backend_writer_lock( + holder=build_writer_lock_holder( + f"app.historical_runner refresh:{server_slug or 'all-servers'}" + ) + ): + refresh_result = run_incremental_refresh( + server_slug=server_slug, + max_pages=max_pages, + page_size=page_size, + rebuild_snapshots=False, + ) + snapshot_result = generate_historical_snapshots( + server_slug=server_slug, + run_number=run_number, + ) return { "status": "ok", "attempts_used": attempt, diff --git a/backend/app/historical_storage.py b/backend/app/historical_storage.py index 65692f7..71e03d9 100644 --- a/backend/app/historical_storage.py +++ b/backend/app/historical_storage.py @@ -16,6 +16,7 @@ from .config import ( from .historical_models import HistoricalServerDefinition from .monthly_mvp import build_monthly_mvp_rankings from .monthly_mvp_v2 import build_monthly_mvp_v2_rankings +from .sqlite_utils import connect_sqlite_writer DEFAULT_HISTORICAL_SERVERS = ( @@ -1903,11 +1904,7 @@ def _build_player_event_scope_sql(server_id: str | None) -> tuple[str, list[obje def _connect(db_path: Path) -> sqlite3.Connection: - connection = sqlite3.connect(db_path, timeout=30.0) - connection.row_factory = sqlite3.Row - connection.execute("PRAGMA journal_mode=WAL") - connection.execute("PRAGMA busy_timeout = 30000") - return connection + return connect_sqlite_writer(db_path) def _resolve_match_winner(allied_score: object, axis_score: object) -> str | None: diff --git a/backend/app/player_event_storage.py b/backend/app/player_event_storage.py index f1ed692..c55f5b4 100644 --- a/backend/app/player_event_storage.py +++ b/backend/app/player_event_storage.py @@ -9,6 +9,7 @@ from pathlib import Path from .config import get_player_event_refresh_overlap_hours, get_storage_path from .player_event_models import PlayerEventRecord +from .sqlite_utils import connect_sqlite_writer def initialize_player_event_storage(*, db_path: Path | None = None) -> Path: @@ -416,10 +417,7 @@ def get_player_event_refresh_cutoff_for_server( def _connect(db_path: Path) -> sqlite3.Connection: - connection = sqlite3.connect(db_path) - connection.row_factory = sqlite3.Row - connection.execute("PRAGMA foreign_keys = ON") - return connection + return connect_sqlite_writer(db_path) def _utc_now_iso() -> str: diff --git a/backend/app/player_event_worker.py b/backend/app/player_event_worker.py index 56ddb0a..84162c4 100644 --- a/backend/app/player_event_worker.py +++ b/backend/app/player_event_worker.py @@ -30,6 +30,7 @@ from .player_event_storage import ( start_player_event_ingestion_run, upsert_player_events, ) +from .writer_lock import backend_writer_lock, build_writer_lock_holder @dataclass(slots=True) @@ -55,105 +56,110 @@ def run_player_event_refresh( overlap_hours: int | None = None, ) -> dict[str, object]: """Refresh recent player event summaries from the configured historical source.""" - initialize_player_event_storage() - data_source = get_historical_data_source() - event_source = get_player_event_source() - resolved_page_size = page_size or get_historical_crcon_page_size() - resolved_detail_workers = detail_workers or get_historical_crcon_detail_workers() - resolved_overlap_hours = ( - get_player_event_refresh_overlap_hours() - if overlap_hours is None - else overlap_hours - ) - if resolved_overlap_hours < 0: - raise ValueError("--overlap-hours must be zero or positive.") - selected_servers = _select_servers(server_slug) - processed_servers: list[dict[str, object]] = [] - active_runs: dict[str, int] = {} + with backend_writer_lock( + holder=build_writer_lock_holder( + f"app.player_event_worker refresh:{server_slug or 'all-servers'}" + ) + ): + initialize_player_event_storage() + data_source = get_historical_data_source() + event_source = get_player_event_source() + resolved_page_size = page_size or get_historical_crcon_page_size() + resolved_detail_workers = detail_workers or get_historical_crcon_detail_workers() + resolved_overlap_hours = ( + get_player_event_refresh_overlap_hours() + if overlap_hours is None + else overlap_hours + ) + if resolved_overlap_hours < 0: + raise ValueError("--overlap-hours must be zero or positive.") + selected_servers = _select_servers(server_slug) + processed_servers: list[dict[str, object]] = [] + active_runs: dict[str, int] = {} - try: - for server in selected_servers: - current_server_slug = str(server["slug"]) - run_id = start_player_event_ingestion_run( - mode="refresh", - target_server_slug=current_server_slug, - ) - active_runs[current_server_slug] = run_id - cutoff = get_player_event_refresh_cutoff_for_server( - current_server_slug, - overlap_hours=resolved_overlap_hours, - ) - mark_player_event_progress_started( - server_slug=current_server_slug, - mode="refresh", - run_id=run_id, - cutoff_occurred_at=cutoff, - ) - server_stats = _ingest_server( - server=server, - run_id=run_id, - data_source=data_source, - event_source=event_source, - page_size=resolved_page_size, - max_pages=max_pages, - start_page=_resolve_start_page( + try: + for server in selected_servers: + current_server_slug = str(server["slug"]) + run_id = start_player_event_ingestion_run( + mode="refresh", + target_server_slug=current_server_slug, + ) + active_runs[current_server_slug] = run_id + cutoff = get_player_event_refresh_cutoff_for_server( + current_server_slug, + overlap_hours=resolved_overlap_hours, + ) + mark_player_event_progress_started( server_slug=current_server_slug, - start_page=start_page, - ), - detail_workers=resolved_detail_workers, - cutoff=cutoff, - ) - finalize_player_event_ingestion_run( - run_id, - status="success", - pages_processed=server_stats["pages_processed"], - matches_seen=server_stats["matches_seen"], - matches_fetched=server_stats["matches_fetched"], - events_inserted=server_stats["events_inserted"], - duplicate_events=server_stats["duplicate_events"], - notes=f"source={data_source.source_kind};adapter={event_source.source_kind}", - ) - finalize_player_event_progress( - server_slug=current_server_slug, - mode="refresh", - run_id=run_id, - status="success", - archive_exhausted=bool(server_stats["archive_exhausted"]), - ) - processed_servers.append(server_stats) - active_runs.pop(current_server_slug, None) - except Exception as exc: - for active_server_slug, run_id in active_runs.items(): - finalize_player_event_ingestion_run( - run_id, - status="failed", - pages_processed=0, - matches_seen=0, - matches_fetched=0, - events_inserted=0, - duplicate_events=0, - notes=str(exc), - ) - finalize_player_event_progress( - server_slug=active_server_slug, - mode="refresh", - run_id=run_id, - status="failed", - error_message=str(exc), - ) - raise + mode="refresh", + run_id=run_id, + cutoff_occurred_at=cutoff, + ) + server_stats = _ingest_server( + server=server, + run_id=run_id, + data_source=data_source, + event_source=event_source, + page_size=resolved_page_size, + max_pages=max_pages, + start_page=_resolve_start_page( + server_slug=current_server_slug, + start_page=start_page, + ), + detail_workers=resolved_detail_workers, + cutoff=cutoff, + ) + finalize_player_event_ingestion_run( + run_id, + status="success", + pages_processed=server_stats["pages_processed"], + matches_seen=server_stats["matches_seen"], + matches_fetched=server_stats["matches_fetched"], + events_inserted=server_stats["events_inserted"], + duplicate_events=server_stats["duplicate_events"], + notes=f"source={data_source.source_kind};adapter={event_source.source_kind}", + ) + finalize_player_event_progress( + server_slug=current_server_slug, + mode="refresh", + run_id=run_id, + status="success", + archive_exhausted=bool(server_stats["archive_exhausted"]), + ) + processed_servers.append(server_stats) + active_runs.pop(current_server_slug, None) + except Exception as exc: + for active_server_slug, run_id in active_runs.items(): + finalize_player_event_ingestion_run( + run_id, + status="failed", + pages_processed=0, + matches_seen=0, + matches_fetched=0, + events_inserted=0, + duplicate_events=0, + notes=str(exc), + ) + finalize_player_event_progress( + server_slug=active_server_slug, + mode="refresh", + run_id=run_id, + status="failed", + error_message=str(exc), + ) + raise - return { - "status": "ok", - "mode": "refresh", - "source_provider": data_source.source_kind, - "event_adapter": event_source.source_kind, - "page_size": resolved_page_size, - "detail_workers": resolved_detail_workers, - "overlap_hours": resolved_overlap_hours, - "scope": event_source.describe_scope(), - "servers": processed_servers, - } + return { + "status": "ok", + "mode": "refresh", + "source_provider": data_source.source_kind, + "event_adapter": event_source.source_kind, + "page_size": resolved_page_size, + "detail_workers": resolved_detail_workers, + "overlap_hours": resolved_overlap_hours, + "scope": event_source.describe_scope(), + "servers": processed_servers, + } def run_periodic_player_event_refresh( diff --git a/backend/app/rcon_historical_storage.py b/backend/app/rcon_historical_storage.py index 67fb731..e036b69 100644 --- a/backend/app/rcon_historical_storage.py +++ b/backend/app/rcon_historical_storage.py @@ -9,6 +9,7 @@ from datetime import datetime, timezone from pathlib import Path from .config import get_storage_path +from .sqlite_utils import connect_sqlite_writer def initialize_rcon_historical_storage(*, db_path: Path | None = None) -> Path: @@ -354,10 +355,7 @@ def list_recent_rcon_historical_samples( def _connect(db_path: Path) -> sqlite3.Connection: - connection = sqlite3.connect(db_path) - connection.row_factory = sqlite3.Row - connection.execute("PRAGMA foreign_keys = ON") - return connection + return connect_sqlite_writer(db_path) def _upsert_target(connection: sqlite3.Connection, *, target: Mapping[str, object]) -> int: diff --git a/backend/app/rcon_historical_worker.py b/backend/app/rcon_historical_worker.py index 34ff608..add7d07 100644 --- a/backend/app/rcon_historical_worker.py +++ b/backend/app/rcon_historical_worker.py @@ -23,6 +23,7 @@ from .rcon_historical_storage import ( start_rcon_historical_capture_run, ) from .snapshots import utc_now +from .writer_lock import backend_writer_lock, build_writer_lock_holder @dataclass(slots=True) @@ -38,93 +39,98 @@ def run_rcon_historical_capture( target_key: str | None = None, ) -> dict[str, object]: """Capture one prospective RCON sample for one or all configured targets.""" - initialize_rcon_historical_storage() - selected_targets = _select_targets(target_key) - 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="capture", target_scope=target_scope) - stats = RconHistoricalCaptureStats() - items: list[dict[str, object]] = [] - errors: list[dict[str, object]] = [] - - try: - for target in selected_targets: - target_metadata = _serialize_target(target) - stats.targets_seen += 1 - try: - sample = query_live_server_sample(target) - 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, - "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=str(exc), - ) - errors.append( - { - "target_key": target_metadata["target_key"], - "name": target.name, - "host": target.host, - "port": target.port, - "message": str(exc), - } - ) - - 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=(",", ":")), + with backend_writer_lock( + holder=build_writer_lock_holder( + f"app.rcon_historical_worker capture:{target_key or 'all-targets'}" ) - 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 + ): + initialize_rcon_historical_storage() + selected_targets = _select_targets(target_key) + 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="capture", target_scope=target_scope) + stats = RconHistoricalCaptureStats() + items: list[dict[str, object]] = [] + errors: list[dict[str, object]] = [] - return { - "status": "ok" if items else "error", - "run_status": status, - "captured_at": captured_at, - "target_scope": target_scope, - "targets": items, - "errors": errors, - "storage_status": list_rcon_historical_target_statuses(), - "totals": { - "targets_seen": stats.targets_seen, - "samples_inserted": stats.samples_inserted, - "duplicate_samples": stats.duplicate_samples, - "failed_targets": stats.failed_targets, - }, - } + try: + for target in selected_targets: + target_metadata = _serialize_target(target) + stats.targets_seen += 1 + try: + sample = query_live_server_sample(target) + 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, + "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=str(exc), + ) + errors.append( + { + "target_key": target_metadata["target_key"], + "name": target.name, + "host": target.host, + "port": target.port, + "message": str(exc), + } + ) + + 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, + "targets": items, + "errors": errors, + "storage_status": list_rcon_historical_target_statuses(), + "totals": { + "targets_seen": stats.targets_seen, + "samples_inserted": stats.samples_inserted, + "duplicate_samples": stats.duplicate_samples, + "failed_targets": stats.failed_targets, + }, + } def run_periodic_rcon_historical_capture( diff --git a/backend/app/sqlite_utils.py b/backend/app/sqlite_utils.py new file mode 100644 index 0000000..7bdf40f --- /dev/null +++ b/backend/app/sqlite_utils.py @@ -0,0 +1,41 @@ +"""Shared SQLite connection helpers for backend persistence layers.""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +from .config import get_sqlite_busy_timeout_ms, get_sqlite_writer_timeout_seconds + + +def connect_sqlite_writer( + db_path: Path, + *, + timeout_seconds: float | None = None, + busy_timeout_ms: int | None = None, +) -> sqlite3.Connection: + """Open one SQLite connection with the common writer policy.""" + resolved_timeout_seconds = ( + get_sqlite_writer_timeout_seconds() + if timeout_seconds is None + else timeout_seconds + ) + resolved_busy_timeout_ms = ( + get_sqlite_busy_timeout_ms() + if busy_timeout_ms is None + else busy_timeout_ms + ) + + connection = sqlite3.connect(db_path, timeout=resolved_timeout_seconds) + connection.row_factory = sqlite3.Row + connection.execute("PRAGMA foreign_keys = ON") + connection.execute("PRAGMA journal_mode = WAL") + connection.execute(f"PRAGMA busy_timeout = {resolved_busy_timeout_ms}") + return connection + + +def connect_sqlite_readonly(db_path: Path) -> sqlite3.Connection: + """Open one read-only SQLite connection with row access enabled.""" + connection = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) + connection.row_factory = sqlite3.Row + return connection diff --git a/backend/app/storage.py b/backend/app/storage.py index a91c7fe..e99dd4e 100644 --- a/backend/app/storage.py +++ b/backend/app/storage.py @@ -8,6 +8,7 @@ from pathlib import Path from typing import Iterable, Mapping from .config import get_storage_path +from .sqlite_utils import connect_sqlite_readonly, connect_sqlite_writer DEFAULT_GAME_SOURCE = { @@ -265,15 +266,11 @@ def list_server_history( def _connect(db_path: Path) -> sqlite3.Connection: - connection = sqlite3.connect(db_path) - connection.row_factory = sqlite3.Row - return connection + return connect_sqlite_writer(db_path) def _connect_readonly(db_path: Path) -> sqlite3.Connection: - connection = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) - connection.row_factory = sqlite3.Row - return connection + return connect_sqlite_readonly(db_path) def _upsert_game_source( diff --git a/backend/app/writer_lock.py b/backend/app/writer_lock.py new file mode 100644 index 0000000..682a3b9 --- /dev/null +++ b/backend/app/writer_lock.py @@ -0,0 +1,223 @@ +"""Shared single-writer lock coordination for backend automation jobs.""" + +from __future__ import annotations + +import json +import os +import socket +import sys +import time +from contextlib import contextmanager +from datetime import datetime, timezone +from pathlib import Path +from uuid import uuid4 + +from .config import ( + get_storage_path, + get_writer_lock_poll_interval_seconds, + get_writer_lock_timeout_seconds, +) + + +class BackendWriterLockTimeoutError(RuntimeError): + """Raised when the shared backend writer lock cannot be acquired in time.""" + + +_ACTIVE_LOCK_DEPTH_BY_PATH: dict[Path, int] = {} +_ACTIVE_LOCK_TOKEN_BY_PATH: dict[Path, str] = {} + + +def resolve_backend_writer_lock_path(*, storage_path: Path | None = None) -> Path: + """Return the shared lock path derived from the configured SQLite storage path.""" + resolved_storage_path = storage_path or get_storage_path() + return resolved_storage_path.parent / f"{resolved_storage_path.stem}.writer.lock" + + +@contextmanager +def backend_writer_lock( + *, + holder: str, + storage_path: Path | None = None, + timeout_seconds: float | None = None, + poll_interval_seconds: float | None = None, +): + """Acquire the shared backend writer lock with reentrant safety per process.""" + lock_path = resolve_backend_writer_lock_path(storage_path=storage_path).resolve() + if lock_path in _ACTIVE_LOCK_DEPTH_BY_PATH: + _ACTIVE_LOCK_DEPTH_BY_PATH[lock_path] += 1 + try: + yield _read_lock_metadata(lock_path) + finally: + _ACTIVE_LOCK_DEPTH_BY_PATH[lock_path] -= 1 + if _ACTIVE_LOCK_DEPTH_BY_PATH[lock_path] <= 0: + _ACTIVE_LOCK_DEPTH_BY_PATH.pop(lock_path, None) + _ACTIVE_LOCK_TOKEN_BY_PATH.pop(lock_path, None) + return + + metadata = _acquire_backend_writer_lock( + lock_path=lock_path, + holder=holder, + timeout_seconds=get_writer_lock_timeout_seconds() + if timeout_seconds is None + else timeout_seconds, + poll_interval_seconds=get_writer_lock_poll_interval_seconds() + if poll_interval_seconds is None + else poll_interval_seconds, + ) + _ACTIVE_LOCK_DEPTH_BY_PATH[lock_path] = 1 + _ACTIVE_LOCK_TOKEN_BY_PATH[lock_path] = str(metadata["lock_token"]) + try: + yield metadata + finally: + _release_backend_writer_lock(lock_path) + _ACTIVE_LOCK_DEPTH_BY_PATH.pop(lock_path, None) + _ACTIVE_LOCK_TOKEN_BY_PATH.pop(lock_path, None) + + +def build_writer_lock_holder(label: str) -> str: + """Build one readable holder label from the current command line.""" + argv = " ".join(sys.argv).strip() + if argv: + return f"{label} [{argv}]" + return label + + +def _acquire_backend_writer_lock( + *, + lock_path: Path, + holder: str, + timeout_seconds: float, + poll_interval_seconds: float, +) -> dict[str, object]: + if timeout_seconds < 0: + raise ValueError("Writer lock timeout must be zero or positive.") + if poll_interval_seconds <= 0: + raise ValueError("Writer lock poll interval must be positive.") + + lock_path.parent.mkdir(parents=True, exist_ok=True) + deadline = time.monotonic() + timeout_seconds + metadata = _build_lock_metadata(holder=holder) + + while True: + try: + file_descriptor = os.open( + lock_path, + os.O_CREAT | os.O_EXCL | os.O_WRONLY, + ) + except FileExistsError: + existing_metadata = _read_lock_metadata(lock_path) + if _can_clear_stale_lock(existing_metadata): + _remove_lock_file(lock_path) + continue + if time.monotonic() >= deadline: + raise BackendWriterLockTimeoutError( + _build_lock_timeout_message( + lock_path=lock_path, + holder=holder, + timeout_seconds=timeout_seconds, + existing_metadata=existing_metadata, + ) + ) + time.sleep(poll_interval_seconds) + continue + + try: + with os.fdopen(file_descriptor, "w", encoding="utf-8") as handle: + json.dump(metadata, handle, ensure_ascii=True, indent=2) + handle.write("\n") + return metadata + except Exception: + _remove_lock_file(lock_path) + raise + + +def _release_backend_writer_lock(lock_path: Path) -> None: + expected_token = _ACTIVE_LOCK_TOKEN_BY_PATH.get(lock_path) + existing_metadata = _read_lock_metadata(lock_path) + if existing_metadata and expected_token and existing_metadata.get("lock_token") != expected_token: + return + _remove_lock_file(lock_path) + + +def _remove_lock_file(lock_path: Path) -> None: + try: + lock_path.unlink() + except FileNotFoundError: + return + + +def _build_lock_metadata(*, holder: str) -> dict[str, object]: + return { + "lock_token": uuid4().hex, + "holder": holder, + "started_at": _utc_now_iso(), + "hostname": socket.gethostname(), + "pid": os.getpid(), + "cwd": str(Path.cwd()), + } + + +def _read_lock_metadata(lock_path: Path) -> dict[str, object] | None: + try: + return json.loads(lock_path.read_text(encoding="utf-8")) + except (FileNotFoundError, OSError, json.JSONDecodeError): + return None + + +def _can_clear_stale_lock(existing_metadata: dict[str, object] | None) -> bool: + if not existing_metadata: + return False + if str(existing_metadata.get("hostname") or "") != socket.gethostname(): + return False + try: + holder_pid = int(existing_metadata.get("pid")) + except (TypeError, ValueError): + return False + if holder_pid <= 0: + return False + return not _is_process_alive(holder_pid) + + +def _is_process_alive(pid: int) -> bool: + if pid == os.getpid(): + return True + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + except PermissionError: + return True + except OSError as exc: + winerror = getattr(exc, "winerror", None) + if winerror in {3, 87} or exc.errno in {3}: + return False + return True + return True + + +def _build_lock_timeout_message( + *, + lock_path: Path, + holder: str, + timeout_seconds: float, + existing_metadata: dict[str, object] | None, +) -> str: + if not existing_metadata: + return ( + f"Writer lock is busy at {lock_path} and could not be acquired within " + f"{timeout_seconds:.1f}s for {holder}." + ) + + existing_holder = existing_metadata.get("holder") or "unknown-holder" + started_at = existing_metadata.get("started_at") or "unknown-started-at" + hostname = existing_metadata.get("hostname") or "unknown-host" + pid = existing_metadata.get("pid") or "unknown-pid" + return ( + f"Writer lock is busy at {lock_path}. Held by {existing_holder} " + f"since {started_at} on {hostname} (pid {pid}). " + f"Timed out after waiting {timeout_seconds:.1f}s for {holder}." + ) + + +def _utc_now_iso() -> str: + return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")