From d11c8d47e045d7e1bfbfd23487c9f670ffd23129 Mon Sep 17 00:00:00 2001 From: Light1YT Date: Sat, 13 Jun 2026 14:14:56 +0500 Subject: [PATCH] =?UTF-8?q?fix(workers):=20per-prefix=20db.commit()=20?= =?UTF-8?q?=D0=B2=20gknspecial=5Fharvest=20=D0=B4=D0=BB=D1=8F=20durability?= =?UTF-8?q?=20(#1116)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit harvest_gknspecial_zones делал единственный commit в конце всех 8 prefix- циклов: WorkerLostError (OOM / рестарт хоста) в середине прогона → ~20k записанных строк откатывались, прогресс ноль. Mirror ird_harvest.py:216 (per-quarter commit): db.commit() после финального logger.info внутри for-prefix loop. SAVEPOINT-per-row уже изолирует битые фичи; per-prefix commit добавляет durability между префиксами. WFS-fail на префиксе → continue → commit не делается; предыдущие префиксы уже durable. +2 теста (durability, WFS-fail skip). 29 gknspecial тестов зелёные. Closes #1116 --- .../app/workers/tasks/gknspecial_harvest.py | 7 +- .../workers/tasks/test_gknspecial_harvest.py | 65 +++++++++++++++++++ 2 files changed, 71 insertions(+), 1 deletion(-) diff --git a/backend/app/workers/tasks/gknspecial_harvest.py b/backend/app/workers/tasks/gknspecial_harvest.py index dd1234d6..29fdd539 100644 --- a/backend/app/workers/tasks/gknspecial_harvest.py +++ b/backend/app/workers/tasks/gknspecial_harvest.py @@ -15,6 +15,8 @@ Sentinel source_layer_id'ы (вне диапазона НСПД): 990001–99000 Conventions (зеркало ird_harvest.py): • SessionLocal() + try/finally close. • SAVEPOINT per-row (with db.begin_nested():) — одна битая фича не валит батч. + • db.commit() per-prefix (durability) — WorkerLost в середине прогона не теряет + уже записанные префиксы (#1116). • CAST(:x AS type) — НИКОГДА :x::type (psycopg v3). • ON CONFLICT (source_layer_id, geom_data_id) DO UPDATE. • content_* пишем мы же (геопортал как источник); content НЕ трогается НСПД-harvest'ом. @@ -279,8 +281,11 @@ def harvest_gknspecial_zones() -> dict[str, int]: len(feats), n_written, ) + # Durable per-prefix commit: WorkerLostError (OOM / рестарт хоста) в середине + # прогона не теряет уже записанные префиксы (#1116). Mirror ird_harvest.py. + db.commit() - db.commit() + db.commit() # финальный — no-op если всё уже коммитнуто per-prefix (подстраховка) logger.info("gknspecial_harvest: готово, types=%d features=%d", n_types, n_features) return {"types": n_types, "features": n_features} finally: diff --git a/backend/tests/workers/tasks/test_gknspecial_harvest.py b/backend/tests/workers/tasks/test_gknspecial_harvest.py index 8a70e78e..76efe332 100644 --- a/backend/tests/workers/tasks/test_gknspecial_harvest.py +++ b/backend/tests/workers/tasks/test_gknspecial_harvest.py @@ -256,6 +256,71 @@ def test_returns_types_and_features_counts() -> None: assert result["features"] == 2 +def test_per_prefix_commit_durability() -> None: + """db.commit вызван N+1 раз: после каждого из 8 префиксов + финальный (#1116). + + Durability: WorkerLostError в середине прогона не теряет уже записанные префиксы. + """ + feat = _make_feature(zone_state_v="Действующий") + # 8 префиксов: первый с фичей, остальные пустые + feats_per_prefix: list[list[Any]] = [[feat]] + [[] for _ in range(7)] + call_idx: list[int] = [0] + + def fake_by_cql(layer: str, cql_filter: str) -> list[Any]: + idx = call_idx[0] + call_idx[0] += 1 + return feats_per_prefix[idx] if idx < len(feats_per_prefix) else [] + + mock_db = MagicMock() + mock_db.begin_nested.return_value.__enter__ = MagicMock(return_value=None) + mock_db.begin_nested.return_value.__exit__ = MagicMock(return_value=False) + + with ( + patch("app.workers.tasks.gknspecial_harvest.EKBGeoportalClient") as mock_client_cls, + patch("app.workers.tasks.gknspecial_harvest.SessionLocal", return_value=mock_db), + ): + mock_client = MagicMock() + mock_client.features_by_cql.side_effect = fake_by_cql + mock_client_cls.return_value = mock_client + harvest_gknspecial_zones() + + # 8 per-prefix commit + 1 финальный (подстраховка) = 9 + assert mock_db.commit.call_count == 9 + + +def test_per_prefix_commit_skipped_on_wfs_failure() -> None: + """Если WFS падает на префиксе — commit за этот префикс не делается (continue). + + Проверяем что commit'ов столько же, сколько успешно обработанных префиксов + 1 финальный. + """ + feat = _make_feature(zone_state_v="Действующий") + # 8 префиксов: первый успех, второй WFS-fail (sentinel), остальные успех + call_idx: list[int] = [0] + + def fake_by_cql(layer: str, cql_filter: str) -> list[Any]: + idx = call_idx[0] + call_idx[0] += 1 + if idx == 1: + raise RuntimeError("WFS timeout") + return [feat] if idx == 0 else [] + + mock_db = MagicMock() + mock_db.begin_nested.return_value.__enter__ = MagicMock(return_value=None) + mock_db.begin_nested.return_value.__exit__ = MagicMock(return_value=False) + + with ( + patch("app.workers.tasks.gknspecial_harvest.EKBGeoportalClient") as mock_client_cls, + patch("app.workers.tasks.gknspecial_harvest.SessionLocal", return_value=mock_db), + ): + mock_client = MagicMock() + mock_client.features_by_cql.side_effect = fake_by_cql + mock_client_cls.return_value = mock_client + harvest_gknspecial_zones() + + # 7 успешных префиксов × commit + 1 финальный = 8 (failed prefix через continue → нет commit) + assert mock_db.commit.call_count == 8 + + def test_unmatched_feature_written_to_sentinel() -> None: """Фича с full_name не из _PREFIX_MAP → пишется в unmatched (source_layer_id=990099).""" feat = _make_feature(