stabilize worker

iaai_scraper/worker/celery_app.py

@@ -1,11 +1,15 @@
 # Инициализация Celery-приложения и периодических задач.
 
+import logging
+
 from celery import Celery
-from celery.signals import worker_process_init, setup_logging as celery_setup_logging
+from celery.signals import worker_process_init, worker_ready, setup_logging as celery_setup_logging
 
 from ..core.config import settings
 from ..core.logs import setup_logging
 
+logger = logging.getLogger("iaai_scraper.worker.celery_app")
+
 
 @celery_setup_logging.connect
 def _configure_logging(loglevel=None, **kwargs):
@@ -36,14 +40,27 @@ celery_app = Celery(
     backend=_result_backend(),
 )
 
+# Auto-clamp: если hard limit слишком далёк от soft (> soft + 120),
+# ограничиваем, чтобы зависший worker не жил вечно.
+_soft = settings.celery.task_soft_time_limit
+_hard = settings.celery.task_time_limit
+_max_hard = _soft + 120 if _soft else _hard
+if _hard > _max_hard:
+    logger.warning(
+        "CELERY_TASK_TIME_LIMIT=%d too far from CELERY_TASK_SOFT_TIME_LIMIT=%d; "
+        "clamping hard limit to %d",
+        _hard, _soft, _max_hard,
+    )
+    _hard = _max_hard
+
 celery_app.conf.update(
     task_serializer="json",
     accept_content=["json"],
     result_serializer="json",
     timezone="UTC",
     enable_utc=True,
-    task_soft_time_limit=settings.celery.task_soft_time_limit,
+    task_soft_time_limit=_soft,
-    task_time_limit=settings.celery.task_time_limit,
+    task_time_limit=_hard,
     task_acks_late=True,
     task_reject_on_worker_lost=True,
     task_track_started=True,
@@ -73,3 +90,15 @@ celery_app.conf.update(
 )
 
 celery_app.autodiscover_tasks(["iaai_scraper.worker"])
+
+
+@worker_ready.connect
+def _on_worker_ready(**kwargs):
+    """Сразу при старте worker отправляем первую задачу sync_listing,
+    чтобы не ждать час до первого beat-цикла."""
+    logger.info("Worker ready — dispatching initial sync_listing task")
+    celery_app.send_task(
+        "iaai_scraper.worker.tasks.sync_listing_task",
+        kwargs={"limit": settings.celery.beat_sync_limit},
+        queue="scraping",
+    )

iaai_scraper/worker/tasks.py

@@ -6,6 +6,7 @@ from threading import Event, Thread
 import time
 import uuid
 
+from billiard.exceptions import SoftTimeLimitExceeded
 from celery import shared_task
 from redis import Redis
 
@@ -42,15 +43,44 @@ def _retry_with_backoff(func, *, attempts: int = 5, base_delay_s: float = 1.0):
 
 def _run_browser_job(func, *args, **kwargs):
     # Браузерный код запускаем в отдельном потоке без активного loop.
-    with ThreadPoolExecutor(max_workers=1, thread_name_prefix="iaai-browser") as executor:
+    # НЕ используем `with` — иначе shutdown(wait=True) заблокирует main thread
-        future = executor.submit(func, *args, **kwargs)
+    # если SoftTimeLimitExceeded прервёт future.result(), а browser thread ещё работает.
-        return future.result()
+    settings = Settings()
+    soft = settings.celery.task_soft_time_limit
+    hard = settings.celery.task_time_limit
+    # Таймаут для future.result(): берём hard limit (или soft + 120), чтобы не висеть вечно.
+    wait_timeout = min(hard, soft + 120) if soft and hard else None
+
+    executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="iaai-browser")
+    future = executor.submit(func, *args, **kwargs)
+    try:
+        result = future.result(timeout=wait_timeout)
+    except SoftTimeLimitExceeded:
+        # Отпускаем executor без ожидания — thread умрёт когда Celery убьёт процесс (hard limit).
+        future.cancel()
+        executor.shutdown(wait=False, cancel_futures=True)
+        raise
+    except TimeoutError:
+        # future.result(timeout=...) вышел по таймауту — browser thread завис.
+        future.cancel()
+        executor.shutdown(wait=False, cancel_futures=True)
+        raise SoftTimeLimitExceeded("Browser thread did not finish within time limit")
+    except Exception:
+        executor.shutdown(wait=False, cancel_futures=True)
+        raise
+    else:
+        executor.shutdown(wait=True)
+    return result
 
 
 def _sync_listing_lock_ttl_seconds() -> int:
     settings = Settings()
-    # Небольшой запас к лимиту времени, чтобы lock снимался после сбоев.
+    soft = settings.celery.task_soft_time_limit
-    return max(settings.celery.task_time_limit + 120, 300)
+    hard = settings.celery.task_time_limit
+    # Используем clamped hard limit (soft + 120), а не сырой task_time_limit,
+    # чтобы lock не висел 11 дней при CELERY_TASK_TIME_LIMIT=999999.
+    effective_hard = min(hard, soft + 120) if soft else hard
+    return max(effective_hard + 120, 300)
 
 
 def _get_persistence() -> PersistenceService:
@@ -187,7 +217,7 @@ def sync_vehicle_task(self, vehicle_url: str, lane: str = "iaai"):
 @shared_task(
     name="iaai_scraper.worker.tasks.sync_listing_task",
     bind=True,
-    max_retries=1,
+    max_retries=3,
     default_retry_delay=120,
     acks_late=True,
 )
@@ -244,21 +274,48 @@ def sync_listing_task(
         summary = {
             "task_id": task_id,
             "run_id": result.get("run_id"),
+            "status": result.get("status", "success"),
             "cars_upserted": result.get("cars_upserted", 0),
             "cars_failed": result.get("cars_failed", 0),
             "images_upserted": result.get("images_upserted", 0),
             "skipped_existing": result.get("skipped_existing", 0),
             "elapsed_seconds": result.get("elapsed_seconds"),
+            "failures_count": len(result.get("failures") or []),
         }
         logger.info(
-            "sync_listing_task completed: %d upserted, %d failed",
+            "sync_listing_task completed: status=%s, %d upserted, %d failed, failures=%d",
-            summary["cars_upserted"], summary["cars_failed"],
+            summary["status"],
+            summary["cars_upserted"],
+            summary["cars_failed"],
+            summary["failures_count"],
         )
-        return {"status": "success", **summary}
+        return summary
+
+    except SoftTimeLimitExceeded:
+        logger.warning(
+            "sync_listing_task soft timeout exceeded — partial progress already saved to DB"
+        )
+        # Partial progress уже записан в БД через finish_sync_run.
+        # Не retry — следующий beat подхватит новые URL автоматически.
+        return {
+            "status": "timed_out",
+            "task_id": task_id,
+            "reason": "soft_time_limit_exceeded",
+            "note": "partial progress saved to DB; next beat cycle will continue",
+        }
 
     except Exception as exc:
         logger.error("sync_listing_task failed: %s", exc, exc_info=True)
-        raise self.retry(exc=exc)
+        # Retry только на не-таймаутные ошибки (сеть, БД, браузер).
+        try:
+            raise self.retry(exc=exc)
+        except self.MaxRetriesExceededError:
+            logger.error("sync_listing_task max retries exceeded, giving up")
+            return {
+                "status": "failed",
+                "task_id": task_id,
+                "error": str(exc),
+            }
     finally:
         if heartbeat_stop is not None:
             heartbeat_stop.set()