diff --git a/tradein-mvp/backend/tests/test_3480_no_open_transaction_over_http.py b/tradein-mvp/backend/tests/test_3480_no_open_transaction_over_http.py new file mode 100644 index 00000000..8d1e8011 --- /dev/null +++ b/tradein-mvp/backend/tests/test_3480_no_open_transaction_over_http.py @@ -0,0 +1,90 @@ +"""Проверка отмены не оставляет транзакцию открытой на сетевую фазу свипа (#3480). + +Прод 17.09 07:48 UTC, pg_stat_activity БД tradein: pid 1933154 из tradein-scraper, +'idle in transaction' 1 ч 35 мин, последний запрос `SELECT status FROM scrape_runs +WHERE id = $1` (runs.is_cancelled), xact_start через 30 мс после старта +domclick_city_sweep_moskva 7344. За 14 суток 10 из 12 окон «транзакция > 1 ч» +совпали со свипами ДомКлика. + +Сессия настоящая (SQLAlchemy + SQLite): значение `in_transaction()` — то же, что видит +Postgres как 'idle in transaction'. +""" + +from __future__ import annotations + +import os +from types import SimpleNamespace +from typing import Any +from unittest.mock import MagicMock, patch + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db") + +from scraper_kit.orchestration import pipeline as pl +from scraper_kit.orchestration import runs +from sqlalchemy import create_engine, text +from sqlalchemy.orm import Session + + +def _session() -> Session: + db = Session(create_engine("sqlite://")) + db.execute(text("CREATE TABLE scrape_runs (id INTEGER, status TEXT, counters TEXT)")) + db.execute( + text( + "INSERT INTO scrape_runs VALUES " + "(7333, 'done', NULL), (7344, 'running', NULL), (7345, 'cancelled', NULL)" + ) + ) + db.commit() + return db + + +def test_is_cancelled_closes_its_transaction() -> None: + db = _session() + + assert runs.is_cancelled(db, 7344) is False + assert db.in_transaction() is False + assert runs.is_cancelled(db, 7345) is True + assert db.in_transaction() is False + + +async def test_domclick_sweep_http_phase_runs_outside_transaction() -> None: + db = _session() + seen: list[bool] = [] + + class _Scraper: + blocked = False + geo_filtered = fetch_errors = buckets_completed = buckets_total = 0 + bucket_start_index = 0 + completed_buckets: list[str] = [] # noqa: RUF012 + + def __init__(self, *_a: Any, **_kw: Any) -> None: ... + + async def __aenter__(self) -> _Scraper: + return self + + async def __aexit__(self, *_e: Any) -> None: + return None + + async def fetch_city(self, **_kw: Any) -> list[Any]: + seen.append(db.in_transaction()) # момент сетевого обхода + return [] + + with ( + patch.object(pl, "DomClickScraper", _Scraper), + patch.object(runs, "update_heartbeat", MagicMock()), + patch.object(runs, "mark_done", MagicMock()), + patch.object(runs, "mark_failed", MagicMock()), + patch.object(runs, "mark_banned", MagicMock()), + ): + await pl.run_domclick_city_sweep( + db, + run_id=7344, + config=SimpleNamespace(browser_http_endpoint="http://x:9000"), + matcher=MagicMock(), + city_id=4, + pages=1, + request_delay_sec=0.0, + resume_run_id=7333, # резюм-SELECT тоже открывает транзакцию + ) + + assert seen == [False] diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/runs.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/runs.py index 4b38558a..b2d32d91 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/runs.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/runs.py @@ -815,11 +815,19 @@ def honors_cancel(source: str) -> bool: def is_cancelled(db: Session, run_id: int) -> bool: - """Проверить status='cancelled' (cooperative cancel в long-running pipeline).""" + """Проверить status='cancelled' (cooperative cancel в long-running pipeline). + + #3480: SELECT открывает транзакцию (autobegin), а вызывающие сразу уходят в сетевую + фазу. Без commit сессия висела 'idle in transaction' весь обход и держала горизонт + vacuum всей БД: прод 17.09, domclick_city_sweep_moskva 7344 — транзакция 1.6 ч с + последним запросом ровно этим SELECT; 10 из 12 окон > 1 ч за 14 суток — свипы + ДомКлика. Закрываем здесь: проверку отмены зовут перед каждым долгим await. + """ row = db.execute( text("SELECT status FROM scrape_runs WHERE id = :id"), {"id": run_id}, ).fetchone() + db.commit() return row is not None and row.status == "cancelled"