rgonsal/comunidadhll
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Harden sqlite writers and add shared lock

Browse Source
This commit is contained in:
devRaGonSa
2026-03-25 12:07:11 +01:00
parent c70073dbe1
commit 02b3bc9ad1
14 changed files with 788 additions and 231 deletions
Download Patch File Download Diff File Expand all files Collapse all files

85
backend/README.md
View File

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

54
backend/app/config.py
View File

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

54
backend/app/historical_ingestion.py
View File

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

26
backend/app/historical_runner.py
View File

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

7
backend/app/historical_storage.py
View File

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

6
backend/app/player_event_storage.py
View File

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

198
backend/app/player_event_worker.py
View File

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

6
backend/app/rcon_historical_storage.py
View File

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

176
backend/app/rcon_historical_worker.py
View File

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

41
backend/app/sqlite_utils.py Normal file
View File

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

9
backend/app/storage.py
View File

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

223
backend/app/writer_lock.py Normal file
View File

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