test(tradein/auth): покрыть отказ пула и отмену в очереди + записать цену размена по NAT (#2714)

Правки по глубокому ревью.

Тесты:
- отказ пула на `submit` (единственная изменённая строка без покрытия): слот
  отдаётся синхронно, оба счётчика в нуле. Мутант «старый код, ключевой
  счётчик не отдан» краснит и тест, и autouse-сторож.
- отмена ЕЩЁ НЕ НАЧАТОЙ работы возвращает слот ключа — ветка future другая,
  чем у отмены начатой, соседний тест её не покрывал.
- пол `max(1, …)`: при очереди в 1 слот доля не округляется в ноль (иначе
  молчаливый отказ всем).
- цена размена по NAT закреплена явно: третий одновременный вход с того же
  адреса получает 429 при двух занятых слотах из четырёх.

Соседям по NAT стало ХУЖЕ, и теперь это записано в docstring с числами: при
флуде 3 запроса/с свои входят 69% попыток против 94% до правки, порог отказа
падает с ~14 до ~7 запросов/с. Взамен вход с чужих адресов идёт 100% против
37%. Размен сознательный, а не побочный эффект.

Refs #2714

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

@@ -169,7 +169,13 @@ async def verify_password_bounded(plain: str, hashed: str, *, key: str) -> bool:
         (сейчас доверенный хоп ровно один — Caddy, `ratelimit._client_ip` берёт
         правый элемент XFF; появится второй — ключ станет клиентским вводом);
       - разделяется: за NAT/корпоративным шлюзом вся организация приходит с
-        одного адреса и делит одну долю с чужим перебором;
+        одного адреса и делит одну долю с чужим перебором. СОСЕДЯМ ПО АДРЕСУ
+        СТАЛО ХУЖЕ, и это честный размен, а не побочный эффект: при флуде в
+        3 запроса/с с того же адреса свои входят 69% попыток против 94% до
+        правки, а порог, за которым сосед перестаёт входить, падает с ~14 до
+        ~7 запросов/с. Взамен вход С ЧУЖИХ адресов идёт 100% против 37%;
+        размен принят сознательно — офис за одним NAT это единицы адресов,
+        а «все остальные» это все;
       - меняется: ботнет или ротация прокси дают злоумышленнику столько ключей,
         сколько ему нужно, и доля на ключ перестаёт быть ограничением.
     То есть это ПОДНИМАЕТ СТОИМОСТЬ атаки (одного адреса больше не хватает,

tradein-mvp/backend/tests/test_password.py

@@ -6,6 +6,7 @@ import asyncio
 import os
 import threading
 import time
+from concurrent.futures import ThreadPoolExecutor
 
 # С #2665 password.py читает настройки (размер пула проверок) — значит тянет
 # `Settings()`, которому нужен DATABASE_URL. В CI он в env (ci-tradein.yml),
@@ -295,6 +296,13 @@ async def test_one_key_cannot_take_more_than_its_share(monkeypatch: pytest.Monke
     await _wait_inflight(2)
 
     # Третий с ТОГО ЖЕ адреса — отказ, хотя два слота из четырёх свободны.
+    # Это ЦЕНА правки, а не побочный эффект: три одновременных входа из одного
+    # офиса за NAT укладываются в окно одной сверки (282 мс), и третьему
+    # сотруднику теперь отказывают при наполовину пустом пуле — до правки для
+    # этого требовалось пятеро. Закрепляем явно, чтобы размен нельзя было
+    # потерять молча: свои с ЧУЖИХ адресов за это получают 100% вместо 37%.
+    assert password_mod._verify_inflight == 2
+    assert settings.login_password_verify_max_inflight == 4
     with pytest.raises(PasswordVerifyOverloadedError):
         await verify_password_bounded("x", "y", key="10.0.0.1")
 
@@ -306,3 +314,74 @@ async def test_one_key_cannot_take_more_than_its_share(monkeypatch: pytest.Monke
     finish.set()
     assert await legit is False, "вход с другого адреса обязан пройти во время флуда"
     assert [await f for f in flood] == [False, False]
+
+
+def test_per_key_cap_never_rounds_down_to_zero(monkeypatch: pytest.MonkeyPatch) -> None:
+    """При очереди в 1 слот доля не округляется в ноль.
+
+    `1 // 2 == 0` означало бы «ни одному ключу нельзя ни одного слота» —
+    молчаливый отказ ВСЕМ на входе, причём тем более незаметный, что настройка
+    выглядит как безобидное ужесточение. `max(1, …)` — тот же страховочный пол,
+    что `ge=1` у самой настройки, только от деления.
+    """
+    monkeypatch.setattr(settings, "login_password_verify_max_inflight", 1)
+    assert password_mod._per_key_slot_cap() == 1
+
+
+async def test_bounded_frees_slot_when_pool_refuses_work(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Пул не принял работу → слот отдан прямо здесь, колбэка ведь не будет.
+
+    Единственный путь, где освобождение НЕ висит на future: `submit` бросает
+    (пул закрыт на остановке процесса). Утечка тут стоила бы дорого — при пуле
+    в один поток невозвращённый слот это вечный 429 всем на входе.
+    """
+    dead_pool = ThreadPoolExecutor(max_workers=1)
+    dead_pool.shutdown()
+    monkeypatch.setattr(password_mod, "_VERIFY_POOL", dead_pool)
+
+    with pytest.raises(RuntimeError):
+        await verify_password_bounded("x", "y", key="10.0.0.4")
+
+    assert password_mod._verify_inflight == 0
+    assert not password_mod._verify_inflight_by_key
+
+
+async def test_cancelling_queued_work_returns_the_key_slot(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Отмена ЕЩЁ НЕ НАЧАТОЙ работы возвращает слот — и общий, и ключа.
+
+    Ветка future другая, чем у отмены начатой работы (`cancel()` на очереди
+    успевает, и работа не исполняется вовсе), поэтому проверяется отдельно:
+    соседний тест про начатую работу эту не покрывает. Пул из одного потока —
+    настоящий, так что второй запрос гарантированно ЖДЁТ в очереди.
+    """
+    monkeypatch.setattr(settings, "login_password_verify_max_inflight", 4)
+    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)
+
+    running = asyncio.create_task(verify_password_bounded("x", "y", key="10.0.0.5"))
+    await asyncio.to_thread(started.wait, 5)
+
+    queued = asyncio.create_task(verify_password_bounded("x", "y", key="10.0.0.6"))
+    deadline = time.monotonic() + 5
+    while password_mod._verify_inflight_by_key.get("10.0.0.6") != 1:
+        assert time.monotonic() < deadline, "второй запрос не занял слот"
+        await asyncio.sleep(0.005)
+
+    queued.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await queued
+
+    deadline = time.monotonic() + 5
+    while "10.0.0.6" in password_mod._verify_inflight_by_key:
+        assert time.monotonic() < deadline, "слот отменённой очереди не вернулся"
+        await asyncio.sleep(0.005)
+
+    finish.set()
+    assert await running is False