Дробить плитку Overpass только при перегрузке, а не при отказе сети (#3528)

backend/app/services/site_finder/poi_loader.py

@@ -141,6 +141,47 @@ def _split_bbox_quadrants(
 # дефолтного региона не меняется.
 RECURSIVE_SPLIT_MAX_DEPTH = 3
 
+# Дробить тайл имеет смысл ТОЛЬКО когда сервер отказал из-за тяжести запроса:
+# 504/429/503 и таймаут чтения — «я не успел посчитать», четверть посчитается.
+# Отказ на уровне транспорта (connection refused / network unreachable) про размер
+# запроса не говорит ВООБЩЕ: хост нас не принимает, и дробление превращает один
+# отказ в 4, 16, 64 повторных стука. Живой случай 15.09.2026: загрузка Москвы
+# поймала блокировку overpass-api.de по IP и за три минуты выдала 58 отказов на
+# 4 успеха — ровно этот механизм.
+_SPLIT_WORTHY_STATUS = frozenset({429, 503, 504})
+
+# Подряд идущие транспортные отказы = хост нас не принимает. Продолжать прогон
+# бессмысленно и вредно (углубляем блокировку), поэтому после порога — стоп всего
+# прогона с явной ошибкой, а не тихий пропуск категорий.
+_MAX_CONSECUTIVE_TRANSPORT_ERRORS = 5
+
+
+class OverpassUnreachableError(RuntimeError):
+    """Overpass отказывает на уровне соединения подряд — прогон остановлен."""
+
+
+class _RunState:
+    """Счётчик подряд идущих транспортных отказов в рамках одного fetch_overpass."""
+
+    __slots__ = ("consecutive_transport_errors",)
+
+    def __init__(self) -> None:
+        self.consecutive_transport_errors = 0
+
+
+def _is_overload(exc: Exception) -> bool:
+    """True, если отказ говорит «запрос слишком тяжёлый» (есть смысл дробить)."""
+    if isinstance(exc, httpx.TimeoutException):
+        return True
+    if isinstance(exc, httpx.HTTPStatusError):
+        return exc.response.status_code in _SPLIT_WORTHY_STATUS
+    return False
+
+
+def _is_transport_error(exc: Exception) -> bool:
+    """True для отказа на уровне соединения (хост не принимает), не про размер запроса."""
+    return isinstance(exc, httpx.TransportError) and not isinstance(exc, httpx.TimeoutException)
+
 
 def _build_overpass_query(
     tag_filters: tuple[tuple[str, str], ...], bbox: tuple[float, float, float, float]
@@ -175,6 +216,7 @@ async def _fetch_category(
     tag_filters: tuple[tuple[str, str], ...],
     category: str,
     bbox: tuple[float, float, float, float],
+    state: _RunState,
     depth: int = 0,
 ) -> list[dict]:
     """Один per-category Overpass-запрос (для ОДНОГО тайла bbox) с ОДНИМ повтором при
@@ -209,15 +251,34 @@ async def _fetch_category(
             # amenity=pharmacy + shop=supermarket) приходит дважды — каждая копия
             # несёт свою category. Иначе _classify по dict-порядку молча терял бы
             # вторую категорию при UPSERT по UNIQUE(osm_type, osm_id, category). См. #1372.
+            state.consecutive_transport_errors = 0
             for el in elements:
                 el["_gd_category"] = category
             return elements
         except Exception as e:
+            if _is_transport_error(e):
+                state.consecutive_transport_errors += 1
+                if state.consecutive_transport_errors >= _MAX_CONSECUTIVE_TRANSPORT_ERRORS:
+                    raise OverpassUnreachableError(
+                        f"Overpass отказывает на уровне соединения "
+                        f"{state.consecutive_transport_errors} раз подряд ({e}) — прогон "
+                        f"остановлен, чтобы не стучаться в блокирующий хост"
+                    ) from e
+                logger.warning(
+                    "Overpass transport error for %s bbox=%s (подряд %d) — тайл пропущен "
+                    "без дробления: %s",
+                    tag_desc,
+                    bbox,
+                    state.consecutive_transport_errors,
+                    e,
+                )
+                return []
+            state.consecutive_transport_errors = 0
             if attempt == 1:
                 logger.warning("Overpass failed for %s (attempt 1, retrying): %s", tag_desc, e)
                 await asyncio.sleep(3.0)
                 continue
-            if depth < RECURSIVE_SPLIT_MAX_DEPTH:
+            if _is_overload(e) and depth < RECURSIVE_SPLIT_MAX_DEPTH:
                 logger.warning(
                     "Overpass failed for %s bbox=%s twice — splitting into 4 quadrants "
                     "(depth %d→%d) instead of dropping the tile: %s",
@@ -230,13 +291,14 @@ async def _fetch_category(
                 combined: list[dict] = []
                 for quadrant in _split_bbox_quadrants(bbox):
                     combined.extend(
-                        await _fetch_category(client, tag_filters, category, quadrant, depth + 1)
+                        await _fetch_category(
+                            client, tag_filters, category, quadrant, state, depth + 1
+                        )
                     )
                     await asyncio.sleep(1.0)
                 return combined
             logger.warning(
-                "Overpass failed for %s bbox=%s at max split depth %d — tile skipped "
-                "this run: %s",
+                "Overpass failed for %s bbox=%s at max split depth %d — tile skipped this run: %s",
                 tag_desc,
                 bbox,
                 depth,
@@ -260,6 +322,7 @@ async def fetch_overpass(region: str = DEFAULT_REGION) -> list[dict]:
     """
     bbox = REGION_BBOX[region]
     tiles = _bbox_tiles(bbox)
+    state = _RunState()
     headers = {
         "User-Agent": "GenDesign-SiteFinder/1.0 (+https://gendsgn.ru)",
         "Accept": "application/json",
@@ -268,7 +331,7 @@ async def fetch_overpass(region: str = DEFAULT_REGION) -> list[dict]:
     async with httpx.AsyncClient(timeout=60, headers=headers) as client:
         for tag_filters, category in OSM_CATEGORIES.items():
             for tile in tiles:
-                elements = await _fetch_category(client, tag_filters, category, tile)
+                elements = await _fetch_category(client, tag_filters, category, tile, state)
                 all_elements.extend(elements)
                 await asyncio.sleep(1.0)
     logger.info(

backend/tests/test_poi_loader.py

@@ -15,28 +15,39 @@ from __future__ import annotations
 
 from types import SimpleNamespace
 
+import httpx
 import pytest
 
 from app.services.site_finder.poi_loader import (
+    _MAX_CONSECUTIVE_TRANSPORT_ERRORS,
     DEFAULT_REGION,
     EKB_BBOX,
     RECURSIVE_SPLIT_MAX_DEPTH,
     REGION_BBOX,
+    OverpassUnreachableError,
     _bbox_tiles,
     _build_overpass_query,
     _fetch_category,
+    _RunState,
     _split_bbox_quadrants,
 )
 
 
 class _FakeResponse:
+    """Ответ Overpass. ok=False отдаёт 504 — «запрос слишком тяжёлый», единственный
+    класс отказа, ради которого тайл вообще осмысленно дробить."""
+
     def __init__(self, ok: bool, elements: list[dict] | None = None) -> None:
         self._ok = ok
         self._elements = elements or []
 
     def raise_for_status(self) -> None:
         if not self._ok:
-            raise RuntimeError("simulated Overpass failure")
+            raise httpx.HTTPStatusError(
+                "simulated Overpass overload",
+                request=httpx.Request("POST", "https://overpass-api.de/api/interpreter"),
+                response=httpx.Response(504),
+            )
 
     def json(self) -> dict:
         return {"elements": self._elements}
@@ -123,7 +134,9 @@ async def test_fetch_category_retries_then_succeeds(instant_sleep: None) -> None
         return _FakeResponse(ok=True, elements=[{"type": "node", "id": 1, "lat": 1, "lon": 2}])
 
     client = SimpleNamespace(post=fake_post)
-    result = await _fetch_category(client, (("amenity", "pharmacy"),), "pharmacy", (0, 0, 1, 1))
+    result = await _fetch_category(
+        client, (("amenity", "pharmacy"),), "pharmacy", (0, 0, 1, 1), _RunState()
+    )
     assert calls["n"] == 2
     assert len(result) == 1
     assert result[0]["_gd_category"] == "pharmacy"
@@ -149,7 +162,7 @@ async def test_fetch_category_splits_into_quadrants_on_persistent_failure(
 
     client = SimpleNamespace(post=fake_post)
     result = await _fetch_category(
-        client, (("amenity", "pharmacy"),), "pharmacy", (0.0, 0.0, 1.0, 1.0)
+        client, (("amenity", "pharmacy"),), "pharmacy", (0.0, 0.0, 1.0, 1.0), _RunState()
     )
     # верхний тайл: 2 неудачных attempt, затем 4 успешных запроса по четвертям
     assert calls["n"] == 2 + 4
@@ -170,9 +183,75 @@ async def test_fetch_category_gives_up_at_max_depth_without_infinite_recursion(
 
     client = SimpleNamespace(post=fake_post)
     result = await _fetch_category(
-        client, (("amenity", "pharmacy"),), "pharmacy", (0.0, 0.0, 1.0, 1.0)
+        client, (("amenity", "pharmacy"),), "pharmacy", (0.0, 0.0, 1.0, 1.0), _RunState()
     )
     assert result == []
     # sum_{d=0}^{max_depth} 4^d узлов, каждый по 2 attempt — рекурсия конечна
     expected_nodes = sum(4**d for d in range(RECURSIVE_SPLIT_MAX_DEPTH + 1))
     assert calls["n"] == expected_nodes * 2
+
+
+# ── отказ транспорта: НЕ дробим и не стучимся дальше ──────────────────────────
+
+
+async def test_transport_error_does_not_split_the_tile(instant_sleep: None) -> None:
+    """Connection refused говорит «хост нас не принимает», а не «запрос тяжёлый».
+
+    Дробление тут превращает один отказ в 4, 16, 64 повторных стука — ровно это
+    случилось 15.09.2026 на загрузке Москвы (58 отказов на 4 успеха за три минуты).
+    """
+    calls = {"n": 0}
+
+    async def fake_post(_url: str, data: dict) -> _FakeResponse:
+        calls["n"] += 1
+        assert data
+        raise httpx.ConnectError("[Errno 101] Network is unreachable")
+
+    client = SimpleNamespace(post=fake_post)
+    result = await _fetch_category(
+        client, (("amenity", "pharmacy"),), "pharmacy", (0.0, 0.0, 1.0, 1.0), _RunState()
+    )
+    assert result == []
+    assert calls["n"] == 1  # ни retry, ни четвертей
+
+
+async def test_consecutive_transport_errors_abort_the_run(instant_sleep: None) -> None:
+    """Порог подряд идущих транспортных отказов останавливает ВЕСЬ прогон явной ошибкой."""
+
+    async def fake_post(_url: str, data: dict) -> _FakeResponse:
+        assert data
+        raise httpx.ConnectError("connection refused")
+
+    client = SimpleNamespace(post=fake_post)
+    state = _RunState()
+    for _ in range(_MAX_CONSECUTIVE_TRANSPORT_ERRORS - 1):
+        assert (
+            await _fetch_category(
+                client, (("amenity", "pharmacy"),), "pharmacy", (0.0, 0.0, 1.0, 1.0), state
+            )
+            == []
+        )
+    with pytest.raises(OverpassUnreachableError):
+        await _fetch_category(
+            client, (("amenity", "pharmacy"),), "pharmacy", (0.0, 0.0, 1.0, 1.0), state
+        )
+
+
+async def test_successful_tile_resets_transport_error_streak(instant_sleep: None) -> None:
+    """Одиночные сетевые икоты вперемешку с успехами не должны копиться до аварии."""
+    calls = {"n": 0}
+
+    async def fake_post(_url: str, data: dict) -> _FakeResponse:
+        calls["n"] += 1
+        assert data
+        if calls["n"] % 2:
+            raise httpx.ConnectError("hiccup")
+        return _FakeResponse(ok=True, elements=[{"type": "node", "id": calls["n"]}])
+
+    client = SimpleNamespace(post=fake_post)
+    state = _RunState()
+    for _ in range(_MAX_CONSECUTIVE_TRANSPORT_ERRORS * 2):
+        await _fetch_category(
+            client, (("amenity", "pharmacy"),), "pharmacy", (0.0, 0.0, 1.0, 1.0), state
+        )
+    assert state.consecutive_transport_errors < _MAX_CONSECUTIVE_TRANSPORT_ERRORS