diff --git a/iaai_scraper/worker/celery_app.py b/iaai_scraper/worker/celery_app.py index b1c4530..0f4294a 100644 --- a/iaai_scraper/worker/celery_app.py +++ b/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_time_limit=settings.celery.task_time_limit, + task_soft_time_limit=_soft, + task_time_limit=_hard, task_acks_late=True, task_reject_on_worker_lost=True, task_track_started=True, @@ -72,4 +89,16 @@ celery_app.conf.update( }, ) -celery_app.autodiscover_tasks(["iaai_scraper.worker"]) \ No newline at end of file +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", + ) \ No newline at end of file diff --git a/iaai_scraper/worker/tasks.py b/iaai_scraper/worker/tasks.py index 77fa051..22217e7 100644 --- a/iaai_scraper/worker/tasks.py +++ b/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: - future = executor.submit(func, *args, **kwargs) - return future.result() + # НЕ используем `with` — иначе shutdown(wait=True) заблокирует main thread + # если SoftTimeLimitExceeded прервёт future.result(), а browser thread ещё работает. + 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 снимался после сбоев. - return max(settings.celery.task_time_limit + 120, 300) + soft = settings.celery.task_soft_time_limit + 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", - summary["cars_upserted"], summary["cars_failed"], + "sync_listing_task completed: status=%s, %d upserted, %d failed, failures=%d", + 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()