fix(scheduler): atomic _claim_run via advisory xact lock (Refs #750)
Конкурентные 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.
This commit is contained in:
parent
a74591d7cc
commit
9f950a5f17
2 changed files with 149 additions and 6 deletions
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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 # лок освобождён
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue