fix(tradein/backfill): rollback on listings save failure + nonnegative used quota

1. cian_history_backfill.py listings block was missing the same
   db.rollback() the houses block already has: save_detail_enrichment()
   runs several unprotected db.execute() and only commits at the end.
   A single bad row (e.g. a non-standard Cian change_time hitting
   CAST(:ct AS timestamptz)) leaves the session in a failed-transaction
   state, and every subsequent listing in the batch (up to 49) then
   fails with PendingRollbackError -- one real failure looked like N
   independent ones in the logs.

2. account_estimate_usage.used had no floor. Prod audit: praktika
   2026-06 shows used=-3 against 42 real estimates. Migration 185
   already fixed this class of bug for user2 (negative used from a
   retired SQL-runbook bonus hack) but only reset that one username.
   No decrement path exists anywhere in app.services.account_quota --
   increment() only ever does used+1 under a WHERE used < lim guard
   (#747) -- so the minus is external (manual UPDATE), not an app bug.
   Migration 189 resets all remaining negative used rows and adds
   CHECK (used >= 0) so a future manual UPDATE can't reintroduce it.

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

@@ -7,6 +7,11 @@
 - Персональный override: таблица account_quota_overrides (username → monthly_limit),
   см. миграцию 185_account_quota_overrides.sql. Заменяет прежний хак бонусных попыток
   через negative `used` (ломал /quota — «Осталось 50 из 15»).
+- `used` в account_estimate_usage защищён CHECK (used >= 0) на уровне схемы, см.
+  миграцию 189_account_estimate_usage_nonnegative.sql — 185 сбросила негативный
+  used только для user2, 189 закрывает остальные аккаунты + запрещает регресс.
+  В коде декремента `used` НЕТ — increment() только `used + 1` под TOCTOU-guard
+  (#747); любой negative used приходит исключительно извне (ручной UPDATE).
 - Без лимита (unlimited): роль admin ИЛИ username == 'kopylov'.
 - Учитываются ТОЛЬКО успешные оценки (инкремент ПОСЛЕ estimate_quality).
 - Если заголовок X-Authenticated-User отсутствует (dev без Caddy) → unlimited,

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

@@ -158,6 +158,18 @@ async def backfill_cian_history(
                             "cian_detail save failed for listing_id=%s: %s", listing_id, exc
                         )
                         result.listings_failed_save += 1
+                        # Roll back to clean session state so next listing can proceed.
+                        # save_detail_enrichment commits on success; on failure the
+                        # transaction is left open/dirty — rollback to avoid session poison
+                        # (same class of defect as the houses block below).
+                        try:
+                            db.rollback()
+                        except Exception as rb_exc:
+                            logger.warning(
+                                "cian_detail rollback failed for listing_id=%s: %s",
+                                listing_id,
+                                rb_exc,
+                            )
 
                     await asyncio.sleep(delay)
 

tradein-mvp/backend/data/sql/189_account_estimate_usage_nonnegative.sql

@@ -0,0 +1,51 @@
+-- Migration 189: account_estimate_usage.used >= 0 — защита от бонус-хака (negative used)
+--
+-- WHY:
+--   Прежний SQL-runbook хак раздачи бонусных попыток (`UPDATE account_estimate_usage
+--   SET used = used - N`) уводил used в отрицательные значения. Migration 185 сбросила
+--   это ТОЛЬКО для user2 (used=-35 -> 0, WHERE username = 'user2'). На проде остаётся
+--   минимум ещё один затронутый аккаунт тем же классом порчи (praktika, период 2026-06:
+--   used=-3 при 42 фактических успешных оценках — расхождение объясняется именно этим
+--   хаком, не кодовым багом).
+--
+--   Аудит app.services.account_quota подтверждает: декремента `used` в текущем коде
+--   НЕТ. increment() делает только `used + 1` под предикатом `WHERE used < :lim`
+--   (#747, atomic conditional increment) — этот путь не может уйти в минус. Значит
+--   источник отрицательных значений исключительно внешний (ручной UPDATE через
+--   runbook), а не баг в приложении.
+--
+-- WHAT:
+--   1. Сброс ВСЕХ оставшихся negative used -> 0 (не только user2, как в 185) —
+--      закрывает praktika и любой другой пропущенный аккаунт.
+--   2. CHECK (used >= 0) — защита на уровне схемы: любой будущий ручной UPDATE/хак,
+--      уводящий used < 0, теперь падает на уровне БД вместо тихой порчи /quota
+--      (GET /quota мог отдать remaining > limit — «Осталось 50 из 15», см. 185).
+--
+-- IDEMPOTENCY:
+--   - UPDATE ... WHERE used < 0 — no-op при повторном прогоне (после первого раза
+--     условие больше не матчит).
+--   - ADD CONSTRAINT через DO-блок с проверкой pg_constraint — Postgres не
+--     поддерживает `ADD CONSTRAINT IF NOT EXISTS` напрямую для CHECK, поэтому
+--     оборачиваем в идемпотентную проверку по имени constraint.
+--
+-- Dependencies: 076_account_estimate_quota.sql (account_estimate_usage),
+-- 185_account_quota_overrides.sql (первый частичный сброс, только user2).
+
+BEGIN;
+
+UPDATE account_estimate_usage
+SET used = 0, updated_at = now()
+WHERE used < 0;
+
+DO $$
+BEGIN
+    IF NOT EXISTS (
+        SELECT 1 FROM pg_constraint
+        WHERE conname = 'account_estimate_usage_used_nonnegative'
+    ) THEN
+        ALTER TABLE account_estimate_usage
+            ADD CONSTRAINT account_estimate_usage_used_nonnegative CHECK (used >= 0);
+    END IF;
+END $$;
+
+COMMIT;

tradein-mvp/backend/tests/test_cian_history_backfill_listings_rollback.py

@@ -0,0 +1,145 @@
+"""Regression test: listings block of cian_history_backfill must rollback on
+save_detail_enrichment failure, same as the houses block already does.
+
+Bug (audit finding #1, fix/tradein-audit-backfill-quota): save_detail_enrichment()
+runs several unprotected db.execute() calls and only commits at the very end. If
+any of them fails (e.g. a malformed `change_time` from Cian hits
+`CAST(:ct AS timestamptz)` with a DataError), the session is left in a failed
+in-transaction state (PendingRollbackError). Without `db.rollback()` in the except
+branch, EVERY subsequent listing in the batch (up to 49 more) raises the same
+PendingRollbackError — one real failure looks like N independent failures in the
+logs, and the batch effectively stops processing after the first bad row.
+
+This test proves: (a) db.rollback() is called after a save failure, (b) the batch
+continues past the failing row and still succeeds on the next one — the exact
+"cascading failure" scenario this fix prevents.
+"""
+
+from __future__ import annotations
+
+import os
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+from app.tasks import cian_history_backfill
+
+
+class _FakeBrowserFetcher:
+    def __init__(self, **kwargs: Any) -> None:
+        pass
+
+    async def __aenter__(self) -> _FakeBrowserFetcher:
+        return self
+
+    async def __aexit__(self, *_: object) -> None:
+        return None
+
+
+def _enrichment() -> SimpleNamespace:
+    return SimpleNamespace(price_changes=[])
+
+
+async def test_listings_save_failure_rolls_back_session() -> None:
+    """save_detail_enrichment raising -> db.rollback() called (session un-poisoned)."""
+    db = MagicMock()
+    db.execute.return_value.mappings.return_value.all.return_value = [
+        {"id": 1, "source_url": "https://cian.ru/1"},
+    ]
+
+    with (
+        patch.object(cian_history_backfill, "BrowserFetcher", _FakeBrowserFetcher),
+        patch.object(cian_history_backfill, "fetch_detail", AsyncMock(return_value=_enrichment())),
+        patch.object(
+            cian_history_backfill,
+            "save_detail_enrichment",
+            side_effect=Exception("DataError: invalid timestamptz"),
+        ),
+        patch("asyncio.sleep", new_callable=AsyncMock),
+    ):
+        result = await cian_history_backfill.backfill_cian_history(
+            db, do_listings=True, do_houses=False, do_valuations=False
+        )
+
+    db.rollback.assert_called_once()
+    assert result.listings_failed_save == 1
+    assert result.listings_succeeded == 0
+
+
+async def test_listings_batch_continues_after_one_bad_row() -> None:
+    """The exact cascading-failure repro: row 1 fails save, row 2 must still
+    succeed — proves rollback actually un-poisons the session for later rows,
+    not just that rollback was called."""
+    db = MagicMock()
+    db.execute.return_value.mappings.return_value.all.return_value = [
+        {"id": 1, "source_url": "https://cian.ru/1"},
+        {"id": 2, "source_url": "https://cian.ru/2"},
+    ]
+
+    save_mock = MagicMock(side_effect=[Exception("DataError: invalid timestamptz"), None])
+
+    with (
+        patch.object(cian_history_backfill, "BrowserFetcher", _FakeBrowserFetcher),
+        patch.object(cian_history_backfill, "fetch_detail", AsyncMock(return_value=_enrichment())),
+        patch.object(cian_history_backfill, "save_detail_enrichment", save_mock),
+        patch("asyncio.sleep", new_callable=AsyncMock),
+    ):
+        result = await cian_history_backfill.backfill_cian_history(
+            db, do_listings=True, do_houses=False, do_valuations=False
+        )
+
+    assert save_mock.call_count == 2
+    db.rollback.assert_called_once()
+    assert result.listings_failed_save == 1
+    assert result.listings_succeeded == 1
+    assert result.listings_processed == 2
+
+
+async def test_listings_rollback_failure_itself_does_not_crash_the_batch() -> None:
+    """If db.rollback() ALSO raises (e.g. connection already dropped), the batch
+    logs a warning and keeps going — mirrors the houses block's same guard."""
+    db = MagicMock()
+    db.rollback.side_effect = Exception("connection already closed")
+    db.execute.return_value.mappings.return_value.all.return_value = [
+        {"id": 1, "source_url": "https://cian.ru/1"},
+    ]
+
+    with (
+        patch.object(cian_history_backfill, "BrowserFetcher", _FakeBrowserFetcher),
+        patch.object(cian_history_backfill, "fetch_detail", AsyncMock(return_value=_enrichment())),
+        patch.object(
+            cian_history_backfill,
+            "save_detail_enrichment",
+            side_effect=Exception("DataError: invalid timestamptz"),
+        ),
+        patch("asyncio.sleep", new_callable=AsyncMock),
+    ):
+        result = await cian_history_backfill.backfill_cian_history(
+            db, do_listings=True, do_houses=False, do_valuations=False
+        )
+
+    assert result.listings_failed_save == 1
+
+
+async def test_listings_no_rollback_on_success() -> None:
+    """Successful save must NOT trigger a rollback (would discard the commit
+    that save_detail_enrichment already made)."""
+    db = MagicMock()
+    db.execute.return_value.mappings.return_value.all.return_value = [
+        {"id": 1, "source_url": "https://cian.ru/1"},
+    ]
+
+    with (
+        patch.object(cian_history_backfill, "BrowserFetcher", _FakeBrowserFetcher),
+        patch.object(cian_history_backfill, "fetch_detail", AsyncMock(return_value=_enrichment())),
+        patch.object(cian_history_backfill, "save_detail_enrichment", MagicMock()),
+        patch("asyncio.sleep", new_callable=AsyncMock),
+    ):
+        result = await cian_history_backfill.backfill_cian_history(
+            db, do_listings=True, do_houses=False, do_valuations=False
+        )
+
+    db.rollback.assert_not_called()
+    assert result.listings_succeeded == 1

tradein-mvp/backend/tests/test_migration_189_account_estimate_usage_nonnegative.py

@@ -0,0 +1,76 @@
+"""Static guards for migration 189 (account_estimate_usage.used >= 0).
+
+Прод применяет data/sql построчно строго (ON_ERROR_STOP). Полный DB-прогон
+требует живой БД; здесь фиксируем структурные инварианты, которые ГАРАНТИРУЮТ
+идемпотентность и что миграция реально закрывает negative-used класс дефекта
+(прежний SQL-runbook хак `used = used - N`, см. migration 185 которая сбросила
+это только для user2 — praktika осталась с used=-3 за 2026-06).
+"""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+_SQL_DIR = Path(__file__).resolve().parents[1] / "data" / "sql"
+_MIGRATION_189 = _SQL_DIR / "189_account_estimate_usage_nonnegative.sql"
+
+
+def _sql() -> str:
+    return _MIGRATION_189.read_text(encoding="utf-8")
+
+
+def _executable_sql() -> str:
+    """SQL без построчных `--`-комментариев — только исполняемый код."""
+    lines = []
+    for raw in _sql().splitlines():
+        code = raw.split("--", 1)[0]
+        if code.strip():
+            lines.append(code)
+    return "\n".join(lines)
+
+
+def _flat(text: str) -> str:
+    return re.sub(r"\s+", " ", text).strip().lower()
+
+
+def test_migration_189_exists() -> None:
+    assert _MIGRATION_189.exists(), f"missing migration: {_MIGRATION_189}"
+
+
+def test_migration_189_is_transactional() -> None:
+    sql = _sql()
+    assert "BEGIN;" in sql
+    assert "COMMIT;" in sql
+
+
+def test_migration_189_resets_all_negative_used_not_scoped_to_user2() -> None:
+    """UPDATE ... WHERE used < 0 без username-фильтра — закрывает praktika и любой
+    другой пропущенный аккаунт, не только user2 (как было в migration 185)."""
+    flat = _flat(_executable_sql())
+    assert "update account_estimate_usage" in flat
+    assert "set used = 0" in flat
+    assert "where used < 0" in flat
+    # НЕ должен быть заскоуплен на конкретного пользователя.
+    assert "username = 'user2'" not in flat
+
+
+def test_migration_189_adds_nonnegative_check_constraint() -> None:
+    """CHECK (used >= 0) — схема защищает от будущих ручных UPDATE в минус."""
+    flat = _flat(_executable_sql())
+    assert "add constraint account_estimate_usage_used_nonnegative" in flat
+    assert "check (used >= 0)" in flat
+
+
+def test_migration_189_constraint_add_is_idempotent() -> None:
+    """ADD CONSTRAINT обёрнут в идемпотентную проверку по pg_constraint.conname —
+    Postgres не поддерживает `ADD CONSTRAINT IF NOT EXISTS` напрямую для CHECK."""
+    flat = _flat(_executable_sql())
+    assert "pg_constraint" in flat
+    assert "conname = 'account_estimate_usage_used_nonnegative'" in flat
+
+
+def test_migration_189_no_psycopg_trap() -> None:
+    """Никаких :param::type — psycopg v3 требует CAST(... AS type) (не применимо
+    в чистом .sql без bind params, но проверяем на регресс copy-paste)."""
+    assert not re.search(r":\w+::", _sql())