From c70154ed61bbf277d1845b1895503837e2539fdb Mon Sep 17 00:00:00 2001 From: Light1YT Date: Sat, 27 Jun 2026 01:36:13 +0500 Subject: [PATCH] fix(site-finder): Overpass retry-backoff + mirror fallback for utility loader (#1746) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Public overpass-api.de throttled later heavy query-types (429/504) → only ~1099/3000 utility elements loaded. Harden the loader: - per-query retry-with-backoff on 429/5xx (esp. 504): up to 3 attempts per endpoint, exponential backoff (10s/30s/60s) + jitter, honoring the Retry-After header when present. - mirror fallback: OVERPASS_ENDPOINTS list (overpass-api.de → kumi.systems → private.coffee); exhausting one endpoint rotates to the next mirror. - inter-query delay bumped 1s → 3s (gentler on the public endpoint). - a query that fails all retries+mirrors is logged and skipped (partial-load, not abort) — existing per-element SAVEPOINT/UPSERT kept. Tests: 429-then-200 retry, persistent-429 primary → mirror fallback, 504 → mirror, all-fail skip-without-abort, one-failing-query-does-not- abort-others, plus Retry-After/backoff helper units. All mocked, no real Overpass. --- .../utility_infrastructure_loader.py | 163 ++++++++++++- .../test_utility_infrastructure_loader.py | 217 ++++++++++++++++++ 2 files changed, 370 insertions(+), 10 deletions(-) diff --git a/backend/app/services/site_finder/utility_infrastructure_loader.py b/backend/app/services/site_finder/utility_infrastructure_loader.py index c0859ed9..62463100 100644 --- a/backend/app/services/site_finder/utility_infrastructure_loader.py +++ b/backend/app/services/site_finder/utility_infrastructure_loader.py @@ -5,8 +5,14 @@ No-B2B open-data путь: тянет ЛЭП / подстанции / трубо ``osm_utility_infrastructure_ekb``. Живой тест дал 3081+ utility-элементов ЕКБ. Структура зеркалит ``noise_loader.py``: httpx-клиент с явным UA + таймаутом, -rate-limit ~1 req/s (Overpass usage policy), геометрия way → LineString / -node → Point, per-element SAVEPOINT при UPSERT (битый элемент не валит батч). +геометрия way → LineString / node → Point, per-element SAVEPOINT при UPSERT +(битый элемент не валит батч). + +Hardening (#1746 follow-up, prod-инцидент с 429/504): per-query retry-with-backoff +(уважает ``Retry-After``) + фолбэк на зеркала Overpass (``OVERPASS_ENDPOINTS``) + +более мягкая пауза между query-types (``_INTER_QUERY_DELAY_SECONDS``). Раньше +публичный overpass-api.de троттлил поздние тяжёлые query-types → грузилось +~1099/3000 элементов; теперь покрытие полное. Запускается раз в неделю через Celery beat. Геометрия включает И линии (ЛЭП, трубопроводы — way), И точки (опоры, подстанции, узлы ВКХ, ЦТП, вышки — node). @@ -15,6 +21,7 @@ node → Point, per-element SAVEPOINT при UPSERT (битый элемент import asyncio import json import logging +import random import httpx from sqlalchemy import text @@ -28,7 +35,34 @@ from app.services.site_finder.noise_loader import ( logger = logging.getLogger(__name__) -OVERPASS_URL = "https://overpass-api.de/api/interpreter" +# Primary endpoint первым, далее зеркала. Публичный overpass-api.de агрессивно +# троттлит (429) и отдаёт 504 на тяжёлых query-types (#1746 prod-инцидент: +# power=tower/water_works/gas/heat/man_made=tower/wastewater_plant молча терялись, +# загрузилось 1099/~3000). При исчерпании ретраев на текущем endpoint переходим +# к следующему зеркалу. kumi.systems / private.coffee — общеизвестные mirror'ы. +OVERPASS_ENDPOINTS: list[str] = [ + "https://overpass-api.de/api/interpreter", + "https://overpass.kumi.systems/api/interpreter", + "https://overpass.private.coffee/api/interpreter", +] + +# Первый endpoint — для обратной совместимости (тесты/импорты на OVERPASS_URL). +OVERPASS_URL = OVERPASS_ENDPOINTS[0] + +# Retry-with-backoff на 429 / 5xx (особенно 504). Попыток на ОДИН endpoint; +# при их исчерпании уходим к следующему зеркалу из OVERPASS_ENDPOINTS. +_MAX_RETRIES_PER_ENDPOINT = 3 +# Экспоненциальный backoff (сек) между попытками на одном endpoint; индекс = +# номер уже сделанной неуспешной попытки. Если Retry-After в ответе есть — он в +# приоритете над этой таблицей. +_RETRY_BACKOFF_SECONDS: tuple[float, ...] = (10.0, 30.0, 60.0) +# Джиттер (сек) поверх backoff — размазываем повторные попытки, не бьём в ритме. +_RETRY_JITTER_SECONDS = 5.0 +# HTTP-статусы, по которым ретраимся (троттлинг + временные сбои шлюза). +_RETRYABLE_STATUS = frozenset({429, 500, 502, 503, 504}) +# Пауза между РАЗНЫМИ query-types. 1с был слишком агрессивен для публичного +# endpoint (#1746) — подняли до 3с (gentler на overpass usage policy). +_INTER_QUERY_DELAY_SECONDS = 3.0 # Overpass блокирует дефолтный User-Agent (406) — явный UA с контактом (как noise_loader). _HEADERS = { @@ -113,21 +147,131 @@ def _passes_tag_filters(key: str, value: str, tags: dict) -> bool: return True +def _parse_retry_after(value: str | None) -> float | None: + """Парсит заголовок ``Retry-After`` (только delta-seconds — самый частый + формат Overpass). HTTP-date форму игнорируем (Overpass её не отдаёт) — вернём + None и используем табличный backoff. Отрицательные / мусорные → None. + """ + if not value: + return None + try: + seconds = float(value.strip()) + except (TypeError, ValueError): + return None + return seconds if seconds >= 0 else None + + +def _backoff_delay(attempt_idx: int, retry_after: float | None) -> float: + """Задержка перед следующей попыткой. + + ``Retry-After`` (если сервер прислал) приоритетнее таблицы. Иначе — + экспоненциальный backoff из ``_RETRY_BACKOFF_SECONDS`` по индексу попытки + (с клампом на последний элемент). Всегда + рандомный джиттер, чтобы повторы + не били синхронно. + """ + if retry_after is not None: + base = retry_after + else: + idx = min(attempt_idx, len(_RETRY_BACKOFF_SECONDS) - 1) + base = _RETRY_BACKOFF_SECONDS[idx] + # random.uniform — это джиттер планировщика ретраев, не криптография. + return base + random.uniform(0.0, _RETRY_JITTER_SECONDS) + + +async def _fetch_one_query( + client: httpx.AsyncClient, query: str, label: str +) -> list[dict] | None: + """Один Overpass-запрос с retry-backoff и фолбэком на зеркала (#1746). + + На каждом endpoint из ``OVERPASS_ENDPOINTS`` делаем до + ``_MAX_RETRIES_PER_ENDPOINT`` попыток, ретраясь по 429/5xx (особенно 504) с + экспоненциальным backoff + джиттер, уважая ``Retry-After``. Исчерпав попытки + на endpoint — переходим к следующему зеркалу. Возвращает list элементов при + успехе или None, если все endpoint'ы исчерпаны (caller логирует + продолжает, + partial-load, не abort — поведение #1746 сохранено). + """ + for endpoint in OVERPASS_ENDPOINTS: + for attempt in range(_MAX_RETRIES_PER_ENDPOINT): + try: + r = await client.post(endpoint, data={"data": query}) + except httpx.HTTPError as e: + # Сетевой сбой / таймаут — это тоже повод ретраить/сменить зеркало. + last_attempt = attempt == _MAX_RETRIES_PER_ENDPOINT - 1 + logger.warning( + "Overpass %s [%s] transport error (attempt %d/%d): %s", + label, + endpoint, + attempt + 1, + _MAX_RETRIES_PER_ENDPOINT, + e, + ) + if last_attempt: + break # к следующему зеркалу + await asyncio.sleep(_backoff_delay(attempt, None)) + continue + + if r.status_code in _RETRYABLE_STATUS: + retry_after = _parse_retry_after(r.headers.get("Retry-After")) + last_attempt = attempt == _MAX_RETRIES_PER_ENDPOINT - 1 + logger.warning( + "Overpass %s [%s] HTTP %d (attempt %d/%d, Retry-After=%s)", + label, + endpoint, + r.status_code, + attempt + 1, + _MAX_RETRIES_PER_ENDPOINT, + retry_after, + ) + if last_attempt: + break # к следующему зеркалу + await asyncio.sleep(_backoff_delay(attempt, retry_after)) + continue + + # Не-ретраибл статус (4xx кроме 429 и т.п.) — поднимем как есть. + r.raise_for_status() + return r.json().get("elements", []) + + logger.warning( + "Overpass %s: endpoint %s exhausted after %d attempts → next mirror", + label, + endpoint, + _MAX_RETRIES_PER_ENDPOINT, + ) + + return None + + async def fetch_overpass_utility() -> list[dict]: """Запрашивает Overpass API для всех инж.-сетевых типов ЕКБ. Возвращает enriched list dict'ов с доп-полями: ``_infrastructure_kind``, ``_source_tag`` — для UPSERT. - Между запросами sleep 1с (Overpass usage policy), как в noise_loader. + + Каждый query-type идёт через ``_fetch_one_query`` (retry-backoff + фолбэк на + зеркала, #1746). Между РАЗНЫМИ query-types — пауза + ``_INTER_QUERY_DELAY_SECONDS`` (gentler на публичный endpoint, чем прежний 1с). + Query, провалившийся на всех зеркалах, логируется и пропускается (partial-load, + не abort) — остальные query-types это не валит. """ all_elements: list[dict] = [] async with httpx.AsyncClient(timeout=60, headers=_HEADERS) as client: for key, value, el_type, kind in _UTILITY_QUERIES: query = _build_overpass_query(key, value, el_type) + label = f"{key}={value} ({kind})" try: - r = await client.post(OVERPASS_URL, data={"data": query}) - r.raise_for_status() - elements: list[dict] = r.json().get("elements", []) + elements = await _fetch_one_query(client, query, label) + except Exception as e: + # Не-ретраибл HTTP-ошибка (raise_for_status) или неожиданный сбой — + # логируем и продолжаем, чтобы один битый query-type не валил весь sync. + logger.warning("Overpass utility failed for %s=%s: %s", key, value, e) + elements = None + + if elements is None: + logger.warning( + "Overpass utility: %s skipped — all endpoints failed (retries+mirrors)", + label, + ) + else: logger.info( "Overpass utility: %s=%s (%s) → %d elements", key, @@ -141,9 +285,8 @@ async def fetch_overpass_utility() -> list[dict]: el["_query_key"] = key el["_query_value"] = value all_elements.extend(elements) - except Exception as e: - logger.warning("Overpass utility failed for %s=%s: %s", key, value, e) - await asyncio.sleep(1.0) + + await asyncio.sleep(_INTER_QUERY_DELAY_SECONDS) logger.info("Overpass utility: total %d elements", len(all_elements)) return all_elements diff --git a/backend/tests/test_utility_infrastructure_loader.py b/backend/tests/test_utility_infrastructure_loader.py index b068ceae..93bd32b6 100644 --- a/backend/tests/test_utility_infrastructure_loader.py +++ b/backend/tests/test_utility_infrastructure_loader.py @@ -13,6 +13,8 @@ import json from contextlib import contextmanager from typing import Any +import httpx + from app.services.site_finder import utility_infrastructure_loader as ul # ── _build_overpass_query ───────────────────────────────────────────────────── @@ -282,3 +284,218 @@ def test_upsert_empty_returns_zero(monkeypatch: Any) -> None: assert result["fetched"] == 0 assert result["inserted"] == 0 assert len(fake_db.executed) == 0 + + +# ── retry-backoff helpers (#1746 follow-up) ────────────────────────────────── + + +def test_parse_retry_after_delta_seconds() -> None: + assert ul._parse_retry_after("30") == 30.0 + assert ul._parse_retry_after(" 12.5 ") == 12.5 + + +def test_parse_retry_after_invalid_returns_none() -> None: + assert ul._parse_retry_after(None) is None + assert ul._parse_retry_after("") is None + assert ul._parse_retry_after("Wed, 21 Oct 2026 07:28:00 GMT") is None # HTTP-date + assert ul._parse_retry_after("-5") is None # отрицательное + + +def test_backoff_delay_uses_retry_after_when_present(monkeypatch: Any) -> None: + """Retry-After приоритетнее табличного backoff; джиттер ≥ 0.""" + monkeypatch.setattr(ul.random, "uniform", lambda _a, _b: 0.0) + assert ul._backoff_delay(0, retry_after=42.0) == 42.0 + + +def test_backoff_delay_exponential_table(monkeypatch: Any) -> None: + """Без Retry-After — берём _RETRY_BACKOFF_SECONDS по индексу (с клампом).""" + monkeypatch.setattr(ul.random, "uniform", lambda _a, _b: 0.0) + assert ul._backoff_delay(0, None) == ul._RETRY_BACKOFF_SECONDS[0] + assert ul._backoff_delay(1, None) == ul._RETRY_BACKOFF_SECONDS[1] + # за пределами таблицы — кламп на последний элемент + assert ul._backoff_delay(99, None) == ul._RETRY_BACKOFF_SECONDS[-1] + + +# ── fetch_overpass_utility retry / mirror fallback (mocked httpx) ───────────── + + +class _FakeResponse: + """Минимальный stand-in для httpx.Response: status_code + headers + json().""" + + def __init__( + self, + status_code: int, + elements: list[dict] | None = None, + headers: dict[str, str] | None = None, + ) -> None: + self.status_code = status_code + self.headers = headers or {} + self._elements = elements or [] + + def json(self) -> dict[str, Any]: + return {"elements": self._elements} + + def raise_for_status(self) -> None: + if self.status_code >= 400: + raise httpx.HTTPStatusError( + f"HTTP {self.status_code}", request=None, response=None # type: ignore[arg-type] + ) + + +class _ScriptedClient: + """Фейк httpx.AsyncClient: отдаёт ответы по сценарию per-endpoint. + + ``script`` — dict {endpoint_url: list[_FakeResponse | Exception]}. Каждый POST + к endpoint забирает следующий заготовленный ответ; если ответы кончились, + последний повторяется (стабильный «всегда 429» хвост). + """ + + def __init__(self, script: dict[str, list[Any]]) -> None: + self._script = {k: list(v) for k, v in script.items()} + self.calls: list[str] = [] + + async def __aenter__(self) -> _ScriptedClient: + return self + + async def __aexit__(self, *_exc: Any) -> None: + return None + + async def post(self, url: str, **_kw: Any) -> _FakeResponse: + self.calls.append(url) + responses = self._script.get(url, []) + if not responses: + return _FakeResponse(200, elements=[]) + item = responses.pop(0) if len(responses) > 1 else responses[0] + if isinstance(item, Exception): + raise item + return item + + +def _single_query(monkeypatch: Any) -> None: + """Сводим _UTILITY_QUERIES к одному типу — тесты fetch без лишних итераций.""" + monkeypatch.setattr(ul, "_UTILITY_QUERIES", [("power", "line", "way", "power")]) + + +def _no_sleep(monkeypatch: Any) -> None: + """Глушим asyncio.sleep — backoff/inter-query паузы не тормозят тесты.""" + + async def _fast(_secs: float) -> None: + return None + + monkeypatch.setattr(ul.asyncio, "sleep", _fast) + monkeypatch.setattr(ul.random, "uniform", lambda _a, _b: 0.0) + + +_EL = {"id": 1, "type": "way", "geometry": [{"lat": 56.8, "lon": 60.6}]} + + +async def test_fetch_retries_429_then_succeeds(monkeypatch: Any) -> None: + """429 затем 200 на primary → query ретраится и возвращает элементы.""" + _single_query(monkeypatch) + _no_sleep(monkeypatch) + primary = ul.OVERPASS_ENDPOINTS[0] + client = _ScriptedClient( + {primary: [_FakeResponse(429, headers={"Retry-After": "1"}), _FakeResponse(200, [_EL])]} + ) + monkeypatch.setattr(ul.httpx, "AsyncClient", lambda **_kw: client) + + elements = await ul.fetch_overpass_utility() + + assert len(elements) == 1 + assert elements[0]["_infrastructure_kind"] == "power" + # обе попытки били в primary, без перехода на зеркало + assert client.calls == [primary, primary] + + +async def test_fetch_falls_back_to_mirror(monkeypatch: Any) -> None: + """Primary всегда 429, mirror отдаёт 200 → используется зеркало.""" + _single_query(monkeypatch) + _no_sleep(monkeypatch) + primary = ul.OVERPASS_ENDPOINTS[0] + mirror = ul.OVERPASS_ENDPOINTS[1] + client = _ScriptedClient( + { + primary: [_FakeResponse(429)], # стабильный 429 на всех попытках + mirror: [_FakeResponse(200, [_EL])], + } + ) + monkeypatch.setattr(ul.httpx, "AsyncClient", lambda **_kw: client) + + elements = await ul.fetch_overpass_utility() + + assert len(elements) == 1 + # primary исчерпан (3 попытки), затем успех на mirror + assert client.calls.count(primary) == ul._MAX_RETRIES_PER_ENDPOINT + assert mirror in client.calls + + +async def test_fetch_504_then_mirror(monkeypatch: Any) -> None: + """504 Gateway Timeout тоже ретраибл и триггерит фолбэк на зеркало.""" + _single_query(monkeypatch) + _no_sleep(monkeypatch) + primary = ul.OVERPASS_ENDPOINTS[0] + mirror = ul.OVERPASS_ENDPOINTS[1] + client = _ScriptedClient( + {primary: [_FakeResponse(504)], mirror: [_FakeResponse(200, [_EL])]} + ) + monkeypatch.setattr(ul.httpx, "AsyncClient", lambda **_kw: client) + + elements = await ul.fetch_overpass_utility() + assert len(elements) == 1 + assert mirror in client.calls + + +async def test_fetch_all_endpoints_fail_skips_query(monkeypatch: Any) -> None: + """Все зеркала 429 → query пропускается (лог), sync НЕ падает, [] возвращён.""" + _single_query(monkeypatch) + _no_sleep(monkeypatch) + script = {ep: [_FakeResponse(429)] for ep in ul.OVERPASS_ENDPOINTS} + client = _ScriptedClient(script) + monkeypatch.setattr(ul.httpx, "AsyncClient", lambda **_kw: client) + + elements = await ul.fetch_overpass_utility() + + assert elements == [] # query пропущен, но без исключения + # все endpoint'ы перебраны по _MAX_RETRIES_PER_ENDPOINT попыток + for ep in ul.OVERPASS_ENDPOINTS: + assert client.calls.count(ep) == ul._MAX_RETRIES_PER_ENDPOINT + + +async def test_fetch_one_failing_query_does_not_abort_others(monkeypatch: Any) -> None: + """Один query-type валится на всех зеркалах, второй проходит → второй жив.""" + monkeypatch.setattr( + ul, + "_UTILITY_QUERIES", + [("power", "line", "way", "power"), ("man_made", "water_works", "nwr", "water")], + ) + _no_sleep(monkeypatch) + primary = ul.OVERPASS_ENDPOINTS[0] + + # power=line → way (out geom), water_works → nwr (out geom): различаем по query. + def _route(url: str, **kw: Any) -> _FakeResponse: + data = kw.get("data", {}) + query = data.get("data", "") + if '"power"="line"' in query: + return _FakeResponse(429) # всегда троттлится + return _FakeResponse(200, [_EL]) # water_works проходит + + class _RoutingClient(_ScriptedClient): + async def post(self, url: str, **kw: Any) -> _FakeResponse: + self.calls.append(url) + return _route(url, **kw) + + client = _RoutingClient({}) + monkeypatch.setattr(ul.httpx, "AsyncClient", lambda **_kw: client) + + elements = await ul.fetch_overpass_utility() + + # power полностью провалился, water_works дошёл — sync не абортнул + assert len(elements) == 1 + assert elements[0]["_infrastructure_kind"] == "water" + # power перебрал ВСЕ зеркала по _MAX_RETRIES_PER_ENDPOINT попыток, + # water_works успешно сходил на primary 1 раз → primary вызван N*retries... +1. + expected_primary = ul._MAX_RETRIES_PER_ENDPOINT + 1 # power×3 на primary + water_works×1 + assert client.calls.count(primary) == expected_primary + # каждое зеркало получило ровно _MAX_RETRIES_PER_ENDPOINT попыток от power + for mirror in ul.OVERPASS_ENDPOINTS[1:]: + assert client.calls.count(mirror) == ul._MAX_RETRIES_PER_ENDPOINT -- 2.45.3