fix(#231): address re-review blockers — per-instance semaphore + time_limit + test mocks

PR #260 second-opinion review 🔴 BLOCK → fixes:

1. nspd_bulk_client.py: _SEMAPHORE module-level → per-instance self._sem
   создаётся в __aenter__ под текущий event loop. Sequential asyncio.run()
   в search_by_quarter (11 layers × asyncio.run() per layer) bind'ил
   module-level semaphore к first loop'у → RuntimeError на втором вызове →
   все cells return_exceptions=True проглатывали → grid возвращал 0 features.

2. nspd_sync.py: harvest_quarter добавлен time_limit=900 (hard) рядом с
   soft_time_limit=600. Safety net если task игнорит SoftTimeLimitExceeded.

3. tests/test_nspd_client.py: добавлен _make_fake_grid_walk helper +
   monkeypatch NSPDClient.get_features_in_bbox_grid в 3 тестах
   (search_by_quarter_core_only / _with_zouit / _layers_fetched_with_risks).
   Без этого area layers били бы по живому NSPD API → CI flaky.

Tests: 37 passed, 1 skipped (3 area-dispatch + 34 grid-walk).
This commit is contained in:
lekss361 2026-05-17 10:13:34 +03:00
parent fdced94626
commit 8191a85102
3 changed files with 65 additions and 4 deletions

View file

@ -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:

View file

@ -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,

View file

@ -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