diff --git a/backend/app/scrapers/nspd_bulk_client.py b/backend/app/scrapers/nspd_bulk_client.py index 53d93c8e..7475abc9 100644 --- a/backend/app/scrapers/nspd_bulk_client.py +++ b/backend/app/scrapers/nspd_bulk_client.py @@ -57,8 +57,12 @@ _CAT_TO_THEMATIC_SEARCH_ID: dict[int, int] = { 39663: 15, # ЕНК } -# Лимит одновременных запросов (глобальный, shared между всеми instance'ами) -_SEMAPHORE = asyncio.Semaphore(3) +# Concurrency cap per HTTP fetch loop. Issue #260 (Sub-PR B re-review): +# module-level asyncio.Semaphore() bind'ится к event loop первого asyncio.run() +# вызова, на втором — RuntimeError: bound to a different event loop. +# search_by_quarter делает N sequential asyncio.run() per layer → semaphore надо +# создавать per-instance в __aenter__ (под текущим running loop). +_SEMAPHORE_LIMIT = 3 # Таймаут подключения и чтения DEFAULT_TIMEOUT = httpx.Timeout(30.0, connect=10.0) @@ -131,6 +135,10 @@ class NSPDBulkClient: self._headers = headers or DEFAULT_HEADERS self._max_retries = max_retries self._client: httpx.AsyncClient | None = None + # Issue #260: semaphore создаётся в __aenter__ под текущий event loop. + # Module-level не работал — sequential asyncio.run() bind'ил его к + # закрытому loop'у первого вызова. + self._sem: asyncio.Semaphore | None = None async def __aenter__(self) -> NSPDBulkClient: # NSPD prod chain contains a self-signed/internal CA on Beget VPS → @@ -144,6 +152,8 @@ class NSPDBulkClient: follow_redirects=True, verify=False, ) + # Создаём semaphore под running loop — safe для sequential asyncio.run(). + self._sem = asyncio.Semaphore(_SEMAPHORE_LIMIT) return self async def __aexit__(self, *_: Any) -> None: @@ -161,12 +171,12 @@ class NSPDBulkClient: NspdBulkRateLimitError при 429 после исчерпания retries. NspdBulkError при прочих 4xx/5xx. """ - if self._client is None: + if self._client is None or self._sem is None: raise RuntimeError("NSPDBulkClient не инициализирован — используй async with") attempt = 0 while True: - async with _SEMAPHORE: + async with self._sem: # Небольшой jitter чтобы N одновременных запросов не начинались в 0мс await asyncio.sleep(0.05) try: diff --git a/backend/app/workers/tasks/nspd_sync.py b/backend/app/workers/tasks/nspd_sync.py index bb49674c..a7d4989a 100644 --- a/backend/app/workers/tasks/nspd_sync.py +++ b/backend/app/workers/tasks/nspd_sync.py @@ -259,6 +259,8 @@ def _upsert_dump( name="tasks.nspd_sync.harvest_quarter", max_retries=3, soft_time_limit=600, # 10 мин — grid-walk: 11 layers × 49 cells × ~70ms ≈ 40s + retries + time_limit=900, # 15 мин hard kill — safety net (PR #260 re-review): task может игнорить + # SoftTimeLimitExceeded → процесс не освобождается. Hard limit гарантирует worker recovery. autoretry_for=(NspdLiteWafError,), retry_backoff=True, retry_backoff_max=120, diff --git a/backend/tests/test_nspd_client.py b/backend/tests/test_nspd_client.py index b7b7479b..9fa82d81 100644 --- a/backend/tests/test_nspd_client.py +++ b/backend/tests/test_nspd_client.py @@ -507,6 +507,40 @@ def _make_fake_http( return fake_http +def _make_fake_grid_walk( + layer_feature_counts: dict[str, int] | None = None, +) -> Any: + """Возвращает fake `NSPDClient.get_features_in_bbox_grid`. + + После Sub-PR B (#260) area/linear layers (territorial_zones, red_lines, + engineering_structures, zouit_*, risk_*) идут через grid-walk вместо + legacy `_http_get_json` → тесты обязаны мокать оба пути, иначе грид-walk + бьёт по живому NSPD API. + """ + counts = layer_feature_counts or {} + from app.services.scrapers.nspd_client import LAYERS as _LAYERS + from app.services.scrapers.nspd_client import NSPDFeature + + id_to_name: dict[int, str] = {v: k for k, v in _LAYERS.items()} + + def fake_grid( + self: Any, + layer_id: int, + bbox: tuple[float, float, float, float], + *, + grid_n: int = 7, + step_m: float = 50.0, + ) -> list[NSPDFeature]: + layer_name = id_to_name.get(layer_id, "unknown") + n = counts.get(layer_name, 1) + return [ + NSPDFeature.from_raw({"id": f"{layer_name}-{i}", "geometry": None, "properties": {}}) + for i in range(n) + ] + + return fake_grid + + def test_search_by_quarter_core_only(monkeypatch: pytest.MonkeyPatch) -> None: """core_only (include_zouit=False, include_risks=False): 1 search + 5 bulk.""" search_calls: list[str] = [] @@ -520,6 +554,11 @@ def test_search_by_quarter_core_only(monkeypatch: pytest.MonkeyPatch) -> None: "app.services.scrapers.nspd_client._http_get_json", _make_fake_http(_LAYER_FEATURE_COUNTS), ) + # Sub-PR B (#260): area layers идут через grid-walk — мокаем оба пути. + monkeypatch.setattr( + "app.services.scrapers.nspd_client.NSPDClient.get_features_in_bbox_grid", + _make_fake_grid_walk(_LAYER_FEATURE_COUNTS), + ) result = NSPDClient().search_by_quarter( "66:41:0204016", include_zouit=False, include_risks=False @@ -566,6 +605,11 @@ def test_search_by_quarter_with_zouit(monkeypatch: pytest.MonkeyPatch) -> None: "app.services.scrapers.nspd_client._http_get_json", _make_fake_http(), ) + # Sub-PR B (#260): zouit_* и area layers идут через grid-walk. + monkeypatch.setattr( + "app.services.scrapers.nspd_client.NSPDClient.get_features_in_bbox_grid", + _make_fake_grid_walk(), + ) result = NSPDClient().search_by_quarter( "66:41:0204016", include_zouit=True, include_risks=False @@ -628,6 +672,11 @@ def test_search_by_quarter_layers_fetched_with_risks(monkeypatch: pytest.MonkeyP "app.services.scrapers.nspd_client._http_get_json", _make_fake_http(), ) + # Sub-PR B (#260): risk_* и area layers идут через grid-walk. + monkeypatch.setattr( + "app.services.scrapers.nspd_client.NSPDClient.get_features_in_bbox_grid", + _make_fake_grid_walk(), + ) result = NSPDClient().search_by_quarter( "66:41:0204016", include_zouit=False, include_risks=True