diff --git a/backend/app/services/cadastre/bulk_harvest.py b/backend/app/services/cadastre/bulk_harvest.py index 9503bc13..40e7dcae 100644 --- a/backend/app/services/cadastre/bulk_harvest.py +++ b/backend/app/services/cadastre/bulk_harvest.py @@ -111,23 +111,13 @@ async def harvest_quarter( """ result = HarvestResult(quarter=quarter) - # Читаем текущий phase_state из job (для resumability) - job_row = ( - db.execute( - text("SELECT phase_state FROM cadastre_jobs WHERE job_id = :id"), - {"id": job_id}, - ) - .mappings() - .first() - ) - phase_state: dict[str, Any] = {} - if job_row and job_row["phase_state"]: - phase_state = dict(job_row["phase_state"]) - - # Ранний выход если quarter уже полностью обработан - if phase_state.get("phase") == "done": - logger.info("harvest_quarter: quarter=%s уже done (job_id=%s), skip", quarter, job_id) - return result + # NB: `cadastre_jobs.phase_state` шарится между ВСЕМИ parallel Celery workers одного job. + # Раньше тут был early-exit `if phase_state.phase == 'done': return` — но он триггерился + # как только ПЕРВЫЙ quarter записывал phase=done через update_progress callback, и все + # остальные workers (читая ту же shared row) бейлились без работы. Эффект: 95% quarters + # early-exit'или, coverage 1.6% parcels (pilot v7: 5 NSPD req для 50 quarters). + # Idempotency обеспечивается ON CONFLICT DO UPDATE в каждом upsert_* — re-enqueued task + # просто перепишет те же rows, не создавая дубликатов. # ── Phase 1: search_by_quarter snapshot ────────────────────────────────── update_progress({"phase": "snapshot_started", "quarter": quarter}) diff --git a/backend/tests/services/test_cadastre_bulk.py b/backend/tests/services/test_cadastre_bulk.py index 74b6f999..5431bfe6 100644 --- a/backend/tests/services/test_cadastre_bulk.py +++ b/backend/tests/services/test_cadastre_bulk.py @@ -349,36 +349,64 @@ def test_upsert_parcel_uses_cast_not_double_colon() -> None: @pytest.mark.asyncio -async def test_harvest_quarter_phase_done_early_return() -> None: - """Если phase_state='done' — harvest_quarter возвращается ранним выходом.""" +async def test_harvest_quarter_does_not_early_exit_on_shared_phase_done() -> None: + """harvest_quarter НЕ должен бейлиться если другой worker записал phase=done + в shared `cadastre_jobs.phase_state`. Раньше тут был early-exit, который + вызывал 95% workers пропускать работу (race condition между parallel Celery + tasks одного job). + + Idempotency обеспечивается ON CONFLICT DO UPDATE в upsert_*, поэтому даже + если task re-enqueued после shared phase=done, work должна выполниться. + """ from app.services.cadastre.bulk_harvest import harvest_quarter + snapshot = QuarterSnapshot( + quarter="66:41:0303161", + fetched_at="2026-05-15T10:00:00+00:00", + features=[], + meta_counts={}, + ) + db = MagicMock() + # Симулируем shared phase_state с phase=done от ДРУГОГО quarter db.execute = MagicMock( return_value=MagicMock( mappings=MagicMock( return_value=MagicMock( - first=MagicMock(return_value={"phase_state": {"phase": "done"}}) + first=MagicMock( + return_value={"phase_state": {"phase": "done", "quarter": "66:41:9999999"}} + ) ) ) ) ) + db.commit = MagicMock() client = AsyncMock() - progress_calls: list[Any] = [] + client.search_by_quarter = AsyncMock(return_value=snapshot) - await harvest_quarter( - db=db, - client=client, - quarter="66:41:0303161", - job_id=1, - update_progress=lambda s: progress_calls.append(s), - ) + with patch("app.services.cadastre.bulk_harvest.upsert_features") as mock_upsert: + mock_upsert.return_value = { + "parcels": 0, + "buildings": 0, + "constructions": 0, + "oncs": 0, + "enks": 0, + "zouit": 0, + "skipped": 0, + } + result = await harvest_quarter( + db=db, + client=client, + quarter="66:41:0303161", + job_id=1, + update_progress=lambda s: None, + ) - # client.search_by_quarter не должен вызываться - client.search_by_quarter.assert_not_called() - # Нет progress calls - assert len(progress_calls) == 0 + # search_by_quarter ДОЛЖЕН быть вызван несмотря на shared phase=done + client.search_by_quarter.assert_awaited_once_with("66:41:0303161") + assert result.snapshot_requests == 1 + assert result.phase_state == {"phase": "done", "quarter": "66:41:0303161"} @pytest.mark.asyncio