fix(tradein/avito): #487 BLOCK — mark_banned + defensive rollback (Option B)

deep-code-reviewer P0: mark_rate_limited() writes status='rate_limited' but
SQL CHECK constraint (015 + 051) only allows banned/done/failed/etc.
First blocking event would CheckViolation -> cascade through mark_failed ->
runs stuck in 'running' -> zombie cleanup. Frontend never shows badge.

Option B fix (no migration): rename mark_rate_limited -> mark_banned, use
existing 'banned' status (migration 015 doc'd as 'Avito вернул 403/captcha' -
exactly this scenario).

Plus:
- mark_failed + mark_banned now defensive db.rollback() at start (cascade-safe)
- All mark_* now RETURNING id + log warning when no rows updated (M3)
- Callers updated: scrape_pipeline.run_avito_city_sweep + test names

Refs PR #487 review (deep-code-reviewer, 2026-05-23).

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

@@ -14,7 +14,7 @@ Anti-bot hardening (2023-05-23):
   - Shared AsyncSession на весь sweep (один TLS handshake)
   - asyncio.sleep с random +-20% jitter между detail requests
   - AvitoBlockedError/AvitoRateLimitedError propagation → early abort
-  - После 3 consecutive blocks — abort sweep + mark rate_limited
+  - После 3 consecutive blocks — abort sweep + mark_banned (status='banned')
 
 Graceful degradation на каждом step: exception в одной house/listing
 не валит весь pipeline (кроме blocked exceptions — они abort весь sweep).
@@ -334,7 +334,7 @@ async def run_avito_city_sweep(
     """Full city sweep: iterate anchors × pages → save → enrich houses + detail.
 
     - Single shared AsyncSession на весь sweep (один TLS fingerprint)
-    - AvitoBlockedError/AvitoRateLimitedError → ABORT sweep + mark rate_limited
+    - AvitoBlockedError/AvitoRateLimitedError → ABORT sweep + mark_banned (status='banned')
     - Прочие errors per-anchor логируются, не валят весь sweep
     """
     _anchors = anchors if anchors is not None else EKB_ANCHORS
@@ -389,7 +389,7 @@ async def run_avito_city_sweep(
                     counters.errors_count += 1
                     counters.anchors_done = idx
                     scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
-                    scrape_runs.mark_rate_limited(db, run_id, str(e), counters.to_dict())
+                    scrape_runs.mark_banned(db, run_id, str(e), counters.to_dict())
                     return counters
                 except Exception:
                     logger.exception(

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

@@ -63,53 +63,77 @@ def is_cancelled(db: Session, run_id: int) -> bool:
 
 def mark_done(db: Session, run_id: int, counters: dict[str, int]) -> None:
     """Финализация run: status='done', finished_at=NOW(), counters."""
-    db.execute(
+    row = db.execute(
         text(
             """
             UPDATE scrape_runs
             SET status = 'done', finished_at = NOW(), heartbeat_at = NOW(),
                 counters = CAST(:counters AS jsonb)
             WHERE id = :run_id AND status = 'running'
+            RETURNING id
             """
         ),
         {"run_id": run_id, "counters": json.dumps(counters)},
-    )
+    ).first()
+    if row is None:
+        logger.warning("mark_done no-op: run_id=%d not in 'running' state", run_id)
     db.commit()
 
 
 def mark_failed(db: Session, run_id: int, error: str, counters: dict[str, int]) -> None:
-    """Финализация run: status='failed', error (первые 1000 символов)."""
+    """Финализация run: status='failed', error (первые 1000 символов).
-    db.execute(
+
+    Defensive rollback: если до этого вызова в той же транзакции был ошибочный UPDATE,
+    он мог оставить сессию в error state — rollback сбрасывает состояние.
+    """
+    try:
+        db.rollback()  # defensive: cascade-safe if prior UPDATE left txn in error state
+    except Exception:
+        pass
+    row = db.execute(
         text(
             """
             UPDATE scrape_runs
             SET status = 'failed', finished_at = NOW(), heartbeat_at = NOW(),
                 error = :error, counters = CAST(:counters AS jsonb)
             WHERE id = :run_id AND status = 'running'
+            RETURNING id
             """
         ),
         {"run_id": run_id, "error": error[:1000], "counters": json.dumps(counters)},
-    )
+    ).first()
+    if row is None:
+        logger.warning("mark_failed no-op: run_id=%d not in 'running' state", run_id)
     db.commit()
 
 
-def mark_rate_limited(db: Session, run_id: int, error: str, counters: dict[str, int]) -> None:
+def mark_banned(db: Session, run_id: int, error: str, counters: dict[str, int]) -> None:
-    """Финализация run: status='rate_limited' (IP заблокирован Avito).
+    """Финализация run: status='banned' (IP заблокирован Avito — 403/captcha).
 
-    Отличается от 'failed': это external constraint, не наш bug.
+    Per migration 015 — 'banned' задокументирован как 'Avito вернул 403/captcha'.
-    Cooldown 2-4 часа потом retry.
+    Отличается от 'failed': это external constraint, не наш bug. Cooldown 2-4 часа.
+
+    Defensive rollback: если до этого вызова в той же транзакции был ошибочный UPDATE,
+    он мог оставить сессию в error state — rollback сбрасывает состояние.
     """
-    db.execute(
+    try:
+        db.rollback()  # defensive: cascade-safe if prior UPDATE left txn in error state
+    except Exception:
+        pass
+    row = db.execute(
         text(
             """
             UPDATE scrape_runs
-            SET status = 'rate_limited', finished_at = NOW(), heartbeat_at = NOW(),
+            SET status = 'banned', finished_at = NOW(), heartbeat_at = NOW(),
                 error = :error, counters = CAST(:counters AS jsonb)
             WHERE id = :run_id AND status = 'running'
+            RETURNING id
             """
         ),
         {"run_id": run_id, "error": error[:1000], "counters": json.dumps(counters)},
-    )
+    ).first()
+    if row is None:
+        logger.warning("mark_banned no-op: run_id=%d not in 'running' state", run_id)
     db.commit()
 
 

tradein-mvp/backend/tests/scrapers/test_avito_anti_bot.py

@@ -188,15 +188,18 @@ async def test_pipeline_sleep_between_detail_requests() -> None:
 
 
 @pytest.mark.asyncio
-async def test_city_sweep_marks_rate_limited_on_block() -> None:
+async def test_city_sweep_marks_banned_on_block() -> None:
-    """City sweep mark'ит run как rate_limited при AvitoBlockedError."""
+    """City sweep mark'ит run как banned при AvitoBlockedError.
+
+    Migration 015 задокументировал 'banned' как 'Avito вернул 403/captcha' — именно этот сценарий.
+    """
     from app.services.scrape_pipeline import run_avito_city_sweep
 
     mock_db = MagicMock()
     mock_runs = MagicMock()
     mock_runs.is_cancelled.return_value = False
     mock_runs.update_heartbeat = MagicMock()
-    mock_runs.mark_rate_limited = MagicMock()
+    mock_runs.mark_banned = MagicMock()
     mock_runs.mark_done = MagicMock()
 
     block_error = AvitoBlockedError("IP blocked at page=1")
@@ -221,6 +224,6 @@ async def test_city_sweep_marks_rate_limited_on_block() -> None:
             anchors=[(56.84, 60.60, "Центр"), (56.79, 60.53, "ЮЗ")],
         )
 
-    assert mock_runs.mark_rate_limited.called
+    assert mock_runs.mark_banned.called
     assert not mock_runs.mark_done.called
     assert result.anchors_done == 1