From 9f950a5f17871a76f3c658c61680058639b5eed3 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Sat, 30 May 2026 19:01:37 +0300 Subject: [PATCH] fix(scheduler): atomic _claim_run via advisory xact lock (Refs #750) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Конкурентные scheduler-тики одного source (redeploy-overlap старого+нового контейнера / 2 реплики) оба проходили has_running_run pre-check и запускали ДВОЙНОЙ city-sweep → 2x rate к Avito/Cian → бан скрапера. Path A: transaction-scoped pg_try_advisory_xact_lock(hashtext(source)) после pre-check сериализует claim; non-blocking (lock занят → return None). Под локом double-checked has_running_run (running мог появиться в окне pre-check→lock). Лок авто-снимается на commit/rollback. Path B (partial unique index WHERE status='running') отклонён: миграция могла упасть на deploy если dirty-данные уже содержат >=2 running-строк на source (ровно то, что баг и создаёт) — рискованно перед demo. Tests: double-call→один run; lock unavailable→skip; running-under-lock→rollback+skip. --- tradein-mvp/backend/app/services/scheduler.py | 22 +++ tradein-mvp/backend/tests/test_scheduler.py | 133 +++++++++++++++++- 2 files changed, 149 insertions(+), 6 deletions(-) diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 32956e49..83cb5c62 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -133,6 +133,28 @@ def _claim_run(db: Session, schedule_row: dict[str, Any]) -> int | None: logger.info("scheduler: skip — already running for source=%s", source) return None + # #750: атомарный claim через transaction-scoped advisory lock. Сериализует + # конкурентные _claim_run одного source (redeploy-overlap старого+нового контейнера + # / 2 реплики), иначе оба пройдут has_running_run-pre-check и запустят ДВОЙНОЙ sweep + # → 2× rate к Avito/Cian → бан. pg_try_advisory_xact_lock НЕ блокирует: lock занят + # другим тиком → false → return None. Лок авто-снимается на commit/rollback этой + # транзакции (has_running_run выше — быстрый pre-check, обычно без взятия лока). + got_lock = db.execute( + text("SELECT pg_try_advisory_xact_lock(hashtext(:source))"), + {"source": source}, + ).scalar() + if not got_lock: + logger.info("scheduler: skip — concurrent claim in progress for source=%s", source) + return None + + # Double-checked под локом: конкурентный тик мог закоммитить running-run МЕЖДУ + # pre-check и взятием лока (затем освободить lock на commit). Перепроверяем под + # локом — иначе словим двойной run на узком окне. + if has_running_run(db, source): + logger.info("scheduler: skip — running appeared under lock for source=%s", source) + db.rollback() # освобождаем advisory lock (claim не состоялся) + return None + params = schedule_row.get("default_params") or {} run_id = runs_mod.create_run(db, source=source, params=params) diff --git a/tradein-mvp/backend/tests/test_scheduler.py b/tradein-mvp/backend/tests/test_scheduler.py index 2f04b0ed..a977d4bb 100644 --- a/tradein-mvp/backend/tests/test_scheduler.py +++ b/tradein-mvp/backend/tests/test_scheduler.py @@ -1,9 +1,10 @@ """Offline smoke for scheduler logic.""" + import os os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") -from datetime import datetime, timezone +from datetime import UTC, datetime import pytest @@ -11,14 +12,14 @@ from app.services.scheduler import ZOMBIE_THRESHOLD_HOURS, compute_next_run_at def test_compute_next_run_in_normal_window() -> None: - now = datetime(2026, 5, 23, 12, 0, tzinfo=timezone.utc) + now = datetime(2026, 5, 23, 12, 0, tzinfo=UTC) next_at = compute_next_run_at(2, 5, now=now) assert next_at.date() == datetime(2026, 5, 24).date() assert 2 <= next_at.hour <= 4 # window [2, 5) UTC def test_compute_next_run_cross_midnight() -> None: - now = datetime(2026, 5, 23, 12, 0, tzinfo=timezone.utc) + now = datetime(2026, 5, 23, 12, 0, tzinfo=UTC) # Window 22..3 = 22:00-23:59 + 00:00-02:59 next_at = compute_next_run_at(22, 3, now=now) # Должен попасть либо в [22,23] либо в [0,3) range @@ -27,7 +28,7 @@ def test_compute_next_run_cross_midnight() -> None: def test_compute_next_run_window_1_hour() -> None: - now = datetime(2026, 5, 23, 12, 0, tzinfo=timezone.utc) + now = datetime(2026, 5, 23, 12, 0, tzinfo=UTC) # Минимальное окно [3, 4) → ровно 3-й час next_at = compute_next_run_at(3, 4, now=now) assert next_at.hour == 3 @@ -39,13 +40,133 @@ def test_zombie_threshold_constant() -> None: def test_compute_next_run_is_future() -> None: """next_run_at всегда в будущем (завтра) — не в прошлом.""" - now = datetime(2026, 5, 23, 3, 30, tzinfo=timezone.utc) + now = datetime(2026, 5, 23, 3, 30, tzinfo=UTC) next_at = compute_next_run_at(2, 5, now=now) assert next_at > now def test_compute_next_run_timezone_aware() -> None: """Результат всегда timezone-aware (UTC).""" - now = datetime(2026, 5, 23, 12, 0, tzinfo=timezone.utc) + now = datetime(2026, 5, 23, 12, 0, tzinfo=UTC) next_at = compute_next_run_at(2, 5, now=now) assert next_at.tzinfo is not None + + +# ── #750: атомарный _claim_run (advisory lock — anti-double-sweep) ────────── + +from collections import deque # noqa: E402 + +from app.services import scheduler as _sched # noqa: E402 + + +class _Scalar: + def __init__(self, v: object) -> None: + self._v = v + + def scalar(self) -> object: + return self._v + + +class _Fetchone: + def __init__(self, row: object) -> None: + self._row = row + + def fetchone(self) -> object: + return self._row + + +class _FakeSchedulerDB: + """Stateful fake моделирующий advisory-lock claim (#750), без реальной БД. + + - SELECT pg_try_advisory_xact_lock → lock_available. + - SELECT 1 FROM scrape_runs (has_running_run) → из running_results (если задано), + иначе по флагу self.running[source]. + - UPDATE scrape_schedules → no-op. + create_run (монкипатчится) выставляет self.running[source]=True. + """ + + def __init__(self, *, lock_available: bool = True, running_results: list[bool] | None = None): + self.lock_available = lock_available + self.running: dict[str, bool] = {} + self._running_results: deque[bool] = deque(running_results or []) + self.commits = 0 + self.rollbacks = 0 + + def execute(self, stmt: object, params: dict | None = None): + sql = str(stmt) + if "pg_try_advisory_xact_lock" in sql: + return _Scalar(self.lock_available) + if "SELECT 1 FROM scrape_runs" in sql: + if self._running_results: + return _Fetchone(object() if self._running_results.popleft() else None) + src = (params or {}).get("source") + return _Fetchone(object() if self.running.get(src) else None) + return _Fetchone(None) # UPDATE scrape_schedules etc. + + def commit(self) -> None: + self.commits += 1 + + def rollback(self) -> None: + self.rollbacks += 1 + + +_SCHED_ROW = { + "source": "avito_city_sweep", + "window_start_hour": 2, + "window_end_hour": 5, + "default_params": {}, +} + + +def test_claim_run_double_call_creates_one_run(monkeypatch: pytest.MonkeyPatch) -> None: + """#750 DoD: два последовательных _claim_run одного source (нет running) → + первый run_id, второй None (create_run вызван РОВНО один раз).""" + db = _FakeSchedulerDB() + calls = {"n": 0} + + def fake_create_run(_d, *, source, params): + calls["n"] += 1 + db.running[source] = True + return 100 + calls["n"] + + monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run) + + rid1 = _sched._claim_run(db, _SCHED_ROW) + rid2 = _sched._claim_run(db, _SCHED_ROW) + + assert rid1 == 101 + assert rid2 is None + assert calls["n"] == 1 # второй НЕ создал run + + +def test_claim_run_skips_when_advisory_lock_unavailable(monkeypatch: pytest.MonkeyPatch) -> None: + """#750: lock занят конкурентным тиком → return None, create_run НЕ вызван.""" + db = _FakeSchedulerDB(lock_available=False) + calls = {"n": 0} + + def fake_create_run(_d, *, source, params): + calls["n"] += 1 + return 999 + + monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run) + + assert _sched._claim_run(db, _SCHED_ROW) is None + assert calls["n"] == 0 + + +def test_claim_run_rechecks_running_under_lock(monkeypatch: pytest.MonkeyPatch) -> None: + """#750: running появился МЕЖДУ pre-check и взятием лока (double-checked) → + return None + rollback (освобождение лока), create_run НЕ вызван.""" + # pre-check → False (нет running), re-check под локом → True (появился). + db = _FakeSchedulerDB(lock_available=True, running_results=[False, True]) + calls = {"n": 0} + + def fake_create_run(_d, *, source, params): + calls["n"] += 1 + return 999 + + monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run) + + assert _sched._claim_run(db, _SCHED_ROW) is None + assert calls["n"] == 0 + assert db.rollbacks == 1 # лок освобождён