From c75e6a4cfadcdfbdec4bfe37a4aa8d67881a551f Mon Sep 17 00:00:00 2001 From: Light1YT Date: Sat, 13 Jun 2026 10:36:08 +0500 Subject: [PATCH] fix(scrape_kn): Redis-lock owner-token + resume retry on lock_held (#1216) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Два concurrency-бага в scrape_kn: 1. Lock value="1" без owner-токена + TTL=30мин = заявленной длительности sweep'а (нулевой запас). Sweep > TTL → lock истекает, второй sweep стартует, первый при завершении безусловным r.delete сносит ЧУЖОЙ lock → возможен третий параллельный. 2. SIGKILL/redeploy mid-sweep: finally не выполняется, lock живёт до 30 мин. worker_ready метит run 'zombie' и enqueue'ит resume_kn_run БЕЗ countdown, который ловит lock_held и возвращает skipped без retry → run теряется до недельного beat. Patch: - _region_lock: value = uuid4().hex; release через Lua check-and-delete (`if GET == ARGV[1] then DEL else 0 end`) — не сносим чужой lock. - _LOCK_TTL_SECONDS 30 → 45 мин (полтора max sweep duration, запас). - resume_kn_run: при lock_held → raise self.retry(countdown=300, max_retries=12) → ~час окно для подхвата вместо silent skip. - scrape_kn_region (scheduled) намеренно остаётся skipped — beat поднимет в следующий weekly tick (другая семантика). 11 новых юнит-тестов (token uniqueness, Lua guard, retry-not-skip, existing behaviors). 16/16 scrape_kn тестов зелёные. ruff clean. Closes #1216 --- backend/app/workers/tasks/scrape_kn.py | 71 +++- backend/tests/workers/tasks/test_scrape_kn.py | 317 ++++++++++++++++++ 2 files changed, 373 insertions(+), 15 deletions(-) create mode 100644 backend/tests/workers/tasks/test_scrape_kn.py diff --git a/backend/app/workers/tasks/scrape_kn.py b/backend/app/workers/tasks/scrape_kn.py index 0e37afa8..ddfe6038 100644 --- a/backend/app/workers/tasks/scrape_kn.py +++ b/backend/app/workers/tasks/scrape_kn.py @@ -6,6 +6,8 @@ import asyncio import logging import random import time +import uuid +from collections.abc import Iterator from contextlib import contextmanager from pathlib import Path from typing import Any @@ -18,10 +20,23 @@ from app.workers.celery_app import celery_app logger = logging.getLogger(__name__) -# Lock TTL — full Свердл sweep с extras занимает ~25-30 мин. 30 мин — потолок. -# Если worker умер с lock'ом — он истечёт за полчаса, а не за 2 часа (раньше). -# Heartbeat-based autoclean в /queue endpoint всё равно подхватит зомби-runs. -_LOCK_TTL_SECONDS = 30 * 60 +# Lock TTL — full Свердл sweep с extras занимает ~25-30 мин. Держим 45 мин — +# полтора max sweep duration: запас на хвостовые retry внутри sweep'а и не +# даём lock'у истечь под живым worker'ом (фикс P2 issue #1216 — раньше TTL +# равнялся sweep duration → race на границе). +# Если worker умер с lock'ом — он истечёт за 45 мин, а не за 2 часа (раньше), +# heartbeat-based autoclean в /queue endpoint всё равно подхватит зомби-runs. +_LOCK_TTL_SECONDS = 45 * 60 + +# Lua check-and-delete: освобождаем lock ТОЛЬКО если value совпадает с нашим +# owner-токеном. Защита от P2 race: TTL может истечь под живым sweep'ом, второй +# sweep встанет под тем же key с НОВЫМ токеном; финализация первого без guard +# снесла бы чужой lock → возможен третий параллельный sweep (issue #1216). +_RELEASE_LOCK_LUA = ( + 'if redis.call("GET", KEYS[1]) == ARGV[1] ' + 'then return redis.call("DEL", KEYS[1]) ' + "else return 0 end" +) def _lock_key(region_code: int, developers: list[str] | None) -> str: @@ -102,20 +117,31 @@ def force_release_lock(region_code: int, developers: list[str] | None) -> bool: @contextmanager -def _region_lock(region_code: int, developers: list[str] | None): +def _region_lock(region_code: int, developers: list[str] | None) -> Iterator[bool]: """Redis singleton lock so beat double-fire or two workers can't run the - same (region, devs) sweep concurrently. SET NX EX is atomic.""" + same (region, devs) sweep concurrently. SET NX EX is atomic. + + Owner-token (uuid4 hex) кладём в value и снимаем lock через Lua + check-and-delete — иначе истекший по TTL lock, перехваченный другим + sweep'ом, был бы снесён первым worker'ом при finally → возможен третий + параллельный sweep (P2 race, issue #1216). + """ key = _lock_key(region_code, developers) + token = uuid.uuid4().hex r = redis.Redis.from_url(settings.redis_url) - acquired = r.set(key, "1", nx=True, ex=_LOCK_TTL_SECONDS) + acquired = r.set(key, token, nx=True, ex=_LOCK_TTL_SECONDS) try: yield bool(acquired) finally: if acquired: try: - r.delete(key) - except Exception: - pass + # Lua check-and-delete: атомарно, мимо python GIL → нельзя + # снести чужой lock даже если наш TTL уже истёк. + r.eval(_RELEASE_LOCK_LUA, 1, key, token) + except Exception as e: + # Глотаем намеренно: lock истечёт сам по TTL. Логируем, + # чтобы не молча терять Redis-фейлы (не bare except). + logger.warning("release lock %s failed: %s", key, e) @celery_app.task(bind=True, name="tasks.scrape_kn.scrape_kn_region", max_retries=2) @@ -188,11 +214,18 @@ def scrape_kn_region( ) -@celery_app.task(bind=True, name="tasks.scrape_kn.resume_kn_run", max_retries=2) +@celery_app.task(bind=True, name="tasks.scrape_kn.resume_kn_run", max_retries=12) def resume_kn_run(self: Any, run_id: int) -> dict[str, Any]: """Resume a previously interrupted sweep using objects_snapshot + progress_obj_index from kn_scrape_runs. Triggered by worker_ready hook - when a stale 'running' row is found (heartbeat >5 min).""" + when a stale 'running' row is found (heartbeat >5 min). + + На lock_held делаем self.retry(countdown=300, max_retries=12) — окно + в ~час суммарно. Без retry resume пропускался до недельного beat + (P2 issue #1216): worker_ready помечал run 'zombie' и enqueue'ил resume + без countdown; live sweep держал lock → resume получал skipped и run + терялся до следующего планового запуска. + """ from sqlalchemy import text from app.core.db import SessionLocal @@ -234,12 +267,20 @@ def resume_kn_run(self: Any, run_id: int) -> dict[str, Any]: with _region_lock(region_code, developers) as got_lock: if not got_lock: logger.warning( - "resume_kn_run SKIPPED run=%s region=%s — lock held by another sweep", + "resume_kn_run lock_held run=%s region=%s — retry in 5min" + " (live sweep still holds the lock)", run_id, region_code, ) - _log_task_skipped(region_code, developers, self.request.id, "lock_held") - return {"skipped": True, "reason": "lock_held", "run_id": run_id} + _log_task_skipped(region_code, developers, self.request.id, "lock_held_retry") + # 5 мин × 12 retries = 60 мин окно. Покрывает full sweep + # duration (~30 мин) + запас. Celery поднимет Retry exception, + # task поставится обратно в очередь с countdown=300. + raise self.retry( + exc=RuntimeError(f"lock_held for run={run_id} region={region_code}"), + countdown=300, + max_retries=12, + ) logger.info("resume_kn_run start run=%s region=%s devs=%s", run_id, region_code, developers) return asyncio.run( diff --git a/backend/tests/workers/tasks/test_scrape_kn.py b/backend/tests/workers/tasks/test_scrape_kn.py new file mode 100644 index 00000000..88f93a7f --- /dev/null +++ b/backend/tests/workers/tasks/test_scrape_kn.py @@ -0,0 +1,317 @@ +"""Тесты для backend/app/workers/tasks/scrape_kn.py — Redis-lock owner-token +и resume_kn_run retry on lock_held (P2 issue #1216). + +Покрытие: +- `_region_lock` value содержит уникальный uuid (а не литерал "1"); +- два последовательных вызова получают разные токены (uniqueness); +- `r.eval` вызывается с Lua check-and-delete: правильный key + наш token; +- если в Redis под key лежит ЧУЖОЙ token — Lua возвращает 0 (mock-симуляция); +- `_LOCK_TTL_SECONDS` поднят до 45 мин (полтора max sweep duration); +- `resume_kn_run` при lock_held → `self.retry(countdown=300, max_retries=12)`, + а НЕ silent skip (issue #1216 fix #2). +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import MagicMock, patch + +# ── _region_lock owner-token ────────────────────────────────────────────────── + + +def test_region_lock_value_is_uuid_token_not_literal_one() -> None: + """`r.set` должен получить uuid4-hex (32 hex chars) — НЕ литерал '1'. + + Это фикс fix #1 из issue #1216: без owner-токена finally безусловно + делал `r.delete(key)` → можно было снести ЧУЖОЙ lock после TTL-expiry. + """ + from app.workers.tasks import scrape_kn as mod + + mock_redis = MagicMock() + mock_redis.set.return_value = True # acquired + + with patch.object(mod.redis.Redis, "from_url", return_value=mock_redis): + with mod._region_lock(66, None) as got_lock: + assert got_lock is True + + # `r.set(key, token, nx=True, ex=...)` — token positional[1] + set_args, _set_kwargs = mock_redis.set.call_args + token = set_args[1] + assert token != "1", "lock value must be unique owner-token, not literal '1'" + assert isinstance(token, str) + assert len(token) == 32, f"uuid4.hex is 32 chars, got len={len(token)}" + int(token, 16) # raises if not hex + + +def test_region_lock_tokens_are_unique_across_calls() -> None: + """Два последовательных context-manager call → два РАЗНЫХ токена. + + Иначе owner-check (Lua GET == ARGV[1]) станет бессмысленным — + оба worker'а получили бы одинаковый token, и второй снёс бы lock первого. + """ + from app.workers.tasks import scrape_kn as mod + + mock_redis = MagicMock() + mock_redis.set.return_value = True + + captured_tokens: list[str] = [] + with patch.object(mod.redis.Redis, "from_url", return_value=mock_redis): + with mod._region_lock(66, None): + captured_tokens.append(mock_redis.set.call_args[0][1]) + with mod._region_lock(66, None): + captured_tokens.append(mock_redis.set.call_args[0][1]) + + assert len(captured_tokens) == 2 + assert captured_tokens[0] != captured_tokens[1], "tokens must be unique per call" + + +def test_region_lock_release_uses_lua_eval_with_own_token() -> None: + """В finally вызывается `r.eval(LUA, 1, key, token)` — НЕ `r.delete(key)`. + + Это фикс fix #1 из issue #1216 (check-and-delete семантика). + """ + from app.workers.tasks import scrape_kn as mod + + mock_redis = MagicMock() + mock_redis.set.return_value = True + mock_redis.eval.return_value = 1 # пусть Lua говорит «удалил» + + with patch.object(mod.redis.Redis, "from_url", return_value=mock_redis): + with mod._region_lock(66, ["devA"]): + set_token = mock_redis.set.call_args[0][1] + + # eval должен быть вызван ровно один раз с правильными аргументами. + mock_redis.eval.assert_called_once() + eval_args = mock_redis.eval.call_args[0] + assert eval_args[0] == mod._RELEASE_LOCK_LUA + assert eval_args[1] == 1 # numkeys + assert eval_args[2] == mod._lock_key(66, ["devA"]) # KEYS[1] + assert eval_args[3] == set_token # ARGV[1] — наш token + + # delete (без guard) НЕ должен вызываться — был бы регрессией к старому багу. + mock_redis.delete.assert_not_called() + + +def test_region_lock_release_skipped_when_not_acquired() -> None: + """Если acquire не удался — finally НЕ должен делать eval/delete. + + Иначе worker, получивший skipped, снёс бы lock владельца. + """ + from app.workers.tasks import scrape_kn as mod + + mock_redis = MagicMock() + mock_redis.set.return_value = None # nx=True + key exists → None + + with patch.object(mod.redis.Redis, "from_url", return_value=mock_redis): + with mod._region_lock(66, None) as got_lock: + assert got_lock is False + + mock_redis.eval.assert_not_called() + mock_redis.delete.assert_not_called() + + +def test_release_lock_lua_script_contract() -> None: + """Smoke: Lua script содержит check-and-delete семантику. + + Если script отредактируют наугад — этот guard упадёт. + """ + from app.workers.tasks.scrape_kn import _RELEASE_LOCK_LUA + + assert 'redis.call("GET", KEYS[1])' in _RELEASE_LOCK_LUA + assert "ARGV[1]" in _RELEASE_LOCK_LUA + assert 'redis.call("DEL", KEYS[1])' in _RELEASE_LOCK_LUA + assert "else return 0 end" in _RELEASE_LOCK_LUA + + +def test_region_lock_uses_full_ttl_45_min() -> None: + """`r.set` получает `ex=_LOCK_TTL_SECONDS == 45*60 == 2700`. + + Issue #1216 fix: было 30 мин (== max sweep duration → нулевой запас); + поднимаем до 45 мин = полтора sweep duration. + """ + from app.workers.tasks import scrape_kn as mod + + assert mod._LOCK_TTL_SECONDS == 45 * 60 + + mock_redis = MagicMock() + mock_redis.set.return_value = True + + with patch.object(mod.redis.Redis, "from_url", return_value=mock_redis): + with mod._region_lock(66, None): + pass + + _, set_kwargs = mock_redis.set.call_args + assert set_kwargs["ex"] == 2700 + assert set_kwargs["nx"] is True + + +def test_lua_script_simulation_rejects_foreign_token() -> None: + """Симуляция Lua check-and-delete на python-уровне: + если value под key != нашему token → 0 (не удалили), иначе → 1. + + Гарантирует что контракт скрипта корректен; реальная атомарность — + забота Redis (тестируется на integration-уровне). + """ + from app.workers.tasks.scrape_kn import _RELEASE_LOCK_LUA + + def simulate(stored: str | None, our_token: str) -> int: + # Грубая модель: parse-free assertion на ключевые ветки скрипта. + assert 'if redis.call("GET", KEYS[1]) == ARGV[1]' in _RELEASE_LOCK_LUA + assert "else return 0 end" in _RELEASE_LOCK_LUA + if stored == our_token: + return 1 + return 0 + + own = "a" * 32 + foreign = "b" * 32 + assert simulate(stored=own, our_token=own) == 1 + assert simulate(stored=foreign, our_token=own) == 0 + assert simulate(stored=None, our_token=own) == 0 + + +# ── resume_kn_run retry on lock_held ────────────────────────────────────────── + + +def _patch_session_local(mock_db: MagicMock) -> Any: + """Helper: SessionLocal импортируется внутри resume_kn_run из `app.core.db`, + поэтому патчить надо именно там, а не на module-level scrape_kn.""" + return patch("app.core.db.SessionLocal", return_value=mock_db) + + +def test_resume_kn_run_retries_on_lock_held_not_silent_skip() -> None: + """`resume_kn_run` при lock_held → `self.retry(countdown=300, max_retries=12)`. + + Старое поведение (silent skip → return {'skipped': True}) теряло run до + недельного beat — это и есть P2 issue #1216 fix #2. + + Patch'им `mod.resume_kn_run.retry` напрямую: при bind=True Celery передаёт + task-instance как `self`, и self.retry == task.retry. Подменяем его на + side_effect-exception, ловим, проверяем kwargs. + """ + from app.workers.tasks import scrape_kn as mod + + row = {"region_codes": [66], "developer_ids": None} + mock_result = MagicMock() + mock_result.mappings.return_value.first.return_value = row + mock_db = MagicMock() + mock_db.execute.return_value = mock_result + + mock_redis = MagicMock() + mock_redis.set.return_value = None # lock_held + + class FakeRetryError(Exception): + pass + + with ( + _patch_session_local(mock_db), + patch.object(mod.redis.Redis, "from_url", return_value=mock_redis), + patch.object(mod, "_log_task_received"), + patch.object(mod, "_log_task_skipped"), + patch.object( + mod.resume_kn_run, "retry", side_effect=FakeRetryError("re-queue") + ) as mock_retry, + ): + raised = False + try: + mod.resume_kn_run.run(42) + except FakeRetryError: + raised = True + + assert raised, "resume_kn_run must raise (via self.retry), not return" + mock_retry.assert_called_once() + retry_kwargs = mock_retry.call_args.kwargs + assert retry_kwargs["countdown"] == 300 + assert retry_kwargs["max_retries"] == 12 + assert "exc" in retry_kwargs + + +def test_resume_kn_run_lock_held_does_not_silently_return_skipped() -> None: + """Регрессия-guard: при lock_held НЕ возвращаем dict — поднимаем Retry. + + Подменяем self.retry на sentinel-exception; если код пошёл бы по + старой ветке (silent skip → return), exception не возник бы. + """ + from app.workers.tasks import scrape_kn as mod + + row = {"region_codes": [66], "developer_ids": None} + mock_result = MagicMock() + mock_result.mappings.return_value.first.return_value = row + mock_db = MagicMock() + mock_db.execute.return_value = mock_result + + mock_redis = MagicMock() + mock_redis.set.return_value = None # lock_held + + class SentinelRetryError(Exception): + pass + + result_returned: Any = None + raised = False + with ( + _patch_session_local(mock_db), + patch.object(mod.redis.Redis, "from_url", return_value=mock_redis), + patch.object(mod, "_log_task_received"), + patch.object(mod, "_log_task_skipped"), + patch.object(mod.resume_kn_run, "retry", side_effect=SentinelRetryError()), + ): + try: + result_returned = mod.resume_kn_run.run(42) + except SentinelRetryError: + raised = True + + assert raised, "resume_kn_run must call self.retry (raise Retry), not return" + assert result_returned is None + + +def test_resume_kn_run_run_not_found_still_returns_skipped() -> None: + """Если row не найден — возвращаем `{'skipped': True, 'reason': 'run_not_found'}`. + + Это легитимный skip (run уже почистили), retry бесполезен. Проверяем что + fix #2 не сломал валидные ранние выходы. + """ + from app.workers.tasks import scrape_kn as mod + + mock_result = MagicMock() + mock_result.mappings.return_value.first.return_value = None + mock_db = MagicMock() + mock_db.execute.return_value = mock_result + + with ( + _patch_session_local(mock_db), + patch.object(mod.resume_kn_run, "retry") as mock_retry, + ): + result = mod.resume_kn_run.run(999) + + assert result == {"skipped": True, "reason": "run_not_found", "run_id": 999} + mock_retry.assert_not_called() + + +# ── scrape_kn_region остаётся skipped (а не retry) — это by-design ──────────── + + +def test_scrape_kn_region_lock_held_returns_skipped() -> None: + """`scrape_kn_region` (scheduled beat) при lock_held → return skipped. + + В отличие от resume_kn_run, scheduled run крутится по weekly beat — + пропустить разовый запуск ок, beat поднимет в следующий раз. Retry'ить + тут было бы спамом задач. + """ + from app.workers.tasks import scrape_kn as mod + + mock_redis = MagicMock() + mock_redis.set.return_value = None # lock_held + + with ( + patch.object(mod.redis.Redis, "from_url", return_value=mock_redis), + patch.object(mod, "_log_task_received"), + patch.object(mod, "_log_task_skipped"), + patch.object(mod.settings, "scrape_kn_jitter_seconds", 0), + patch.object(mod.settings, "scrape_kn_state_path", ""), + patch.object(mod.scrape_kn_region, "retry") as mock_retry, + ): + result = mod.scrape_kn_region.run(66, None, True) + + assert result["skipped"] is True + assert result["reason"] == "lock_held" + assert result["region_code"] == 66 + mock_retry.assert_not_called()