fix(chat): не держать соединение с БД всю жизнь WS-подключения
Обработчик `WS /conferences/{id}/chat` получает `AsyncSession` через
`Depends(get_session)`, а хендшейк делает четыре SELECT'а (тоггл чата,
конференция, тоггл рук, история). SQLAlchemy открывает транзакцию на
первом из них и держит её — вместе с соединением из пула — всё время,
пока участник сидит в комнате. Соединений в пуле `db_pool_size +
db_max_overflow` = 20 на воркер, то есть 40 на инстанс из двух воркеров:
сороковой вошедший выгребал пул досуха.
Ровно это положило вход на нагрузочном тесте 07.08.2026: 245 ошибок
`QueuePool limit ... timed out`, 170 ответов 500 (из них 123 на резолве
конференции и 21 на гостевом входе), а `pg_stat_activity` показывал рост
`idle in transaction` 3 → 8 → 16 → 26 → 35 → 39 → 40 при одном `active`.
Число открытых WS чата в логах backend растёт синхронно и упирается в
те же 40 ровно к моменту первого таймаута пула.
Соединение освобождается сразу после хендшейка: дальше оба насоса
работают через Redis, а единственная запись в БД (`persist_and_publish`)
открывает и коммитит собственную транзакцию.
Замер на локальном стенде (один воркер, потолок пула 20), 15 посторонних
запросов на каждой ступени:
| участников | idle in transaction | 5xx | p95 |
|------------|---------------------|-----|----------|
| было 20 | 20 | 10 | 10.05 с |
| стало 20 | 0 | 0 | 0.02 с |
| стало 120 | 0 | 0 | 0.03 с |
До правки 21-й участник не мог войти вовсе (500 на guest-join), в логе
40 ошибок `QueuePool limit`; после — ни одной на 120 участниках.
This commit is contained in:
@@ -87,6 +87,26 @@ async def chat_websocket(
|
|||||||
await pubsub.subscribe(channel, room_channel)
|
await pubsub.subscribe(channel, room_channel)
|
||||||
try:
|
try:
|
||||||
history = await service.history(conference)
|
history = await service.history(conference)
|
||||||
|
# 🔑 Вернуть соединение с БД в пул ДО входа в долгоживущие насосы.
|
||||||
|
#
|
||||||
|
# Хендшейк выше сделал несколько SELECT'ов (тоггл чата, конференция,
|
||||||
|
# тоггл рук, история) — SQLAlchemy открыла транзакцию на первом же из
|
||||||
|
# них и держала бы её, а с ней и соединение из пула, ВСЁ время жизни
|
||||||
|
# WS: участник сидит в комнате час — час занято соединение. Пул это
|
||||||
|
# `db_pool_size + db_max_overflow` на воркер (10 + 10), то есть
|
||||||
|
# 40 на инстанс из двух воркеров, и сороковой вошедший выгребал его
|
||||||
|
# досуха: `pg_stat_activity` показывал 40 соединений
|
||||||
|
# `idle in transaction` при одном `active`, а посторонние ручки
|
||||||
|
# (резолв, гостевой вход, логин, refresh) начинали падать в
|
||||||
|
# `QueuePool limit ... timed out` и отдавать 500. Ровно это положило
|
||||||
|
# вход на нагрузочном тесте 07.08.2026 при ~50 участниках.
|
||||||
|
#
|
||||||
|
# Соединение здесь больше не нужно: оба насоса ниже работают через
|
||||||
|
# Redis, а единственная запись в БД (`persist_and_publish`) сама
|
||||||
|
# открывает транзакцию и коммитит её, освобождая соединение сразу.
|
||||||
|
# ⚠️ Любое чтение из БД, добавленное между этой строкой и концом
|
||||||
|
# обработчика, обязано так же завершаться commit/rollback.
|
||||||
|
await session.commit()
|
||||||
await websocket.send_json(ChatHistoryOut(messages=history).model_dump(mode="json"))
|
await websocket.send_json(ChatHistoryOut(messages=history).model_dump(mode="json"))
|
||||||
seen_ids = {item.id for item in history}
|
seen_ids = {item.id for item in history}
|
||||||
|
|
||||||
|
|||||||
@@ -237,6 +237,42 @@ async def test_no_duplicate_when_message_already_in_history(
|
|||||||
assert received["message"]["text"] == "genuinely new"
|
assert received["message"]["text"] == "genuinely new"
|
||||||
|
|
||||||
|
|
||||||
|
# --- Удержание соединения с БД ------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
async def test_handshake_releases_db_connection(
|
||||||
|
db_session: AsyncSession, ws_client: WSFactory
|
||||||
|
) -> None:
|
||||||
|
"""Regression: после хендшейка WS не держит открытую транзакцию БД.
|
||||||
|
|
||||||
|
Обработчик получает `AsyncSession` на ВСЁ время жизни соединения, а
|
||||||
|
SELECT'ы хендшейка (тоггл чата, конференция, тоггл рук, история)
|
||||||
|
открывают транзакцию. Без явного `commit` она висела бы, пока участник
|
||||||
|
сидит в комнате: одно занятое соединение из пула на каждого человека
|
||||||
|
в конференции. На нагрузочном тесте 07.08.2026 это выгребло пул
|
||||||
|
(`db_pool_size + db_max_overflow` = 20 на воркер, 40 на инстанс) при
|
||||||
|
сорока участниках — и вход в систему начал отдавать 500 всем
|
||||||
|
остальным. Проверяем именно отсутствие открытой транзакции, а не
|
||||||
|
состояние пула: тестовая сессия привязана к своему соединению
|
||||||
|
(см. докстринг `tests/conftest.py`) и пул не задействует.
|
||||||
|
"""
|
||||||
|
conference = await _make_conference(db_session)
|
||||||
|
user = await _make_user(db_session)
|
||||||
|
await db_session.commit()
|
||||||
|
|
||||||
|
ws = ws_client(_chat_path(conference.id))
|
||||||
|
await _connect_and_auth(ws, _user_token(conference, user))
|
||||||
|
|
||||||
|
assert not db_session.in_transaction()
|
||||||
|
|
||||||
|
# Запись сообщения открывает транзакцию заново — и тоже обязана её
|
||||||
|
# закрыть, иначе первый же чат вернул бы прежнее поведение.
|
||||||
|
await ws.send_json({"type": "message", "text": "проверка"})
|
||||||
|
echo = await ws.receive_json()
|
||||||
|
assert echo["type"] == "message"
|
||||||
|
assert not db_session.in_transaction()
|
||||||
|
|
||||||
|
|
||||||
# --- Auth: коды закрытия ----------------------------------------------------
|
# --- Auth: коды закрытия ----------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user