fix(gisogd66): карточки без 5xx-ретраев + circuit breaker + периодический commit

Холодный прод-прогон вскрыл: сегмент старых ЕКБ-ключей (~100001007xxx)
детерминированно отдаёт 500 на карточках (портал-сайд, с любого IP),
каждая жгла ~12с на 3 ретраях (181 фейл / 40 мин), а единственный commit
в конце терял всё при soft_time_limit-kill.

- fetch_doc_card: retries_5xx=0 (5xx карточки — битые записи, не
  транзиент); список — с прежними ретраями; сетевые ретраи не тронуты.
- _sync_group: commit каждые 200 строк; load_gisogd_permits — после
  каждой (schema, group): партиал переживает time-limit kill.
- SCHEMAS: регион (agate_sverdregion) первым — карточки стабильны,
  покрывает ЕКБ-доки через агрегат; ekbgo вторым (freshness).
- circuit breaker: 50 подряд card-фейлов прерывают группу (битый
  сегмент), успех/skip сбрасывает. Новые счётчики: card_failed,
  group_aborted.

Refs #2367

backend/app/services/scrapers/gisogd66.py

@@ -13,15 +13,21 @@
     WGS84 lon/lat, может отсутствовать — hasGeometry=false), approvedOrganization, docNum,
     dateDoc, dateReg.
 
-Две схемы (тянем обе): ``agate_ekbgo`` (ЕКБ, свежая) и ``agate_sverdregion`` (регион-
-агрегат, отстаёт ~неделю). Один документ может встретиться в обеих → бизнес-ключ
-``(doc_group, doc_num)``; при конфликте предпочитаем запись с более поздним ``date_reg``.
+Две схемы (тянем обе, sverdregion первым): ``agate_sverdregion`` (регион-агрегат,
+стабильные карточки, покрывает и ЕКБ) и ``agate_ekbgo`` (ЕКБ, свежее на ~неделю, но
+часть старых ключей битая на портале). Один документ может встретиться в обеих →
+бизнес-ключ ``(doc_group, doc_num)``; при конфликте предпочитаем запись с более
+поздним ``date_reg``.
 
 Инкрементальность: список группы сверяется с БД по (source_schema, source_key, date_reg);
 карточки (второй запрос — дорогой) тянутся ТОЛЬКО для новых / изменившихся документов.
 
-Politeness: единый httpx-клиент с limits(max_connections=5), timeout=20с, 2 ретрая на 5xx,
-малый sleep между карточками. UPSERT — per-row SAVEPOINT (битая строка не валит батч).
+Politeness: единый httpx-клиент с limits(max_connections=5), timeout=20с, 2 ретрая на
+сетевых ошибках. Карточки (5xx-битые записи портала) НЕ ретраятся — deterministically
+broken, не транзиент. Circuit breaker: 50 карточек подряд с ошибкой → прерываем группу
+(битый сегмент). UPSERT — per-row SAVEPOINT (битая строка не валит батч). Периодический
+db.commit() (каждые 200 записей + после каждой schema/group), чтобы партиал пережил
+soft_time_limit-kill.
 """
 
 from __future__ import annotations
@@ -46,17 +52,26 @@ logger = logging.getLogger(__name__)
 GISOGD_BASE = "https://gisogd66.midural.ru/api/v1"
 _SECTION = "razdel13"
 
-# Схемы источника — обе тянем (ekbgo свежее, sverdregion — регион-агрегат).
-SCHEMAS: tuple[str, ...] = ("agate_ekbgo", "agate_sverdregion")
+# Схемы источника — обе тянем. Порядок ВАЖЕН: sverdregion первым.
+# sverdregion (регион-агрегат) отдаёт карточки стабильно и покрывает в т.ч. ЕКБ-доки
+# через агрегацию, поэтому он — базовый источник. ekbgo вторым как freshness-энхансер
+# (свежее на ~неделю), НО у него большой сегмент старых ключей (префикс ~100001007xxx)
+# детерминированно отдаёт 500 на карточках (сломано на стороне портала, 2026-07-04) —
+# гоняем его после того, как регион уже покрыл основную массу (см. circuit breaker ниже).
+SCHEMAS: tuple[str, ...] = ("agate_sverdregion", "agate_ekbgo")
 
 # Группа источника (doc group key) → наш doc_group-код в БД (CHECK IN ('RS','RV')).
 GROUP_CODE: dict[str, str] = {"DocRS": "RS", "DocRV": "RV"}
 
 _HTTP_TIMEOUT = 20.0
 _MAX_CONNECTIONS = 5
-_MAX_RETRIES = 2  # ретраи на 5xx (helper делает суммарно до 1 + _MAX_RETRIES попыток)
+_MAX_RETRIES = 2  # ретраи (helper делает суммарно до 1 + _MAX_RETRIES попыток)
 _RETRY_BACKOFF_S = 1.5
 _CARD_SLEEP_S = 0.05  # вежливая пауза между карточками (второй, «дорогой» запрос)
+_COMMIT_EVERY = 200  # периодический commit внутри группы (партиал переживает time-limit)
+# Circuit breaker: столько карточек ПОДРЯД с ошибкой fetch → прерываем группу (портал
+# отдаёт битый сегмент старых ключей детерминированным 5xx; остаток догоним будущими ранами).
+_CARD_FAIL_BREAKER = 50
 
 USER_AGENT = "GenDesign/1.0 (+https://gendsgn.ru) Site Finder gisogd permits loader"
 
@@ -78,20 +93,25 @@ def _make_client() -> httpx.Client:
     )
 
 
-def _get_json(client: httpx.Client, path: str) -> Any:
+def _get_json(client: httpx.Client, path: str, *, retries_5xx: int = _MAX_RETRIES) -> Any:
     """GET path → JSON с ретраями на 5xx / сетевых ошибках.
 
-    До (1 + _MAX_RETRIES) попыток с линейным backoff. 4xx → сразу raise (не ретраим).
+    Сетевые Timeout/TransportError ретраятся всегда (до 1 + _MAX_RETRIES попыток) —
+    это транзиент. 5xx ретраятся ``retries_5xx`` раз: списки группы — с дефолтом
+    (транзиентная нагрузка портала), карточки — с ``retries_5xx=0`` (5xx карточки
+    детерминированно битые записи сегмента портала, не транзиент). 4xx → сразу raise.
     """
     last_exc: Exception | None = None
     for attempt in range(_MAX_RETRIES + 1):
         try:
             resp = client.get(path)
             if resp.status_code >= 500:
-                # 5xx — транзиентная ошибка сервера, ретраим.
                 last_exc = httpx.HTTPStatusError(
                     f"{resp.status_code} on {path}", request=resp.request, response=resp
                 )
+                # 5xx: ретраим только пока attempt в пределах retries_5xx (0 → сразу raise).
+                if attempt >= retries_5xx:
+                    raise last_exc
                 logger.warning(
                     "gisogd66: %s (попытка %d) на %s", resp.status_code, attempt + 1, path
                 )
@@ -128,8 +148,11 @@ def fetch_doc_card(client: httpx.Client, schema: str, key: str | int) -> dict[st
 
     Поля: spatialUnit, infosetGeoJson, hasGeometry, approvedOrganization, docNum,
     dateDoc, dateReg, name.
+
+    5xx карточки НЕ ретраятся (retries_5xx=0): битый сегмент старых ключей портала
+    отдаёт детерминированный 500 — 3 попытки с backoff = ~12с потерянного времени/карточку.
     """
-    data = _get_json(client, f"/{schema}/gisogddocs/{_SECTION}/{key}")
+    data = _get_json(client, f"/{schema}/gisogddocs/{_SECTION}/{key}", retries_5xx=0)
     if not isinstance(data, dict):
         logger.warning("gisogd66: карточка %s/%s не-dict (%s)", schema, key, type(data).__name__)
         return None
@@ -304,7 +327,7 @@ def load_gisogd_permits(db: Session | None = None) -> dict[str, int]:
     SessionLocal.
 
     Returns: счётчики {listed, fetched_details, inserted, updated, skipped_unchanged,
-    no_cad, no_geom}.
+    no_cad, no_geom, card_failed, group_aborted}.
     """
     owns_session = db is None
     if db is None:
@@ -318,6 +341,8 @@ def load_gisogd_permits(db: Session | None = None) -> dict[str, int]:
         "skipped_unchanged": 0,
         "no_cad": 0,
         "no_geom": 0,
+        "card_failed": 0,  # каждая упавшая карточка (5xx / сеть / не-dict)
+        "group_aborted": 0,  # групп прервано circuit breaker'ом (битый сегмент портала)
     }
     client = _make_client()
     try:
@@ -331,6 +356,8 @@ def load_gisogd_permits(db: Session | None = None) -> dict[str, int]:
                     continue
                 counts["listed"] += len(items)
                 _sync_group(db, client, schema, doc_group, items, existing, counts)
+                # Commit после КАЖДОЙ (schema, group), а не один раз в конце: партиал
+                # должен пережить soft_time_limit-kill (иначе теряем весь прогон).
                 db.commit()
     except Exception as exc:
         db.rollback()
@@ -354,7 +381,15 @@ def _sync_group(
     existing: dict[str, date | None],
     counts: dict[str, int],
 ) -> None:
-    """Обработать одну (schema, group): инкрементальная сверка + карточки + UPSERT."""
+    """Обработать одну (schema, group): инкрементальная сверка + карточки + UPSERT.
+
+    Периодический commit каждые ``_COMMIT_EVERY`` обработанных записей (партиал переживёт
+    time-limit kill). Circuit breaker: ``_CARD_FAIL_BREAKER`` карточек ПОДРЯД с ошибкой
+    fetch → прерываем группу (битый сегмент портала — остаток догоним будущими ранами).
+    Любая успешная карточка сбрасывает счётчик подряд-фейлов.
+    """
+    processed = 0  # успешно обработанных (upsert/skip) с последнего commit
+    consecutive_fail = 0  # карточек подряд с ошибкой fetch (для circuit breaker)
     for item in items:
         key = item.get("key")
         if key is None:
@@ -365,17 +400,36 @@ def _sync_group(
         # Инкрементальность: пропускаем карточку, если key уже есть И date_reg совпал.
         if source_key in existing and existing[source_key] == list_reg:
             counts["skipped_unchanged"] += 1
+            consecutive_fail = 0  # skip — не фейл карточки, сбрасываем breaker
+            processed += 1
+            if processed % _COMMIT_EVERY == 0:
+                db.commit()
             continue
 
+        card: dict[str, Any] | None
         try:
             card = fetch_doc_card(client, schema, source_key)
         except Exception as exc:
             logger.warning("gisogd66: карточка %s/%s упала: %s", schema, source_key, exc)
-            continue
+            card = None
         finally:
             time.sleep(_CARD_SLEEP_S)
         if card is None:
+            # Неуспех fetch (5xx / сеть / не-dict): считаем и двигаем circuit breaker.
+            counts["card_failed"] += 1
+            consecutive_fail += 1
+            if consecutive_fail >= _CARD_FAIL_BREAKER:
+                logger.error(
+                    "gisogd66: %s/%s: %d карточек подряд 5xx — прерываю группу "
+                    "(битый сегмент портала)",
+                    schema,
+                    doc_group,
+                    _CARD_FAIL_BREAKER,
+                )
+                counts["group_aborted"] += 1
+                return
             continue
+        consecutive_fail = 0  # успешная карточка сбрасывает breaker
         counts["fetched_details"] += 1
 
         rec = _build_record(item, card, schema, doc_group, source_key, list_reg)
@@ -393,6 +447,9 @@ def _sync_group(
             logger.warning("gisogd66: upsert %s %r упал: %s", doc_group, rec["doc_num"], exc)
             continue
         counts[outcome] += 1
+        processed += 1
+        if processed % _COMMIT_EVERY == 0:
+            db.commit()
 
 
 def _build_record(

backend/tests/test_gisogd66_loader.py

@@ -18,6 +18,8 @@ from contextlib import contextmanager
 from datetime import date
 from typing import Any
 
+import httpx
+
 from app.services.scrapers import gisogd66 as g
 
 # ── Реальные фрагменты API (разведка 2026-07-04) ──────────────────────────────
@@ -221,6 +223,7 @@ class _FakeSession:
         self._scalar = scalar_value
         self._select_rows = select_rows or []
         self.committed = False
+        self.commit_count = 0
 
     @contextmanager
     def begin_nested(self):  # type: ignore[no-untyped-def]
@@ -235,6 +238,7 @@ class _FakeSession:
 
     def commit(self) -> None:
         self.committed = True
+        self.commit_count += 1
 
     def rollback(self) -> None:
         pass
@@ -284,6 +288,23 @@ def test_upsert_permit_no_geom_passes_null() -> None:
 # ── Инкрементальность через _sync_group ───────────────────────────────────────
 
 
+_COUNT_KEYS = (
+    "inserted",
+    "updated",
+    "skipped_unchanged",
+    "fetched_details",
+    "no_cad",
+    "no_geom",
+    "card_failed",
+    "group_aborted",
+)
+
+
+def _fresh_counts() -> dict[str, int]:
+    """Свежий словарь счётчиков _sync_group (все ключи, что лоадер инкрементит)."""
+    return dict.fromkeys(_COUNT_KEYS, 0)
+
+
 class _RecordingClient:
     """Заглушка httpx-клиента: считает, сколько карточек запрошено (fetch_doc_card)."""
 
@@ -302,9 +323,7 @@ def test_sync_group_incremental_skip(monkeypatch: Any) -> None:
     _patch_card_fetch(monkeypatch, fetched)
     db = _FakeSession(scalar_value=True)
     existing = {"1000210001947595": date(2026, 6, 19)}  # тот же dateReg, что в списке
-    counts = dict.fromkeys(
-        ("inserted", "updated", "skipped_unchanged", "fetched_details", "no_cad", "no_geom"), 0
-    )
+    counts = _fresh_counts()
     g._sync_group(db, _RecordingClient(), "agate_ekbgo", "RS", [dict(_LIST_ITEM)], existing, counts)  # type: ignore[arg-type]
     assert fetched == []  # карточку не тянули
     assert counts["skipped_unchanged"] == 1
@@ -317,9 +336,7 @@ def test_sync_group_changed_datereg_fetches(monkeypatch: Any) -> None:
     _patch_card_fetch(monkeypatch, fetched)
     db = _FakeSession(scalar_value=False)  # update
     existing = {"1000210001947595": date(2026, 6, 10)}  # старее списочного 06-19
-    counts = dict.fromkeys(
-        ("inserted", "updated", "skipped_unchanged", "fetched_details", "no_cad", "no_geom"), 0
-    )
+    counts = _fresh_counts()
     g._sync_group(db, _RecordingClient(), "agate_ekbgo", "RS", [dict(_LIST_ITEM)], existing, counts)  # type: ignore[arg-type]
     assert fetched == ["1000210001947595"]  # карточку тянули
     assert counts["fetched_details"] == 1
@@ -331,9 +348,7 @@ def test_sync_group_new_key_inserts_and_counts_geom(monkeypatch: Any) -> None:
     fetched: list[Any] = []
     _patch_card_fetch(monkeypatch, fetched)
     db = _FakeSession(scalar_value=True)  # insert
-    counts = dict.fromkeys(
-        ("inserted", "updated", "skipped_unchanged", "fetched_details", "no_cad", "no_geom"), 0
-    )
+    counts = _fresh_counts()
     g._sync_group(db, _RecordingClient(), "agate_ekbgo", "RS", [dict(_LIST_ITEM)], {}, counts)  # type: ignore[arg-type]
     assert counts["inserted"] == 1
     assert counts["fetched_details"] == 1
@@ -350,15 +365,142 @@ def test_sync_group_no_cad_no_geom_counts(monkeypatch: Any) -> None:
     monkeypatch.setattr(g, "fetch_doc_card", _fake)
     db = _FakeSession(scalar_value=True)
     item = {**_LIST_ITEM, "key": 42, "dateReg": "2026-06-01T10:00:00"}
-    counts = dict.fromkeys(
-        ("inserted", "updated", "skipped_unchanged", "fetched_details", "no_cad", "no_geom"), 0
-    )
+    counts = _fresh_counts()
     g._sync_group(db, _RecordingClient(), "agate_ekbgo", "RV", [item], {}, counts)  # type: ignore[arg-type]
     assert counts["no_cad"] == 1
     assert counts["no_geom"] == 1
     assert counts["inserted"] == 1
 
 
+# ── Хотфикс #2370: 5xx-карточки без ретраев, circuit breaker, периодический commit ──
+
+
+class _FakeResponse:
+    """Мини-заглушка httpx.Response: только status_code + json/raise_for_status/request."""
+
+    def __init__(self, status_code: int, payload: Any = None) -> None:
+        self.status_code = status_code
+        self._payload = payload
+        self.request = httpx.Request("GET", "https://example/x")
+
+    def json(self) -> Any:
+        return self._payload
+
+    def raise_for_status(self) -> None:
+        if self.status_code >= 400:
+            raise httpx.HTTPStatusError(
+                f"{self.status_code}",
+                request=self.request,
+                response=None,  # type: ignore[arg-type]
+            )
+
+
+class _CountingClient:
+    """Заглушка httpx.Client: считает GET-и, всегда отдаёт заданный status_code."""
+
+    def __init__(self, status_code: int, payload: Any = None) -> None:
+        self.status_code = status_code
+        self.payload = payload
+        self.get_calls: list[str] = []
+
+    def get(self, path: str) -> _FakeResponse:
+        self.get_calls.append(path)
+        return _FakeResponse(self.status_code, self.payload)
+
+
+def test_fetch_doc_card_500_single_attempt(monkeypatch: Any) -> None:
+    """Хотфикс: карточка отдаёт 500 → ровно 1 HTTP-попытка (retries_5xx=0), затем raise.
+
+    Битый сегмент старых ключей портала детерминированно 5xx — ретраи = ~12с/карточку
+    впустую. Убираем sleep, чтоб тест был мгновенным.
+    """
+    monkeypatch.setattr(g.time, "sleep", lambda _s: None)
+    client = _CountingClient(status_code=500)
+    raised = False
+    try:
+        g.fetch_doc_card(client, "agate_ekbgo", "100001007000123")  # type: ignore[arg-type]
+    except httpx.HTTPStatusError:
+        raised = True
+    assert raised is True
+    assert len(client.get_calls) == 1  # НИ одного ретрая на 5xx карточки
+
+
+def test_fetch_group_list_500_retries(monkeypatch: Any) -> None:
+    """Список группы 5xx ретраится с дефолтом (1 + _MAX_RETRIES попыток) — контраст с карточкой."""
+    monkeypatch.setattr(g.time, "sleep", lambda _s: None)
+    client = _CountingClient(status_code=500)
+    raised = False
+    try:
+        g.fetch_group_list(client, "agate_ekbgo", "DocRS")  # type: ignore[arg-type]
+    except httpx.HTTPStatusError:
+        raised = True
+    assert raised is True
+    assert len(client.get_calls) == 1 + g._MAX_RETRIES  # 3 попытки
+
+
+def test_sync_group_circuit_breaker_aborts(monkeypatch: Any) -> None:
+    """50 карточек ПОДРЯД с 5xx → группа прервана: group_aborted=1, остаток НЕ запрошен."""
+    monkeypatch.setattr(g.time, "sleep", lambda _s: None)
+    requested: list[Any] = []
+
+    def _always_fail(client: Any, schema: str, key: Any) -> dict[str, Any] | None:
+        requested.append(key)
+        raise httpx.HTTPStatusError("500", request=httpx.Request("GET", "x"), response=None)  # type: ignore[arg-type]
+
+    monkeypatch.setattr(g, "fetch_doc_card", _always_fail)
+    db = _FakeSession(scalar_value=True)
+    # 60 новых ключей — breaker (50) должен сработать раньше, чем дойдём до 60-го.
+    items = [{**_LIST_ITEM, "key": 100001007000000 + i} for i in range(60)]
+    counts = _fresh_counts()
+    g._sync_group(db, _RecordingClient(), "agate_ekbgo", "RS", items, {}, counts)  # type: ignore[arg-type]
+    assert counts["group_aborted"] == 1
+    assert counts["card_failed"] == g._CARD_FAIL_BREAKER  # ровно 50 фейлов до аборта
+    assert len(requested) == g._CARD_FAIL_BREAKER  # 51-й ключ НЕ запрашивался
+    assert counts["inserted"] == 0
+
+
+def test_sync_group_success_resets_breaker(monkeypatch: Any) -> None:
+    """Успешная карточка между фейлами сбрасывает breaker → группа НЕ прерывается.
+
+    49 фейлов, потом успех (reset), потом ещё 49 фейлов — суммарно 98 фейлов, но НИ разу
+    50 подряд → group_aborted=0, вся группа обработана.
+    """
+    monkeypatch.setattr(g.time, "sleep", lambda _s: None)
+    # Индексы 49 и 99 (0-based) — успех; остальные — фейл. Между ними ≤49 подряд.
+    good_idx = {49, 99}
+    seq: list[int] = []
+
+    def _mostly_fail(client: Any, schema: str, key: Any) -> dict[str, Any] | None:
+        idx = len(seq)
+        seq.append(idx)
+        if idx in good_idx:
+            return dict(_CARD)
+        raise httpx.HTTPStatusError("500", request=httpx.Request("GET", "x"), response=None)  # type: ignore[arg-type]
+
+    monkeypatch.setattr(g, "fetch_doc_card", _mostly_fail)
+    db = _FakeSession(scalar_value=True)
+    items = [{**_LIST_ITEM, "key": 200000000 + i} for i in range(100)]
+    counts = _fresh_counts()
+    g._sync_group(db, _RecordingClient(), "agate_ekbgo", "RS", items, {}, counts)  # type: ignore[arg-type]
+    assert counts["group_aborted"] == 0  # breaker сбрасывался — не прервались
+    assert counts["card_failed"] == 98  # 100 − 2 успешных
+    assert counts["fetched_details"] == 2  # 2 успешные карточки
+    assert len(seq) == 100  # прошли ВСЮ группу
+
+
+def test_sync_group_periodic_commit(monkeypatch: Any) -> None:
+    """>200 обработанных записей → db.commit() вызывается периодически (каждые 200)."""
+    monkeypatch.setattr(g.time, "sleep", lambda _s: None)
+    _patch_card_fetch(monkeypatch, [])
+    db = _FakeSession(scalar_value=True)  # каждый upsert → inserted
+    # 450 новых ключей → 2 периодических commit (на 200-й и 400-й записи).
+    items = [{**_LIST_ITEM, "key": 300000000 + i} for i in range(450)]
+    counts = _fresh_counts()
+    g._sync_group(db, _RecordingClient(), "agate_ekbgo", "RS", items, {}, counts)  # type: ignore[arg-type]
+    assert counts["inserted"] == 450
+    assert db.commit_count == 2  # 450 // 200 = 2 периодических commit
+
+
 # ── Конфликт двух схем: поздний date_reg параметризуется в UPSERT ──────────────