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