fix(tradein/deactivate): cap effective TTL floor at 2x configured value

Revisit-floor (#2659) raises effective TTL via max(ttl_days, floor) with no
upper bound -- a positive feedback loop confirmed on prod: slow crawl raises
the floor, a high floor keeps stale listings marked active longer than a
fresh sweep needs to return, the "active" pool bloats with rot, and the next
floor measurement on that bloated pool comes out even higher. Yandex counters
sat at ttl_days_effective=75/75/75/39/52/54 for six runs straight with
deactivated=0; 23,687/44,744 "active" avito listings hadn't been confirmed in
>7 days, cian 10,572/19,514 and yandex 7,178/15,790 were >30 days stale, the
oldest "active" row hadn't been seen in 86 days.

CAP_MULT=2 caps the floor's upward push without disabling it -- the floor
still protects against premature deactivation during genuinely slow (but
alive) crawl cycles, it just can no longer grow unbounded. Beyond 2x, a
persistently low crawl rate is better handled by the existing health gate
(min_confirmations), which disables deactivation outright instead of
stretching TTL forever.

When the cap binds, counters gain ttl_floor_capped=1 + ttl_days_floor_raw
(the uncapped value) so it's visible in the run-history dashboard, not just
logs -- counters are stored as-is in scrape_runs.counters.

Single fix point: all four sources (avito/yandex/cian/domklik) route through
this one deactivate_stale_listings() via the product_handlers wildcard
"deactivate_stale_*" handler, so no other task file needed the change.

tradein-mvp/backend/app/tasks/deactivate_stale_avito.py

@@ -190,6 +190,32 @@ DEFAULT_REVISIT_FLOOR_QUANTILE = 0.99
 _REVISIT_FLOOR_SEGMENT_FILTER = "\n          AND l.listing_segment = ANY(CAST(:segments AS text[]))"
 
 
+# ── Потолок эффективного TTL (положительная обратная связь пола, найдено 2026-08-15) ──
+# У пола выше нет верхней границы: max(ttl_days, пол) может расти неограниченно.
+# ЗАМЕР НА ПРОДЕ, из-за которого этот потолок существует: 23 687 из 44 744 «активных»
+# avito-объявлений не подтверждались >7 суток; cian 10 572/19 514 и yandex 7 178/15 790
+# — старше 30 суток; самая старая «активная» запись не видена 86 суток. В пуле
+# сравнимых 3 497 просроченных строк. У yandex counters держали ttl_days_effective
+# 75/75/75/39/52/54 шесть прогонов подряд при deactivated=0.
+#
+# МЕХАНИЗМ ПЕТЛИ: медленный обход поднимает пол (он же квантиль разрывов переобхода)
+# -> высокий пол продлевает жизнь снятым лотам дольше, чем к ним успевает вернуться
+# свежий обход -> пул «активных» раздувается «протухшими» строками -> следующий замер
+# пола на том же раздутом пуле оказывается ещё выше. Без верхней границы это не
+# самокорректирующийся пол, а положительная обратная связь.
+#
+# CAP_MULT = 2 -- эффективный TTL не может превысить удвоенный заданный оператором
+# ttl_days. Пол по-прежнему может его поднять (ради #2659 -- см. комментарий выше:
+# ложные снятия при неполном покрытии обхода), но не бесконечно. Почему именно 2, а
+# не 3 или 1.5: вдвое — это ещё «мы искренне не уверены, что молчание значит
+# снятие», не «источник вообще умер». Дальнейший рост пола сигнализирует не о
+# медленном, но живом обходе, а о мёртвом источнике -- для ЭТОГО случая уже есть
+# отдельный гейт по здоровью (min_confirmations) выше в этой же функции, который
+# выключает деактивацию целиком, а не растягивает TTL до бесконечности. Калибровочная
+# ручка, не догма -- при новом замере можно пересмотреть, как и revisit_floor_quantile.
+CAP_MULT = 2
+
+
 def _build_revisit_floor_sql(staleness_column: str, *, with_segments: bool) -> Any:
     """Квантиль возраста, при котором свип за окно ДОКАЗАЛ, что строка жива.
 
@@ -334,7 +360,8 @@ def deactivate_stale_listings(
         health_window_days: окно подтверждений для гейта, суток. Дефолт 3.
         revisit_floor_quantile: пол TTL по измеренному циклу переобхода (#2659).
             Квантиль возраста, при котором свип за окно ДОКАЗАЛ строку живой;
-            эффективный TTL = max(ttl_days, этот пол). 0 -> пол выключен (так
+            эффективный TTL = min(max(ttl_days, этот пол), ttl_days * CAP_MULT) --
+            пол поднимает TTL, но не выше потолка. 0 -> пол выключен (так
             вызывают старые тесты и совместимая обёртка), рабочее значение —
             DEFAULT_REVISIT_FLOOR_QUANTILE, см. комментарий выше.
 
@@ -345,7 +372,9 @@ def deactivate_stale_listings(
     Returns {"deactivated": N} -- количество обновлённых строк (1:1 со снимками).
     Если гейт не пропустил прогон: {"deactivated": 0, "confirmations": N,
     "skipped_unhealthy": 1} и НИ ОДНА строка не тронута. Если пол переобхода поднял
-    TTL: дополнительно {"revisit_floor_days": N, "ttl_days_effective": N}.
+    TTL: дополнительно {"revisit_floor_days": N, "ttl_days_effective": N}. Если пол
+    упёрся в потолок CAP_MULT: дополнительно {"ttl_floor_capped": 1,
+    "ttl_days_floor_raw": N} -- N это то, во что пол поднял бы TTL БЕЗ потолка.
 
     Raises:
         ValueError: если staleness_column не входит в whitelist (проверка ДО SQL,
@@ -402,7 +431,8 @@ def deactivate_stale_listings(
                 return counters
 
         # Пол TTL по измеренному циклу переобхода (#2659) — тоже ДО UPDATE и по тому же
-        # срезу. Поднимает порог, никогда не опускает: max(), а не замена.
+        # срезу. Поднимает порог (max), но не выше потолка CAP_MULT * ttl_days (min) —
+        # см. комментарий у CAP_MULT про петлю с положительной обратной связью.
         effective_ttl_days = ttl_days
         if revisit_floor_quantile > 0:
             floor_params: dict[str, Any] = {
@@ -420,9 +450,37 @@ def deactivate_stale_listings(
             # Тогда пола нет и TTL остаётся как задан: выдумывать пол не из чего.
             if floor_days is not None:
                 counters["revisit_floor_days"] = ceil(float(floor_days))
-                effective_ttl_days = max(ttl_days, counters["revisit_floor_days"])
+                # Пол поднимает TTL (max), потолок CAP_MULT его не пускает выше
+                # ttl_days * CAP_MULT (min) — без этого пол растёт без ограничения
+                # (см. комментарий у CAP_MULT).
+                raw_effective_ttl_days = max(ttl_days, counters["revisit_floor_days"])
+                capped_ttl_days = ttl_days * CAP_MULT
+                effective_ttl_days = min(raw_effective_ttl_days, capped_ttl_days)
                 counters["ttl_days_effective"] = effective_ttl_days
-                if effective_ttl_days > ttl_days:
+
+                if raw_effective_ttl_days > capped_ttl_days:
+                    # Пол упёрся в потолок -- оба числа в counters (не только в логе),
+                    # чтобы это было видно в витрине прогонов, а не только в логах.
+                    # 1, а не True -- counters типизирован dict[str, int] (тот же
+                    # идиом, что skipped_unhealthy выше).
+                    counters["ttl_floor_capped"] = 1
+                    counters["ttl_days_floor_raw"] = raw_effective_ttl_days
+                    logger.warning(
+                        "deactivate_stale source=%s run_id=%d TTL пол упёрся в потолок "
+                        "CAP_MULT=%d: пол поднял бы TTL до %d сут, потолок ограничивает "
+                        "заданные %d сут значением %d (квантиль %.3f, segments=%r) — "
+                        "растущий без ограничения пол это петля с положительной обратной "
+                        "связью, см. комментарий у CAP_MULT",
+                        listing_source,
+                        run_id,
+                        CAP_MULT,
+                        raw_effective_ttl_days,
+                        ttl_days,
+                        effective_ttl_days,
+                        revisit_floor_quantile,
+                        segments,
+                    )
+                elif effective_ttl_days > ttl_days:
                     logger.warning(
                         "deactivate_stale source=%s run_id=%d TTL поднят с %d до %d сут: "
                         "свип за %d сут доказал живой строку, молчавшую %d сут "

tradein-mvp/backend/tests/test_deactivate_stale_revisit_floor.py

@@ -126,13 +126,21 @@ def _run(db: _FakeDB, monkeypatch: pytest.MonkeyPatch, **kwargs: Any) -> dict[st
 
 
 def test_effective_ttl_covers_every_proven_false_kill(monkeypatch: pytest.MonkeyPatch) -> None:
-    """Ни одно из 127 доказанно ложных снятий не должно повториться.
+    """Ни одно из 127 доказанно ложных снятий (cian/yandex) не должно повториться.
 
     Все они произошли на возрасте 29.9..30.3 суток. Эффективный TTL обязан быть
-    строго выше этого возраста на КАЖДОМ прод-срезе — иначе следующий прогон
+    строго выше этого возраста на cian/yandex-срезах — иначе следующий прогон
-    снимет ту же строку снова.
+    снимет ту же строку снова. avito пропущен намеренно: 127 доказанных ложных
+    снятий (_FALSE_KILLS_BY_CITY) измерены только по cian/yandex, а гипотетический
+    замер пола avito=69.7 при ttl=10 -- ровно тот случай, для которого заведён
+    потолок CAP_MULT (#TTL-CAP, 2026-08-15): без потолка пол растёт без
+    ограничения (петля с положительной обратной связью, найдена на проде),
+    покрытие такого выброса потолком намеренно НЕ гарантируется -- см.
+    test_deactivate_stale_ttl_cap.py.
     """
     for slice_name, (source, segments, ttl_days, floor) in _PROD_FLOORS.items():
+        if source == "avito":
+            continue
         db = _FakeDB(floor_days=floor)
         out = _run(
             db,

tradein-mvp/backend/tests/test_deactivate_stale_ttl_cap.py

@@ -0,0 +1,198 @@
+"""Потолок эффективного TTL деактивации (найдено на проде 2026-08-15).
+
+Пол TTL по измеренному циклу переобхода (#2659, deactivate_stale_avito.py) поднимает
+эффективный TTL через max(ttl_days, пол) без верхней границы. На проде это оказалось
+петлёй с положительной обратной связью: медленный обход поднимает пол, высокий пол
+продлевает жизнь снятым лотам дольше, чем к ним успевает вернуться свежий обход, пул
+«активных» раздувается протухшими строками -- 23 687 из 44 744 avito-строк не
+подтверждались >7 суток; cian 10 572/19 514 и yandex 7 178/15 790 -- старше 30 суток;
+самая старая «активная» запись не видена 86 суток. У yandex ttl_days_effective держали
+75/75/75/39/52/54 шесть прогонов подряд при deactivated=0.
+
+Этот файл проверяет CAP_MULT -- потолок, не пускающий эффективный TTL выше
+ttl_days * CAP_MULT, независимо от того, насколько высоко посчитанный пол.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any
+
+import pytest
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from app.tasks import deactivate_stale_avito as task_mod
+
+# ── Фейковая сессия (тот же контракт, что в test_deactivate_stale_revisit_floor.py) ──
+
+
+class _FakeResult:
+    def __init__(self, rowcount: int = 0, scalar_value: Any = None) -> None:
+        self.rowcount = rowcount
+        self._scalar = scalar_value
+
+    def scalar(self) -> Any:
+        return self._scalar
+
+
+class _FakeDB:
+    """Session-заглушка: percentile_disc -> пол, count(*) -> подтверждения, UPDATE -> rowcount."""
+
+    def __init__(
+        self,
+        *,
+        floor_days: float | None,
+        confirmations: int = 10_000,
+        rowcount: int = 137,
+    ) -> None:
+        self._floor = floor_days
+        self._confirmations = confirmations
+        self._rowcount = rowcount
+        self.executed: list[tuple[str, dict[str, Any] | None]] = []
+        self.committed = False
+        self.rolled_back = False
+
+    def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> _FakeResult:
+        sql = str(stmt.text)
+        self.executed.append((sql, params))
+        if "percentile_disc" in sql:
+            return _FakeResult(scalar_value=self._floor)
+        if "SELECT count(*)" in sql:
+            return _FakeResult(scalar_value=self._confirmations)
+        return _FakeResult(rowcount=self._rowcount)
+
+    def commit(self) -> None:
+        self.committed = True
+
+    def rollback(self) -> None:
+        self.rolled_back = True
+
+    @property
+    def update_query(self) -> tuple[str, dict[str, Any] | None]:
+        return next((e for e in self.executed if "UPDATE listings" in e[0]), ("", None))
+
+
+def _run(db: _FakeDB, monkeypatch: pytest.MonkeyPatch, **kwargs: Any) -> dict[str, int]:
+    monkeypatch.setattr(task_mod.runs_mod, "mark_done", lambda *a, **k: None)
+    monkeypatch.setattr(task_mod.runs_mod, "mark_failed", lambda *a, **k: None)
+    return task_mod.deactivate_stale_listings(
+        db,  # type: ignore[arg-type]
+        1,
+        listing_source=kwargs.pop("listing_source", "avito"),
+        ttl_days=kwargs.pop("ttl_days", 10),
+        **kwargs,
+    )
+
+
+# ── Контракт из задачи ─────────────────────────────────────────────────────────
+
+
+def test_high_floor_is_capped_at_double_ttl(monkeypatch: pytest.MonkeyPatch) -> None:
+    """revisit_floor=75, ttl_days=10 -> итог 20 (потолок 2x), НЕ 75."""
+    db = _FakeDB(floor_days=75.0)
+    out = _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99)
+    assert out["ttl_days_effective"] == 20
+    assert out["ttl_floor_capped"] == 1
+    assert out["ttl_days_floor_raw"] == 75
+    _, update_params = db.update_query
+    assert update_params is not None
+    assert update_params["ttl_days"] == 20, "UPDATE обязан получить капнутый TTL, не сырой пол"
+
+
+def test_low_floor_leaves_ttl_unchanged(monkeypatch: pytest.MonkeyPatch) -> None:
+    """revisit_floor=5, ttl_days=10 -> итог 10 (пол ниже заданного TTL, max() его не поднимает)."""
+    db = _FakeDB(floor_days=5.0)
+    out = _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99)
+    assert out["ttl_days_effective"] == 10
+    assert "ttl_floor_capped" not in out
+    assert "ttl_days_floor_raw" not in out
+    _, update_params = db.update_query
+    assert update_params is not None
+    assert update_params["ttl_days"] == 10
+
+
+# ── Контракт потолка ────────────────────────────────────────────────────────────
+
+
+def test_cap_mult_is_named_module_constant_equal_two() -> None:
+    assert task_mod.CAP_MULT == 2
+
+
+def test_floor_between_ttl_and_cap_is_not_flagged_capped(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Пол поднял TTL, но не дотянулся до потолка -- capped-флаг НЕ выставляется."""
+    db = _FakeDB(floor_days=15.0)
+    out = _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99)
+    assert out["ttl_days_effective"] == 15
+    assert "ttl_floor_capped" not in out
+
+
+def test_floor_exactly_at_cap_boundary_is_not_flagged_capped(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Пол ровно на потолке (2x ttl) -- это ещё "поднят до потолка", не "срезан выше него".
+
+    Формула -- min(raw, cap): при raw == cap срезания не происходит (raw > cap ложно),
+    капнутый флаг предназначен сигналить именно "потолок реально что-то отрезал".
+    """
+    db = _FakeDB(floor_days=20.0)
+    out = _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99)
+    assert out["ttl_days_effective"] == 20
+    assert "ttl_floor_capped" not in out
+
+
+def test_cap_logs_warning_containing_both_numbers(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """WARNING при срезании содержит и сырой пол, и капнутый результат -- не только counters."""
+    db = _FakeDB(floor_days=75.0)
+    with caplog.at_level("WARNING", logger=task_mod.logger.name):
+        _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99)
+    messages = " ".join(r.getMessage() for r in caplog.records)
+    assert "75" in messages, "лог обязан называть сырой пол"
+    assert "20" in messages, "лог обязан называть итоговый (капнутый) TTL"
+
+
+def test_cap_never_lowers_ttl_below_configured_value(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Потолок -- верхняя граница, не альтернативный источник истины: заданный TTL
+    (10) остаётся нижней границей независимо от того, насколько низко ушёл пол."""
+    db = _FakeDB(floor_days=1.0)
+    out = _run(db, monkeypatch, ttl_days=10, revisit_floor_quantile=0.99)
+    assert out["ttl_days_effective"] == 10
+
+
+# ── avito self-descend (52 -> ... -> 10) не должен ломаться потолком ────────────
+
+
+def test_avito_high_transient_floor_is_capped_not_left_unbounded(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Наблюдённый на проде транзиентный пик avito (счётчики видели ttl_days_effective=52)
+    теперь капается на 2x ttl=20, а не пропускается в UPDATE как есть."""
+    db = _FakeDB(floor_days=52.0)
+    out = _run(db, monkeypatch, listing_source="avito", ttl_days=10, revisit_floor_quantile=0.99)
+    assert out["ttl_days_effective"] == 20
+    assert out["ttl_floor_capped"] == 1
+    assert out["ttl_days_floor_raw"] == 52
+
+
+def test_avito_recovered_low_floor_still_reaches_configured_ttl(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """После восстановления обхода (пол опустился ниже ttl_days=10, как на проде 52->10)
+    потолок не мешает нормальному пути -- эффективный TTL просто равен заданному."""
+    db = _FakeDB(floor_days=9.0)
+    out = _run(db, monkeypatch, listing_source="avito", ttl_days=10, revisit_floor_quantile=0.99)
+    assert out["ttl_days_effective"] == 10
+    assert "ttl_floor_capped" not in out
+
+
+def test_avito_floor_above_ttl_but_under_cap_passes_through_uncapped(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Промежуточная точка того же самопонижения (пол между ttl и потолком, например 18)
+    поднимает TTL как раньше -- потолок не мешает нормальному постепенному пути."""
+    db = _FakeDB(floor_days=18.0)
+    out = _run(db, monkeypatch, listing_source="avito", ttl_days=10, revisit_floor_quantile=0.99)
+    assert out["ttl_days_effective"] == 18
+    assert "ttl_floor_capped" not in out