fix(#20): потокобезопасные чтения, BaseException в apply, old_value внутри транзакции (#54, #55)

Закрывает замечания ревью QA по PR #50: * #54: все чтения Store идут через Database.read() под тем же RLock, что транзакции; контракт «одно соединение, все операции под локом» задокументирован в db.py и README модуля. Устраняет TypeError/IndexError на общем соединении, грязные чтения и выдачу состояния, которое затем откатывается падающим apply. * #55 D-1: apply ловит BaseException — отмена/CancelledError/ KeyboardInterrupt вызывают Applier.revert(previous) и пишут failed в audit_log (без «висящей» записи staged); падение revert попадает в detail. * #55 D-2: set_geoip_schedule читает old_value внутри той же транзакции. Refs #20, Refs #54, Refs #55
2026-10-28 21:12:09 +03:00
parent 7e10ffaf8a
commit 249f32e3f9
4 changed files with 174 additions and 72 deletions
@@ -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
@@ -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]
@@ -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:
@@ -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