diff --git a/backend/awg_api/store/README.md b/backend/awg_api/store/README.md index 970014d..631e124 100644 --- a/backend/awg_api/store/README.md +++ b/backend/awg_api/store/README.md @@ -21,10 +21,29 @@ SQLite (WAL) в `/data/awg-dash.db`: правила маршрутизации, - **Пустой `/data`** — каталог создаётся, миграции применяются, схема v1 появляется без ручных действий. - **Версия схемы** — при старте схема проверяется: новее ожидаемой или не мигрированная → `SchemaError` (без «автоотката»). -- **Аудит** — каждая мутация пишет `actor/ts/action/entity/old_value/new_value` в `audit_log` в той же транзакции. -- **Атомарность** — применяемое состояние проходит `validate → stage → apply`; падение внешнего применителя откатывает транзакцию БД и вызывает `Applier.revert(previous)`, `ApplyError.restored=True`. +- **Аудит** — каждая мутация пишет `actor/ts/action/entity/old_value/new_value` в `audit_log` в той же транзакции; `old_value` читается внутри той же транзакции, поэтому конкурентные обновления не рвут цепочку `old→new`. +- **Атомарность** — применяемое состояние проходит `validate → stage → apply`; падение внешнего применителя откатывает транзакцию БД и вызывает `Applier.revert(previous)`, `ApplyError.restored=True`. Ловится `BaseException`, поэтому отмена (`asyncio.CancelledError`), `KeyboardInterrupt` и `SystemExit` тоже дают `revert` и запись `failed` в журнале. - **Откат** — `ApplyEngine.rollback(plan_id)` возвращает состояние из снимка. +## Потокобезопасность (#54) + +Контракт: **одно соединение на процесс, все операции под одним `RLock`**. + +| Операция | API | Поведение | +|---|---|---| +| Мутация | `Database.transaction()` | `BEGIN IMMEDIATE` … `COMMIT`/`ROLLBACK` под локом | +| Чтение | `Database.read()` | берёт тот же лок; ждёт завершения чужой транзакции | +| Прямой доступ | `Database.connection` / `Store.raw()` | лок **не** берётся — только однопоточный код (миграции, тесты, отладка) | + +Следствия, на которые можно опираться в API-слое (#19/#21): + +- обычные чтения (`GET /rules`, `GET /mode`, …) не падают от конкурентных мутаций и не отдают перемешанный результат операторов; +- читатель **не видит** незафиксированных строк другой транзакции; +- во время `apply` наружу не видно состояние, которое затем будет откачено (AC «не частично применённая конфигурация»); +- конкурентные запросы ждут, а не получают ошибку; потокобезопасность подтверждена `backend/tests/test_store_concurrency.py`. + +WAL остаётся включённым: он даёт устойчивость к падению процесса и позволяет читать БД снаружи (`sqlite3` CLI) во время работы шлюза. Внутрипроцессного параллелизма чтений контракт не обещает: если позже понадобится параллельная пропускная способность, меняем контракт на «соединение на поток» (`threading.local`), а не полагаемся на неявное поведение. + ## Пример ```python diff --git a/backend/awg_api/store/apply.py b/backend/awg_api/store/apply.py index df239c8..3ecc2fe 100644 --- a/backend/awg_api/store/apply.py +++ b/backend/awg_api/store/apply.py @@ -12,13 +12,15 @@ * **rollback** — явный возврат к снимку по id. Ошибка применения поднимает :class:`ApplyError` (поле ``restored=True``): API -должен ответить ошибкой, не «полуприменённой» конфигурацией. +должен ответить ошибкой, не «полуприменённой» конфигурацией. Ловится +``BaseException`` (#55 D-1), поэтому отмена запроса тоже вызывает +``Applier.revert(previous)`` и закрывает запись аудита исходом ``failed`` — +«висящих» записей ``staged`` без исхода не остаётся. """ from __future__ import annotations from collections.abc import Mapping -from contextlib import suppress from dataclasses import dataclass, field from typing import Any, Protocol, runtime_checkable @@ -229,6 +231,15 @@ class ApplyEngine: :raises ApplyError: если применитель упал; ``restored=True`` означает, что предыдущее состояние подтверждённо восстановлено. + + Ловится ``BaseException``, не только ``Exception`` (#55 D-1): отмена + запроса (``asyncio.CancelledError``), ``KeyboardInterrupt`` и + ``SystemExit`` тоже обязаны вернуть внешнюю подсистему в предыдущее + состояние и закрыть запись аудита исходом ``failed``. Иначе iptables + останется в новом состоянии, тогда как ``audit_log`` сохранит + «висящую» запись ``staged`` без исхода. Исключения из ``Exception`` + преобразуются в :class:`ApplyError`, остальные пробрасываются как есть + (семантика отмены сохраняется). """ payload = plan.desired.to_payload() audit_id: int @@ -247,8 +258,10 @@ class ApplyEngine: status=AUDIT_STATUS_OK, detail=plan.reason, ) - except Exception as exc: + except BaseException as exc: restored_ok = self._rollback_failed_apply(plan, exc) + if not isinstance(exc, Exception): + raise raise ApplyError( f"применение состояния не удалось: {exc}", restored=restored_ok, @@ -269,10 +282,22 @@ class ApplyEngine: return state.rollback_to_snapshot(conn, plan_id, actor=self._actor, reason=reason) # -- внутреннее ----------------------------------------------------- - def _rollback_failed_apply(self, plan: StagedPlan, exc: Exception) -> bool: - """Вернуть внешнюю подсистему и убедиться, что БД = предыдущее состояние.""" - with suppress(Exception): # откат внешней системы — best-effort + def _rollback_failed_apply(self, plan: StagedPlan, exc: BaseException) -> bool: + """Вернуть внешнюю подсистему и убедиться, что БД = предыдущее состояние. + + Откат внешней подсистемы — best-effort: падение ``revert`` не мешает + восстановить БД и записать исход в журнал (причина падения revert + попадает в ``detail``), и наоборот — падение revert не должно скрывать + исходное исключение. + """ + revert_error: BaseException | None = None + try: self._applier.revert(plan.previous) + except BaseException as revert_exc: # откат best-effort: исход важнее падения revert + revert_error = revert_exc + detail = f"{type(exc).__name__}: {exc}" + if revert_error is not None: + detail += f"; revert: {type(revert_error).__name__}: {revert_error}" with self._store.database.transaction() as conn: state.restore_state(conn, plan.previous) audit.record( @@ -284,14 +309,16 @@ class ApplyEngine: old_value=plan.desired.to_payload(), new_value=plan.previous, status=AUDIT_STATUS_FAILED, - detail=f"{type(exc).__name__}: {exc}", + detail=detail, ) - return state.capture_state(self._store.raw()) == plan.previous + with self._store.database.read() as conn: + return state.capture_state(conn) == plan.previous def _last_failed_audit_id(self, plan: StagedPlan) -> int | None: - entries = audit.list_entries( - self._store.raw(), entity=Entity.APPLY.value, entity_id=plan.plan_id, limit=1 - ) + with self._store.database.read() as conn: + entries = audit.list_entries( + conn, entity=Entity.APPLY.value, entity_id=plan.plan_id, limit=1 + ) if not entries: return None last = entries[0] diff --git a/backend/awg_api/store/db.py b/backend/awg_api/store/db.py index 8d0743a..bf03f2d 100644 --- a/backend/awg_api/store/db.py +++ b/backend/awg_api/store/db.py @@ -5,9 +5,25 @@ Каталог создаётся автоматически, поэтому на пустом ``/data`` схема v1 поднимается без ручных действий. -Соединение открывается в режиме WAL: читатели не блокируют писателя, -``synchronous=NORMAL`` достаточен для WAL (риск потери только последних -транзакций при падении операционной системы, БД при этом не разрушается). +Соединение открывается в режиме WAL: ``synchronous=NORMAL`` достаточен для WAL +(риск потери только последних транзакций при падении операционной системы, БД +при этом не разрушается). + +Контракт потокобезопасности (#54): соединение одно на процесс, все операции — +и записи (:meth:`Database.transaction`), и чтения (:meth:`Database.read`) — +выполняются под одним ``RLock``. Поэтому: + +* результаты операторов из разных потоков не перемешиваются (нет + ``TypeError``/``IndexError``); +* читатель не видит незафиксированных строк чужой транзакции и состояния, + которое затем будет откачено (падение внешнего применителя); +* конкурентные запросы ждут, не падают. + +WAL при этом остаётся: он даёт устойчивость к падению процесса и позволяет +читать БД снаружи (``sqlite3`` CLI) во время работы шлюза. Если позже +понадобится параллельная пропускная способность чтений, контракт меняется на +«соединение на поток» (``threading.local``) — это отдельное решение, не +молчаливое поведение. """ from __future__ import annotations @@ -83,12 +99,18 @@ class Database: @property def connection(self) -> sqlite3.Connection: - """Открытое соединение (для сервисного слоя).""" + """Прямой доступ к соединению — только для однопоточного кода. + + Рантайм API/сервисного слоя использует :meth:`read` (чтения) и + :meth:`transaction` (мутации): они берут общий лок и гарантируют + согласованность результата (#54). + """ return self._conn @property def journal_mode(self) -> str: - row = self._conn.execute("PRAGMA journal_mode").fetchone() + with self._lock: + row = self._conn.execute("PRAGMA journal_mode").fetchone() return str(row[0]) if row is not None else "" @property @@ -97,7 +119,8 @@ class Database: @property def schema_version(self) -> int: - return current_version(self._conn) + with self._lock: + return current_version(self._conn) # -- схема ---------------------------------------------------------- def migrate(self) -> MigrationReport: @@ -109,9 +132,10 @@ class Database: def verify_schema(self) -> None: """Проверить соответствие фактической схемы ожидаемой версии.""" - verify_schema(self._conn) + with self._lock: + verify_schema(self._conn) - # -- транзакции ----------------------------------------------------- + # -- транзакции и чтения -------------------------------------------- @contextmanager def transaction(self) -> Iterator[sqlite3.Connection]: """Транзакция ALL-OR-NOTHING (``BEGIN IMMEDIATE`` + rollback при ошибке).""" @@ -125,6 +149,18 @@ class Database: else: self._conn.commit() + @contextmanager + def read(self) -> Iterator[sqlite3.Connection]: + """Согласованное чтение под общим локом (#54). + + Ждёт завершения чужой транзакции и потому не видит незафиксированных + строк и состояния, которое затем будет откачено. Внутри уже открытой + транзакции того же потока лок реентерабелен, поэтому читать можно и из + неё. + """ + with self._lock: + yield self._conn + def close(self) -> None: with self._lock: try: diff --git a/backend/awg_api/store/repository.py b/backend/awg_api/store/repository.py index 7faaa13..a21dcb2 100644 --- a/backend/awg_api/store/repository.py +++ b/backend/awg_api/store/repository.py @@ -1,8 +1,10 @@ """Репозиторий сущностей состояния (issue #20). Каждая мутация выполняется в одной транзакции, запись в ``audit_log`` -(«кто/когда/что/old→new») коммитится там же. Чтения идут по соединению WAL -и не блокируют писателя. +(«кто/когда/что/old→new») коммитится там же. Чтения идут через +:meth:`Database.read` — под тем же ``RLock``, что и транзакции (#54), поэтому +читатель не видит незафиксированных строк и состояния, которое затем будет +откачено, и не может получить перемешанный результат операторов другого потока. """ from __future__ import annotations @@ -79,6 +81,20 @@ def _device_from_row(row: sqlite3.Row) -> Device: ) +def _geoip_from_row(conn: sqlite3.Connection) -> GeoIpSchedule: + row = conn.execute( + "SELECT enabled, cron, provider, updated_at FROM geoip_schedule WHERE id = 1" + ).fetchone() + if row is None: + return GeoIpSchedule() + return GeoIpSchedule( + enabled=bool(row["enabled"]), + cron=str(row["cron"]), + provider=str(row["provider"]), + updated_at=str(row["updated_at"]), + ) + + class Store: """Доступ к состоянию: чтение + аудируемые мутации.""" @@ -90,80 +106,75 @@ class Store: return self._db # -- чтение --------------------------------------------------------- + # Чтения идут через Database.read() под тем же RLock, что транзакции + # (#54): это исключает перемешивание операторов разных потоков и грязные + # чтения незафиксированных строк. def get_mode(self) -> Mode: - row = self._db.connection.execute("SELECT mode FROM routing_mode WHERE id = 1").fetchone() + with self._db.read() as conn: + row = conn.execute("SELECT mode FROM routing_mode WHERE id = 1").fetchone() if row is None: return Mode.OFF return Mode(str(row["mode"])) def get_geoip_schedule(self) -> GeoIpSchedule: - row = self._db.connection.execute( - "SELECT enabled, cron, provider, updated_at FROM geoip_schedule WHERE id = 1" - ).fetchone() - if row is None: - return GeoIpSchedule() - return GeoIpSchedule( - enabled=bool(row["enabled"]), - cron=str(row["cron"]), - provider=str(row["provider"]), - updated_at=str(row["updated_at"]), - ) + with self._db.read() as conn: + return _geoip_from_row(conn) def list_rules(self, *, enabled_only: bool = False) -> list[Rule]: sql = "SELECT * FROM routing_rules" if enabled_only: sql += " WHERE enabled = 1" sql += " ORDER BY priority, id" - return [_rule_from_row(row) for row in self._db.connection.execute(sql).fetchall()] + with self._db.read() as conn: + rows = conn.execute(sql).fetchall() + return [_rule_from_row(row) for row in rows] def get_rule(self, rule_id: int) -> Rule | None: - row = self._db.connection.execute( - "SELECT * FROM routing_rules WHERE id = ?", (rule_id,) - ).fetchone() + with self._db.read() as conn: + row = conn.execute("SELECT * FROM routing_rules WHERE id = ?", (rule_id,)).fetchone() return None if row is None else _rule_from_row(row) def get_rule_by_name(self, name: str) -> Rule | None: - row = self._db.connection.execute( - "SELECT * FROM routing_rules WHERE name = ?", (name,) - ).fetchone() + with self._db.read() as conn: + row = conn.execute("SELECT * FROM routing_rules WHERE name = ?", (name,)).fetchone() return None if row is None else _rule_from_row(row) def list_exceptions(self) -> list[RuleException]: - rows = self._db.connection.execute("SELECT * FROM rule_exceptions ORDER BY id").fetchall() + with self._db.read() as conn: + rows = conn.execute("SELECT * FROM rule_exceptions ORDER BY id").fetchall() return [_exception_from_row(row) for row in rows] def get_exception(self, match: str) -> RuleException | None: - row = self._db.connection.execute( - "SELECT * FROM rule_exceptions WHERE match = ?", (match,) - ).fetchone() + with self._db.read() as conn: + row = conn.execute("SELECT * FROM rule_exceptions WHERE match = ?", (match,)).fetchone() return None if row is None else _exception_from_row(row) def list_configs(self) -> list[ConfigMeta]: - rows = self._db.connection.execute("SELECT * FROM configs ORDER BY name").fetchall() + with self._db.read() as conn: + rows = conn.execute("SELECT * FROM configs ORDER BY name").fetchall() return [_config_from_row(row) for row in rows] def get_config(self, name: str) -> ConfigMeta | None: - row = self._db.connection.execute( - "SELECT * FROM configs WHERE name = ?", (name,) - ).fetchone() + with self._db.read() as conn: + row = conn.execute("SELECT * FROM configs WHERE name = ?", (name,)).fetchone() return None if row is None else _config_from_row(row) def list_devices(self) -> list[Device]: - rows = self._db.connection.execute("SELECT * FROM devices ORDER BY name").fetchall() + with self._db.read() as conn: + rows = conn.execute("SELECT * FROM devices ORDER BY name").fetchall() return [_device_from_row(row) for row in rows] def get_device(self, mac: str) -> Device | None: - row = self._db.connection.execute( - "SELECT * FROM devices WHERE mac = ?", (validation.normalize_mac(mac),) - ).fetchone() + normalized = validation.normalize_mac(mac) + with self._db.read() as conn: + row = conn.execute("SELECT * FROM devices WHERE mac = ?", (normalized,)).fetchone() return None if row is None else _device_from_row(row) def list_audit( self, *, entity: str | None = None, entity_id: str | int | None = None, limit: int = 100 ) -> list[AuditEntry]: - return audit.list_entries( - self._db.connection, entity=entity, entity_id=entity_id, limit=limit - ) + with self._db.read() as conn: + return audit.list_entries(conn, entity=entity, entity_id=entity_id, limit=limit) # -- мутации: правила ------------------------------------------------ def upsert_rule(self, rule: Rule, *, actor: str) -> Rule: @@ -358,22 +369,25 @@ class Store: actor: str, ) -> GeoIpSchedule: """Обновить расписание GeoIP (частичное обновление) и записать аудит.""" - current = self.get_geoip_schedule() - updated = GeoIpSchedule( - enabled=current.enabled if enabled is None else enabled, - cron=validation.validate_cron(current.cron if cron is None else cron), - provider=validation.validate_provider( - current.provider if provider is None else provider - ), - ) now = utcnow() with self._db.transaction() as conn: + # чтение old_value — внутри той же транзакции, что и запись (#55 D-2): + # иначе при конкурентном обновлении журнал заявит устаревшее «старое». + current = _geoip_from_row(conn) + stored = GeoIpSchedule( + enabled=current.enabled if enabled is None else enabled, + cron=validation.validate_cron(current.cron if cron is None else cron), + provider=validation.validate_provider( + current.provider if provider is None else provider + ), + updated_at=now, + ) conn.execute( "INSERT INTO geoip_schedule (id, enabled, cron, provider, updated_at)" " VALUES (1, ?, ?, ?, ?)" " ON CONFLICT(id) DO UPDATE SET enabled = excluded.enabled, cron = excluded.cron," " provider = excluded.provider, updated_at = excluded.updated_at", - (1 if updated.enabled else 0, updated.cron, updated.provider, now), + (1 if stored.enabled else 0, stored.cron, stored.provider, now), ) audit.record( conn, @@ -382,9 +396,9 @@ class Store: entity=Entity.GEOIP_SCHEDULE.value, entity_id=1, old_value=current.as_dict(), - new_value=updated.as_dict(), + new_value=stored.as_dict(), ) - return replace(updated, updated_at=now) + return stored # -- мутации: конфиги ------------------------------------------------- def upsert_config(self, config: ConfigMeta, *, actor: str) -> ConfigMeta: @@ -569,11 +583,17 @@ class Store: "snapshots": "state_snapshots", } result: dict[str, int] = {} - for key, table in tables.items(): - row = self._db.connection.execute(f"SELECT COUNT(*) FROM {table}").fetchone() # noqa: S608 - result[key] = int(row[0]) if row is not None else 0 + with self._db.read() as conn: # один снимок на все таблицы: счётчики согласованы + for key, table in tables.items(): + row = conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone() # noqa: S608 + result[key] = int(row[0]) if row is not None else 0 return result def raw(self) -> sqlite3.Connection: - """Соединение для кода, которому нужен доступ ниже уровня репозитория.""" + """Соединение для кода, которому нужен доступ ниже уровня репозитория. + + Лок не берётся: в рантайме используйте :meth:`Database.read` / + :meth:`Database.transaction` (#54), иначе возможны гонки на общем + соединении. Метод оставлен для однопоточного кода и миграций. + """ return self._db.connection