Дробить плитку Overpass только при перегрузке, а не при отказе сети #3528
2 changed files with 151 additions and 9 deletions
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue