fix(tradein/proxy): backup-узел должен быть пригоден + рычаг снятия бана (#2600 п.2)

Правки по deep-review PR #2654.

MEDIUM. Внутренний EXISTS считал backup'ом любой enabled-узел affinity. До п.2 это
было эквивалентно «пригоден», потому что бан выключал узел глобально; теперь узел
бывает enabled и одновременно забанен СВОИМ же источником. Fallback мог увести
последний реально рабочий узел выделенной affinity (два domclick-узла, один забанен
domclick'ом → второй уходит под avito → domclick без прокси). Добавлено требование,
что backup не забанен своим источником — в acquire и зеркально в защите mark_banned.

MEDIUM. У оператора не осталось способа снять бан: в п.1 ложное срабатывание
лечилось PATCH enabled=true (он обнулял disabled_reason), теперь бан живёт в
отдельной таблице и истекает только по таймеру, до 72ч при эскалации. Добавлен
proxy_pool.clear_source_bans; зовётся из patch_proxy при ручном включении и после
УСПЕШНОЙ ротации exit-IP (бан привязан к proxy_id, а банился IP — после смены
адреса строка держала бы узел вне выдачи без причины).

LOW. Тест защиты дублировал логику вместо её проверки: ban-предикаты в фейксессии
теперь гейтятся по подстрокам боевого SQL (как в acquire-ветке) — проверено
мутацией, тесты краснеют при удалении NOT EXISTS из запроса.

LOW. Конверсия в миграции 210 матчила disabled_reason по LIKE 'banned:%' и могла
отменить ручное выключение оператора (формат подсказан комментарием 209-й) — сужено
до точного списка значений домена provider_affinity.

LOW. Docstring report_ban в browser_fetcher описывал старую модель (enabled=false);
формула в COMMENT ON COLUMN была на шаг мимо (срок ТЕКУЩЕГО бана, не следующего).
Расхождение с acquire по leased_by зафиксировано в докстринге как осознанное.

Refs #2600

tradein-mvp/backend/app/api/v1/admin.py

@@ -73,6 +73,7 @@ from app.services import domclick_session as domclick_session_svc
 from app.services import proxy_rotation as proxy_rotation_svc
 from app.services import scrape_runs as runs_mod
 from app.services.geocoder import geocode
+from app.services.proxy_pool import clear_source_bans
 from app.services.scheduler import has_running_run
 from app.services.scraper_adapters import (
     RealEnrichmentJobs,
@@ -2922,6 +2923,11 @@ def patch_proxy(
     if row is None:
         raise HTTPException(status_code=404, detail=f"proxy id={proxy_id} not found")
     db.commit()
+    if payload.enabled:
+        # Ручное включение = чистый лист, как и обнуление disabled_reason выше (#2610).
+        # Иначе узел вернулся бы enabled=true, но по-прежнему невыдаваемым источникам с
+        # активным баном — и оператор не имел бы способа снять ложный бан (#2600 п.2).
+        clear_source_bans(db, proxy_id, reason="manual enable via admin API")
     if not payload.enabled:
         logger.info(
             "proxy_pool: proxy id=%d manually disabled via admin API (reason=%r) — "

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

@@ -52,6 +52,9 @@ Self-healing (#2600):
   - Защита последнего узла сохранена, но теперь ПО ИСТОЧНИКУ: если после записи бана
     у acquire(source) не останется ни одного кандидата — бан не пишется, только
     WARNING (пул надо пополнять, #2638).
+  - Ручное снятие — `clear_source_bans` (ложный бан детектора капчи, #2642) плюс
+    автоматическое после успешной ротации exit-IP: бан привязан к proxy_id, а банился
+    IP, поэтому смена адреса делает строку недействительной.
 
 Ручное выключение vs авто-выключение (#2610):
   - scrape_proxies.disabled_reason (миграция 209) различает ДВЕ разные причины
@@ -100,6 +103,7 @@ __all__ = [
     "STALE_LEASE_MINUTES",
     "ProxyLease",
     "acquire",
+    "clear_source_bans",
     "mark_banned",
     "mark_health",
     "reap_stale_leases",
@@ -257,12 +261,26 @@ def acquire(db: Session, provider: str, *, run_id: int | None = None) -> ProxyLe
                       )
                       AND (
                           sp.provider_affinity = 'any'
+                          -- backup обязан быть ПРИГОДЕН для своей affinity, а не просто
+                          -- enabled (#2600 п.2 deep-review): после перехода на per-source
+                          -- баны узел бывает enabled и одновременно забанен СВОИМ же
+                          -- источником. Засчитывать такой как backup — значит разрешить
+                          -- fallback увести последний реально рабочий узел выделенной
+                          -- affinity и обрушить её (два domclick-узла, один забанен
+                          -- domclick'ом → второй уходит под avito → domclick без прокси).
                           OR EXISTS (
                               SELECT 1
                               FROM scrape_proxies AS other
                               WHERE other.provider_affinity = sp.provider_affinity
                                 AND other.enabled
                                 AND other.id <> sp.id
+                                AND NOT EXISTS (
+                                    SELECT 1
+                                    FROM scrape_proxy_source_bans b2
+                                    WHERE b2.proxy_id = other.id
+                                      AND b2.source = other.provider_affinity
+                                      AND b2.banned_until > now()
+                                )
                           )
                       )
                     ORDER BY sp.last_ok_at NULLS LAST, sp.id
@@ -494,6 +512,13 @@ def mark_banned(db: Session, proxy_id: int, *, source: str) -> None:
     наивный `COUNT(*) WHERE enabled`. Голодать без прокси хуже, чем ходить через
     забаненный: капча хотя бы иногда пропускает, отсутствие узла — нет.
 
+    `leased_by IS NULL` защита НАМЕРЕННО не проверяет (в отличие от acquire) — так было
+    и в п.1, и это не оплошность: lease живёт минуты-часы и снимается сам (release /
+    reap_stale_leases), т.е. занятый узел — это доступный узел через мгновение, а вот
+    отказ записать бан из-за чужого lease был бы вечным (узел так и остался бы в выдаче
+    забаненным). Точность здесь не бесплатна: с проверкой lease защита срабатывала бы
+    ложно при каждом параллельном прогоне.
+
     КОНКУРЕНТНОСТЬ (deep-review fix 2 из #2600 п.1, сохранено): один
     `INSERT ... WHERE EXISTS(...)` — НЕ атомарная гарантия поперёк СТРОК. EXISTS читает
     состояние других строк на момент своего снапшота (READ COMMITTED), но не лочит их —
@@ -557,12 +582,23 @@ def mark_banned(db: Session, proxy_id: int, *, source: str) -> None:
                             -- other.id <> sp.id (а не NOT IN (sp.id, :proxy_id), как в
                             -- п.1): банимый узел остаётся enabled и по-прежнему обслуживает
                             -- СВОЮ affinity — значит он и есть валидный backup для неё.
+                            -- NOT EXISTS b2 — тот же критерий пригодности, что в acquire()
+                            -- fallback: enabled-узел, забаненный СВОИМ источником, backup'ом
+                            -- не считается (иначе защита сочла бы affinity живой, когда она
+                            -- уже нет).
                             OR EXISTS (
                                 SELECT 1
                                 FROM scrape_proxies other
                                 WHERE other.provider_affinity = sp.provider_affinity
                                   AND other.enabled
                                   AND other.id <> sp.id
+                                  AND NOT EXISTS (
+                                      SELECT 1
+                                      FROM scrape_proxy_source_bans b2
+                                      WHERE b2.proxy_id = other.id
+                                        AND b2.source = other.provider_affinity
+                                        AND b2.banned_until > now()
+                                  )
                             )
                         )
                   )
@@ -629,6 +665,53 @@ def mark_banned(db: Session, proxy_id: int, *, source: str) -> None:
         )
 
 
+def clear_source_bans(db: Session, proxy_id: int, *, source: str | None = None, reason: str) -> int:
+    """Снять баны узла по источникам (#2600 п.2). Returns число снятых строк.
+
+    ЗАЧЕМ ОТДЕЛЬНАЯ РУЧКА: до п.2 ложный бан лечился оператором через
+    `PATCH /proxies/{id} enabled=true` — включение обнуляло `disabled_reason`, и узел
+    возвращался в строй. Теперь бан живёт в отдельной таблице и сам по себе истекает
+    только по таймеру, вплоть до 72 часов при эскалации. Без этой функции ложное
+    срабатывание детектора капчи (#2642) парковало бы узел на часы, а снять это можно
+    было бы только руками в SQL.
+
+    ГДЕ ВЫЗЫВАЕТСЯ:
+      - `admin.patch_proxy` при ручном включении узла — «оператор включил» означает
+        чистый лист, ровно как обнуление disabled_reason рядом (#2610);
+      - после УСПЕШНОЙ ротации exit-IP (`proxy_rotation.rotate_proxy`) — площадка
+        банила IP, а строка бана привязана к proxy_id и пережила бы смену адреса,
+        держа узел вне выдачи уже без причины.
+
+    source=None — снять все баны узла; конкретный source — только его. DELETE, а не
+    `banned_until = now()`: строка живёт ещё и ради `ban_count` (память об эскалации),
+    а здесь мы как раз объявляем историю недействительной — новый бан начнётся с базовых
+    SOURCE_BAN_BASE_HOURS.
+
+    `reason` идёт только в лог (человекочитаемый повод — «manual enable», «ip rotated»).
+    """
+    rows = db.execute(
+        text(
+            """
+            DELETE FROM scrape_proxy_source_bans
+            WHERE proxy_id = CAST(:proxy_id AS bigint)
+              AND (CAST(:source AS text) IS NULL OR source = CAST(:source AS text))
+            RETURNING source
+            """
+        ),
+        {"proxy_id": proxy_id, "source": source},
+    ).fetchall()
+    db.commit()
+    if rows:
+        logger.info(
+            "proxy_pool: cleared %d source ban(s) for proxy id=%d (%s) — reason=%s",
+            len(rows),
+            proxy_id,
+            [r.source for r in rows],
+            reason,
+        )
+    return len(rows)
+
+
 def reap_stale_leases(db: Session, older_than_minutes: int = STALE_LEASE_MINUTES) -> int:
     """Освободить lease'ы старше older_than_minutes (упавший sweep не вызвал release).
 

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

@@ -72,6 +72,7 @@ from sqlalchemy import text
 from sqlalchemy.orm import Session
 
 from app.core.config import settings
+from app.services.proxy_pool import clear_source_bans
 
 logger = logging.getLogger(__name__)
 
@@ -363,6 +364,10 @@ async def rotate_proxy(db: Session, proxy_id: int) -> RotationResult:
     new_ip = _extract_new_ip(resp)
     logger.info("proxy_rotation: rotated proxy_id=%d status=%d new_ip=%s", proxy_id, status, new_ip)
     _record_attempt(db, proxy_id, success=True, http_status=status, note=None)
+    # Площадки банили СТАРЫЙ exit-IP, а строка бана привязана к proxy_id (#2600 п.2) —
+    # после смены адреса она держала бы узел вне выдачи уже без причины, вплоть до 72ч
+    # при эскалации. Ротация прошла → история банов этого узла недействительна.
+    clear_source_bans(db, proxy_id, reason=f"exit ip rotated (status={status})")
     return RotationResult(
         ok=True,
         reason=None,

tradein-mvp/backend/data/sql/210_scrape_proxy_source_bans.sql

@@ -29,8 +29,9 @@
 --   disabled_reason (#2610) — узел завис бы выключенным навсегда, до ручного PATCH.
 --   Поэтому здесь каждый такой узел конвертируется в per-source бан на 6 часов
 --   (тот же SOURCE_BAN_BASE_HOURS) и возвращается в строй: enabled=true,
---   disabled_reason=NULL. Ручные выключения (disabled_reason без префикса
---   'banned:') НЕ трогаются — это решение оператора.
+--   disabled_reason=NULL. Матчинг по ТОЧНОМУ списку 'banned:<источник>', а не по
+--   LIKE — ручные тексты оператора (в т.ч. начинающиеся с 'banned:', этот формат
+--   подсказан комментарием 209-й) НЕ трогаются, это его решение.
 --
 -- IDEMPOTENCY / SAFETY:
 --   - Весь файл в одной транзакции BEGIN/COMMIT.
@@ -63,9 +64,11 @@ COMMENT ON TABLE scrape_proxy_source_bans IS
     'транспортных сбоев.';
 
 COMMENT ON COLUMN scrape_proxy_source_bans.ban_count IS
-    'Сколько раз эта пара банилась. Срок следующего бана = base * 2^(ban_count-1), '
-    'потолок SOURCE_BAN_MAX_HOURS. Сбрасывается только удалением строки purge''ем '
-    'через SOURCE_BAN_PURGE_DAYS после истечения бана.';
+    'Сколько раз эта пара банилась. Срок ТЕКУЩЕГО бана (banned_until - banned_at) = '
+    'base * 2^(ban_count-1), потолок SOURCE_BAN_MAX_HOURS: ban_count=1 → 6ч, 2 → 12ч, '
+    '3 → 24ч и т.д. Сбрасывается удалением строки — либо purge''ем через '
+    'SOURCE_BAN_PURGE_DAYS после истечения, либо proxy_pool.clear_source_bans '
+    '(ручное включение узла оператором / успешная ротация exit-IP).';
 
 -- Горячий путь — NOT EXISTS-фильтр в acquire(): (proxy_id, source) уже покрыт PK,
 -- этот индекс закрывает purge/листинг активных банов по времени.
@@ -74,18 +77,21 @@ CREATE INDEX IF NOT EXISTS idx_scrape_proxy_source_bans_until
 
 -- ── конверсия старых глобальных банов (#2600 п.1 → п.2) ─────────────────────
 --
--- substring(... from 8) — отрезает префикс 'banned:' (7 символов). LIKE 'banned:_%'
--- (а не 'banned:%') гарантирует непустой source для NOT NULL-колонки; вырожденное
--- 'banned:' без источника строки бана не получает, но узел ниже всё равно вернётся
--- в строй — висеть вечно выключенным он не должен ни в каком случае.
+-- ТОЧНЫЙ список значений, а не LIKE 'banned:%': 209-я миграция сама предлагает этот
+-- формат в комментарии, поэтому оператор мог написать руками что-то вроде
+-- 'banned:avito вручную'. LIKE тогда дал бы source='avito вручную' (бан-строка, которая
+-- ни с чем не сматчится) и МОЛЧА отменил бы ручное выключение. Домен ниже — тот же, что
+-- у scrape_proxies.provider_affinity (на практике mark_banned п.1 писал только
+-- avito/cian/yandex/domclick — это значения BrowserFetcher._source).
 INSERT INTO scrape_proxy_source_bans (proxy_id, source, banned_until, reason)
 SELECT id,
-       substring(disabled_reason from 8),
+       substring(disabled_reason from 8),  -- отрезает префикс 'banned:' (7 символов)
        now() + interval '6 hours',
        'migrated from disabled_reason (210)'
 FROM scrape_proxies
 WHERE NOT enabled
-  AND disabled_reason LIKE 'banned:_%'
+  AND disabled_reason IN ('banned:avito', 'banned:cian', 'banned:yandex',
+                          'banned:domclick', 'banned:generic', 'banned:any')
 ON CONFLICT (proxy_id, source) DO NOTHING;
 
 UPDATE scrape_proxies
@@ -93,6 +99,7 @@ SET enabled = true,
     disabled_reason = NULL,
     updated_at = now()
 WHERE NOT enabled
-  AND disabled_reason LIKE 'banned:%';
+  AND disabled_reason IN ('banned:avito', 'banned:cian', 'banned:yandex',
+                          'banned:domclick', 'banned:generic', 'banned:any');
 
 COMMIT;

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

@@ -35,7 +35,10 @@ reap_stale_leases проверяются по фактическому изме
       * защита последнего узла: бан НЕ записывается, если у acquire(source) не
         останется кандидатов;
       * run_proxy_healthcheck сносит бан-строки, истёкшие дольше SOURCE_BAN_PURGE_DAYS,
-        и НЕ трогает истёкшие недавно (в них живёт ban_count для эскалации).
+        и НЕ трогает истёкшие недавно (в них живёт ban_count для эскалации);
+      * backup выделенной affinity засчитывается, только если он сам не забанен своим
+        источником (иначе fallback уводил бы последний рабочий узел, deep-review);
+      * clear_source_bans снимает баны узла (все / один source) и обнуляет эскалацию.
 """
 
 from __future__ import annotations
@@ -157,6 +160,10 @@ class FakeSession:
                 # кода и не смог бы отличить старый (незащищённый) fallback-запрос от
                 # нового. Тот же класс бага, что был с "enabled" в mark_health-моке.
                 protects_last_node = "EXISTS" in sql
+                # backup засчитывается, только если он ПРИГОДЕН для своей affinity —
+                # тоже гейтим по подстроке (b2-подзапрос), иначе мок «чинил» бы
+                # незащищённый SQL сам.
+                backup_must_be_usable = "b2.banned_until > now()" in sql
 
                 def _has_backup(row: dict[str, Any]) -> bool:
                     if row["provider_affinity"] == "any":
@@ -165,6 +172,10 @@ class FakeSession:
                         other["provider_affinity"] == row["provider_affinity"]
                         and other["enabled"]
                         and other["id"] != row["id"]
+                        and not (
+                            backup_must_be_usable
+                            and self._has_active_ban(other["id"], other["provider_affinity"])
+                        )
                         for other in self.rows
                     )
 
@@ -274,19 +285,30 @@ class FakeSession:
             if self._by_id(proxy_id) is None:
                 return _FakeResult([])  # узла нет — no-op
 
+            # Оба ban-предиката гейтим по подстрокам самого SQL (как в acquire-ветке):
+            # иначе мок реализовывал бы защиту сам и тест оставался бы зелёным даже
+            # после удаления NOT EXISTS из боевого запроса.
+            filters_bans = "b.banned_until > now()" in sql
+            backup_must_be_usable = "b2.banned_until > now()" in sql
+
             def _is_candidate(sp: dict[str, Any]) -> bool:
                 if not (sp["enabled"] and sp["consecutive_fails"] < max_fails):
                     return False
-                if self._has_active_ban(sp["id"], source):
+                if filters_bans and self._has_active_ban(sp["id"], source):
                     return False  # уже забанен этим же источником — не кандидат
                 if sp["provider_affinity"] in (source, "any"):
                     return True
-                # fallback-safe: другой enabled узел ТОЙ ЖЕ affinity (банимый узел
-                # остаётся enabled и тоже считается — бан теперь per-source).
+                # fallback-safe: другой ПРИГОДНЫЙ узел ТОЙ ЖЕ affinity (банимый узел
+                # остаётся enabled и тоже считается — бан теперь per-source; а вот
+                # забаненный своим же источником backup'ом не считается).
                 return any(
                     other["provider_affinity"] == sp["provider_affinity"]
                     and other["enabled"]
                     and other["id"] != sp["id"]
+                    and not (
+                        backup_must_be_usable
+                        and self._has_active_ban(other["id"], other["provider_affinity"])
+                    )
                     for other in self.rows
                 )
 
@@ -315,6 +337,17 @@ class FakeSession:
                 [{"ban_count": ban["ban_count"], "banned_until": ban["banned_until"]}]
             )
 
+        if "DELETE FROM scrape_proxy_source_bans" in sql and "proxy_id = CAST" in sql:
+            # clear_source_bans: снять баны узла (все либо один source), #2600 п.2
+            cleared = [
+                b
+                for b in self.bans
+                if b["proxy_id"] == p["proxy_id"]
+                and (p["source"] is None or b["source"] == p["source"])
+            ]
+            self.bans = [b for b in self.bans if b not in cleared]
+            return _FakeResult([{"source": b["source"]} for b in cleared])
+
         if "DELETE FROM scrape_proxy_source_bans" in sql:  # purge истёкших (#2600 п.2)
             cutoff = datetime.now(UTC) - timedelta(days=p["days"])
             purged = [{"proxy_id": b["proxy_id"]} for b in self.bans if b["banned_until"] < cutoff]
@@ -474,6 +507,33 @@ def test_acquire_fallback_allows_when_dedicated_affinity_has_backup() -> None:
     assert db._by_id(2)["leased_by"] is None  # у domclick остался живой запасной узел
 
 
+def test_acquire_fallback_backup_must_be_usable_for_its_own_source() -> None:
+    """deep-review #2600 п.2: «backup» — это ПРИГОДНЫЙ узел, а не просто enabled.
+
+    Два узла domclick: node1 забанен САМИМ domclick'ом (законно — node2 тогда был жив) и
+    сейчас занят чужим прогоном, node2 свободен. Для domclick node2 — последний рабочий.
+    Раньше EXISTS видел node1 как backup (он ведь enabled) и разрешал fallback увести
+    node2 под avito — domclick оставался бы без прокси вообще при двух включённых узлах.
+    """
+    db = FakeSession(
+        [_proxy(1, affinity="domclick", leased_by=99), _proxy(2, affinity="domclick")],
+        bans=[
+            {
+                "proxy_id": 1,
+                "source": "domclick",
+                "ban_count": 1,
+                "banned_until": datetime.now(UTC) + timedelta(hours=SOURCE_BAN_BASE_HOURS),
+                "reason": "banned:domclick",
+            }
+        ],
+    )
+    assert acquire(db, "avito", run_id=1) is None  # type: ignore[arg-type]
+    assert db._by_id(2)["leased_by"] is None  # последний рабочий узел domclick не тронут
+    # сам domclick при этом обслуживается: node2 свободен и не забанен
+    lease = acquire(db, "domclick", run_id=2)  # type: ignore[arg-type]
+    assert lease is not None and lease.id == 2
+
+
 # ── release ──────────────────────────────────────────────────────────────────
 
 
@@ -932,6 +992,36 @@ def test_mark_banned_dedicated_affinity_with_backup_counts_as_fallback() -> None
     assert db._ban(1, "avito") is not None
 
 
+def test_mark_banned_backup_of_dedicated_affinity_must_be_usable() -> None:
+    """Зеркало acquire-правила в защите (deep-review #2600 п.2).
+
+    Узел 3 (domclick) забанен САМИМ domclick'ом и вдобавок в карантине по fails, т.е.
+    сам заменой для avito быть не может. Узел 2 — последний РАБОЧИЙ узел domclick,
+    fallback не имеет права его забрать. Значит замены для avito нет вообще → бан
+    avito-узла НЕ пишется. Со старым правилом («backup = любой enabled той же affinity»)
+    узел 3 засчитался бы бэкапом, узел 2 стал бы «доступным» и бан бы записался.
+    """
+    assert mark_banned is not None
+    db = FakeSession(
+        [
+            _proxy(1, affinity="avito"),
+            _proxy(2, affinity="domclick"),
+            _proxy(3, affinity="domclick", fails=MAX_CONSECUTIVE_FAILS),
+        ],
+        bans=[
+            {
+                "proxy_id": 3,
+                "source": "domclick",
+                "ban_count": 1,
+                "banned_until": datetime.now(UTC) + timedelta(hours=SOURCE_BAN_BASE_HOURS),
+                "reason": "banned:domclick",
+            }
+        ],
+    )
+    mark_banned(db, 1, source="avito")  # type: ignore[arg-type]
+    assert db._ban(1, "avito") is None
+
+
 def test_mark_banned_unhealthy_candidate_not_counted_as_backup() -> None:
     """Кандидат формально enabled, но consecutive_fails>=MAX_CONSECUTIVE_FAILS (карантин,
     acquire() его не выдаёт) — НЕ считается доступной заменой, защита срабатывает."""
@@ -1109,3 +1199,68 @@ async def test_healthcheck_purges_long_expired_bans_only(
 
     assert counters["bans_purged"] == 1
     assert [b["source"] for b in db.bans] == ["cian"]
+
+
+# ── clear_source_bans (#2600 п.2 — рычаг оператора против ложного бана) ────────
+#
+# До п.2 ложный бан лечился PATCH enabled=true (обнулял disabled_reason). Теперь бан
+# в отдельной таблице и истекает только по таймеру (до 72ч при эскалации) — без этой
+# ручки ложное срабатывание детектора капчи (#2642) снималось бы только руками в SQL.
+
+
+def _active_ban(pid: int, source: str, ban_count: int = 1) -> dict[str, Any]:
+    return {
+        "proxy_id": pid,
+        "source": source,
+        "ban_count": ban_count,
+        "banned_until": datetime.now(UTC) + timedelta(hours=SOURCE_BAN_MAX_HOURS),
+        "reason": f"banned:{source}",
+    }
+
+
+def test_clear_source_bans_removes_all_bans_of_node() -> None:
+    db = FakeSession(
+        [_proxy(1, affinity="any"), _proxy(2, affinity="any")],
+        bans=[_active_ban(1, "avito"), _active_ban(1, "cian"), _active_ban(2, "avito")],
+    )
+    cleared = proxy_pool.clear_source_bans(db, 1, reason="manual enable")  # type: ignore[arg-type]
+    assert cleared == 2
+    assert [(b["proxy_id"], b["source"]) for b in db.bans] == [(2, "avito")]  # чужой цел
+    # узел снова выдаётся источнику, который его банил
+    lease = acquire(db, "avito", run_id=1)  # type: ignore[arg-type]
+    assert lease is not None and lease.id == 1
+
+
+def test_clear_source_bans_single_source_keeps_others() -> None:
+    db = FakeSession(
+        [_proxy(1, affinity="any")],
+        bans=[_active_ban(1, "avito"), _active_ban(1, "cian")],
+    )
+    cleared = proxy_pool.clear_source_bans(  # type: ignore[arg-type]
+        db, 1, source="avito", reason="ip rotated"
+    )
+    assert cleared == 1
+    assert [b["source"] for b in db.bans] == ["cian"]
+
+
+def test_clear_source_bans_noop_when_nothing_to_clear() -> None:
+    db = FakeSession([_proxy(1, affinity="any")])
+    assert proxy_pool.clear_source_bans(db, 1, reason="manual enable") == 0  # type: ignore[arg-type]
+
+
+def test_clear_source_bans_resets_escalation() -> None:
+    """DELETE, а не banned_until=now(): снятие обнуляет и ban_count — следующий бан
+    начинается с базовых SOURCE_BAN_BASE_HOURS, а не продолжает эскалацию."""
+    assert mark_banned is not None
+    db = FakeSession(
+        [_proxy(1, affinity="any"), _proxy(2, affinity="any")],
+        bans=[_active_ban(1, "avito", ban_count=4)],
+    )
+    proxy_pool.clear_source_bans(db, 1, reason="manual enable")  # type: ignore[arg-type]
+
+    mark_banned(db, 1, source="avito")  # type: ignore[arg-type]
+
+    ban = db._ban(1, "avito")
+    assert ban["ban_count"] == 1
+    expected = datetime.now(UTC) + timedelta(hours=SOURCE_BAN_BASE_HOURS)
+    assert abs((ban["banned_until"] - expected).total_seconds()) < 60

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

@@ -48,6 +48,10 @@ class _FakeResult:
     def fetchone(self) -> dict[str, Any] | None:
         return self._rows[0] if self._rows else None
 
+    def fetchall(self) -> list[Any]:
+        # RETURNING source у clear_source_bans — код читает r.source (attribute access)
+        return [type("Row", (), r)() for r in self._rows]
+
 
 class FakeSession:
     """Эмуляция Session: одна строка scrape_proxies + append-only
@@ -58,9 +62,13 @@ class FakeSession:
         self,
         proxy_row: dict[str, Any] | None,
         rotations: list[dict[str, Any]] | None = None,
+        source_bans: list[dict[str, Any]] | None = None,
     ):
         self.proxy_row = proxy_row
         self.rotations: list[dict[str, Any]] = rotations or []
+        # #2600 п.2: успешная ротация снимает баны узла (IP сменился — бан старого
+        # адреса недействителен), см. proxy_pool.clear_source_bans.
+        self.source_bans: list[dict[str, Any]] = source_bans or []
         self.commits = 0
 
     def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult:
@@ -96,6 +104,11 @@ class FakeSession:
             )
             return _FakeResult([])
 
+        if "DELETE FROM scrape_proxy_source_bans" in sql:  # clear_source_bans (#2600 п.2)
+            cleared = [b for b in self.source_bans if b["proxy_id"] == p["proxy_id"]]
+            self.source_bans = [b for b in self.source_bans if b not in cleared]
+            return _FakeResult([{"source": b["source"]} for b in cleared])
+
         raise AssertionError(f"unhandled SQL: {sql}")
 
     def commit(self) -> None:
@@ -328,6 +341,43 @@ async def test_successful_rotation_writes_history_row(monkeypatch: pytest.Monkey
     assert db.commits >= 1
 
 
+async def test_successful_rotation_clears_source_bans(monkeypatch: pytest.MonkeyPatch) -> None:
+    """(#2600 п.2) Сменился exit-IP → баны площадок на СТАРОМ адресе недействительны.
+
+    Строка бана привязана к proxy_id, а не к IP — без снятия узел остался бы вне выдачи
+    источнику до 72 часов уже без причины.
+    """
+    monkeypatch.setattr(proxy_rotation.settings, "asocks_api_token", SECRET_TOKEN)
+    fake_client, _calls = _fake_async_client(response=(200, {"ip": "9.9.9.9"}), exception=None)
+    monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", fake_client)
+
+    db = FakeSession(
+        _proxy_row(),
+        source_bans=[
+            {"proxy_id": 1, "source": "avito"},
+            {"proxy_id": 1, "source": "cian"},
+            {"proxy_id": 2, "source": "avito"},  # чужой узел — не трогаем
+        ],
+    )
+    result = await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]
+
+    assert result.ok is True
+    assert db.source_bans == [{"proxy_id": 2, "source": "avito"}]
+
+
+async def test_failed_rotation_keeps_source_bans(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Провайдер ответил ошибкой — IP НЕ сменился, баны обязаны остаться."""
+    monkeypatch.setattr(proxy_rotation.settings, "asocks_api_token", SECRET_TOKEN)
+    fake_client, _calls = _fake_async_client(response=(500, None), exception=None)
+    monkeypatch.setattr(proxy_rotation.httpx, "AsyncClient", fake_client)
+
+    db = FakeSession(_proxy_row(), source_bans=[{"proxy_id": 1, "source": "avito"}])
+    result = await proxy_rotation.rotate_proxy(db, 1)  # type: ignore[arg-type]
+
+    assert result.ok is False
+    assert db.source_bans == [{"proxy_id": 1, "source": "avito"}]
+
+
 # ── 401 → loud failure ───────────────────────────────────────────────────────
 
 

tradein-mvp/backend/tests/test_admin_proxies.py

@@ -51,6 +51,17 @@ def _scalar_result(value: object) -> MagicMock:
     return res
 
 
+def _cleared_bans_result(rows: list[dict[str, Any]] | None = None) -> MagicMock:
+    """Ответ на DELETE ... RETURNING source (proxy_pool.clear_source_bans, #2600 п.2).
+
+    PATCH enabled=true снимает баны узла по источникам — «ручное включение = чистый
+    лист», как и обнуление disabled_reason рядом.
+    """
+    res = MagicMock()
+    res.fetchall.return_value = [type("Row", (), r)() for r in (rows or [])]
+    return res
+
+
 def _bans_result(rows: list[dict[str, Any]] | None = None) -> MagicMock:
     """Ответ на ВТОРОЙ execute в /proxies-ручках — активные баны по источникам (#2600 п.2).
 
@@ -305,7 +316,7 @@ def test_patch_enable_clears_disabled_reason(client: TestClient, db: MagicMock)
     result.mappings.return_value.fetchone.return_value = _proxy_db_row(
         enabled=True, disabled_reason=None
     )
-    db.execute.side_effect = [result, _bans_result()]
+    db.execute.side_effect = [result, _cleared_bans_result(), _bans_result()]
 
     r = client.patch("/api/v1/admin/proxies/1", json={"enabled": True})
     assert r.status_code == 200, r.text
@@ -314,6 +325,37 @@ def test_patch_enable_clears_disabled_reason(client: TestClient, db: MagicMock)
     assert params["enabled"] is True
 
 
+def test_patch_enable_clears_source_bans(client: TestClient, db: MagicMock) -> None:
+    """(#2600 п.2) Ручное включение = чистый лист: снимаются и per-source баны, иначе у
+    оператора нет способа отменить ложный бан (детектор капчи, #2642) — узел был бы
+    enabled=true и всё равно невыдаваемым источнику до 72 часов."""
+    result = MagicMock()
+    result.mappings.return_value.fetchone.return_value = _proxy_db_row(
+        enabled=True, disabled_reason=None
+    )
+    db.execute.side_effect = [result, _cleared_bans_result([{"source": "avito"}]), _bans_result()]
+
+    r = client.patch("/api/v1/admin/proxies/1", json={"enabled": True})
+    assert r.status_code == 200, r.text
+    delete_sql = str(db.execute.call_args_list[1].args[0])
+    assert "DELETE FROM scrape_proxy_source_bans" in delete_sql
+    assert db.execute.call_args_list[1].args[1]["proxy_id"] == 1
+
+
+def test_patch_disable_keeps_source_bans(client: TestClient, db: MagicMock) -> None:
+    """Выключение узла бан-строки НЕ снимает — снятие это «оператор говорит, что узел
+    в порядке», а выключение утверждает обратное."""
+    result = MagicMock()
+    result.mappings.return_value.fetchone.return_value = _proxy_db_row(enabled=False)
+    db.execute.side_effect = [result, _bans_result()]
+
+    r = client.patch("/api/v1/admin/proxies/1", json={"enabled": False})
+    assert r.status_code == 200, r.text
+    assert not any(
+        "DELETE FROM scrape_proxy_source_bans" in str(c.args[0]) for c in db.execute.call_args_list
+    )
+
+
 # ── POST /proxies/bulk — не глушит ручной disable (#2610) ──────────────────
 
 

tradein-mvp/packages/scraper-kit/src/scraper_kit/browser_fetcher.py

@@ -323,7 +323,7 @@ class BrowserFetcher:
             )
 
     def report_ban(self, reason: str) -> None:
-        """Пометить ТЕКУЩИЙ lease забаненным площадкой (#2600 п.1).
+        """Пометить ТЕКУЩИЙ lease забаненным площадкой (#2600 п.1, п.2).
 
         Вызывать из точки детекта бана (заглушка HTTP 200 / капча / QRATOR-маркер),
         ПОКА lease ещё держится (до `__aexit__`/`_release_lease`) — `fetch()` уже
@@ -336,8 +336,11 @@ class BrowserFetcher:
         подключён — best-effort, как touch/mark_health/release: проблема пула не должна
         ронять сбор. Lease НЕ освобождается и НЕ ротируется здесь — вызывающий код обычно
         сразу поднимает исключение и завершает сессию (release произойдёт как обычно в
-        `__aexit__`); пометка узла (`enabled=false`) переживает release — `acquire()`
-        фильтрует по `enabled`, свежий lease его больше не возьмёт.
+        `__aexit__`); бан переживает release — с #2600 п.2 это строка в
+        `scrape_proxy_source_bans` для пары (узел, `self._source`), и `acquire(source)`
+        её фильтрует, так что свежий lease ЭТОГО источника узел больше не возьмёт. Узел
+        при этом остаётся `enabled` и продолжает работать на другие источники: площадка
+        забанила IP, а не сломала прокси.
         """
         if self._lease is None or self._proxy_provider is None:
             return