harden: cache PG/Redis singletons, safe orphan-lock cleanup, ScraperAbortedError on lock loss

iaai_scraper/worker/tasks.py

@@ -11,6 +11,7 @@ from celery import shared_task
 from redis import Redis
 
 from ..core.config import Settings, parse_listing_segments
+from ..core.exceptions import ScraperAbortedError
 from ..scraper import IAAIScraper
 from ..storage.db import PersistenceService
 
@@ -26,6 +27,11 @@ SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_TTL_SECONDS = 24 * 60 * 60
 SYNC_LISTING_FOLLOWUP_PENDING_KEY = "iaai:state:sync_listing_followup_pending"
 SYNC_LISTING_TASK_NAME = "iaai_scraper.worker.tasks.sync_listing_task"
 
+# Process-level caches (по одному на prefork-child) — пересоздаются при ошибках
+# или при старте нового child (max_tasks_per_child).
+_CACHED_PERSISTENCE: PersistenceService | None = None
+_CACHED_REDIS: Redis | None = None
+
 
 def _retry_with_backoff(func, *, attempts: int = 5, base_delay_s: float = 1.0):
     last_exc: Exception | None = None
@@ -92,32 +98,75 @@ def _sync_listing_lock_ttl_seconds() -> int:
 
 
 def _get_persistence() -> PersistenceService:
-    settings = Settings()
-    persistence = PersistenceService(settings)
+    # Кэшируем engine на уровне процесса — иначе каждая task создаёт новый
+    # SQLAlchemy pool (5+10 коннектов), и за сутки worker исчерпает
+    # max_connections Postgres. С max_tasks_per_child=5 пул переиспользуется
+    # для всех 5 тасок, после чего child перезапустится и пул пересоздастся.
+    global _CACHED_PERSISTENCE
+    if _CACHED_PERSISTENCE is None:
+        settings = Settings()
+        _CACHED_PERSISTENCE = PersistenceService(settings)
+
+    persistence = _CACHED_PERSISTENCE
 
     def _ping_db() -> None:
         with persistence.engine.connect() as conn:
             conn.exec_driver_sql("SELECT 1")
 
-    _retry_with_backoff(_ping_db, attempts=5, base_delay_s=1.0)
+    try:
+        _retry_with_backoff(_ping_db, attempts=3, base_delay_s=1.0)
+    except Exception:
+        # Engine мог стать невалидным (PG рестарт) — пересоздаём.
+        logger.warning("Cached DB engine ping failed; recreating engine", exc_info=True)
+        try:
+            _CACHED_PERSISTENCE.engine.dispose()
+        except Exception:
+            pass
+        _CACHED_PERSISTENCE = PersistenceService(Settings())
+        persistence = _CACHED_PERSISTENCE
+        _retry_with_backoff(_ping_db, attempts=5, base_delay_s=1.0)
     return persistence
 
 
 def _get_redis() -> Redis:
-    settings = Settings()
-    redis_client = Redis.from_url(
-        settings.redis.url,
-        decode_responses=True,
-        socket_connect_timeout=settings.redis.socket_connect_timeout_seconds,
-        socket_timeout=settings.redis.socket_timeout_seconds,
-        health_check_interval=settings.redis.health_check_interval_seconds,
-        retry_on_timeout=True,
-    )
+    # Кэшируем клиент на уровне процесса — Redis().from_url каждый раз создаёт
+    # отдельный connection pool. См. _get_persistence для аналогичного обоснования.
+    global _CACHED_REDIS
+    if _CACHED_REDIS is None:
+        settings = Settings()
+        _CACHED_REDIS = Redis.from_url(
+            settings.redis.url,
+            decode_responses=True,
+            socket_connect_timeout=settings.redis.socket_connect_timeout_seconds,
+            socket_timeout=settings.redis.socket_timeout_seconds,
+            health_check_interval=settings.redis.health_check_interval_seconds,
+            retry_on_timeout=True,
+        )
+
+    redis_client = _CACHED_REDIS
 
     def _ping_redis() -> None:
         redis_client.ping()
 
-    _retry_with_backoff(_ping_redis, attempts=5, base_delay_s=1.0)
+    try:
+        _retry_with_backoff(_ping_redis, attempts=3, base_delay_s=1.0)
+    except Exception:
+        logger.warning("Cached Redis ping failed; recreating client", exc_info=True)
+        try:
+            _CACHED_REDIS.close()
+        except Exception:
+            pass
+        settings = Settings()
+        _CACHED_REDIS = Redis.from_url(
+            settings.redis.url,
+            decode_responses=True,
+            socket_connect_timeout=settings.redis.socket_connect_timeout_seconds,
+            socket_timeout=settings.redis.socket_timeout_seconds,
+            health_check_interval=settings.redis.health_check_interval_seconds,
+            retry_on_timeout=True,
+        )
+        redis_client = _CACHED_REDIS
+        _retry_with_backoff(_ping_redis, attempts=5, base_delay_s=1.0)
     return redis_client
 
 
@@ -179,7 +228,10 @@ def _release_lock_if_owner(redis_client: Redis, key: str, owner_token: str) -> N
 
 def _has_running_sync_listing_tasks(celery_app, *, exclude_task_id: str | None = None) -> bool:
     try:
-        inspector = celery_app.control.inspect(timeout=1.0)
+        # 10s — под нагрузкой workers могут отвечать с задержкой; 1s давал
+        # ложноположительные "пусто" → удаление валидного lock и параллельный
+        # запуск sync_listing с дублированием запросов к IAAI.
+        inspector = celery_app.control.inspect(timeout=10.0)
         snapshots = [
             inspector.active() or {},
             inspector.reserved() or {},
@@ -208,6 +260,7 @@ def _clear_orphan_sync_listing_lock(redis_client: Redis, celery_app, *, current_
         owner_token = redis_client.get(SYNC_LISTING_LOCK_KEY)
         if not owner_token:
             return False
+        ttl = redis_client.ttl(SYNC_LISTING_LOCK_KEY)
     except Exception:
         logger.warning("Failed to read sync listing lock before cleanup", exc_info=True)
         return False
@@ -216,8 +269,21 @@ def _clear_orphan_sync_listing_lock(redis_client: Redis, celery_app, *, current_
         logger.info("sync_listing lock preserved: active task still detected")
         return False
 
+    # Защита от race: если активный owner_token изменился между inspect и delete —
+    # значит другой worker уже acquire'нул lock; удалять его нельзя.
+    try:
+        owner_token_after = redis_client.get(SYNC_LISTING_LOCK_KEY)
+        if owner_token_after != owner_token:
+            logger.info(
+                "sync_listing lock owner changed during cleanup (%s -> %s); preserving",
+                owner_token, owner_token_after,
+            )
+            return False
+    except Exception:
+        logger.warning("Failed to re-check sync listing lock owner before cleanup", exc_info=True)
+        return False
+
     try:
-        ttl = redis_client.ttl(SYNC_LISTING_LOCK_KEY)
         redis_client.delete(SYNC_LISTING_LOCK_KEY)
         logger.warning(
             "Removed orphan sync_listing lock owner=%s ttl=%s after worker restart",
@@ -344,20 +410,29 @@ def _start_lock_heartbeat(
     key: str,
     owner_token: str,
     ttl_seconds: int,
-) -> tuple[Event, Thread]:
+) -> tuple[Event, Event, Thread]:
+    # Возвращает (stop_event, lock_lost_event, thread).
+    # lock_lost_event ставится в True, когда heartbeat обнаружил, что lock
+    # принадлежит другому owner'у (orphan-cleanup сработал ошибочно).
+    # Главная задача должна периодически проверять этот event.
     stop_event = Event()
+    lock_lost_event = Event()
     interval_seconds = max(5.0, min(30.0, ttl_seconds / 3))
 
     def _heartbeat() -> None:
         while not stop_event.wait(interval_seconds):
             refreshed = _refresh_lock_if_owner(redis_client, key, owner_token, ttl_seconds)
             if refreshed is False:
-                logger.warning("Lost sync_listing lock ownership for %s", owner_token)
+                logger.error(
+                    "Lost sync_listing lock ownership for %s — signaling abort",
+                    owner_token,
+                )
+                lock_lost_event.set()
                 return
 
     thread = Thread(target=_heartbeat, name="sync-listing-lock-heartbeat", daemon=True)
     thread.start()
-    return stop_event, thread
+    return stop_event, lock_lost_event, thread
 
 
 @shared_task(
@@ -416,29 +491,40 @@ def sync_listing_task(
     lock_acquired = False
     lock_ttl = _sync_listing_lock_ttl_seconds()
     heartbeat_stop: Event | None = None
+    heartbeat_lock_lost: Event | None = None
     heartbeat_thread: Thread | None = None
     force_bootstrap_full_scan = False
 
     def _enqueue_bootstrap_followup(
         reason: str,
-        delay_seconds: int = 5,
+        delay_seconds: int | None = None,
         *,
         count_as_failure: bool = False,
     ) -> None:
-        flag_ttl = max(lock_ttl, delay_seconds + 300)
+        # Всегда растим стрик (даже на soft timeout) — иначе в патологическом сценарии
+        # follow-up уходит каждые N секунд бесконечно. Сбрасывается только полным успехом.
+        streak, should_enqueue = _bump_bootstrap_failure_streak(redis_client, reason=reason)
+        if not should_enqueue:
+            logger.error(
+                "Bootstrap follow-up suppressed by circuit breaker (streak=%d, reason=%s); "
+                "next attempt will go through Celery beat schedule",
+                streak, reason,
+            )
+            return
+
+        # Экспоненциальный backoff: 30s, 60s, 120s, ... до 1 часа.
+        # Для count_as_failure=False (soft timeout с прогрессом) — короче.
+        base = 15 if not count_as_failure else 30
+        computed_delay = min(3600, base * (2 ** max(0, streak - 1)))
+        effective_delay = computed_delay if delay_seconds is None else max(int(delay_seconds), computed_delay)
+
+        flag_ttl = max(lock_ttl, effective_delay + 300)
         if not _try_set_followup_pending(redis_client, ttl_seconds=flag_ttl):
             logger.info(
                 "Bootstrap follow-up already pending; skip enqueue (reason=%s)",
                 reason,
             )
             return
-        if count_as_failure:
-            _, should_enqueue = _bump_bootstrap_failure_streak(redis_client, reason=reason)
-            if not should_enqueue:
-                _clear_followup_pending(redis_client)
-                return
-        else:
-            _clear_bootstrap_failure_streak(redis_client)
         try:
             self.app.send_task(
                 "iaai_scraper.worker.tasks.sync_listing_task",
@@ -450,12 +536,13 @@ def sync_listing_task(
                     "only_new": only_new,
                 },
                 queue="scraping",
-                countdown=max(0, int(delay_seconds)),
+                countdown=effective_delay,
             )
             logger.info(
-                "Bootstrap follow-up sync queued in %ss (reason=%s)",
-                delay_seconds,
+                "Bootstrap follow-up sync queued in %ss (reason=%s, streak=%d)",
+                effective_delay,
                 reason,
+                streak,
             )
         except Exception:
             _clear_followup_pending(redis_client)
@@ -505,7 +592,7 @@ def sync_listing_task(
 
         _clear_followup_pending(redis_client)
 
-        heartbeat_stop, heartbeat_thread = _start_lock_heartbeat(
+        heartbeat_stop, heartbeat_lock_lost, heartbeat_thread = _start_lock_heartbeat(
             redis_client,
             SYNC_LISTING_LOCK_KEY,
             owner_token,
@@ -537,6 +624,10 @@ def sync_listing_task(
 
         def _job():
             with IAAIScraper() as scraper:
+                # Прокидываем abort-сигнал: scraper будет проверять его
+                # между сегментами/страницами и поднимет ScraperAbortedError.
+                if heartbeat_lock_lost is not None:
+                    scraper.set_abort_callback(heartbeat_lock_lost.is_set)
                 if use_segmented:
                     return scraper.sync_listing_segmented(
                         segments=segments,
@@ -618,13 +709,27 @@ def sync_listing_task(
             "note": "partial progress saved to DB; bootstrap continuation queued",
         }
 
+    except ScraperAbortedError as exc:
+        # Lock потерян (orphan-cleanup в другом worker'е). НЕ освобождаем lock
+        # принудительно — он принадлежит другому owner'у. Не retry'имся, не
+        # планируем follow-up: новый владелец lock'а уже работает.
+        logger.warning("sync_listing_task aborted: %s", exc)
+        return {
+            "status": "aborted",
+            "task_id": task_id,
+            "reason": "lock_lost",
+        }
+
     except Exception as exc:
         logger.error("sync_listing_task failed: %s", exc, exc_info=True)
         # Retry только на не-таймаутные ошибки (сеть, БД, браузер).
         try:
             if force_bootstrap_full_scan:
                 _set_full_scan_done(redis_client, False)
-                raise self.retry(exc=exc, countdown=5)
+                # Не делаем мгновенный retry в bootstrap — иначе зацикливание при
+                # стабильно падающем сегменте. Backoff: 60s × 2^attempt.
+                attempt = int(getattr(self.request, "retries", 0) or 0)
+                raise self.retry(exc=exc, countdown=min(600, 60 * (2 ** attempt)))
             raise self.retry(exc=exc)
         except self.MaxRetriesExceededError:
             logger.error("sync_listing_task max retries exceeded, giving up")