fix(tradein/proxy): не отдавать в fallback последний узел выделенной affinity (#2600)

Ревью PR #2609: domclick — ровно один узел (прод scrape_proxies.id=1),
намеренно вырезанный из общего пула через provider_affinity='domclick'
(см. 173_scrape_proxies_add_domclick_affinity.sql) — QRATOR банит всё,
кроме этого одного чистого residential-адреса. Fallback-запрос из
предыдущего коммита мог законно забрать его под avito/cian/yandex,
оставив domclick (сейчас исправно собирает: 6501 активных объявлений,
368/сутки) без прокси вообще — чинили бы один источник ценой полной
поломки другого.

- acquire(): fallback-SELECT дополнен условием "affinity='any' ИЛИ есть
  ДРУГОЙ enabled-узел той же affinity" через коррелированный EXISTS-
  подзапрос (WHERE + FOR UPDATE SKIP LOCKED + ORDER BY last_ok_at NULLS
  LAST, id — сохранены). Кандидат с единственным enabled-узлом своей
  выделенной affinity в fallback не участвует.
- Тесты: единственный domclick-узел → acquire('avito') возвращает None;
  второй enabled domclick-узел появляется — fallback снова срабатывает.
- Починен мок FakeSession (tests/services/test_proxy_pool.py):
  ветка "mark_health ok" раньше ставила enabled=True безусловно по
  совпадению общей подстроки "SET consecutive_fails = 0" (одинаковой в
  старом и новом SQL) — test_mark_health_ok_revives_disabled_proxy
  проходил бы и против кода без реанимации. Теперь ставит enabled=True
  только если в тексте SQL реально есть "enabled". Та же проблема была
  и в fallback-ветке (protects_last_node переопределял логику в Python
  независимо от SQL) — исправлено аналогично: применяется, только если
  в SQL реально есть EXISTS-подзапрос.

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

@@ -30,7 +30,9 @@ Self-healing (#2600):
     Без этого auto-disable необратим: транзиентный сбой = вечный приговор узлу.
   - acquire, не найдя свободного здорового узла нужной provider_affinity, вторым заходом
     берёт любой свободный здоровый узел ЛЮБОЙ affinity (WARNING-лог) — иначе источник
-    голодает при живых свободных узлах чужой affinity.
+    голодает при живых свободных узлах чужой affinity. Fallback НЕ забирает последний
+    enabled-узел выделенной affinity (пример — domclick, один узел на всё, см. acquire
+    docstring) — иначе чинили бы один источник ценой полной поломки другого.
 
 psycopg v3 / SQLAlchemy text(): все параметры через CAST(:x AS type), НЕ :x::type.
 """
@@ -115,6 +117,15 @@ def acquire(db: Session, provider: str, *, run_id: int | None = None) -> ProxyLe
     чужая — только запасной вариант, чтобы источник не голодал при живых свободных узлах
     чужой affinity (#2600).
 
+    Fallback НЕ трогает последний enabled-узел выделенной (не-'any') affinity — см.
+    173_scrape_proxies_add_domclick_affinity.sql: у domclick ровно один узел (id=1),
+    намеренно вырезанный из общего пула, потому что QRATOR банит все прокси кроме этого
+    одного чистого residential-адреса. Если fallback заберёт его под avito/cian/yandex,
+    domclick останется без прокси вообще — хуже, чем голодание исходного источника,
+    которое фикс призван устранить. Кандидат участвует в fallback, только если его
+    affinity='any' ИЛИ у этой affinity есть ДРУГОЙ enabled-узел (EXISTS-подзапрос) —
+    т.е. выдача не обнулит доступность выделенной affinity целиком.
+
     Конкурентные acquire не дерутся за одну строку: SKIP LOCKED пропускает залоченную
     другим вызовом строку, второй параллельный acquire берёт следующую свободную.
 
@@ -145,19 +156,30 @@ def acquire(db: Session, provider: str, *, run_id: int | None = None) -> ProxyLe
 
     fallback_used = False
     if row is None:
-        # Нет своих (provider/'any') — запасной заход: любой свободный здоровый узел,
-        # affinity не важна. Лучше выдать источнику чужой прокси, чем оставить его без
-        # прокси при живых свободных узлах.
+        # Нет своих (provider/'any') — запасной заход: любой свободный здоровый узел
+        # ЛЮБОЙ affinity, кроме последнего enabled-узла выделенной affinity (domclick и
+        # т.п.) — EXISTS-подзапрос требует хотя бы ОДИН ДРУГОЙ enabled-узел той же
+        # affinity, иначе affinity='any' достаточно.
         row = (
             db.execute(
                 text(
                     """
-                    SELECT id, url, kind, rotate_url
-                    FROM scrape_proxies
-                    WHERE enabled
-                      AND consecutive_fails < CAST(:max_fails AS integer)
-                      AND leased_by IS NULL
-                    ORDER BY last_ok_at NULLS LAST, id
+                    SELECT sp.id, sp.url, sp.kind, sp.rotate_url
+                    FROM scrape_proxies AS sp
+                    WHERE sp.enabled
+                      AND sp.consecutive_fails < CAST(:max_fails AS integer)
+                      AND sp.leased_by IS NULL
+                      AND (
+                          sp.provider_affinity = 'any'
+                          OR EXISTS (
+                              SELECT 1
+                              FROM scrape_proxies AS other
+                              WHERE other.provider_affinity = sp.provider_affinity
+                                AND other.enabled
+                                AND other.id <> sp.id
+                          )
+                      )
+                    ORDER BY sp.last_ok_at NULLS LAST, sp.id
                     FOR UPDATE SKIP LOCKED
                     LIMIT 1
                     """

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

@@ -88,13 +88,31 @@ class FakeSession:
                     and r["provider_affinity"] in (provider, "any")
                     and r["leased_by"] is None
                 ]
-            else:  # fallback: любая affinity (#2600 п.3)
+            else:  # fallback: любая affinity, но не последний узел выделенной affinity
+                # (domclick и т.п. — #2600 review). ВАЖНО: применяем эту фильтрацию,
+                # только если сама SQL реально содержит защиту (EXISTS-подзапрос) —
+                # иначе мок реализовывал бы бизнес-логику независимо от проверяемого
+                # кода и не смог бы отличить старый (незащищённый) fallback-запрос от
+                # нового. Тот же класс бага, что был с "enabled" в mark_health-моке.
+                protects_last_node = "EXISTS" in sql
+
+                def _has_backup(row: dict[str, Any]) -> bool:
+                    if row["provider_affinity"] == "any":
+                        return True
+                    return any(
+                        other["provider_affinity"] == row["provider_affinity"]
+                        and other["enabled"]
+                        and other["id"] != row["id"]
+                        for other in self.rows
+                    )
+
                 cands = [
                     r
                     for r in self.rows
                     if r["enabled"]
                     and r["consecutive_fails"] < max_fails
                     and r["leased_by"] is None
+                    and (not protects_last_node or _has_backup(r))
                 ]
             # ORDER BY last_ok_at NULLS LAST, id
             cands.sort(
@@ -142,7 +160,12 @@ class FakeSession:
                 row["latency_ms"] = p["latency_ms"]
                 row["last_ok_at"] = datetime.now(UTC)
                 row["last_check_at"] = datetime.now(UTC)
-                row["enabled"] = True  # реанимация выключенного узла (#2600 п.1)
+                # "SET consecutive_fails = 0" — общая подстрока старого И нового SQL,
+                # НЕ различает их сама по себе. Реанимация (enabled=true) — только если
+                # в тексте запроса реально есть присвоение enabled (#2600 review: старый
+                # мок ставил enabled=True безусловно и не ловил регресс).
+                if "enabled" in sql:
+                    row["enabled"] = True
             return _FakeResult([])
 
         if "consecutive_fails = consecutive_fails + 1" in sql:  # mark_health fail
@@ -235,19 +258,6 @@ def test_acquire_empty_pool_returns_none() -> None:
     assert acquire(db, "avito", run_id=1) is None  # type: ignore[arg-type]
 
 
-def test_acquire_affinity_filter_falls_back_instead_of_none() -> None:
-    """До #2600 такой сетап возвращал None (голодный источник); теперь — fallback-выдача.
-
-    Поведение намеренно изменено п.3 issue #2600: чужой прокси лучше, чем никакого при
-    живом свободном узле. Дублирующее покрытие того же сценария —
-    test_acquire_falls_back_to_other_affinity_when_no_own_free.
-    """
-    db = FakeSession([_proxy(1, affinity="cian")])
-    lease = acquire(db, "avito", run_id=1)  # type: ignore[arg-type]
-    assert lease is not None
-    assert lease.id == 1
-
-
 def test_acquire_skips_disabled() -> None:
     db = FakeSession([_proxy(1, affinity="avito", enabled=False)])
     assert acquire(db, "avito", run_id=1) is None  # type: ignore[arg-type]
@@ -277,8 +287,12 @@ def test_acquire_prefers_own_affinity_when_available() -> None:
 
 
 def test_acquire_falls_back_to_other_affinity_when_no_own_free() -> None:
-    """Свободных avito/any нет, но есть свободный здоровый cian → fallback, а не None."""
-    db = FakeSession([_proxy(1, affinity="cian")])
+    """Свободных avito/any нет, но есть свободный здоровый cian с бэкапом → fallback, а не None.
+
+    Два cian-узла — забрать один через fallback безопасно: у cian остаётся другой
+    enabled-узел (protection на "последний узел affinity" не срабатывает).
+    """
+    db = FakeSession([_proxy(1, affinity="cian"), _proxy(2, affinity="cian")])
     lease = acquire(db, "avito", run_id=1)  # type: ignore[arg-type]
     assert lease is not None
     assert lease.id == 1
@@ -291,6 +305,32 @@ def test_acquire_no_fallback_when_nothing_free_at_all() -> None:
     assert acquire(db, "avito", run_id=1) is None  # type: ignore[arg-type]
 
 
+# ── acquire: fallback НЕ забирает последний узел выделенной affinity (review #2609) ──
+#
+# domclick — ровно один узел (прод scrape_proxies.id=1), намеренно вырезанный из общего
+# пула через provider_affinity='domclick': QRATOR банит всё, кроме этого одного чистого
+# residential-адреса (см. 173_scrape_proxies_add_domclick_affinity.sql). Если fallback
+# заберёт его под avito/cian/yandex — domclick (сейчас исправно собирает: 6501 активных
+# объявлений, 368/сутки) останется без прокси вообще. Починка одного источника ценой
+# полной поломки другого недопустима.
+
+
+def test_acquire_fallback_protects_last_node_of_dedicated_affinity() -> None:
+    """Единственный enabled-узел domclick НЕ отдаётся avito через fallback — None."""
+    db = FakeSession([_proxy(1, affinity="domclick")])
+    assert acquire(db, "avito", run_id=1) is None  # type: ignore[arg-type]
+    assert db._by_id(1)["leased_by"] is None  # узел не тронут
+
+
+def test_acquire_fallback_allows_when_dedicated_affinity_has_backup() -> None:
+    """Второй enabled-узел domclick есть → fallback как и раньше отдаёт свободный."""
+    db = FakeSession([_proxy(1, affinity="domclick"), _proxy(2, affinity="domclick")])
+    lease = acquire(db, "avito", run_id=1)  # type: ignore[arg-type]
+    assert lease is not None
+    assert lease.id == 1
+    assert db._by_id(2)["leased_by"] is None  # у domclick остался живой запасной узел
+
+
 # ── release ──────────────────────────────────────────────────────────────────