(medium) Sber pull выполняет блокирующий DB/CPU на event loop (async-таск без run_in_executor) #1348

New issue

Closed

opened 2026-06-14 18:08:52 +00:00 by bot-backend · 1 comment

bot-backend commented

2026-06-14 18:08:52 +00:00

Collaborator

Описание

pull_sber_indices объявлен async def и запускается scheduler.trigger_sber_index_pull_run через asyncio.create_task → await run_sber_index_pull, т.е. исполняется НА event loop FastAPI/scheduler. Внутри — полностью синхронный блокирующий psycopg: per-row db.execute(...) UPSERT-цикл (потенциально сотни строк × 3 города × 3 дашборда), db.commit() per series, синхронный benchmark SELECT и CPU-bound base64/JSON декодирование в decode_sber_response. Ничего из этого не обёрнуто в run_in_executor/to_thread.

Корень

Синхронный SQLAlchemy Session (backed sync create_engine, core/db.py:8-9) используется внутри async def, исполняющегося на loop. Все остальные sync DB-таски в scheduler.py (snapshot_listing_sources, recompute_asking_to_sold_ratios, deactivate_stale_avito_listings, import_rosreestr_dkp — строки 489/524/567/604/756) корректно диспатчатся через await loop.run_in_executor(None, ...); sber — единственное исключение.

Evidence

scheduler.py:640:

await run_sber_index_pull(run_db, run_id=run_id, params=params)   # на loop, НЕ run_in_executor

sber_index.py:338,370:

db.execute(text("""INSERT INTO sber_price_index ..."""), {...})   # sync, в async def, per row
...
db.commit()

HTTP-fetch действительно async (httpx.AsyncClient), но upsert-цикл — нет.

Suggested fix

Вынести синхронный DB-блок из coroutine: либо диспатчить весь pull через loop.run_in_executor (как остальные sync-таски) и сделать pull_sber_indices plain def, либо оставить async HTTP-fetch, но offload upsert-цикл + commit через asyncio.to_thread.

Обоснование severity

medium: задача ежемесячная (низкочастотный триггер, не per-request), отдельные UPSERT обычно мс; практическая блокировка — сотни мс до низких секунд за прогон. Была бы high при per-request/высокочастотном пути. confidence 0.92.

week ревью 1 · multi-agent аудит 2026-06-14 · severity=medium · category=async-concurrency

Затронутые файлы:

/Users/anton/Птица/gendesign/tradein-mvp/backend/app/services/sber_index.py
/Users/anton/Птица/gendesign/tradein-mvp/backend/app/services/scheduler.py

## Описание `pull_sber_indices` объявлен `async def` и запускается `scheduler.trigger_sber_index_pull_run` через `asyncio.create_task` → `await run_sber_index_pull`, т.е. исполняется НА event loop FastAPI/scheduler. Внутри — полностью синхронный блокирующий psycopg: per-row `db.execute(...)` UPSERT-цикл (потенциально сотни строк × 3 города × 3 дашборда), `db.commit()` per series, синхронный benchmark SELECT и CPU-bound base64/JSON декодирование в `decode_sber_response`. Ничего из этого не обёрнуто в `run_in_executor`/`to_thread`. ## Корень Синхронный SQLAlchemy `Session` (backed sync `create_engine`, core/db.py:8-9) используется внутри `async def`, исполняющегося на loop. Все остальные sync DB-таски в scheduler.py (snapshot_listing_sources, recompute_asking_to_sold_ratios, deactivate_stale_avito_listings, import_rosreestr_dkp — строки 489/524/567/604/756) корректно диспатчатся через `await loop.run_in_executor(None, ...)`; sber — единственное исключение. ## Evidence `scheduler.py:640`: ```python await run_sber_index_pull(run_db, run_id=run_id, params=params) # на loop, НЕ run_in_executor ``` `sber_index.py:338,370`: ```python db.execute(text("""INSERT INTO sber_price_index ..."""), {...}) # sync, в async def, per row ... db.commit() ``` HTTP-fetch действительно async (`httpx.AsyncClient`), но upsert-цикл — нет. ## Suggested fix Вынести синхронный DB-блок из coroutine: либо диспатчить весь pull через `loop.run_in_executor` (как остальные sync-таски) и сделать `pull_sber_indices` plain `def`, либо оставить async HTTP-fetch, но offload upsert-цикл + commit через `asyncio.to_thread`. ## Обоснование severity medium: задача ежемесячная (низкочастотный триггер, не per-request), отдельные UPSERT обычно мс; практическая блокировка — сотни мс до низких секунд за прогон. Была бы high при per-request/высокочастотном пути. confidence 0.92. --- _week ревью 1 · multi-agent аудит 2026-06-14 · severity=**medium** · category=`async-concurrency`_ Затронутые файлы: - `/Users/anton/Птица/gendesign/tradein-mvp/backend/app/services/sber_index.py` - `/Users/anton/Птица/gendesign/tradein-mvp/backend/app/services/scheduler.py`

bot-backend added the

week ревью 1

label 2026-06-14 18:08:52 +00:00

bot-backend added the

scope/backend

label 2026-06-15 11:01:50 +00:00

lekss361 commented

2026-06-15 11:13:45 +00:00

Owner

🔍 week ревью 1 — верификация (ПОДТВЕРЖДЕНО)

Текущее состояние кода: sber_index.py:284 — async def pull_sber_indices(...); внутри синхронный psycopg per-row UPSERT db.execute(...) :338-367 + db.commit() :370 + benchmark SELECT :395. scheduler.py:640 (forgejo/main) — await run_sber_index_pull(run_db, run_id=run_id, params=params) исполняется НА event loop, без run_in_executor. Ни run_in_executor, ни to_thread в sber_index.py нет (grep — 0 совпадений).

Корень: Синхронный Session используется внутри async def, запущенного на loop. Остальные sync DB-таски в scheduler.py диспатчатся через await loop.run_in_executor(None, ...) (:489 rosreestr_dkp, :524 snapshot, :567 matview, :604 deactivate_stale, :756 asking_ratios) — sber единственное исключение.

Что править:

Либо: сделать pull_sber_indices plain def и диспатчить весь pull через loop.run_in_executor в trigger_sber_index_pull_run (по образцу :604). HTTP-fetch (httpx.AsyncClient) тогда вынести/переписать на sync, либо
Оставить async HTTP-fetch (fetch_sber_index :216), но offload синхронный upsert-цикл + commit через asyncio.to_thread.

Что посмотреть:

Код: scheduler.py:616-642 (trigger_sber_index_pull_run → create_task → _run → await run_sber_index_pull), app/tasks/sber_index_pull.py (run_sber_index_pull); CPU-bound decode_sber_response sber_index.py:149. Эталон run_in_executor: scheduler.py:604.
Vault: релевантных заметок не найдено (sberindex-заметки в inbox про per-dashboard filter, не про async-блокировку).

**🔍 week ревью 1 — верификация (ПОДТВЕРЖДЕНО)** **Текущее состояние кода:** `sber_index.py:284` — `async def pull_sber_indices(...)`; внутри синхронный psycopg per-row UPSERT `db.execute(...)` `:338-367` + `db.commit()` `:370` + benchmark SELECT `:395`. `scheduler.py:640` (forgejo/main) — `await run_sber_index_pull(run_db, run_id=run_id, params=params)` исполняется НА event loop, без `run_in_executor`. Ни `run_in_executor`, ни `to_thread` в `sber_index.py` нет (grep — 0 совпадений). **Корень:** Синхронный `Session` используется внутри `async def`, запущенного на loop. Остальные sync DB-таски в `scheduler.py` диспатчатся через `await loop.run_in_executor(None, ...)` (`:489` rosreestr_dkp, `:524` snapshot, `:567` matview, `:604` deactivate_stale, `:756` asking_ratios) — sber единственное исключение. **Что править:** - Либо: сделать `pull_sber_indices` plain `def` и диспатчить весь pull через `loop.run_in_executor` в `trigger_sber_index_pull_run` (по образцу `:604`). HTTP-fetch (`httpx.AsyncClient`) тогда вынести/переписать на sync, либо - Оставить async HTTP-fetch (`fetch_sber_index` `:216`), но offload синхронный upsert-цикл + commit через `asyncio.to_thread`. **Что посмотреть:** - Код: `scheduler.py:616-642` (`trigger_sber_index_pull_run` → `create_task` → `_run` → `await run_sber_index_pull`), `app/tasks/sber_index_pull.py` (`run_sber_index_pull`); CPU-bound `decode_sber_response` `sber_index.py:149`. Эталон run_in_executor: `scheduler.py:604`. - Vault: релевантных заметок не найдено (sberindex-заметки в inbox про per-dashboard filter, не про async-блокировку).