add Redis lock for sync_listing

iaai_scraper/worker/tasks.py

@@ -1,9 +1,11 @@
 # Celery-задачи для синхронизации автомобилей и листинга IAAI.
 
+from concurrent.futures import ThreadPoolExecutor
 import json
 import logging
 
 from celery import shared_task
+from redis import Redis
 
 from ..core.config import Settings
 from ..scraper import IAAIScraper
@@ -11,11 +13,64 @@ from ..storage.db import PersistenceService
 
 logger = logging.getLogger("iaai_scraper.worker.tasks")
 
+SYNC_LISTING_LOCK_KEY = "iaai:locks:sync_listing"
+
+
+def _run_browser_job(func, *args, **kwargs):
+    # Playwright Sync API нельзя запускать в потоке с активным asyncio loop.
+    # Celery/зависимости могут поднимать loop в worker-процессе, поэтому
+    # браузерный код выполняем в отдельном thread без loop.
+    with ThreadPoolExecutor(max_workers=1, thread_name_prefix="iaai-browser") as executor:
+        future = executor.submit(func, *args, **kwargs)
+        return future.result()
+
+
+def _sync_listing_lock_ttl_seconds() -> int:
+    settings = Settings()
+    # Небольшой запас к hard time limit задачи, чтобы lock самоснимался после сбоев.
+    return max(settings.celery.task_time_limit + 120, 300)
+
 
 def _get_persistence() -> PersistenceService:
     return PersistenceService(Settings())
 
 
+def _get_redis() -> Redis:
+    settings = Settings()
+    return Redis.from_url(settings.redis.url, decode_responses=True)
+
+
+def _release_lock_if_owner(redis_client: Redis, key: str, owner: str) -> None:
+    try:
+        current_owner = redis_client.get(key)
+        if current_owner == owner:
+            redis_client.delete(key)
+    except Exception as exc:
+        logger.warning("Failed to release lock %s: %s", key, exc)
+
+
+def _has_other_active_sync_listing_task(task) -> bool:
+    try:
+        inspector = task.app.control.inspect(timeout=1.0)
+        active_map = inspector.active() or {}
+    except Exception as exc:
+        logger.warning("Failed to inspect active tasks: %s", exc)
+        return False
+
+    current_task_id = task.request.id
+    for worker_tasks in active_map.values():
+        for item in worker_tasks or []:
+            name = str(item.get("name") or "")
+            task_id = str(item.get("id") or "")
+            if (
+                name == "iaai_scraper.worker.tasks.sync_listing_task"
+                and task_id
+                and task_id != current_task_id
+            ):
+                return True
+    return False
+
+
 @shared_task(
     name="iaai_scraper.worker.tasks.sync_vehicle_task",
     bind=True,
@@ -29,8 +84,11 @@ def sync_vehicle_task(self, vehicle_url: str, lane: str = "iaai"):
     persistence.create_tables()
 
     try:
-        with IAAIScraper() as scraper:
-            result = scraper.sync_vehicle(vehicle_url, lane=lane)
+        def _job():
+            with IAAIScraper() as scraper:
+                return scraper.sync_vehicle(vehicle_url, lane=lane)
+
+        result = _run_browser_job(_job)
         logger.info("sync_vehicle_task completed: %s", vehicle_url)
         return {
             "status": "success",
@@ -62,18 +120,71 @@ def sync_listing_task(
     # Полный цикл: листинг + sync всех найденных машин.
     persistence = _get_persistence()
     persistence.create_tables()
+    task_id = self.request.id or "unknown"
+    redis_client = _get_redis()
+
+    lock_acquired = False
+    lock_ttl = _sync_listing_lock_ttl_seconds()
+    try:
+        lock_acquired = bool(
+            redis_client.set(
+                SYNC_LISTING_LOCK_KEY,
+                task_id,
+                nx=True,
+                ex=lock_ttl,
+            )
+        )
+    except Exception as exc:
+        logger.warning("Failed to acquire sync lock in Redis: %s", exc)
+
+    if not lock_acquired:
+        # Возможен stale lock после рестарта worker. Если активного sync_listing нет —
+        # снимаем lock и пытаемся взять его заново.
+        if not _has_other_active_sync_listing_task(self):
+            try:
+                stale_owner = redis_client.get(SYNC_LISTING_LOCK_KEY)
+                if stale_owner:
+                    logger.warning(
+                        "Removing stale sync lock held by task %s",
+                        stale_owner,
+                    )
+                redis_client.delete(SYNC_LISTING_LOCK_KEY)
+                lock_acquired = bool(
+                    redis_client.set(
+                        SYNC_LISTING_LOCK_KEY,
+                        task_id,
+                        nx=True,
+                        ex=lock_ttl,
+                    )
+                )
+            except Exception as exc:
+                logger.warning("Failed to recover stale sync lock: %s", exc)
+
+    if not lock_acquired:
+        logger.info("sync_listing_task skipped: another sync is already running")
+        return {
+            "status": "skipped",
+            "reason": "sync_already_running",
+            "task_id": task_id,
+        }
 
     try:
-        with IAAIScraper() as scraper:
-            result = scraper.sync_listing(
-                make=make,
-                model=model,
-                lane=lane,
-                limit=limit,
-                only_new=only_new,
-            )
+        self.update_state(state="STARTED", meta={"stage": "sync_listing_started", "task_id": task_id})
+        def _job():
+            with IAAIScraper() as scraper:
+                return scraper.sync_listing(
+                    make=make,
+                    model=model,
+                    lane=lane,
+                    limit=limit,
+                    only_new=only_new,
+                )
+
+        result = _run_browser_job(_job)
 
         summary = {
+            "task_id": task_id,
+            "run_id": result.get("run_id"),
             "cars_upserted": result.get("cars_upserted", 0),
             "cars_failed": result.get("cars_failed", 0),
             "images_upserted": result.get("images_upserted", 0),
@@ -89,3 +200,6 @@ def sync_listing_task(
     except Exception as exc:
         logger.error("sync_listing_task failed: %s", exc)
         raise self.retry(exc=exc)
+    finally:
+        if lock_acquired:
+            _release_lock_if_owner(redis_client, SYNC_LISTING_LOCK_KEY, task_id)