fix(tradein/auth): слот проверки пароля освобождает работа, а не отмена запроса

Колбэк висел на обёртке `run_in_executor`: у неё «готово» наступает и при
ОТМЕНЕ корутины, а подхваченная пулом задача при этом продолжает занимать
поток свои 282 мс. Значит отваливающийся клиент получал свежий слот на каждую
отмену и мог набивать очередь пула быстрее, чем та разгребается — темп bcrypt
по-прежнему держал бы пул, но очередь и память росли бы без границы.

Колбэк перевешен на future ПУЛА (`submit`), декремент возвращается в поток
цикла через `call_soon_threadsafe` — счётчик остаётся собственностью цикла и
живёт без лока. Тест на отмену ловит ровно эту разницу: он краснел на
предыдущей реализации.

Refs #2665

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

@@ -131,9 +131,44 @@ async def verify_password_bounded(plain: str, hashed: str) -> bool:
     if _verify_inflight >= settings.login_password_verify_max_inflight:
         raise PasswordVerifyOverloadedError
 
+    loop = asyncio.get_running_loop()
     _verify_inflight += 1
     try:
-        loop = asyncio.get_running_loop()
-        return await loop.run_in_executor(_VERIFY_POOL, verify_password, plain, hashed)
-    finally:
+        work = _VERIFY_POOL.submit(verify_password, plain, hashed)
+    except BaseException:
+        # Работа в пул НЕ встала — колбэка не будет, слот отдаём здесь. Иначе
+        # утёкший слот навсегда отнимает у входа часть и без того малой ёмкости.
         _verify_inflight -= 1
+        raise
+
+    # Слот освобождает ЗАВЕРШЕНИЕ РАБОТЫ, а не выход из этой корутины. Отмена
+    # (клиент отвалился, таймаут) прекращает корутину, но уже подхваченную пулом
+    # задачу не отменяет — она всё равно займёт поток на свои 282 мс. Отдавай мы
+    # слот в `finally`, отменяющий клиент получал бы свежий слот на каждую
+    # отмену и набивал очередь пула быстрее, чем та разгребается: темп bcrypt
+    # по-прежнему держал бы пул, но очередь и память росли бы без границы.
+    #
+    # Именно поэтому колбэк висит на 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_password.py

@@ -153,3 +153,44 @@ async def test_bounded_rejects_surplus_instead_of_queueing(monkeypatch: pytest.M
 
     # Слоты возвращаются: после отработки очереди вход снова доступен.
     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:
+    """Отмена запроса не возвращает слот раньше времени.
+
+    Отменённая корутина работу из пула не забирает: bcrypt всё равно займёт
+    поток на свои 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