process_nspd_geo_job (nspd_geo.py:374-408, 82-97) не имеет атомарного claim → дублирующие task-сообщения для одного job_id могут идти параллельно (двойные WAF-хиты + зат… #1621

New issue

Closed

opened 2026-06-16 17:21:52 +00:00 by bot-backend · 0 comments

bot-backend commented

2026-06-16 17:21:52 +00:00

Collaborator

Severity: medium · Категория: concurrency/idempotency · Файл: backend/app/workers/tasks/nspd_geo.py:374-408, 82-97

process_nspd_geo_job загружает строку job (SELECT * FROM nspd_geo_jobs WHERE job_id, nspd_geo.py:377-384) и вызывает _start_job, который безусловно делает UPDATE ... SET status='running' WHERE job_id=:id (82-95) — НЕТ проверки текущего status='queued', НЕТ FOR UPDATE, НЕТ advisory/Redis lock, и НЕТ acks_late/dedup в celery_app.py (подтверждено grep'ом — конфиг пуст). Два независимых пути порождают дублирующие сообщения на один job_id: (a) worker_ready resume + per-minute cleanup_zombies beat одновременно (оба делают apply_async после UPDATE→'queued'; cleanup_zombies при этом фильтрует по heartbeat, worker_ready — нет, см. findings 0/1), и (b) перекрытие контейнеров при rolling-redeploy. Prod gunicorn-worker один, но --concurrency=8 (docker-compose.prod.yml:128) → два prefork-child могут снять оба сообщения и крутить один job параллельно. Per-target SELECT (409-424) — LIMIT 1 WHERE status='pending' ORDER BY target_id БЕЗ FOR UPDATE SKIP LOCKED, так что оба runner'а берут один и тот же pending-target. ИМПАКТ: (1) один cad_num тянется из WAF-throttled NSPD двумя runner'ами сразу → удвоение rate и риска бана; (2) счётчики job-строки (targets_done/waf_blocked_count/requests_count) ведутся как независимые in-memory аккумуляторы на runner и флашатся через _heartbeat/_finish_job → last-writer-wins clobber, done может уехать за targets_total → телеметрия/прогресс в admin неверны. _persist_target UPSERT'ит идемпотентно, поэтому сами гео-данные не повреждаются — отсюда medium. ФИКС: атомарный claim в начале process_nspd_geo_job — UPDATE nspd_geo_jobs SET status='running' WHERE job_id=:id AND status IN ('queued','running'-с-устаревшим-hb) RETURNING; при 0 строк — выходить; либо Redis-lock per job_id по образцу _region_lock в resume_kn_run (scrape_kn.py:120). Для target-select добавить FOR UPDATE SKIP LOCKED.

Почему баг: Concurrent execution of one job means: (1) the same cad_num is fetched from the WAF-throttled NSPD service by two runners at once, doubling request rate and ban risk; (2) job-row counters (targets_done, waf_blocked_count, requests_count) are maintained as independent in-memory accumulators per runner and flushed via _heartbeat/_finish_job — last-writer-wins clobbers and can drive done past targets_total, so the admin progress/telemetry is wrong. _persist_target UPSERTs make the stored geo data itself idempotent, which limits this to wasted/risky requests + corrupted counters rather than data corruption, hence medium.

Как чинить: Make the start an atomic claim: UPDATE nspd_geo_jobs SET status='running', heartbeat_at=NOW() WHERE job_id=:id AND status IN ('queued','running') AND (heartbeat_at IS NULL OR heartbeat_at < NOW() - INTERVAL '6 minutes') RETURNING job_id — if no row returned, another worker owns it, return {skipped:true}. Additionally use 'FOR UPDATE SKIP LOCKED' on the per-target SELECT to prevent two runners claiming the same pending target.

_{deep-audit backend v2 (после PR #1543) · verify+harden, confidence 0.7 · unit B42}

**Severity:** medium · **Категория:** concurrency/idempotency · **Файл:** `backend/app/workers/tasks/nspd_geo.py:374-408, 82-97` process_nspd_geo_job загружает строку job (SELECT * FROM nspd_geo_jobs WHERE job_id, nspd_geo.py:377-384) и вызывает _start_job, который безусловно делает UPDATE ... SET status='running' WHERE job_id=:id (82-95) — НЕТ проверки текущего status='queued', НЕТ FOR UPDATE, НЕТ advisory/Redis lock, и НЕТ acks_late/dedup в celery_app.py (подтверждено grep'ом — конфиг пуст). Два независимых пути порождают дублирующие сообщения на один job_id: (a) worker_ready resume + per-minute cleanup_zombies beat одновременно (оба делают apply_async после UPDATE→'queued'; cleanup_zombies при этом фильтрует по heartbeat, worker_ready — нет, см. findings 0/1), и (b) перекрытие контейнеров при rolling-redeploy. Prod gunicorn-worker один, но --concurrency=8 (docker-compose.prod.yml:128) → два prefork-child могут снять оба сообщения и крутить один job параллельно. Per-target SELECT (409-424) — LIMIT 1 WHERE status='pending' ORDER BY target_id БЕЗ FOR UPDATE SKIP LOCKED, так что оба runner'а берут один и тот же pending-target. ИМПАКТ: (1) один cad_num тянется из WAF-throttled NSPD двумя runner'ами сразу → удвоение rate и риска бана; (2) счётчики job-строки (targets_done/waf_blocked_count/requests_count) ведутся как независимые in-memory аккумуляторы на runner и флашатся через _heartbeat/_finish_job → last-writer-wins clobber, done может уехать за targets_total → телеметрия/прогресс в admin неверны. _persist_target UPSERT'ит идемпотентно, поэтому сами гео-данные не повреждаются — отсюда medium. ФИКС: атомарный claim в начале process_nspd_geo_job — UPDATE nspd_geo_jobs SET status='running' WHERE job_id=:id AND status IN ('queued','running'-с-устаревшим-hb) RETURNING; при 0 строк — выходить; либо Redis-lock per job_id по образцу _region_lock в resume_kn_run (scrape_kn.py:120). Для target-select добавить FOR UPDATE SKIP LOCKED. **Почему баг:** Concurrent execution of one job means: (1) the same cad_num is fetched from the WAF-throttled NSPD service by two runners at once, doubling request rate and ban risk; (2) job-row counters (targets_done, waf_blocked_count, requests_count) are maintained as independent in-memory accumulators per runner and flushed via _heartbeat/_finish_job — last-writer-wins clobbers and can drive done past targets_total, so the admin progress/telemetry is wrong. _persist_target UPSERTs make the stored geo data itself idempotent, which limits this to wasted/risky requests + corrupted counters rather than data corruption, hence medium. **Как чинить:** Make the start an atomic claim: UPDATE nspd_geo_jobs SET status='running', heartbeat_at=NOW() WHERE job_id=:id AND status IN ('queued','running') AND (heartbeat_at IS NULL OR heartbeat_at < NOW() - INTERVAL '6 minutes') RETURNING job_id — if no row returned, another worker owns it, return {skipped:true}. Additionally use 'FOR UPDATE SKIP LOCKED' on the per-target SELECT to prevent two runners claiming the same pending target. --- <sub>deep-audit backend v2 (после PR #1543) · verify+harden, confidence 0.7 · unit B42</sub>