Merge pull request 'МЕРА: у запросов к БД появился потолок по времени и по ожиданию блокировки (#3463)' (#3508) from fix/3463-db-statement-timeout into main

Правки по deep-ревью PR #3508.

1. HIGH. `app/core/auth_db.py` строил второй движок БЕЗ `connect_args`, а моё
обоснование пропуска было ложным: на проде `IDENTITY_STORE=auth` во всех трёх
сервисах образа и `AUTH_DB_PASSWORD` задан (сверено `printenv` в контейнерах),
то есть реестр живой. Путь горячий: `core/rbac.py` резолвит session-cookie в
middleware, синхронно на event loop'е, на каждом запросе с cookie — значит
`ACCESS EXCLUSIVE` на `auth.sessions` вешал бы не четыре слота `/estimate`, а
весь uvicorn-воркер (он один), включая `/health`. Потолок — та же константа
`DB_CONNECT_ARGS`: одна на оба движка, а не защита на одном и мина на втором.
Срабатывание безопасно — вызов уже под `except Exception` с фолбэком.

2. MEDIUM. Испорченный `options` (`statement_timeout=30000zz`) проходил ЗЕЛЁНЫМ:
статическая проверка искала ПОДСТРОКУ (а `…=30000` — префикс испорченного), а
живая глушила отказ коннекта голым `except` → skip → запись в allowlist. На
проде это `FATAL: invalid value for parameter` на КАЖДОМ коннекте, то есть
полный отказ продукта при зелёном сьюте. Теперь: сравнение `options` на
РАВЕНСТВО, и `_live_engine` сначала пробует коннект БЕЗ `connect_args` — сервера
нет это пропуск, а «сервер есть, наши options он не принял» это падение.

3. LOW. У проверки согласованности был пол и не было крыши: `300_000` (пять
минут) зеленел. Добавлена симметричная граница `<= 2 ×` самого длинного
объявленного бюджета.

4. LOW. Три факта в комментариях исправлены:
   * `pg_stat_statements` НЕ опора по планировщику — вытесняет записи с calls=1
     (`dealloc` вырос за десять минут, из топа пропал `REFRESH MATERIALIZED VIEW`
     30.85 с). Основная опора — `scrape_runs`;
   * самый длинный set-based statement через движок — матч ГАР→houses: 2.07 с с
     городским фильтром и 6.46 с без. Запас ~3×, а не 7×;
   * `idle in transaction` 29 с — это tgbot (`services/tgbot/bridge.py`,
     транзакция поверх long-poll Telegram; сверено 3 пробами: одна и та же
     сессия, `SELECT value FROM tg_support_state …`), а не свипы. Решение не
     ставить потолок на простой от этого только крепче.

Мутационная проверка (обе лэйны краснеют на каждой): испорченный `options`,
`_STATEMENT_TIMEOUT_MS = 300_000`, снятый `connect_args`.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

tradein-mvp/backend/app/core/auth_db.py

@@ -47,6 +47,7 @@ from sqlalchemy.exc import ArgumentError
 from sqlalchemy.orm import Session, sessionmaker
 
 from app.core.config import settings
+from app.core.db import DB_CONNECT_ARGS
 
 
 class AuthDatabaseNotConfiguredError(RuntimeError):
@@ -101,6 +102,18 @@ def _build() -> tuple[Engine, sessionmaker[Session]]:
             # НЕ закрывает: текст ошибки самого драйвера (Postgres DETAIL со значением)
             # и сырые psycopg-подключения мимо движков — это отдельный класс.
             hide_parameters=True,
+            # #3463. Те же потолки, что у продуктового движка, — ОДНОЙ константой на оба:
+            # потолок на одном движке и мина на втором это не починка, а половина.
+            # Этот движок живёт на ГОРЯЧЕМ пути: `core/rbac.py` резолвит session-cookie
+            # в middleware, синхронно на event loop'е, на КАЖДОМ запросе с cookie
+            # (на проде IDENTITY_STORE=auth во всех трёх сервисах образа — сверено 12.09,
+            # `printenv` в контейнерах). Без потолка `ACCESS EXCLUSIVE` на `auth.sessions`
+            # вешает не четыре слота `/estimate`, а весь uvicorn-воркер (он один, без
+            # --workers) — включая `/health`.
+            # Срабатывание потолка безопасно: вызов в rbac.py уже под `except Exception`
+            # с фолбэком на заголовочную аутентификацию, то есть отмена запроса даёт тот
+            # же путь, что и любой другой сбой реестра, а не 500.
+            connect_args=DB_CONNECT_ARGS,
         )
     except (ArgumentError, ValueError):
         # ValueError — не паранойя: на «почти URL» разбор SQLAlchemy доходит до

tradein-mvp/backend/app/core/db.py

@@ -10,10 +10,59 @@ from app.core.config import settings
 
 logger = logging.getLogger(__name__)
 
+# #3463. Потолок ОДНОГО statement'а, секунды·1000. Ставится на КОННЕКТЕ (libpq
+# `options`), а не в питоновской обёртке: обёртка (`run_db_thread` ниже) при
+# отмене обязана ДОЖДАТЬСЯ потока, иначе поток остаётся сиротой в общей
+# `Session` — ровно то, ради чего писался #3449. Значит верхняя граница ожидания
+# = длительность самого запроса, и задать её может только сервер.
+#
+# 30 с выбраны так, чтобы потолок НИКОГДА не стал биндящим ограничением для
+# честной работы, но остался конечным:
+#   * самый длинный ОБЪЯВЛЕННЫЙ бюджет на `/estimate` — 20 с (`estimate_avito_imv_timeout_s`,
+#     config.py:852); дальше 12 с геокод, 8 с Yandex/Cian/house_meta. 30 с = 1.5× от максимума;
+#   * ОСНОВНАЯ опора по планировщику — `scrape_runs` (длительности целых прогонов, они
+#     не вытесняются): самая долгая ЧИСТО-БД задача за 14 суток — listing_source_snapshot,
+#     9.7 с ЦЕЛИКОМ (и у неё сверх того свой `SET LOCAL statement_timeout = 900000`,
+#     который перекрывает это значение — гейт tests/test_3463_db_timeouts.py);
+#   * самый длинный set-based statement ЧЕРЕЗ движок из замеренных — матч ГАР→houses
+#     (`services/gar_flats_loader._MATCH_SQL`): 2.07 с с городским фильтром и 6.46 с без
+#     него (`city_filter=None`, флаг CLI). Запас ~3×, и это СЧИТАЮЩИЙ запрос, а не ждущий.
+#
+# `pg_stat_statements` опорой по планировщику НЕ является: при `max = 5000` он вытесняет
+# редкие записи (проверено 12.09 — `dealloc` вырос на единицу за десять минут, и из топа
+# пропали ВСЕ записи с `calls = 1`, включая `REFRESH MATERIALIZED VIEW` 30.85 с и KNN
+# `cadastral_geo_match` 2.45 с). Суточная задача до следующих суток там не доживает, так
+# что «самый долгий запрос 4.27 с» верно только для ВЫСОКОЧАСТОТНЫХ запросов.
+_STATEMENT_TIMEOUT_MS = 30_000
+
+# Ожидание БЛОКИРОВКИ — заведомо меньше: ждать лок дольше секунд смысла нет, лучше
+# деградировать. 5 с — та же величина, что у миграций проекта
+# (`SET LOCAL lock_timeout = '5s'` в data/sql/250,251,260,272,277…), снизу ограничена
+# deadlock_timeout (на проде 1 с — сверено 12.09). Именно этот потолок закрывает
+# сценарий #3463: под `ACCESS EXCLUSIVE` на `geocode_cache` запрос ЖДЁТ лок, а не
+# считает, — statement_timeout тут только страховка от «считает вечно».
+_LOCK_TIMEOUT_MS = 5_000
+
+# idle_in_transaction_session_timeout НАМЕРЕННО не трогаем: тем же движком живёт tgbot,
+# и `services/tgbot/bridge.py` держит транзакцию открытой ПОВЕРХ long-poll Telegram
+# (замер на проде 12.09, 3 пробы с шагом 7 с: одна и та же сессия, запрос
+# `SELECT value FROM tg_support_state …`, возраст транзакции циклически растёт до ~29 с).
+# Сессионный потолок на простой в транзакции ронял бы long-poll КАЖДЫЙ цикл —
+# гарантированно, а не в редком случае.
+DB_CONNECT_ARGS = {
+    "options": f"-c statement_timeout={_STATEMENT_TIMEOUT_MS} -c lock_timeout={_LOCK_TIMEOUT_MS}"
+}
+
 engine = create_engine(
     settings.database_url,
     pool_pre_ping=True,
     future=True,
+    # #3463. Накрывает ВСЕ три сервиса образа (backend / scraper / tgbot — один и тот
+    # же `app.core.db`, см. docker-compose.prod.yml) и обе стороны: продуктовый путь
+    # `/estimate` и задачи планировщика. Миграции идут мимо (psql из
+    # .forgejo/workflows/deploy-tradein.yml, не этот движок) — их DDL под своим
+    # `SET LOCAL lock_timeout` и потолком не ограничен.
+    connect_args=DB_CONNECT_ARGS,
     # #3194: SQLAlchemy печатает ВСЕ bind-параметры в тексте StatementError —
     # через них в GlitchTip уезжали ключ шифрования кук и сами куки
     # (pgp_sym_encrypt(:cookies_json, :key)). Флаг на УРОВНЕ ДВИЖКА кроет все

tradein-mvp/backend/tests/skip_allowlist.txt

@@ -129,3 +129,18 @@ tests/test_revisit_floor_lateral_lookup.py::test_missing_history_before_anchor_s
 # (postgres-сервис) — там прогон и был зелёным; в deploy-tradein.yml БД нет вовсе.
 tests/test_3063_seller_fields_not_eroded.py::test_poor_rescrape_does_not_erase_seller_fields
 tests/test_3063_seller_fields_not_eroded.py::test_real_change_still_overwrites
+
+# Потолки БД на движке (#3463) — поведенческая половина: проверяют, что запрос
+# длиннее потолка ОБРЫВАЕТСЯ (pg_sleep), что ожидание блокировки отваливается по
+# lock_timeout, и что `SET LOCAL` задачи планировщика перекрывает сессионный
+# потолок и не течёт за свою транзакцию. Всё это нельзя проверить без сервера:
+# потолок применяет Postgres, а не питон. Статическая половина того же файла
+# (test_engine_opens_connections_with_both_ceilings +
+# test_ceilings_are_coherent_with_declared_estimate_budgets) идёт на ОБОИХ лэйнах
+# и краснеет от снятия `connect_args` — проверено вручную 12.09. В ci-tradein.yml
+# эти пять бегут по-настоящему (postgres-сервис, #2745).
+tests/test_3463_db_timeouts.py::test_live_session_reports_both_ceilings
+tests/test_3463_db_timeouts.py::test_statement_over_ceiling_is_cancelled_not_hung
+tests/test_3463_db_timeouts.py::test_lock_wait_over_ceiling_is_aborted
+tests/test_3463_db_timeouts.py::test_set_local_statement_timeout_overrides_session_ceiling
+tests/test_3463_db_timeouts.py::test_set_local_is_scoped_to_its_transaction

tradein-mvp/backend/tests/test_3463_db_timeouts.py

@@ -0,0 +1,276 @@
+"""#3463 — у запросов к БД обязан быть потолок по времени и по ожиданию блокировки.
+
+Отказ, который тут закрывается (см. issue): под `ACCESS EXCLUSIVE` на таблице шаг БД
+на пути `/estimate` ЖДЁТ блокировку; бюджет источника истекает, обёртка `run_db_thread`
+уходит ждать свой поток (иначе он останется сиротой в общей `Session` — #3449), а
+верхняя граница этого ожидания = длительность самого запроса. Границы у запроса не
+было → слот `_estimate_slots` не возвращался → `_ESTIMATE_CONCURRENCY = 4` исчерпывался
+и `/estimate` отдавал 429 всем остальным.
+
+Потолок поэтому стоит на КОННЕКТЕ (libpq `options`), а не в питоновской обёртке:
+таймаут в обёртке вернул бы ровно ту сироту, ради которой писался #3449.
+
+Проверки по значению, а не по тексту:
+  * потолки реально доехали до параметров подключения движка и согласованы с
+    объявленными бюджетами `/estimate` (без БД — падают от снятия `connect_args`);
+  * живая сессия этого движка сообщает оба потолка (без БД пропускается);
+  * запрос длиннее потолка ОБРЫВАЕТСЯ за отведённое время, а не висит;
+  * ожидание блокировки длиннее потолка обрывается — это и есть сценарий #3463;
+  * задача со своим `SET LOCAL statement_timeout` (планировщик: 900 с в
+    `app/tasks/listing_source_snapshot.py`) новым сессионным потолком НЕ обрезается.
+"""
+
+from __future__ import annotations
+
+import inspect
+import os
+import time
+from collections.abc import Iterator
+
+import psycopg
+import pytest
+from sqlalchemy import Engine, create_engine, text
+from sqlalchemy.exc import OperationalError
+
+from app.core.db import _LOCK_TIMEOUT_MS, _STATEMENT_TIMEOUT_MS, DB_CONNECT_ARGS, engine
+
+# Потолки для ПОВЕДЕНЧЕСКИХ проверок — намеренно маленькие: проверяется механизм
+# (потолок из `options` реально обрывает запрос / ожидание лока), а прод-ВЕЛИЧИНЫ
+# проверяет `test_live_session_reports_both_ceilings` на самом прод-движке. Иначе
+# каждый прогон сьюта стоил бы 30 с ожидания pg_sleep.
+_PROBE_STATEMENT_TIMEOUT_MS = 1_000
+_PROBE_LOCK_TIMEOUT_MS = 500
+
+_LOCK_PROBE_TABLE = "t3463_lock_probe"
+
+
+def _engine_connect_options(eng: Engine) -> str:
+    """Строка libpq `options`, с которой движок РЕАЛЬНО открывает коннекты.
+
+    `connect_args` в движке не хранятся полем: `create_engine` вливает их в `cparams`
+    замыкания `pool._creator`. Читаем оттуда, а не из `DB_CONNECT_ARGS`, — иначе тест
+    остался бы зелёным после снятия `connect_args=` у `create_engine`.
+
+    Наружу отдаём ТОЛЬКО `options`: в `cparams` лежит пароль роли, и текст
+    провалившегося assert'а уехал бы с ним в лог CI.
+    """
+    creator = getattr(eng.pool, "_creator", None)
+    assert creator is not None, "у пула движка нет _creator — SQLAlchemy сменила устройство"
+    cparams = inspect.getclosurevars(creator).nonlocals.get("cparams")
+    assert cparams is not None, (
+        "в замыкании pool._creator нет cparams — SQLAlchemy сменила устройство, "
+        "проверку параметров подключения надо переписать, а не удалять"
+    )
+    return str(cparams.get("options", ""))
+
+
+def _live_engine(connect_args: dict[str, str]) -> Engine | None:
+    """Движок против живой Postgres с заданными `connect_args`, иначе None.
+
+    Тот же способ добыть DSN, что у `_live_session()` в tests/test_house_dedup_merge.py
+    и tests/test_purge_expired_trade_in_data.py: в CI Postgres есть (ci-tradein.yml),
+    на ноутбуке без БД тест пропускается (учтён в tests/skip_allowlist.txt).
+    """
+    dsn = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL", "")
+    if not dsn or "localhost:5432/test" in dsn:
+        return None
+
+    # Сначала проба БЕЗ connect_args: она отделяет «сервера нет» (честный пропуск)
+    # от «сервер есть, но наши `options` он не принял». Глушить второе нельзя —
+    # именно так испорченное значение (`statement_timeout=30000zz`) проходило
+    # зелёным: коннект падал, тест пропускался, запись в allowlist гасила сигнал,
+    # а на проде это FATAL на КАЖДОМ коннекте.
+    try:
+        probe = create_engine(dsn, future=True)
+    except Exception:
+        return None
+    try:
+        with probe.connect() as conn:
+            conn.execute(text("SELECT 1"))
+    except Exception:
+        return None
+    finally:
+        probe.dispose()
+
+    eng = create_engine(dsn, future=True, connect_args=connect_args)
+    with eng.connect() as conn:  # НЕ под except: сервер живой, виноваты connect_args
+        conn.execute(text("SELECT 1"))
+    return eng
+
+
+@pytest.fixture
+def probe_engine() -> Iterator[Engine]:
+    """Живой движок с МАЛЕНЬКИМИ потолками — проверка механизма, не прод-величин."""
+    eng = _live_engine(
+        {
+            "options": (
+                f"-c statement_timeout={_PROBE_STATEMENT_TIMEOUT_MS} "
+                f"-c lock_timeout={_PROBE_LOCK_TIMEOUT_MS}"
+            )
+        }
+    )
+    if eng is None:
+        pytest.skip("живой Postgres недоступен (DATABASE_URL-заглушка)")
+    try:
+        yield eng
+    finally:
+        eng.dispose()
+
+
+# ── Проводка и согласованность величин (без БД) ───────────────────────────────
+
+
+def test_engine_opens_connections_with_both_ceilings() -> None:
+    """Оба потолка доехали до параметров подключения ПРОД-движка.
+
+    Фальсификация: убрать `connect_args=DB_CONNECT_ARGS` из `create_engine` —
+    `options` станет пустой, тест краснеет.
+    """
+    expected = f"-c statement_timeout={_STATEMENT_TIMEOUT_MS} -c lock_timeout={_LOCK_TIMEOUT_MS}"
+    options = _engine_connect_options(engine)
+    # РАВЕНСТВО, а не `in`: подстрочная проверка пропускала испорченный хвост
+    # (`…=30000zz` содержит `…=30000`), а Postgres на такое значение отвечает
+    # `FATAL: invalid value for parameter "statement_timeout"` — ни одного коннекта
+    # ни в одном из трёх сервисов образа, полный отказ продукта.
+    assert options == expected, (
+        f"движок открывает коннекты с options={options!r}, ожидалось {expected!r}: "
+        "либо потолка нет вовсе (заблокированный запрос снова висит и жжёт слот "
+        "/estimate, #3463), либо значение испорчено — тогда libpq отвергнет КАЖДЫЙ коннект"
+    )
+    assert DB_CONNECT_ARGS["options"] == options
+
+
+def test_ceilings_are_coherent_with_declared_estimate_budgets() -> None:
+    """Потолок выше САМОГО ДЛИННОГО объявленного бюджета `/estimate`, но конечен."""
+    from app.core.config import settings
+
+    declared_budgets_s = (
+        settings.estimate_avito_imv_timeout_s,  # 20 с — самый длинный
+        settings.estimate_geocode_budget_s,  # 12 с
+        settings.estimate_house_meta_timeout_s,  # 8 с
+        settings.estimate_yandex_valuation_timeout_s,  # 8 с
+        settings.estimate_cian_valuation_timeout_s,  # 8 с
+    )
+    longest_ms = max(declared_budgets_s) * 1000
+    assert _STATEMENT_TIMEOUT_MS > longest_ms, (
+        f"потолок запроса {_STATEMENT_TIMEOUT_MS} мс не выше самого длинного объявленного "
+        f"бюджета {longest_ms:.0f} мс — потолок стал бы биндящим ограничением честной работы"
+    )
+    # Потолок сверху — иначе у проверки есть пол и нет крыши: `_STATEMENT_TIMEOUT_MS =
+    # 300_000` (пять минут) зеленел бы, а пять минут ожидания это тот же отказ, только
+    # медленнее: четыре таких запроса всё так же выедают `_ESTIMATE_CONCURRENCY`.
+    assert _STATEMENT_TIMEOUT_MS <= 2 * longest_ms, (
+        f"потолок запроса {_STATEMENT_TIMEOUT_MS} мс больше чем вдвое превышает самый "
+        f"длинный объявленный бюджет {longest_ms:.0f} мс — это уже не защита, а отсрочка: "
+        "слот /estimate держится всё это время"
+    )
+    assert 0 < _LOCK_TIMEOUT_MS < _STATEMENT_TIMEOUT_MS, (
+        "ожидание блокировки обязано обрываться РАНЬШЕ потолка на сам запрос: "
+        "деградировать лучше, чем держать слот"
+    )
+
+
+# ── Поведение против живой Postgres ──────────────────────────────────────────
+
+
+def test_live_session_reports_both_ceilings() -> None:
+    """Прод-ВЕЛИЧИНЫ на живой сессии ПРОД-движка: сервер их принял, а не проигнорировал.
+
+    Коннект берётся у самого `app.core.db.engine`, а не у собранного здесь двойника:
+    двойник остался бы зелёным после снятия `connect_args=` в `create_engine`.
+    """
+    if _live_engine(DB_CONNECT_ARGS) is None:
+        pytest.skip("живой Postgres недоступен (DATABASE_URL-заглушка)")
+    with engine.connect() as conn:
+        statement_timeout = conn.execute(
+            text("SELECT current_setting('statement_timeout')")
+        ).scalar_one()
+        lock_timeout = conn.execute(text("SELECT current_setting('lock_timeout')")).scalar_one()
+
+    assert statement_timeout == "30s", (
+        f"сессия сообщает statement_timeout={statement_timeout!r} — "
+        f"ожидалось 30s (_STATEMENT_TIMEOUT_MS={_STATEMENT_TIMEOUT_MS})"
+    )
+    assert lock_timeout == "5s", (
+        f"сессия сообщает lock_timeout={lock_timeout!r} — "
+        f"ожидалось 5s (_LOCK_TIMEOUT_MS={_LOCK_TIMEOUT_MS})"
+    )
+
+
+def test_statement_over_ceiling_is_cancelled_not_hung(probe_engine: Engine) -> None:
+    """`pg_sleep` длиннее потолка обрывается ОТМЕНОЙ за отведённое время."""
+    sleep_s = _PROBE_STATEMENT_TIMEOUT_MS / 1000 * 5
+    started = time.monotonic()
+    with probe_engine.connect() as conn, pytest.raises(OperationalError) as excinfo:
+        conn.execute(text(f"SELECT pg_sleep({sleep_s})"))
+    elapsed = time.monotonic() - started
+
+    assert isinstance(excinfo.value.orig, psycopg.errors.QueryCanceled), (
+        f"запрос упал не отменой по таймауту, а {type(excinfo.value.orig).__name__}"
+    )
+    assert elapsed < sleep_s, (
+        f"запрос шёл {elapsed:.1f} с при потолке {_PROBE_STATEMENT_TIMEOUT_MS} мс — "
+        "потолок не сработал, он висел до конца pg_sleep"
+    )
+
+
+def test_lock_wait_over_ceiling_is_aborted(probe_engine: Engine) -> None:
+    """Сценарий #3463: под ACCESS EXCLUSIVE читатель ОТВАЛИВАЕТСЯ, а не ждёт вечно."""
+    with probe_engine.connect() as blocker:
+        blocker.execute(text(f"CREATE TABLE IF NOT EXISTS {_LOCK_PROBE_TABLE} (id int)"))
+        blocker.commit()
+        try:
+            blocker.execute(text(f"LOCK TABLE {_LOCK_PROBE_TABLE} IN ACCESS EXCLUSIVE MODE"))
+            started = time.monotonic()
+            with probe_engine.connect() as victim, pytest.raises(OperationalError) as excinfo:
+                victim.execute(text(f"SELECT count(*) FROM {_LOCK_PROBE_TABLE}"))
+            elapsed = time.monotonic() - started
+        finally:
+            blocker.rollback()
+            blocker.execute(text(f"DROP TABLE IF EXISTS {_LOCK_PROBE_TABLE}"))
+            blocker.commit()
+
+    assert isinstance(excinfo.value.orig, psycopg.errors.LockNotAvailable), (
+        f"читатель упал не по ожиданию блокировки, а {type(excinfo.value.orig).__name__} — "
+        "сработал не тот потолок"
+    )
+    assert elapsed < _PROBE_STATEMENT_TIMEOUT_MS / 1000, (
+        f"ожидание блокировки длилось {elapsed:.2f} с при lock_timeout "
+        f"{_PROBE_LOCK_TIMEOUT_MS} мс — оборвал не lock_timeout"
+    )
+
+
+def test_set_local_statement_timeout_overrides_session_ceiling(probe_engine: Engine) -> None:
+    """Задача со своим `SET LOCAL` НЕ обрезается сессионным потолком.
+
+    Это и есть проверка обещания «задачи планировщика не пострадают»: у
+    `app/tasks/listing_source_snapshot.py:288` стоит `SET LOCAL statement_timeout = 900000`,
+    и он обязан ПЕРЕКРЫВАТЬ значение из `connect_args`, а не наоборот.
+    """
+    own_budget_ms = _PROBE_STATEMENT_TIMEOUT_MS * 10
+    sleep_s = _PROBE_STATEMENT_TIMEOUT_MS / 1000 * 2  # заведомо больше сессионного потолка
+
+    with probe_engine.connect() as conn:
+        conn.execute(text(f"SET LOCAL statement_timeout = {own_budget_ms}"))
+        effective = conn.execute(text("SELECT current_setting('statement_timeout')")).scalar_one()
+        assert effective == "10s", f"SET LOCAL не применился: current_setting={effective!r}"
+        # Не просто current_setting: запрос длиннее СЕССИОННОГО потолка обязан дойти до конца.
+        conn.execute(text(f"SELECT pg_sleep({sleep_s})"))
+        conn.rollback()
+
+
+def test_set_local_is_scoped_to_its_transaction(probe_engine: Engine) -> None:
+    """Обратная сторона: чужой `SET LOCAL` не снимает потолок со всей сессии.
+
+    Иначе одна задача с 900-секундным бюджетом отключала бы защиту у всех, кому
+    достанется тот же коннект из пула.
+    """
+    with probe_engine.connect() as conn:
+        conn.execute(text(f"SET LOCAL statement_timeout = {_PROBE_STATEMENT_TIMEOUT_MS * 10}"))
+        conn.rollback()
+        after = conn.execute(text("SELECT current_setting('statement_timeout')")).scalar_one()
+
+    assert after == "1s", (
+        f"после завершения транзакции statement_timeout={after!r} — "
+        "SET LOCAL протёк за пределы своей транзакции, коннект вернулся в пул без потолка"
+    )