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

Merged
lekss361 merged 1 commit from fix/poi-no-split-on-transport-error into main 2026-09-15 07:33:57 +00:00
2 changed files with 151 additions and 9 deletions

View file

@ -141,6 +141,47 @@ def _split_bbox_quadrants(
# дефолтного региона не меняется. # дефолтного региона не меняется.
RECURSIVE_SPLIT_MAX_DEPTH = 3 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( def _build_overpass_query(
tag_filters: tuple[tuple[str, str], ...], bbox: tuple[float, float, float, float] 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], ...], tag_filters: tuple[tuple[str, str], ...],
category: str, category: str,
bbox: tuple[float, float, float, float], bbox: tuple[float, float, float, float],
state: _RunState,
depth: int = 0, depth: int = 0,
) -> list[dict]: ) -> list[dict]:
"""Один per-category Overpass-запрос (для ОДНОГО тайла bbox) с ОДНИМ повтором при """Один per-category Overpass-запрос (для ОДНОГО тайла bbox) с ОДНИМ повтором при
@ -209,15 +251,34 @@ async def _fetch_category(
# amenity=pharmacy + shop=supermarket) приходит дважды — каждая копия # amenity=pharmacy + shop=supermarket) приходит дважды — каждая копия
# несёт свою category. Иначе _classify по dict-порядку молча терял бы # несёт свою category. Иначе _classify по dict-порядку молча терял бы
# вторую категорию при UPSERT по UNIQUE(osm_type, osm_id, category). См. #1372. # вторую категорию при UPSERT по UNIQUE(osm_type, osm_id, category). См. #1372.
state.consecutive_transport_errors = 0
for el in elements: for el in elements:
el["_gd_category"] = category el["_gd_category"] = category
return elements return elements
except Exception as e: 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: if attempt == 1:
logger.warning("Overpass failed for %s (attempt 1, retrying): %s", tag_desc, e) logger.warning("Overpass failed for %s (attempt 1, retrying): %s", tag_desc, e)
await asyncio.sleep(3.0) await asyncio.sleep(3.0)
continue continue
if depth < RECURSIVE_SPLIT_MAX_DEPTH: if _is_overload(e) and depth < RECURSIVE_SPLIT_MAX_DEPTH:
logger.warning( logger.warning(
"Overpass failed for %s bbox=%s twice — splitting into 4 quadrants " "Overpass failed for %s bbox=%s twice — splitting into 4 quadrants "
"(depth %d%d) instead of dropping the tile: %s", "(depth %d%d) instead of dropping the tile: %s",
@ -230,13 +291,14 @@ async def _fetch_category(
combined: list[dict] = [] combined: list[dict] = []
for quadrant in _split_bbox_quadrants(bbox): for quadrant in _split_bbox_quadrants(bbox):
combined.extend( 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) await asyncio.sleep(1.0)
return combined return combined
logger.warning( logger.warning(
"Overpass failed for %s bbox=%s at max split depth %d — tile skipped " "Overpass failed for %s bbox=%s at max split depth %d — tile skipped this run: %s",
"this run: %s",
tag_desc, tag_desc,
bbox, bbox,
depth, depth,
@ -260,6 +322,7 @@ async def fetch_overpass(region: str = DEFAULT_REGION) -> list[dict]:
""" """
bbox = REGION_BBOX[region] bbox = REGION_BBOX[region]
tiles = _bbox_tiles(bbox) tiles = _bbox_tiles(bbox)
state = _RunState()
headers = { headers = {
"User-Agent": "GenDesign-SiteFinder/1.0 (+https://gendsgn.ru)", "User-Agent": "GenDesign-SiteFinder/1.0 (+https://gendsgn.ru)",
"Accept": "application/json", "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: async with httpx.AsyncClient(timeout=60, headers=headers) as client:
for tag_filters, category in OSM_CATEGORIES.items(): for tag_filters, category in OSM_CATEGORIES.items():
for tile in tiles: 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) all_elements.extend(elements)
await asyncio.sleep(1.0) await asyncio.sleep(1.0)
logger.info( logger.info(

View file

@ -15,28 +15,39 @@ from __future__ import annotations
from types import SimpleNamespace from types import SimpleNamespace
import httpx
import pytest import pytest
from app.services.site_finder.poi_loader import ( from app.services.site_finder.poi_loader import (
_MAX_CONSECUTIVE_TRANSPORT_ERRORS,
DEFAULT_REGION, DEFAULT_REGION,
EKB_BBOX, EKB_BBOX,
RECURSIVE_SPLIT_MAX_DEPTH, RECURSIVE_SPLIT_MAX_DEPTH,
REGION_BBOX, REGION_BBOX,
OverpassUnreachableError,
_bbox_tiles, _bbox_tiles,
_build_overpass_query, _build_overpass_query,
_fetch_category, _fetch_category,
_RunState,
_split_bbox_quadrants, _split_bbox_quadrants,
) )
class _FakeResponse: class _FakeResponse:
"""Ответ Overpass. ok=False отдаёт 504 — «запрос слишком тяжёлый», единственный
класс отказа, ради которого тайл вообще осмысленно дробить."""
def __init__(self, ok: bool, elements: list[dict] | None = None) -> None: def __init__(self, ok: bool, elements: list[dict] | None = None) -> None:
self._ok = ok self._ok = ok
self._elements = elements or [] self._elements = elements or []
def raise_for_status(self) -> None: def raise_for_status(self) -> None:
if not self._ok: 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: def json(self) -> dict:
return {"elements": self._elements} 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}]) return _FakeResponse(ok=True, elements=[{"type": "node", "id": 1, "lat": 1, "lon": 2}])
client = SimpleNamespace(post=fake_post) 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 calls["n"] == 2
assert len(result) == 1 assert len(result) == 1
assert result[0]["_gd_category"] == "pharmacy" 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) client = SimpleNamespace(post=fake_post)
result = await _fetch_category( 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 успешных запроса по четвертям # верхний тайл: 2 неудачных attempt, затем 4 успешных запроса по четвертям
assert calls["n"] == 2 + 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) client = SimpleNamespace(post=fake_post)
result = await _fetch_category( 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 == [] assert result == []
# sum_{d=0}^{max_depth} 4^d узлов, каждый по 2 attempt — рекурсия конечна # sum_{d=0}^{max_depth} 4^d узлов, каждый по 2 attempt — рекурсия конечна
expected_nodes = sum(4**d for d in range(RECURSIVE_SPLIT_MAX_DEPTH + 1)) expected_nodes = sum(4**d for d in range(RECURSIVE_SPLIT_MAX_DEPTH + 1))
assert calls["n"] == expected_nodes * 2 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