Проверка отмены закрывает свою транзакцию: свип ДомКлика больше не держит горизонт vacuum на весь обход (#3480)
Виновник по живому снимку pg_stat_activity 17.09 07:48 UTC: 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. По Prometheus за 03-17.09 10 из 12 окон «транзакция > 1 ч» совпадают по началу и концу со свипами ДомКлика. is_cancelled делал SELECT без commit, а вызывающие сразу уходят в сетевую фазу (у ДомКлика — весь fetch_city, до 3 ч). Теперь commit после чтения — в общей точке для всех 12 вызовов; незакоммиченных записей, рассчитанных на откат, перед вызовами нет (перед каждым — update_heartbeat/save_listings с commit или чтения). Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
parent
facdbb2566
commit
fb02e41d77
2 changed files with 99 additions and 1 deletions
|
|
@ -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]
|
||||||
|
|
@ -815,11 +815,19 @@ def honors_cancel(source: str) -> bool:
|
||||||
|
|
||||||
|
|
||||||
def is_cancelled(db: Session, run_id: int) -> 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(
|
row = db.execute(
|
||||||
text("SELECT status FROM scrape_runs WHERE id = :id"),
|
text("SELECT status FROM scrape_runs WHERE id = :id"),
|
||||||
{"id": run_id},
|
{"id": run_id},
|
||||||
).fetchone()
|
).fetchone()
|
||||||
|
db.commit()
|
||||||
return row is not None and row.status == "cancelled"
|
return row is not None and row.status == "cancelled"
|
||||||
|
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue