From edbaca1e8761c118f7d83e8e1dbcd9dc9bab22b8 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Thu, 20 Aug 2026 08:54:26 +0000 Subject: [PATCH] =?UTF-8?q?fix(ptica):=20=D1=83=D1=81=D0=BF=D0=B5=D1=85?= =?UTF-8?q?=D0=B8=20=D0=B1=D0=B0=D1=82=D1=87=D0=B0=20=D0=BA=D0=B0=D1=82?= =?UTF-8?q?=D0=B0=D0=BB=D0=BE=D0=B3=D0=B0=20=D1=84=D0=B8=D0=BA=D1=81=D0=B8?= =?UTF-8?q?=D1=80=D1=83=D1=8E=D1=82=D1=81=D1=8F=20=D0=BF=D0=BE=20=D1=85?= =?UTF-8?q?=D0=BE=D0=B4=D1=83,=20=D0=B0=20=D0=BD=D0=B5=20=D0=BE=D0=B4?= =?UTF-8?q?=D0=BD=D0=B8=D0=BC=20commit'=D0=BE=D0=BC=20=D0=B2=20=D0=BA?= =?UTF-8?q?=D0=BE=D0=BD=D1=86=D0=B5=20(#2464)=20(#2960)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../services/scrapers/domrf_catalog_object.py | 21 ++- .../test_2464_catalog_batch_commits.py | 138 ++++++++++++++++++ 2 files changed, 156 insertions(+), 3 deletions(-) create mode 100644 backend/tests/services/scrapers/test_2464_catalog_batch_commits.py diff --git a/backend/app/services/scrapers/domrf_catalog_object.py b/backend/app/services/scrapers/domrf_catalog_object.py index ad7ef832..86d1310b 100644 --- a/backend/app/services/scrapers/domrf_catalog_object.py +++ b/backend/app/services/scrapers/domrf_catalog_object.py @@ -462,12 +462,27 @@ async def scrape_catalog_objects( ok = await scrape_catalog_object(db, session, obj_id, snapshot_date) if ok: stats["succeeded"] += 1 + # Фиксируем сразу, а не одним commit'ом в конце (#2464). Раньше весь + # батч жил в одной незакоммиченной транзакции, и любой отказ ПОСЛЕ + # цикла — исключение в BrowserSession.__aexit__, снятие Celery-таски, + # перезапуск контейнера — обнулял все уже успешные UPDATE'ы. + # Это не теория: беговой режим здесь force=True («Загрузить все»), + # то есть SQL без LIMIT. На 20.08.2026 в очереди 13200 объектов из + # 13801 — многочасовой прогон, где отказ в конце стоил бы всего. + # SAVEPOINT внутри scrape_catalog_object к этому моменту уже снят, + # поэтому commit здесь корректен. + try: + db.commit() + except Exception: + db.rollback() + raise else: stats["failed"] += 1 - # Commit outer transaction: SAVEPOINT (`begin_nested`) releases внутри loop, - # но outer tx остаётся autobegin'd — без commit() все UPDATE'ы откатятся - # при db.close() в Celery task. + # Финальный commit. Успешные строки зафиксированы по ходу цикла (см. выше), но + # этот вызов остаётся: он закрывает транзакцию, которую могли autobegin'ить + # неудачные итерации (их SAVEPOINT откатился, а внешняя транзакция открыта), + # и сохраняет прежнее поведение для вызывающих, которые на него полагались. try: db.commit() except Exception: diff --git a/backend/tests/services/scrapers/test_2464_catalog_batch_commits.py b/backend/tests/services/scrapers/test_2464_catalog_batch_commits.py new file mode 100644 index 00000000..a44d7e03 --- /dev/null +++ b/backend/tests/services/scrapers/test_2464_catalog_batch_commits.py @@ -0,0 +1,138 @@ +"""Успешные объекты каталога фиксируются по ходу батча, а не одним commit'ом в конце (#2464). + +`scrape_catalog_objects` держал весь батч в одной незакоммиченной транзакции: `db.commit()` +стоял ПОСЛЕ блока `async with BrowserSession(...)`. Любой отказ после цикла — исключение в +`BrowserSession.__aexit__`, снятие Celery-таски, перезапуск контейнера — обнулял все уже +успешные UPDATE'ы. + +Масштаб не гипотетический. Беговой режим здесь `force=True` («Загрузить все»), то есть SQL +без LIMIT: на 20.08.2026 в очереди 13200 объектов из 13801 (последний успешный скрейп — +19.05, до блокировки DOM.РФ, #2443). Многочасовой прогон, где отказ в конце стоил бы всего. + +Тест не ходит в сеть и в БД: `BrowserSession` и `scrape_catalog_object` подменяются, сессия +БД — счётчик вызовов. Проверяется поведение — доживают ли успехи до фиксации, — а не +наличие новых символов. +""" + +from __future__ import annotations + +import os + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +import asyncio +from datetime import date +from typing import Any +from unittest.mock import patch + +import pytest + + +class _RecordingSession: + """Сессия-счётчик: помнит порядок commit/rollback.""" + + def __init__(self) -> None: + self.events: list[str] = [] + + def commit(self) -> None: + self.events.append("commit") + + def rollback(self) -> None: + self.events.append("rollback") + + @property + def commits(self) -> int: + return self.events.count("commit") + + +class _FakeBrowserSession: + """Подмена BrowserSession: ничего не делает, но умеет упасть на выходе.""" + + raise_on_exit: BaseException | None = None + + def __init__(self, *_a: Any, **_kw: Any) -> None: + pass + + async def __aenter__(self) -> _FakeBrowserSession: + return self + + async def __aexit__(self, *_exc: Any) -> None: + if _FakeBrowserSession.raise_on_exit is not None: + exc = _FakeBrowserSession.raise_on_exit + _FakeBrowserSession.raise_on_exit = None + raise exc + + async def warm_up(self) -> None: + return None + + +def _run(db: _RecordingSession, obj_ids: list[int], *, fail_on_exit: bool = False) -> Any: + from app.services.scrapers import domrf_catalog_object as mod + + async def _fake_scrape(_db: Any, _s: Any, obj_id: int, _d: date) -> bool: + # Чётные — успех, нечётные — неудача: проверяем, что фиксируются именно успехи. + return obj_id % 2 == 0 + + _FakeBrowserSession.raise_on_exit = ( + RuntimeError("падение на выходе из сессии") if fail_on_exit else None + ) + + with ( + patch.object(mod, "BrowserSession", _FakeBrowserSession), + patch.object(mod, "scrape_catalog_object", _fake_scrape), + ): + return asyncio.run( + mod.scrape_catalog_objects( + db=db, obj_ids=obj_ids, snapshot_date=date(2026, 8, 20), region_code=66 + ) + ) + + +def test_successes_survive_a_late_failure() -> None: + """Отказ ПОСЛЕ цикла не должен стоить уже собранных объектов. + + На origin/main commit стоит после `async with`, поэтому исключение в __aexit__ + случается ДО единственной фиксации — ни один UPDATE не доживает. + """ + db = _RecordingSession() + + with pytest.raises(RuntimeError, match="падение на выходе"): + _run(db, [2, 4, 6, 8], fail_on_exit=True) + + assert db.commits >= 4, ( + f"зафиксировано {db.commits} раз при 4 успешных объектах — успехи не пережили " + "отказ после цикла: весь батч висел в одной транзакции" + ) + + +def test_commit_happens_per_success_not_per_object() -> None: + """Фиксируются успехи, а не каждая итерация — неудачные строки коммитить нечего.""" + db = _RecordingSession() + stats = _run(db, [1, 2, 3, 4, 5, 6]) + + assert stats["succeeded"] == 3 + assert stats["failed"] == 3 + # 3 успеха по ходу + финальная фиксация в конце. + assert db.commits == 4, f"ожидали 3 по ходу + 1 финальную, получили {db.commits}" + + +def test_empty_batch_touches_nothing() -> None: + """Контроль: пустой список не открывает и не фиксирует ничего.""" + db = _RecordingSession() + stats = _run(db, []) + + assert stats["processed"] == 0 + assert db.events == [], f"пустой батч тронул сессию: {db.events}" + + +def test_all_failed_still_closes_transaction() -> None: + """Контроль: батч без единого успеха всё равно закрывает транзакцию. + + Неудачная итерация откатывает свой SAVEPOINT, но внешняя транзакция остаётся + autobegin'нутой — финальный commit обязан остаться на месте. + """ + db = _RecordingSession() + stats = _run(db, [1, 3, 5]) + + assert stats["succeeded"] == 0 + assert db.commits == 1, f"ожидали одну финальную фиксацию, получили {db.commits}"