tradein-mvp/backend/app/api/v1/auth.py

@@ -28,6 +28,13 @@ Security:
     username с `:` внутри мог бы схлопнуть бюджет с другой (username, ip)
     парой (IPv6-адреса тоже содержат `:`, так что просто эскейпить разделитель
     в username недостаточно — паразитная граница возможна с обеих сторон).
+  - Настоящий ПОТОЛОК ТЕМПА — `verify_password_bounded` (#2665): bcrypt считает
+    282 мс, и ровно столько же он раньше держал заблокированным единственный
+    событийный цикл, кладя вместе с логином ВЕСЬ API. Теперь bcrypt крутится в
+    пуле из `login_password_verify_workers` потоков, а число потоков и есть
+    потолок (проверок/с не больше workers/282мс). Убрать одно без другого
+    нельзя: вынос без потолка ускорил бы перебор вчетверо, потолок без выноса
+    оставил бы отказ в обслуживании. Сверх очереди — 429, не ожидание.
   - Поверх него — ГЛОБАЛЬНЫЙ счётчик неудач на ИМЯ, без IP в ключе (#2571):
     лимит по паре (username, IP) распределённый перебор обходит целиком, просто
     меняя адрес. Превышение порога не блокирует вход, а замедляет ответ
@@ -49,7 +56,7 @@ from pydantic import BaseModel, Field
 from sqlalchemy.orm import Session
 
 from app.core.config import settings
-from app.core.password import hash_password, verify_password
+from app.core.password import PasswordVerifyOverloadedError, hash_password, verify_password_bounded
 from app.core.ratelimit import SlidingWindowLimiter, _client_ip
 from app.services.auth_session import create_session, get_user_by_username, revoke_session
 from app.services.identity_store import AccessState, get_identity_db
@@ -252,7 +259,21 @@ async def login(
     )
     # ВСЕГДА вызывается — dummy-хеш при отсутствующем юзере/NULL password_hash
     # держит время ответа одинаковым независимо от существования аккаунта.
-    password_ok = verify_password(body.password, hash_to_check)
+    try:
+        password_ok = await verify_password_bounded(body.password, hash_to_check)
+    except PasswordVerifyOverloadedError:
+        # Настоящий потолок темпа (#2665): слоты проверки заняты, ждать нельзя —
+        # ждущий держит соединение к БД. Отказ ОДИНАКОВ для любого имени и
+        # случается ДО сверки, поэтому оракулом существования учётки не служит и
+        # бюджет неудач по имени не тратит (это не попытка входа: пароль не
+        # проверялся). Retry-After 1с — порядок времени одной проверки, не окно
+        # соседнего `_LOGIN_LIMITER`.
+        logger.warning("login rejected: password verify saturated ip=%s", ip)
+        raise HTTPException(
+            status_code=429,
+            detail="слишком много попыток входа, попробуйте позже",
+            headers={"Retry-After": "1"},
+        ) from None
 
     # Пароль проверен ВЫШЕ и безусловно — только теперь смотрим на состояние
     # доступа. Порядок несущий, а не стилистический: см. модульный docstring.

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

@@ -116,6 +116,40 @@ class Settings(BaseSettings):
     login_username_throttle_max_delay_s: float = Field(
         default=8.0, validation_alias="LOGIN_USERNAME_THROTTLE_MAX_DELAY_S"
     )
+    # ── #2665: проверка пароля вне событийного цикла + СОЗНАТЕЛЬНЫЙ потолок ────
+    # Замер в прод-контейнере 2026-08-06: bcrypt cost 12 (все живые хеши —
+    # `$2b$12$`) = 282 мс медиана. Пока `verify_password` звался прямо в
+    # `async def login`, эти 282 мс были простоем ВСЕГО API, и они же были
+    # единственным настоящим потолком темпа логинов — замерено 3.6 попытки/с при
+    # стойле событийного цикла до 836 мс. Обе половины чинятся вместе, см.
+    # `app.core.password.verify_password_bounded`.
+    #
+    # `workers` — это и есть потолок темпа: не больше workers/282мс проверок в
+    # секунду, сколько бы соединений ни пришло. Дефолт 1 выбран так, чтобы
+    # ПОСЛЕ выноса в пул потолок остался тем же (~3.5/с), что случайно давала
+    # блокировка цикла: вынос не должен ускорять перебор. Поднимать имеет смысл
+    # только вместе с осознанным ответом «во сколько раз мы согласны ускорить
+    # перебор ради параллельных входов».
+    # ge=1: 0 или -1 роняют ThreadPoolExecutor прямо НА ИМПОРТЕ («max_workers must
+    # be greater than 0») — контейнер уходит в crash-loop, и причина видна только
+    # в трейсбеке старта. Пусть отказ будет на валидации настроек, с именем поля.
+    login_password_verify_workers: int = Field(
+        default=1, ge=1, validation_alias="LOGIN_PASSWORD_VERIFY_WORKERS"
+    )
+    # Сколько запросов одновременно допускаются к проверке (считая тех, кто ждёт
+    # очереди в пуле). Сверх — сразу 429, без ожидания. Не режет темп (его режут
+    # workers), а держит конечной ОЧЕРЕДЬ: каждый ждущий запрос удерживает
+    # соединение к БД (сессия реестра открыта после SELECT в
+    # `get_user_by_username`), а в QueuePool их всего 5+10. Неограниченная
+    # очередь выбрала бы пул и положила API ровно так же, как блокировка цикла,
+    # только другим способом. 4 из 15 соединений и худшее ожидание
+    # 4/1×282мс ≈ 1.1с — цена, которую живой вход переживает.
+    # ge=1: 0 читается как «выключить лимит», а означал бы обратное — КАЖДЫЙ вход
+    # получает 429 навсегда и молча (слотов нет ни одного). Выключать тут нечего:
+    # потолок — это workers, а очередь без границы выбирает пул соединений к БД.
+    login_password_verify_max_inflight: int = Field(
+        default=4, ge=1, validation_alias="LOGIN_PASSWORD_VERIFY_MAX_INFLIGHT"
+    )
 
     # ── Эпик «единый вход»: общий реестр людей в БД `auth` ─────────────────────
     # DSN БД `auth` (роль auth_app) — единый реестр людей «Меры» (trade-in) и

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

@@ -5,14 +5,28 @@ bcrypt тихо обрезает пароли длиннее 72 байт (UTF-8)
 `hash_password` явно ловит это и падает с ValueError вместо тихого поведения.
 `verify_password` на длинном пароле возвращает False (не raise) — сравнение
 паролей не должно ронять запрос авторизации.
+
+#2665: из `async def` зови ТОЛЬКО `verify_password_bounded` — см. её docstring.
+Синхронный `verify_password` остаётся для sync-кода (сидов, тестов, CLI) и как
+тело, которое исполняется в пуле.
+
+Правило про пул относится к СВЕРКЕ, не к хешированию. `hash_password` — тот же
+cost 12 и те же ~282 мс на цикле — сознательно остаётся синхронным в
+`app/api/v1/team.py` (заведение сотрудника, смена пароля): это редкая операция
+АУТЕНТИФИЦИРОВАННОГО менеджера, её нельзя вызвать анонимно и потому нельзя
+превратить в поток. Станет их много — переносить тем же приёмом.
 """
 
 from __future__ import annotations
 
+import asyncio
 import logging
+from concurrent.futures import ThreadPoolExecutor
 
 import bcrypt
 
+from app.core.config import settings
+
 logger = logging.getLogger(__name__)
 
 _BCRYPT_MAX_BYTES = 72
@@ -59,3 +73,111 @@ def verify_password(plain: str, hashed: str) -> bool:
         # Malformed hash (напр. не-bcrypt строка в БД) — не должно ронять login.
         logger.warning("verify_password: malformed hash rejected: %s", e)
         return False
+
+
+class PasswordVerifyOverloadedError(RuntimeError):
+    """Свободных слотов на проверку пароля нет. Вызывающий обязан ответить 429."""
+
+
+# Пул, в котором крутится bcrypt. `max_workers` — не тюнинг пропускной
+# способности, а САМ ПОТОЛОК ТЕМПА: проверок в секунду не больше, чем
+# workers / 282мс, независимо от числа соединений. Читается один раз на импорте
+# — размер пула по определению статичен (см. `login_password_verify_workers`).
+_VERIFY_POOL = ThreadPoolExecutor(
+    max_workers=settings.login_password_verify_workers,
+    thread_name_prefix="pw-verify",
+)
+
+# Сколько проверок сейчас в работе ИЛИ ждут очереди в пуле. Обычный int без
+# лока — намеренно: и инкремент, и декремент выполняются в потоке событийного
+# цикла, между чтением и записью нет ни одного `await`, так что чередования
+# внутри пары нет. Счётчик, а не `asyncio.Semaphore`: мы никогда не ЖДЁМ на нём
+# (сверх лимита — сразу отказ), а int не имеет привязки к конкретному циклу и
+# потому одинаково честен под несколькими event loop'ами в тестах.
+_verify_inflight = 0
+
+
+async def verify_password_bounded(plain: str, hashed: str) -> bool:
+    """`verify_password`, унесённая с событийного цикла И с сознательным потолком темпа (#2665).
+
+    ДВЕ ПОЛОВИНЫ ОДНОЙ ПРАВКИ, И ЖИВУТ ОНИ ЗДЕСЬ ВМЕСТЕ НЕ ИЗ ЛЮБВИ К ПОРЯДКУ.
+    Порознь каждая делает хуже, чем было:
+      - вынести bcrypt в пул, не поставив потолок → перебор УСКОРЯЕТСЯ (замер
+        ниже: 3.6/с → 16/с на дефолтном executor'е);
+      - поставить потолок, не вынося bcrypt → 282 мс простоя всего API на каждую
+        попытку остаются.
+    Поэтому единственная точка выноса в поток и единственная точка учёта слотов —
+    одна и та же функция: состояние «вынесено, но потолка нет» невыразимо.
+
+    Замер в прод-контейнере (2026-08-06, cost 12, все живые хеши `$2b$12$`):
+      verify_password = 282 мс медиана;
+      вызов прямо в `async def` — 3.6 проверки/с, стойло событийного цикла 836 мс
+        (это и был «потолок» — случайный, ценой отказа в обслуживании всего API);
+      `asyncio.to_thread` без потолка — 16 проверок/с, стойло 6 мс.
+    Отсюда дефолт `workers=1`: потолок остаётся тем же ~3.5/с, что был, а API
+    перестаёт стоять. Числа перепроверяемы: tests/test_password.py.
+
+    Потолок держится ПРОЦЕССОМ, а не общим хранилищем. Это проверено, а не
+    предположено: прод-бэкенд запущен `uvicorn app.main:app` без `--workers`
+    (один процесс), а `REDIS_URL` в окружении tradein-backend НЕ ЗАДАН вовсе
+    (`printenv | grep -c ^REDIS_URL=` → 0, находка эпика #2674 — кэш поиска всю
+    жизнь стучится в localhost и получает отказ). Потолок на Redis был бы
+    потолком, который молча не работает.
+    Ceiling: появятся `--workers N` (или `WEB_CONCURRENCY=N` в `.env.runtime` —
+    uvicorn читает число процессов и оттуда, а файл правится руками на VPS) —
+    темп множится на N, как и у соседних in-memory лимитеров в
+    app/api/v1/auth.py; тогда потолок надо переносить в общее хранилище,
+    предварительно убедившись, что оно реально доступно.
+
+    Raises:
+        PasswordVerifyOverloadedError: очередь на проверку заполнена
+            (`login_password_verify_max_inflight`). Отказ мгновенный: ждать
+            нельзя, ждущий запрос держит соединение к БД.
+    """
+    global _verify_inflight
+
+    if _verify_inflight >= settings.login_password_verify_max_inflight:
+        raise PasswordVerifyOverloadedError
+
+    loop = asyncio.get_running_loop()
+    _verify_inflight += 1
+    try:
+        work = _VERIFY_POOL.submit(verify_password, plain, hashed)
+    except BaseException:
+        # Работа в пул НЕ встала — колбэка не будет, слот отдаём здесь. Иначе
+        # утёкший слот навсегда отнимает у входа часть и без того малой ёмкости.
+        _verify_inflight -= 1
+        raise
+
+    # Слот освобождает ЗАВЕРШЕНИЕ РАБОТЫ, а не выход из этой корутины. Отмена
+    # (клиент отвалился, таймаут) прекращает корутину, но УЖЕ НАЧАТУЮ сверку не
+    # снимает — поток занят ею все 282 мс. Отдавай мы слот в `finally`, на это
+    # время слот считался бы свободным: одновременно работающих сверок стало бы
+    # больше, чем разрешено, и очередь пула поехала бы вслед за ними.
+    # (Ещё не начатую работу отмена как раз снимает — `cancel()` пробрасывается
+    # на future пула, — так что вреда от неё нет; проблема ровно в начатой.)
+    #
+    # Именно поэтому колбэк висит на future ПУЛА, а не на обёртке из
+    # `run_in_executor`: у обёртки «готово» наступает и при отмене — тест
+    # `test_bounded_slot_freed_by_the_work_not_by_cancellation` ловит эту разницу.
+    work.add_done_callback(lambda _f: _schedule_verify_slot_release(loop))
+    return await asyncio.wrap_future(work)
+
+
+def _schedule_verify_slot_release(loop: asyncio.AbstractEventLoop) -> None:
+    """Возвращает слот по факту завершения работы в пуле (см. вызывающую).
+
+    Колбэк future пула исполняется В ПОТОКЕ ПУЛА, а счётчик — собственность
+    потока событийного цикла (на том и держится арифметика без лока), поэтому
+    декремент переносим в цикл через `call_soon_threadsafe`.
+    """
+    try:
+        loop.call_soon_threadsafe(_release_verify_slot)
+    except RuntimeError:
+        # Цикл уже закрыт (остановка процесса) — освобождать нечего и некому.
+        logger.debug("verify slot release skipped: event loop is closed")
+
+
+def _release_verify_slot() -> None:
+    global _verify_inflight
+    _verify_inflight -= 1

tradein-mvp/backend/tests/test_auth_api.py

@@ -31,6 +31,7 @@ in-memory fake DB standing in for the identity registry:
 
 from __future__ import annotations
 
+import asyncio
 import os
 import re
 import time
@@ -40,6 +41,7 @@ from typing import Annotated, Any
 
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
 
+import httpx
 import pytest
 from fastapi import FastAPI, Header
 from fastapi.testclient import TestClient
@@ -48,6 +50,7 @@ from app.api.v1 import auth as auth_router
 from app.api.v1 import me as me_router
 from app.core import auth as auth_mod
 from app.core import auth_db, config
+from app.core import password as password_mod
 from app.core.db import get_db
 from app.core.password import hash_password
 from app.core.rbac import rbac_guard
@@ -365,13 +368,16 @@ def test_login_always_calls_verify_password_timing_oracle_guard(
     store.add_user("nullhash", None, role="employee")
 
     calls: list[str] = []
-    real_verify = auth_router.verify_password
+    real_verify = password_mod.verify_password
 
     def _counting_verify(plain: str, hashed: str) -> bool:
         calls.append(hashed)
         return real_verify(plain, hashed)
 
-    monkeypatch.setattr(auth_router, "verify_password", _counting_verify)
+    # Патчим тело в app.core.password, а не имя в auth: с #2665 хендлер зовёт
+    # `verify_password_bounded`, а та ищет `verify_password` в своём модуле на
+    # каждый вызов — так счётчик считает РЕАЛЬНЫЕ bcrypt-сверки, а не обёртку.
+    monkeypatch.setattr(password_mod, "verify_password", _counting_verify)
 
     resp_unknown = client.post("/api/v1/auth/login", json={"username": "ghost", "password": "x"})
     assert resp_unknown.status_code == 401
@@ -707,6 +713,144 @@ def test_failed_login_events_reach_audit_with_counter_state(
     assert "s3cret-typo" not in str(failed)
 
 
+# ---------------------------------------------------------------------------
+# #2665 — настоящий потолок ТЕМПА проверок пароля + свободный событийный цикл
+# ---------------------------------------------------------------------------
+
+
+async def test_login_flood_capped_by_rate_while_api_stays_responsive(
+    store: _Store, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Сто одновременных соединений не получают больше N попыток В СЕКУНДУ, и при
+    этом остальной API продолжает отвечать.
+
+    ОБА утверждения в одном тесте намеренно — по отдельности каждое зелено на
+    сломанной системе:
+      - только про темп: сегодняшний код (bcrypt прямо в `async def`) тоже
+        держит темп низким — ценой того, что весь API стоит;
+      - только про отзывчивость: `asyncio.to_thread` без потолка освобождает
+        цикл и одновременно РАЗГОНЯЕТ перебор (замер на проде: 3.6 → 16
+        проверок/с).
+    Убери любую половину правки — тест обязан покраснеть.
+
+    Проверяем ТЕМП, а не латентность: задержка из #2571 (`await asyncio.sleep`)
+    латентность растит, а темп не ограничивает вовсе — сто соединений отспят её
+    параллельно. Поэтому меряем ЧИСЛО состоявшихся bcrypt-сверок за секунду
+    непрерывного флуда, а не время одного ответа.
+
+    Каждый запрос идёт со СВОЕЙ парой (username, ip). Это худший случай для
+    защит #2571 и он же реалистичный: при credential stuffing ни лимит на
+    (username, IP), ни счётчик неудач на имя не срабатывают ни разу — с чужого
+    адреса и с новым именем бюджет всегда свежий. Значит меряем ровно новый
+    потолок, а не соседний лимитер.
+    """
+    verify_s = 0.05
+    # Потолок = размер пула / время одной сверки. Значение берётся из ТОЙ ЖЕ
+    # настройки, что его задаёт, поэтому этот тест проверяет только МЕХАНИКУ
+    # (потолок работает и равен пулу), но НЕ величину дефолта: подними
+    # login_password_verify_workers — поднимется и ожидание, тест останется
+    # зелёным. Сам дефолт стережёт
+    # tests/test_password.py::test_verify_ceiling_defaults_stay_within_the_db_pool.
+    ceiling_per_s = config.settings.login_password_verify_workers / verify_s
+
+    patch_identity_sessions(monkeypatch, lambda: _FakeDB(store))
+    _capture_events(monkeypatch)
+    app = _build_test_app(store)
+
+    attempts: list[float] = []
+
+    def _slow_verify(plain: str, hashed: str) -> bool:
+        """Стенд-двойник bcrypt: столько же БЛОКИРУЮЩЕГО времени, только меньше.
+
+        Блокирующий `time.sleep`, а не `await` — суть проблемы в том, что bcrypt
+        не отпускает поток; двойник с `await` проверял бы не то.
+        """
+        attempts.append(time.monotonic())
+        time.sleep(verify_s)
+        return False
+
+    monkeypatch.setattr(password_mod, "verify_password", _slow_verify)
+
+    probe_latencies: list[float] = []
+    flood_over = asyncio.Event()
+
+    async def probe(client: httpx.AsyncClient) -> None:
+        """Сторонний (не login) запрос раз в 10мс — детектор занятости цикла."""
+        while not flood_over.is_set():
+            t0 = time.monotonic()
+            await client.get("/api/v1/trade-in/dummy")
+            probe_latencies.append(time.monotonic() - t0)
+            await asyncio.sleep(0.01)
+
+    duration_s = 1.0
+    connections = 100
+
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=app), base_url="https://testserver"
+    ) as client:
+
+        async def attacker(n: int) -> list[int]:
+            codes: list[int] = []
+            i = 0
+            while time.monotonic() < deadline:
+                i += 1
+                resp = await client.post(
+                    "/api/v1/auth/login",
+                    json={"username": f"spray{n}x{i}", "password": "guess"},
+                    headers={"x-forwarded-for": f"10.{n % 250}.{i % 250}.7"},
+                )
+                codes.append(resp.status_code)
+                await asyncio.sleep(0.005)
+            return codes
+
+        started = time.monotonic()
+        deadline = started + duration_s
+        probe_task = asyncio.create_task(probe(client))
+        code_lists = await asyncio.gather(*(attacker(n) for n in range(connections)))
+        elapsed = time.monotonic() - started
+        flood_over.set()
+        await probe_task
+
+    codes = [c for lst in code_lists for c in lst]
+    attempts_per_s = len(attempts) / elapsed
+
+    # 1. Событийный цикл СВОБОДЕН всё это время. С bcrypt внутри `async def`
+    #    сторонний запрос ждёт столько, сколько длится очередь сверок.
+    #    Проверяется ПЕРВЫМ: если цикл занят, встаёт и сам флуд, и тогда
+    #    остальные числа мерят не потолок, а паралич — их надо читать после
+    #    этого вердикта, а не вместо него.
+    assert probe_latencies, "проба не сделала ни одного запроса"
+    probe_latencies.sort()
+    assert (
+        probe_latencies[-1] < 0.5
+    ), f"худший сторонний запрос {probe_latencies[-1] * 1000:.0f}мс — API встаёт под флудом входа"
+    median_probe = probe_latencies[len(probe_latencies) // 2]
+    assert median_probe < verify_s, (
+        f"медиана стороннего запроса {median_probe * 1000:.0f}мс ≥ времени одной "
+        f"сверки — цикл занят проверкой пароля, API стоит"
+    )
+    # Мало проб за секунду — тоже занятый цикл: проба просыпается раз в 10мс.
+    assert (
+        len(probe_latencies) >= 10
+    ), f"проба успела всего {len(probe_latencies)} раз за {elapsed:.2f}с — цикл был занят"
+
+    # 2. ТЕМП ограничен. Флуд предлагал тысячи попыток в секунду — до bcrypt их
+    #    доехало не больше потолка (запас ×1.5 на планировщик).
+    assert len(codes) > connections, (
+        "флуд не состоялся: на каждое соединение вышло не больше одного ответа — "
+        "мерить потолок не на чем"
+    )
+    assert attempts_per_s <= ceiling_per_s * 1.5, (
+        f"{attempts_per_s:.0f} сверок/с при потолке {ceiling_per_s:.0f}/с "
+        f"({len(attempts)} за {elapsed:.2f}с) — потолок темпа не работает"
+    )
+    # 3. Лишнее ОТКЛОНЯЕТСЯ, а не копится в очереди: очередь держала бы
+    #    соединения к БД и выбрала бы пул (QueuePool 5+10).
+    assert codes.count(429) > codes.count(401), "избыток должен получать 429, а не ждать"
+    # 4. Вход не заблокирован совсем: потолок — это темп, а не «ноль попыток».
+    assert len(attempts) >= 2
+
+
 # ---------------------------------------------------------------------------
 # POST /logout
 # ---------------------------------------------------------------------------

tradein-mvp/backend/tests/test_password.py

@@ -2,9 +2,26 @@
 
 from __future__ import annotations
 
+import asyncio
+import os
+import threading
+import time
+
+# С #2665 password.py читает настройки (размер пула проверок) — значит тянет
+# `Settings()`, которому нужен DATABASE_URL. В CI он в env (ci-tradein.yml),
+# локально подставляем заглушку, как это делает tests/test_auth_api.py.
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
 import pytest
 
-from app.core.password import hash_password, verify_password
+from app.core import password as password_mod
+from app.core.config import settings
+from app.core.password import (
+    PasswordVerifyOverloadedError,
+    hash_password,
+    verify_password,
+    verify_password_bounded,
+)
 
 
 def test_roundtrip() -> None:
@@ -77,3 +94,124 @@ def test_hash_is_unique_due_to_salt() -> None:
 def test_verify_malformed_hash_returns_false() -> None:
     """Некорректный (не-bcrypt) хеш в verify_password → False, не raise."""
     assert verify_password("some password", "not-a-bcrypt-hash") is False
+
+
+# ---------------------------------------------------------------------------
+# #2665 — verify_password_bounded: вне событийного цикла + потолок темпа
+# ---------------------------------------------------------------------------
+
+
+def test_verify_ceiling_defaults_stay_within_the_db_pool() -> None:
+    """Дефолты — часть защиты, а не тюнинг. Стережём их здесь.
+
+    Тест про темп (test_auth_api.py) вычисляет ожидаемый потолок из той же
+    настройки, которую охраняет, поэтому подъём дефолта он не заметит. А
+    наступит ослабление именно через настройку: не правкой кода и не ревью, а
+    строчкой `LOGIN_PASSWORD_VERIFY_WORKERS=32` в `.env.runtime` под предлогом
+    «входы тормозят». Пусть тогда краснеет хотя бы этот тест.
+    """
+    from app.core.db import engine
+
+    # max_inflight ждущих ДЕРЖАТ по соединению к БД (сессия реестра открыта
+    # после SELECT в get_user_by_username) — очередь обязана быть уже пула.
+    assert (
+        settings.login_password_verify_max_inflight < engine.pool.size() + engine.pool._max_overflow
+    )
+    assert settings.login_password_verify_workers == 1, (
+        "потолок перебора = workers/282мс. Подъём — осознанное решение "
+        "«во сколько раз ускоряем перебор», а не рефакторинг: правь вместе с тестом"
+    )
+
+
+async def test_bounded_gives_same_answer_as_sync() -> None:
+    """Обёртка не меняет вердикт — она меняет только ГДЕ он считается."""
+    hashed = hash_password("correct horse battery staple")
+    assert await verify_password_bounded("correct horse battery staple", hashed) is True
+    assert await verify_password_bounded("wrong password", hashed) is False
+
+
+async def test_bounded_runs_off_the_event_loop_thread(monkeypatch: pytest.MonkeyPatch) -> None:
+    """bcrypt считается В ДРУГОМ ПОТОКЕ, а не в потоке событийного цикла.
+
+    Замер на проде: сверка = 282 мс, и ровно столько цикл не обслуживал никого.
+    Проверка идентичности потока — самая прямая формулировка «цикл свободен»;
+    таймингом её подменять нельзя, тайминг в CI флейкует.
+    """
+    loop_thread = threading.get_ident()
+    seen: list[int] = []
+
+    def _spy(plain: str, hashed: str) -> bool:
+        seen.append(threading.get_ident())
+        return True
+
+    monkeypatch.setattr(password_mod, "verify_password", _spy)
+    assert await verify_password_bounded("x", "y") is True
+    assert seen and seen[0] != loop_thread
+
+
+async def test_bounded_rejects_surplus_instead_of_queueing(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Сверх лимита — немедленный отказ, а не ожидание в очереди.
+
+    Ожидание выглядело бы безобиднее, но каждый ждущий запрос держит соединение
+    к БД (сессия реестра открыта после SELECT), а в QueuePool их 5+10:
+    неограниченная очередь выбрала бы пул и положила API — тем же концом, каким
+    его клала блокировка цикла.
+    """
+    monkeypatch.setattr(settings, "login_password_verify_max_inflight", 2)
+
+    def _slow(plain: str, hashed: str) -> bool:
+        time.sleep(0.2)
+        return False
+
+    monkeypatch.setattr(password_mod, "verify_password", _slow)
+
+    results = await asyncio.gather(
+        *(verify_password_bounded("x", "y") for _ in range(6)), return_exceptions=True
+    )
+    rejected = [r for r in results if isinstance(r, PasswordVerifyOverloadedError)]
+    admitted = [r for r in results if r is False]
+    assert len(admitted) == 2, results
+    assert len(rejected) == 4, results
+
+    # Слоты возвращаются: после отработки очереди вход снова доступен.
+    assert await verify_password_bounded("x", "y") is False
+
+
+async def test_bounded_slot_freed_by_the_work_not_by_cancellation(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Отмена запроса не возвращает слот раньше времени.
+
+    Отмена снимает работу, которая ещё НЕ началась, — с ней проблем нет. Но уже
+    начатую сверку она не забирает: поток занят ею все 282 мс. Освобождай мы
+    слот по выходу из корутины, на это время он числился бы свободным, и
+    одновременно работающих сверок стало бы больше, чем разрешено.
+    """
+    monkeypatch.setattr(settings, "login_password_verify_max_inflight", 1)
+    started = threading.Event()
+    finish = threading.Event()
+
+    def _blocked(plain: str, hashed: str) -> bool:
+        started.set()
+        finish.wait(5)
+        return False
+
+    monkeypatch.setattr(password_mod, "verify_password", _blocked)
+
+    task = asyncio.create_task(verify_password_bounded("x", "y"))
+    await asyncio.to_thread(started.wait, 5)
+
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+
+    # Работа всё ещё занимает поток — слот занят, следующий получает отказ.
+    with pytest.raises(PasswordVerifyOverloadedError):
+        await verify_password_bounded("x", "y")
+
+    finish.set()
+    for _ in range(100):  # дать колбэку доехать до цикла
+        await asyncio.sleep(0.01)
+        if settings.login_password_verify_max_inflight > password_mod._verify_inflight:
+            break
+    assert await verify_password_bounded("x", "y") is False