fix(tradein/proxy): упавшая проба присваивала себе бан боевого сбора (#2800) (#2805)

tradein-mvp/backend/app/services/proxy_pool.py

@@ -725,14 +725,43 @@ def mark_browser_health(
     return "fail"
 
 
-def mark_banned(db: Session, proxy_id: int, *, source: str, reason: str | None = None) -> None:
+def mark_banned(db: Session, proxy_id: int, *, source: str, reason: str | None = None) -> str:
     """Записать бан узла площадкой `source` — по ПАРЕ (proxy_id, source), #2600 п.2.
 
+    Returns: "banned" (строка записана/продлена) | "deferred" (активная строка пары
+    принадлежит другому вердикту, владельца не меняем) | "protected" (защита последнего
+    узла) | "missing" (нет такого proxy_id).
+
     `reason` попадает в одноимённую колонку и служит МЕТКОЙ ВЛАДЕЛЬЦА строки: по
     умолчанию 'banned:<source>' (бан распознан боевым сбором), у браузерной пробы —
     _PROBE_BAN_REASON (#2800). Снимать чужую строку никто не должен, поэтому
     clear_source_bans умеет фильтровать по ней (`only_reason`).
 
+    ВЛАДЕЛЬЦА АКТИВНОЙ СТРОКИ НЕ МЕНЯЕМ (дефект #2803, реализовался на проде 09.08.2026:
+    пара (1, cian) была `banned:cian, ban_count=1, до 00:21`, упавшая проба через
+    ON CONFLICT переписала её в `probe:browser, ban_count=2, до 07:43`). Фильтр
+    «снимаю только своё» защищает лишь до тех пор, пока чужую строку нельзя ПРИСВОИТЬ:
+    присвоенная строка становится «своей», и следующая успешная проба снимает ею бан,
+    который поставил боевой сбор по настоящему отказу площадки. Плюс теряется
+    происхождение: 'banned:cian' («площадка нас отбила») и 'probe:browser' («наша проба
+    не смогла») — разные факты с разными последствиями (ровно ловушка #2764), а ban_count
+    начинает считать события РАЗНОГО рода одной эскалацией (на проде это удлинило отдых
+    пары с 6 ч до 12 ч).
+
+    Правило в `WHERE` у DO UPDATE: строку берём, если она ИСТЕКЛА (живого владельца нет),
+    ИЛИ она уже наша (та же метка — обычная эскалация), ИЛИ мы боевой сбор (`live_reason`).
+    Иначе — ничего: ни reason, ни ban_count, ни срок. Продлевать чужой бан «безвредно»
+    только на словах: срок пересчитывается от now() по НАШЕЙ эскалации и способен
+    УКОРОТИТЬ уже эскалированный чужой бан. Бан и так стоит — делать нечего.
+
+    АСИММЕТРИЯ НАМЕРЕННАЯ: боевой сбор строку пробы перехватывает. Его вердикт сильнее
+    (площадка реально отбила именно сейчас), пара остаётся забаненной, а метка становится
+    ТОЧНЕЕ. Запретить ему это значило бы оставить строку за пробой — и её же зелёный
+    robots.txt снёс бы настоящий бан площадки, то есть тот самый дефект, только зеркально
+    и хуже. Цена перехвата — ban_count наследуется (отдых чуть длиннее заслуженного);
+    обнулять его на смене владельца нельзя: тогда запись пробы стирала бы память об
+    эскалации боевых банов пары.
+
     Отличается от `mark_health(ok=False)`: та инкрементит consecutive_fails и
     авто-disable'ит только после DISABLE_THRESHOLD ПОДРЯД неудач (мягкая деградация —
     транзиентный сбой должен пережить пару неудач). Здесь причина УЖЕ надёжно
@@ -791,6 +820,10 @@ def mark_banned(db: Session, proxy_id: int, *, source: str, reason: str | None =
     сюда попадают уже обёрнутыми в try/except, но сам mark_banned ошибки БД не глотает
     (падает как обычно) — caller решает, ловить или нет.
     """
+    # Метка боевого сбора: право перехватить АКТИВНУЮ строку пары есть только у неё
+    # (см. докстринг "ВЛАДЕЛЬЦА АКТИВНОЙ СТРОКИ НЕ МЕНЯЕМ").
+    live_reason = f"banned:{source}"
+    effective_reason = reason or live_reason
     # Сериализует check+insert ниже с другими конкурентными mark_banned (см. докстринг
     # "КОНКУРЕНТНОСТЬ"). Держится до db.commit()/rollback() этой транзакции.
     db.execute(
@@ -862,13 +895,20 @@ def mark_banned(db: Session, proxy_id: int, *, source: str, reason: str | None =
                         ) AS integer)),
                     reason       = CAST(:reason AS text),
                     updated_at   = now()
+                -- Владельца АКТИВНОЙ строки не меняем: берём истёкшую (владельца нет),
+                -- свою же (обычная эскалация) или перебиваем боевым сбором — он сильнее
+                -- пробы. Иначе 0 rows и ветка "deferred" ниже (дефект #2803).
+                WHERE scrape_proxy_source_bans.banned_until <= now()
+                   OR scrape_proxy_source_bans.reason = CAST(:reason AS text)
+                   OR CAST(:reason AS text) = CAST(:live_reason AS text)
                 RETURNING ban_count, banned_until
                 """
             ),
             {
                 "proxy_id": proxy_id,
                 "source": source,
-                "reason": reason or f"banned:{source}",
+                "reason": effective_reason,
+                "live_reason": live_reason,
                 "base_hours": SOURCE_BAN_BASE_HOURS,
                 "max_hours": SOURCE_BAN_MAX_HOURS,
                 "max_fails": MAX_CONSECUTIVE_FAILS,
@@ -888,10 +928,40 @@ def mark_banned(db: Session, proxy_id: int, *, source: str, reason: str | None =
             row["banned_until"],
             row["ban_count"],
         )
-        return
+        return "banned"
+
+    # 0 rows — ТРИ разные причины, и путать их нельзя: чужой активный владелец, защита
+    # последнего узла, отсутствующий узел. Читаем состояние ТОЛЬКО ради точного лога
+    # (на решение уже не влияет), но диагноз должен называть то, что произошло.
+    holder = (
+        db.execute(
+            text(
+                """
+                SELECT reason, banned_until
+                FROM scrape_proxy_source_bans
+                WHERE proxy_id = CAST(:proxy_id AS bigint)
+                  AND source = CAST(:source AS text)
+                  AND banned_until > now()
+                """
+            ),
+            {"proxy_id": proxy_id, "source": source},
+        )
+        .mappings()
+        .fetchone()
+    )
+    if holder is not None and holder["reason"] != effective_reason:
+        logger.info(
+            "proxy_pool: proxy id=%d source=%s — бан пары уже стоит от %r до %s; вердикт "
+            "%r его НЕ перебивает (владельца активной строки меняет только боевой сбор, "
+            "иначе проба присвоила бы чужой бан и потом сняла бы его как свой)",
+            proxy_id,
+            source,
+            holder["reason"],
+            holder["banned_until"],
+            effective_reason,
+        )
+        return "deferred"
 
-    # 0 rows: либо узла нет, либо защита последнего узла отменила запись бана — читаем
-    # текущее состояние ТОЛЬКО для точного лога (на решение уже не влияет).
     current = (
         db.execute(
             text(
@@ -904,14 +974,15 @@ def mark_banned(db: Session, proxy_id: int, *, source: str, reason: str | None =
     )
     if current is None:
         logger.warning("proxy_pool: mark_banned id=%d not found — no-op", proxy_id)
-    else:
-        logger.warning(
-            "proxy_pool: proxy id=%d — бан не записан: это последний узел, достижимый для "
-            "source=%s; нужны новые прокси (см. #2638). Узел продолжит выдаваться этому "
-            "источнику (голодание хуже, чем работа через забаненный узел).",
-            proxy_id,
-            source,
-        )
+        return "missing"
+    logger.warning(
+        "proxy_pool: proxy id=%d — бан не записан: это последний узел, достижимый для "
+        "source=%s; нужны новые прокси (см. #2638). Узел продолжит выдаваться этому "
+        "источнику (голодание хуже, чем работа через забаненный узел).",
+        proxy_id,
+        source,
+    )
+    return "protected"
 
 
 def clear_source_bans(
@@ -1004,13 +1075,18 @@ def mark_source_probe(
 
     Успех снимает ТОЛЬКО строку, написанную пробой (`only_reason`). Бан, распознанный
     боевым сбором, остаётся: robots.txt площадка отдаёт и забаненному IP, и разрешить
-    дешёвой пробе гасить дорогой вердикт значило бы повторить #2723 на паре.
+    дешёвой пробе гасить дорогой вердикт значило бы повторить #2723 на паре. Обратная
+    половина того же правила живёт в `mark_banned`: чужую АКТИВНУЮ строку проба не
+    присваивает (дефект #2803) — иначе фильтр `only_reason` перестаёт защищать, ведь
+    присвоенная строка уже «своя».
 
     Защита последнего узла и эскалация срока — целиком из `mark_banned`, здесь ничего
     своего: если после бана у `acquire(source)` не осталось бы кандидатов, бан не
     пишется (голодание хуже работы через плохой узел).
 
-    Returns: "ok" | "cleared" (сняли свой бан) | "banned" | "ignored".
+    Returns: "ok" | "cleared" (сняли свой бан) | "ignored" | исход `mark_banned`
+    ("banned" | "deferred" | "protected" | "missing") — счётчик пар считает баном
+    только реально записанный бан.
     """
     if ok:
         cleared = clear_source_bans(
@@ -1041,8 +1117,7 @@ def mark_source_probe(
         fail_kind,
         detail,
     )
-    mark_banned(db, proxy_id, source=source, reason=_PROBE_BAN_REASON)
-    return "banned"
+    return mark_banned(db, proxy_id, source=source, reason=_PROBE_BAN_REASON)
 
 
 def reap_stale_leases(db: Session, older_than_minutes: int = STALE_LEASE_MINUTES) -> int:

tradein-mvp/backend/tests/services/test_proxy_pool.py

@@ -387,6 +387,17 @@ class FakeSession:
                 }
                 self.bans.append(ban)
             else:
+                # #2803-follow-up: активную строку чужого владельца не перехватываем.
+                # Гейтим по подстроке боевого SQL (как ban-предикаты выше) — иначе мок
+                # реализовал бы защиту сам и тест был бы зелёным на сломанном коде.
+                defends_owner = "scrape_proxy_source_bans.banned_until <= now()" in sql
+                if (
+                    defends_owner
+                    and ban["banned_until"] > now
+                    and ban.get("reason") != p["reason"]
+                    and p["reason"] != p.get("live_reason")
+                ):
+                    return _FakeResult([])  # владельца активной строки не меняем
                 # эскалация: срок = base * 2^(новый ban_count - 1), потолок max_hours
                 ban["ban_count"] += 1
                 hours = min(p["base_hours"] * 2 ** (ban["ban_count"] - 1), p["max_hours"])
@@ -419,6 +430,12 @@ class FakeSession:
             self.bans = [b for b in self.bans if b["banned_until"] >= cutoff]
             return _FakeResult(purged)
 
+        if "SELECT reason, banned_until" in sql:  # mark_banned: кто держит активный бан пары
+            ban = self._ban(p["proxy_id"], p["source"])
+            if ban is None or ban["banned_until"] <= datetime.now(UTC):
+                return _FakeResult([])
+            return _FakeResult([{"reason": ban.get("reason"), "banned_until": ban["banned_until"]}])
+
         if "SELECT enabled, disabled_reason FROM scrape_proxies" in sql:  # mark_banned diag read
             row = self._by_id(p["id"])
             if row is None:

tradein-mvp/backend/tests/test_2800_per_source_probe.py

@@ -283,11 +283,85 @@ async def test_probe_clears_only_its_own_ban(monkeypatch: pytest.MonkeyPatch) ->
 
     counters = await proxy_pool.run_proxy_healthcheck(db)  # type: ignore[arg-type]
 
-    assert db._ban(1, "avito") is not None, "чужой бан проба снимать не имеет права"
+    avito = db._ban(1, "avito")
+    assert avito is not None, "чужой бан проба снимать не имеет права"
+    assert (avito["reason"], avito["ban_count"]) == ("banned:avito", 1), "и не переписывать"
     assert db._ban(1, "cian") is None, "свой вердикт проба обязана снять"
     assert counters["pair_cleared"] == 1
 
 
+# ── 4b. …и не присваивает чужую (дефект #2803, реализовался на проде) ─────────
+
+
+async def test_probe_does_not_steal_a_live_ban(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Упавшая проба НЕ переписывает активный бан, поставленный боевым сбором.
+
+    Прод 09.08.2026, пара (1, cian): строка `banned:cian, ban_count=1, до 00:21` после
+    упавшей пробы стала `probe:browser, ban_count=2, до 07:43`. Фильтр «снимаю только
+    своё» при этом цел, но защищать перестаёт: присвоенная строка уже «своя», и
+    следующая успешная проба сняла бы ею бан, который площадка поставила по-настоящему.
+    Плюс сама метка перестаёт быть свидетельством («нас отбили» неотличимо от «мы не
+    смогли», #2764), а ban_count складывает события разного рода в одну эскалацию —
+    отдых пары вырос с 6 ч до 12 ч.
+    """
+    until = datetime.now(UTC) + timedelta(hours=6)
+    db = FakeSession(
+        [_proxy(1), _proxy(2)],
+        bans=[
+            {
+                "proxy_id": 1,
+                "source": "cian",
+                "banned_until": until,
+                "ban_count": 1,
+                "reason": "banned:cian",  # боевой сбор: Циан отдал заглушку
+            }
+        ],
+    )
+    _patch_probes(
+        monkeypatch,
+        {("http://u:p@h1:8080", "cian"): (False, "page", "not robots.txt (html_len=374168)")},
+    )
+
+    counters = await proxy_pool.run_proxy_healthcheck(db)  # type: ignore[arg-type]
+
+    ban = db._ban(1, "cian")
+    assert ban is not None
+    assert ban["reason"] == "banned:cian", "проба присвоила себе бан боевого сбора"
+    assert ban["ban_count"] == 1, "два события разного рода посчитаны одной эскалацией"
+    assert ban["banned_until"] == until, "чужой срок проба не пересчитывает (может и укоротить)"
+    assert counters["pair_banned"] == 0, "счётчик не должен объявлять баном то, чего не записал"
+
+
+async def test_live_ban_takes_over_the_probe_row(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Зеркало намеренно НЕ симметрично: боевой сбор строку пробы перехватывает.
+
+    Его вердикт сильнее — площадка отбила нас именно сейчас, — пара остаётся забаненной,
+    а метка становится точнее. Если запретить и ему, строка останется за пробой, и её же
+    зелёный robots.txt снесёт настоящий бан площадки: тот же дефект, только зеркально.
+    """
+    db = FakeSession(
+        [_proxy(1), _proxy(2)],
+        bans=[
+            {
+                "proxy_id": 1,
+                "source": "cian",
+                "banned_until": datetime.now(UTC) + timedelta(hours=6),
+                "ban_count": 1,
+                "reason": "probe:browser",
+            }
+        ],
+    )
+
+    proxy_pool.mark_banned(db, 1, source="cian")  # type: ignore[arg-type]
+
+    assert db._ban(1, "cian")["reason"] == "banned:cian"
+
+    # …и с этой минуты зелёная проба его не снимет — ради чего перехват и нужен.
+    _patch_probes(monkeypatch, {})
+    await proxy_pool.run_proxy_healthcheck(db)  # type: ignore[arg-type]
+    assert db._ban(1, "cian") is not None
+
+
 # ── 5-6. чужие отказы ────────────────────────────────────────────────────────