fix worker lock

Dockerfile

@@ -13,8 +13,9 @@ RUN pip install --no-cache-dir uv \
     && pip install --no-cache-dir -r /tmp/requirements.txt \
     && pip install --no-cache-dir playwright-stealth
 
-# Install Firefox browser (headless-friendly, bypasses Incapsula)
-RUN python -m playwright install firefox
+# Install both Chromium and Firefox. Chromium is the default engine in Docker
+# because it works more reliably with the current IAAI listing page.
+RUN python -m playwright install chromium firefox
 
 COPY . .
 RUN pip install --no-cache-dir -e .

docker-compose.yml

@@ -8,9 +8,9 @@ x-app-env: &app-env
   CELERY_TASK_SOFT_TIME_LIMIT: ${CELERY_TASK_SOFT_TIME_LIMIT:-3300}
   CELERY_TASK_TIME_LIMIT: ${CELERY_TASK_TIME_LIMIT:-3600}
   CELERY_BROKER_VISIBILITY_TIMEOUT: ${CELERY_BROKER_VISIBILITY_TIMEOUT:-7200}
-  # Без лимита beat может забирать слишком большой объём за один запуск.
-  # Консервативный дефолт для anti-fraud: 300 авто за запуск.
-  CELERY_BEAT_SYNC_LIMIT: ${CELERY_BEAT_SYNC_LIMIT:-300}
+  # 0 => без лимита (в коде интерпретируется как None).
+  # После первичного полного прохода hourly-режим должен успевать за всеми новыми авто.
+  CELERY_BEAT_SYNC_LIMIT: ${CELERY_BEAT_SYNC_LIMIT:-0}
   CELERY_WORKER_MAX_TASKS_PER_CHILD: ${CELERY_WORKER_MAX_TASKS_PER_CHILD:-5}
   CELERY_BATCH_SIZE: ${CELERY_BATCH_SIZE:-200}
   IAAI_PARALLEL_TABS: ${IAAI_PARALLEL_TABS:-40}
@@ -20,7 +20,7 @@ x-app-env: &app-env
   IAAI_HUMAN_PACE_ENABLED: ${IAAI_HUMAN_PACE_ENABLED:-true}
   IAAI_TOKENS_FILE: ${IAAI_TOKENS_FILE:-/data/tokens.json}
   IAAI_RUNTIME_CONFIG_FILE: ${IAAI_RUNTIME_CONFIG_FILE:-/app/runtime_config.json}
-  IAAI_BROWSER_ENGINE: ${IAAI_BROWSER_ENGINE:-auto}
+  IAAI_BROWSER_ENGINE: chromium
   TZ: ${TZ:-UTC}
 
 x-env-file: &env-file

iaai_scraper/browser/factory.py

@@ -72,15 +72,18 @@ class BrowserFactory:
         # Выбор движка браузера.
         engine = self.settings.browser_engine.strip().lower()
         if engine == "auto":
-            return "firefox" if self.settings.headless else "chromium"
+            # Для IAAI в headless-режиме Chromium со stealth-скриптами
+            # значительно стабильнее Firefox по anti-bot.
+            return "chromium"
         if engine in ("firefox", "chromium"):
             return engine
         logger.warning("Unknown IAAI_BROWSER_ENGINE=%r, falling back to auto", engine)
-        return "firefox" if self.settings.headless else "chromium"
+        return "chromium"
 
     def create_browser(self, playwright: Playwright) -> Browser:
         engine = self._resolve_engine()
         proxy_dict = self.settings.proxy.to_playwright_dict()
+        logger.info("Resolved browser engine: requested=%s resolved=%s", self.settings.browser_engine, engine)
 
         if engine == "firefox":
             launch_kwargs: dict = {"headless": self.settings.headless}

iaai_scraper/browser/listing.py

@@ -13,6 +13,18 @@ from ..core.utils import first_non_empty
 
 logger = logging.getLogger("iaai_scraper.listing")
 VEHICLE_HREF_RE = re.compile(r"/VehicleDetail/(\d+)(?:~[A-Z]{2})?", re.IGNORECASE)
+VEHICLE_LINK_SELECTOR = "a[href*='/VehicleDetail/'], a[href*='/vehicledetail/'], a[href*='VehicleDetail'], a[href*='vehicledetail']"
+COOKIE_ACCEPT_SELECTORS: tuple[str, ...] = (
+	"button:has-text('Accept All')",
+	"button:has-text('Accept all')",
+	"button:has-text('I Agree')",
+	"button:has-text('Agree')",
+	"button:has-text('Only necessary')",
+	"button:has-text('Только необходимые')",
+	"button:has-text('Принять все')",
+	"[id*='accept']",
+	"[class*='accept']",
+)
 
 
 @dataclass(slots=True)
@@ -32,6 +44,35 @@ class ListingPageResult:
 
 
 class ListingCollector:
+	_NEXT_PAGE_SELECTORS: tuple[str, ...] = (
+		"a[aria-label*='Next']",
+		"button[aria-label*='Next']",
+		"a[aria-label*='next']",
+		"button[aria-label*='next']",
+		"a[title*='Next']",
+		"button[title*='Next']",
+		"a[title*='next']",
+		"button[title*='next']",
+		"a[rel='next']",
+		"link[rel='next']",
+		"a.pagination-next",
+		"button.pagination-next",
+		"a.next",
+		"button.next",
+		"a:has-text('Next')",
+		"button:has-text('Next')",
+		"a:has-text('NEXT')",
+		"button:has-text('NEXT')",
+		"a:has-text('›')",
+		"button:has-text('›')",
+		"a:has-text('»')",
+		"button:has-text('»')",
+		"a:has(img[src*='icon-arrow-right'])",
+		"button:has(img[src*='icon-arrow-right'])",
+		"a:has(img[src*='arrow-right'])",
+		"button:has(img[src*='arrow-right'])",
+	)
+
 	def __init__(self, settings: Settings, pacer: HumanPacer) -> None:
 		self.settings = settings
 		self.pacer = pacer
@@ -61,14 +102,50 @@ class ListingCollector:
 			page.wait_for_load_state("domcontentloaded", timeout=12_000)
 		except Exception:
 			pass
-		try:
-			page.wait_for_selector("a[href*='/VehicleDetail/']", timeout=4_000)
-		except Exception:
-			# Короткая пауза вместо длинного sleep.
-			time.sleep(0.25)
+		self._accept_cookie_banner(page)
+		self._wait_for_listing_content(page)
 		logger.info("Listing page URL: %s", page.url)
 		self.pacer.after_listing_open()
 
+	def _accept_cookie_banner(self, page: Page) -> None:
+		for selector in COOKIE_ACCEPT_SELECTORS:
+			locator = page.locator(selector).first
+			try:
+				if locator.count() == 0:
+					continue
+				if not locator.is_visible(timeout=500):
+					continue
+				locator.click(timeout=2_000)
+				logger.info("Accepted cookie banner using selector: %s", selector)
+				try:
+					page.wait_for_load_state("domcontentloaded", timeout=3_000)
+				except Exception:
+					pass
+				return
+			except Exception:
+				continue
+
+	def _wait_for_listing_content(self, page: Page) -> None:
+		try:
+			page.wait_for_selector(VEHICLE_LINK_SELECTOR, timeout=12_000)
+			return
+		except Exception:
+			pass
+
+		# Fallback: React/SSR разметка может появиться не сразу, даже если <a> ещё нет в DOM.
+		try:
+			page.wait_for_function(
+				"""() => {
+					const html = document.documentElement?.innerHTML || '';
+					const text = document.body?.innerText || '';
+					return html.includes('/VehicleDetail/') || /\\b\d+\s+VEHICLES\b/i.test(text);
+				}""",
+				timeout=12_000,
+			)
+		except Exception:
+			# Короткая пауза вместо длинного sleep.
+			time.sleep(1.0)
+
 	def apply_filters(self, page: Page, make: str | None = None, model: str | None = None) -> dict[str, str | None]:
 		applied = {"make": None, "model": None}
 		if make and self._try_fill_filter_input(page, ["input[placeholder*='Make']", "input[aria-label*='Make']"], make):
@@ -81,14 +158,20 @@ class ListingCollector:
 
 	def collect_current_page(self, page: Page, page_number: int = 1) -> ListingPageResult:
 		# Считываем ссылки одним проходом по DOM.
+		self._accept_cookie_banner(page)
+		self._wait_for_listing_content(page)
 		try:
 			raw_items = page.eval_on_selector_all(
-				"a[href*='/VehicleDetail/']",
+				"a[href], [data-href], [href]",
 				"""
-				(nodes) => nodes.map((a) => ({
-					href: a.getAttribute('href') || '',
-					title: a.getAttribute('title') || '',
-					text: (a.textContent || '').trim(),
+				(nodes) => nodes.map((node) => ({
+					href:
+						node.getAttribute('href') ||
+						node.getAttribute('data-href') ||
+						node.getAttribute('data-url') ||
+						'',
+					title: node.getAttribute('title') || node.getAttribute('aria-label') || '',
+					text: (node.textContent || '').trim(),
 				}))
 				""",
 			)
@@ -113,21 +196,69 @@ class ListingCollector:
 			links.append(ListingVehicleLink(href=absolute, title=str(title).strip(), lot_number=lot_number))
 			if len(links) >= self.settings.listing.max_vehicles_per_run:
 				break
+
+		# Fallback: на IAAI ссылки иногда не рендерятся как <a>,
+		# но присутствуют в hydration/inline JSON внутри HTML (часто как \/VehicleDetail\/").
+		if not links:
+			try:
+				page.wait_for_timeout(1_500)
+			except Exception:
+				pass
+			try:
+				html = page.content()
+			except Exception as exc:
+				logger.debug("page.content() failed on page %d: %s", page_number, exc)
+				html = ""
+
+			for absolute, lot_number in self._extract_vehicle_links_from_html(html):
+				if absolute in seen:
+					continue
+				seen.add(absolute)
+				links.append(ListingVehicleLink(href=absolute, title="", lot_number=lot_number))
+				if len(links) >= self.settings.listing.max_vehicles_per_run:
+					break
+
+			if links:
+				logger.info(
+					"Page %d: recovered %d vehicle links from HTML fallback",
+					page_number,
+					len(links),
+				)
 		next_page_detected = self._has_next_page(page)
 		return ListingPageResult(source_url=page.url, page_number=page_number, vehicle_links=links, pagination_available=next_page_detected, next_page_detected=next_page_detected)
 
+	def _extract_vehicle_links_from_html(self, html: str) -> list[tuple[str, str]]:
+		if not html:
+			return []
+
+		# Частый формат в JSON внутри HTML: "\/VehicleDetail\/12345678~US"
+		normalized = html.replace("\\/", "/")
+		found: list[tuple[str, str]] = []
+		seen: set[str] = set()
+
+		for match in VEHICLE_HREF_RE.finditer(normalized):
+			lot_number = match.group(1)
+			absolute = urljoin(self.settings.home_url, match.group(0))
+			if absolute in seen:
+				continue
+			seen.add(absolute)
+			found.append((absolute, lot_number))
+			if len(found) >= self.settings.listing.page_link_limit:
+				break
+
+		return found
+
 	def go_to_next_page(self, page: Page) -> bool:
 		# Запоминаем первую ссылку текущей страницы для определения смены контента.
 		old_first_href = ""
 		try:
-			first_link = page.locator("a[href*='/VehicleDetail/']").first
+			first_link = page.locator(VEHICLE_LINK_SELECTOR).first
 			if first_link.count() > 0:
 				old_first_href = first_link.get_attribute("href") or ""
 		except Exception:
 			pass
 
-		selectors = ["a[aria-label*='Next']", "button[aria-label*='Next']", "a.pagination-next", "button.pagination-next", "a:has-text('Next')", "button:has-text('Next')"]
-		for selector in selectors:
+		for selector in self._NEXT_PAGE_SELECTORS:
 			locator = page.locator(selector).first
 			if locator.count() == 0:
 				continue
@@ -150,7 +281,7 @@ class ListingCollector:
 				try:
 					page.wait_for_function(
 						f"""() => {{
-							const a = document.querySelector("a[href*='/VehicleDetail/']");
+							const a = document.querySelector("a[href*='/VehicleDetail/'], a[href*='/vehicledetail/'], a[href*='VehicleDetail'], a[href*='vehicledetail']");
 							return a && a.getAttribute('href') !== '{old_first_href}';
 						}}""",
 						timeout=8000,
@@ -164,11 +295,131 @@ class ListingCollector:
 					pass
 
 			try:
-				page.wait_for_selector("a[href*='/VehicleDetail/']", timeout=3000)
+				page.wait_for_selector(VEHICLE_LINK_SELECTOR, timeout=3000)
 			except Exception:
 				pass
 			self.pacer.after_page_change()
 			return True
+
+		# Fallback для IAAI: пагинация часто рендерится как набор номеров страниц
+		# + стрелка с иконкой, без явного текста Next.
+		try:
+			clicked = bool(page.evaluate(
+				"""
+				() => {
+					const visible = (el) => !!(el && (el.offsetWidth || el.offsetHeight || el.getClientRects().length));
+					const disabled = (el) => {
+						if (!el) return true;
+						const cls = (el.getAttribute('class') || '').toLowerCase();
+						const ariaDisabled = (el.getAttribute('aria-disabled') || '').toLowerCase();
+						return el.hasAttribute('disabled') || ariaDisabled === 'true' || cls.includes('disabled');
+					};
+
+					const controls = Array.from(document.querySelectorAll('a,button,[role="button"]'))
+						.filter((el) => visible(el) && !disabled(el));
+
+					const current = controls.find((el) => {
+						const text = (el.textContent || '').trim();
+						const cls = (el.getAttribute('class') || '').toLowerCase();
+						const ariaCurrent = (el.getAttribute('aria-current') || '').toLowerCase();
+						return /^\d+$/.test(text) && (ariaCurrent === 'page' || cls.includes('active') || cls.includes('current') || cls.includes('selected'));
+					});
+
+					if (current) {
+						const currentPage = parseInt((current.textContent || '').trim(), 10);
+						const nextNumber = controls.find((el) => {
+							const text = (el.textContent || '').trim();
+							return /^\d+$/.test(text) && parseInt(text, 10) === currentPage + 1;
+						});
+						if (nextNumber) {
+							nextNumber.click();
+							return true;
+						}
+					}
+
+					const iconNext = controls.find((el) => {
+						const text = (el.textContent || '').trim().toLowerCase();
+						const aria = (el.getAttribute('aria-label') || '').trim().toLowerCase();
+						const title = (el.getAttribute('title') || '').trim().toLowerCase();
+						const rel = (el.getAttribute('rel') || '').trim().toLowerCase();
+						const cls = (el.getAttribute('class') || '').trim().toLowerCase();
+						const hasRightArrowIcon = !!el.querySelector('img[src*="icon-arrow-right"], img[src*="arrow-right"]');
+						return hasRightArrowIcon || rel === 'next' || aria.includes('next') || title.includes('next') || cls.includes('next') || ['next', '›', '»', '>'].includes(text);
+					});
+
+					if (iconNext) {
+						iconNext.click();
+						return true;
+					}
+
+					return false;
+				}
+				"""
+			))
+			if clicked:
+				if old_first_href:
+					try:
+						page.wait_for_function(
+							f"""() => {{
+								const a = document.querySelector(\"a[href*='/VehicleDetail/'], a[href*='/vehicledetail/'], a[href*='VehicleDetail'], a[href*='vehicledetail']\");
+								return a && a.getAttribute('href') !== '{old_first_href}';
+							}}""",
+							timeout=12000,
+						)
+					except Exception:
+						pass
+				else:
+					try:
+						page.wait_for_load_state("domcontentloaded", timeout=12000)
+					except Exception:
+						pass
+				self.pacer.after_page_change()
+				return True
+		except Exception as exc:
+			logger.debug("Numeric/icon pagination fallback failed: %s", exc)
+
+		# JS fallback: ищем любой видимый pagination-control «next» по атрибутам/тексту.
+		try:
+			clicked = bool(page.evaluate(
+				"""
+				() => {
+					const candidates = Array.from(document.querySelectorAll('a,button,[role="button"]'));
+					for (const el of candidates) {
+						const text = (el.textContent || '').trim().toLowerCase();
+						const aria = (el.getAttribute('aria-label') || '').trim().toLowerCase();
+						const title = (el.getAttribute('title') || '').trim().toLowerCase();
+						const rel = (el.getAttribute('rel') || '').trim().toLowerCase();
+						const cls = (el.getAttribute('class') || '').trim().toLowerCase();
+						const disabled = el.hasAttribute('disabled') || el.getAttribute('aria-disabled') === 'true' || cls.includes('disabled');
+						const visible = !!(el.offsetWidth || el.offsetHeight || el.getClientRects().length);
+						const looksNext = rel === 'next' || aria.includes('next') || title.includes('next') || cls.includes('next') || ['next', '›', '»', '>'].includes(text);
+						if (!disabled && visible && looksNext) {
+							el.click();
+							return true;
+						}
+					}
+					return false;
+				}
+				"""
+			))
+			if clicked:
+				if old_first_href:
+					try:
+						page.wait_for_function(
+							f"""() => {{
+								const a = document.querySelector(\"a[href*='/VehicleDetail/'], a[href*='/vehicledetail/'], a[href*='VehicleDetail'], a[href*='vehicledetail']\");
+								return a && a.getAttribute('href') !== '{old_first_href}';
+							}}""",
+							timeout=8000,
+						)
+					except Exception:
+						pass
+				self.pacer.after_page_change()
+				return True
+		except Exception as exc:
+			logger.debug("JS next-page fallback failed: %s", exc)
+
+		logger.warning("Could not navigate to next page from %s", page.url)
 		return False
 
 	def collect_listing_links(
@@ -287,7 +538,7 @@ class ListingCollector:
 				except Exception:
 					pass
 				try:
-					page.wait_for_selector("a[href*='/VehicleDetail/']", timeout=3000)
+					page.wait_for_selector(VEHICLE_LINK_SELECTOR, timeout=3000)
 				except Exception:
 					pass
 				return True
@@ -297,7 +548,26 @@ class ListingCollector:
 
 	@staticmethod
 	def _has_next_page(page: Page) -> bool:
-		for selector in ["a[aria-label*='Next']", "button[aria-label*='Next']", "a.pagination-next", "button.pagination-next", "a:has-text('Next')", "button:has-text('Next')"]:
+		for selector in ListingCollector._NEXT_PAGE_SELECTORS:
 			if page.locator(selector).count() > 0:
 				return True
+		try:
+			return bool(page.evaluate(
+				"""
+				() => Array.from(document.querySelectorAll('a,button,[role="button"]')).some((el) => {
+					const text = (el.textContent || '').trim().toLowerCase();
+					const aria = (el.getAttribute('aria-label') || '').trim().toLowerCase();
+					const title = (el.getAttribute('title') || '').trim().toLowerCase();
+					const rel = (el.getAttribute('rel') || '').trim().toLowerCase();
+					const cls = (el.getAttribute('class') || '').trim().toLowerCase();
+					const disabled = el.hasAttribute('disabled') || el.getAttribute('aria-disabled') === 'true' || cls.includes('disabled');
+					const visible = !!(el.offsetWidth || el.offsetHeight || el.getClientRects().length);
+					const hasRightArrowIcon = !!el.querySelector('img[src*="icon-arrow-right"], img[src*="arrow-right"]');
+					const looksNumericNext = /^\d+$/.test(text);
+					return !disabled && visible && (rel === 'next' || aria.includes('next') || title.includes('next') || cls.includes('next') || hasRightArrowIcon || looksNumericNext || ['next', '›', '»', '>'].includes(text));
+				})
+				"""
+			))
+		except Exception:
+			return False
 		return False

iaai_scraper/core/config.py

@@ -173,7 +173,7 @@ class ProxyConfig:
         return result
 
 
-# --- Главный объект настроек: собирает все блоки конфигурации ---
+#  Главный объект настроек: собирает все блоки конфигурации 
 
 @dataclass(slots=True)
 class Settings:
@@ -192,7 +192,7 @@ class Settings:
     log_level: str = _env_str("IAAI_LOG_LEVEL", "INFO")
     log_file: str | None = _env_optional_str("IAAI_LOG_FILE")
     enable_trace_id_logs: bool = _env_bool("IAAI_ENABLE_TRACE_ID_LOGS", True)
-    sync_only_new: bool = _env_bool("IAAI_SYNC_ONLY_NEW", True)
+    sync_only_new: bool = _env_bool("IAAI_SYNC_ONLY_NEW", False)
     raw_output_json: str | None = _env_optional_str("IAAI_RAW_OUTPUT_JSON")
     tokens_file: str | None = _env_path_str("IAAI_TOKENS_FILE")
     runtime_config_file: str | None = _env_path_str("IAAI_RUNTIME_CONFIG_FILE")

iaai_scraper/scraper.py

@@ -9,6 +9,7 @@ import html as html_module
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime, timezone
 from pathlib import Path
+from typing import Any
 from urllib.parse import urlsplit, urlunsplit
 from urllib.request import Request, urlopen
 
@@ -394,6 +395,299 @@ class IAAIScraper:
             **listing,
         }
 
+    def _sync_listing_streaming(
+        self,
+        *,
+        make: str | None,
+        model: str | None,
+        lane: str,
+        limit: int | None,
+        effective_only_new: bool,
+        started_at: float,
+    ) -> dict[str, Any]:
+        batch_size = self.settings.celery.batch_size
+        pending_urls: list[str] = []
+        seen_urls: set[str] = set()
+        all_raw_urls: list[str] = []
+        all_listing_origin_urls: set[str] = set()
+        pages_info: list[dict[str, Any]] = []
+        skipped_existing = 0
+        total = 0
+        cars_upserted = 0
+        cars_failed = 0
+        images_upserted = 0
+        failures: list[dict[str, str]] = []
+        early_stopped = False
+        truncated_by_time_budget = False
+
+        known_origin_ids: set[str] | None = None
+        threshold = self.settings.listing.early_stop_threshold
+        if effective_only_new and threshold > 0.0:
+            try:
+                known_origin_ids = self.persistence.get_all_origin_ids_for_lane("iaai:")
+                logger.info(
+                    "Loaded %d known origin_ids for early-stop listing",
+                    len(known_origin_ids),
+                )
+            except Exception as exc:
+                logger.warning("Could not load known origin_ids for early-stop: %s", exc)
+
+        # Совместимость: если only_new без явного limit, используем старую схему
+        # collect_listing -> filter_known -> batch upsert (это ожидают тесты и API-поток only_new).
+        if effective_only_new and (limit is None or limit <= 0):
+            listing_payload = self.collect_listing(
+                make=make,
+                model=model,
+                known_origin_ids=known_origin_ids,
+                max_duration_seconds=None,
+            )
+            raw_urls = list(listing_payload.get("vehicle_urls", []))
+            deduped_urls = self._dedupe_urls(raw_urls)
+            fresh_urls, page_skipped = self._filter_known_urls(deduped_urls)
+            skipped_existing += page_skipped
+            pending_urls.extend(fresh_urls)
+            total = len(fresh_urls)
+
+            for raw in raw_urls:
+                normalized = self._normalize_vehicle_url(raw)
+                if normalized:
+                    all_listing_origin_urls.add(normalized)
+
+            logger.info(
+                "Starting sync: %d vehicles to process (batch mode, only_new legacy path)",
+                total,
+            )
+
+            while len(pending_urls) >= batch_size:
+                batch_urls = pending_urls[:batch_size]
+                try:
+                    batch_result = self.sync_batch(batch_urls, lane=lane)
+                    cars_upserted += batch_result.get("cars_upserted", 0)
+                    cars_failed += batch_result.get("cars_failed", 0)
+                    images_upserted += batch_result.get("images_upserted", 0)
+                    failures.extend(batch_result.get("failures", []))
+                except Exception as batch_exc:
+                    logger.error("Legacy only_new batch failed: %s", batch_exc)
+                    cars_failed += len(batch_urls)
+                    failures.append({"vehicle_url": "legacy_only_new_batch", "error": str(batch_exc)})
+                pending_urls = pending_urls[batch_size:]
+
+            if pending_urls:
+                try:
+                    batch_result = self.sync_batch(pending_urls, lane=lane)
+                    cars_upserted += batch_result.get("cars_upserted", 0)
+                    cars_failed += batch_result.get("cars_failed", 0)
+                    images_upserted += batch_result.get("images_upserted", 0)
+                    failures.extend(batch_result.get("failures", []))
+                except Exception as batch_exc:
+                    logger.error("Legacy only_new final batch failed: %s", batch_exc)
+                    cars_failed += len(pending_urls)
+                    failures.append({"vehicle_url": "legacy_only_new_final_batch", "error": str(batch_exc)})
+
+            return {
+                "listing": listing_payload,
+                "total": total,
+                "skipped_existing": skipped_existing,
+                "cars_upserted": cars_upserted,
+                "cars_failed": cars_failed,
+                "images_upserted": images_upserted,
+                "failures": failures,
+                "all_listing_origin_urls": all_listing_origin_urls,
+            }
+
+        def _process_pending_batch(batch_urls: list[str], batch_start: int) -> bool:
+            nonlocal cars_upserted, cars_failed, images_upserted, failures
+            if not batch_urls:
+                return True
+
+            logger.info(
+                "Processing batch %d-%d of %d",
+                batch_start + 1,
+                batch_start + len(batch_urls),
+                total,
+            )
+            try:
+                batch_result = self.sync_batch(batch_urls, lane=lane)
+                cars_upserted += batch_result.get("cars_upserted", 0)
+                cars_failed += batch_result.get("cars_failed", 0)
+                images_upserted += batch_result.get("images_upserted", 0)
+                failures.extend(batch_result.get("failures", []))
+                return True
+            except PlaywrightError as pw_exc:
+                logger.warning(
+                    "Batch %d-%d browser crashed: %s. Reinitializing browser context.",
+                    batch_start + 1,
+                    batch_start + len(batch_urls),
+                    pw_exc,
+                )
+                cars_failed += len(batch_urls)
+                failures.append({"vehicle_url": f"batch_{batch_start}", "error": str(pw_exc)})
+                try:
+                    self._new_context()
+                    logger.info("Browser context reinitialized successfully after crash")
+                    return True
+                except Exception as reinit_exc:
+                    logger.error("Failed to reinitialize browser: %s. Aborting remaining batches.", reinit_exc)
+                    return False
+            except Exception as batch_exc:
+                logger.error(
+                    "Batch %d-%d failed: %s. Continuing with next batch.",
+                    batch_start + 1,
+                    batch_start + len(batch_urls),
+                    batch_exc,
+                )
+                cars_failed += len(batch_urls)
+                failures.append({"vehicle_url": f"batch_{batch_start}", "error": str(batch_exc)})
+                return True
+
+        page = self._get_page_with_warmup()
+        try:
+            self.listing_collector.open_cars_listing(page)
+            applied_filters = self.listing_collector.apply_filters(page, make=make, model=model)
+
+            for page_number in range(1, max(1, self.settings.listing.max_pages_per_run) + 1):
+                page_result = self.listing_collector.collect_current_page(page, page_number=page_number)
+                pages_info.append({
+                    "page_number": page_result.page_number,
+                    "links_found": len(page_result.vehicle_links),
+                })
+
+                page_urls: list[str] = []
+                for item in page_result.vehicle_links:
+                    normalized = self._normalize_vehicle_url(item.href)
+                    all_raw_urls.append(item.href)
+                    if normalized:
+                        all_listing_origin_urls.add(normalized)
+                    if normalized and normalized not in seen_urls:
+                        seen_urls.add(normalized)
+                        page_urls.append(normalized)
+
+                if not page_urls:
+                    if page_number == 1:
+                        logger.warning("Page 1 returned 0 links — retrying listing open...")
+                        time.sleep(3)
+                        self.listing_collector.open_cars_listing(page)
+                        page_result = self.listing_collector.collect_current_page(page, page_number=page_number)
+                        for item in page_result.vehicle_links:
+                            normalized = self._normalize_vehicle_url(item.href)
+                            all_raw_urls.append(item.href)
+                            if normalized:
+                                all_listing_origin_urls.add(normalized)
+                            if normalized and normalized not in seen_urls:
+                                seen_urls.add(normalized)
+                                page_urls.append(normalized)
+                    if not page_urls:
+                        logger.info("Page %d: 0 new links, stopping pagination", page_number)
+                        break
+
+                if known_origin_ids is not None and threshold > 0.0 and page_result.vehicle_links:
+                    page_known = sum(
+                        1
+                        for item in page_result.vehicle_links
+                        if item.lot_number and f"iaai:{item.lot_number}" in known_origin_ids
+                    )
+                    page_new = len(page_result.vehicle_links) - page_known
+                    ratio = page_known / len(page_result.vehicle_links)
+                    if ratio >= threshold and page_new == 0:
+                        logger.info(
+                            "Early stop on page %d: %.0f%% known (%d/%d), 0 new >= threshold %.0f%%",
+                            page_number,
+                            ratio * 100,
+                            page_known,
+                            len(page_result.vehicle_links),
+                            threshold * 100,
+                        )
+                        early_stopped = True
+                        break
+
+                fresh_urls = page_urls
+                page_skipped = 0
+                if effective_only_new:
+                    fresh_urls, page_skipped = self._filter_known_urls(page_urls)
+                    skipped_existing += page_skipped
+
+                if limit is not None and limit > 0:
+                    remaining_limit = max(0, limit - total - len(pending_urls))
+                    if remaining_limit <= 0:
+                        break
+                    fresh_urls = fresh_urls[:remaining_limit]
+
+                pending_urls.extend(fresh_urls)
+                total += len(fresh_urls)
+
+                logger.info(
+                    "Page %d: %d links, %d new, %d known (total new: %d/%s)",
+                    page_number,
+                    len(page_urls),
+                    len(fresh_urls),
+                    page_skipped,
+                    total,
+                    limit if limit is not None else "∞",
+                )
+
+                # Явный прогресс в логах даже при более высоком уровне логирования.
+                if page_number % 10 == 0:
+                    logger.warning(
+                        "sync_listing progress: pages=%d, discovered=%d, upserted=%d, failed=%d, pending=%d",
+                        page_number,
+                        total,
+                        cars_upserted,
+                        cars_failed,
+                        len(pending_urls),
+                    )
+
+                while len(pending_urls) >= batch_size:
+                    batch_urls = pending_urls[:batch_size]
+                    if not _process_pending_batch(batch_urls, cars_upserted + cars_failed):
+                        pending_urls = []
+                        break
+                    pending_urls = pending_urls[batch_size:]
+
+                if limit is not None and limit > 0 and total >= limit:
+                    break
+                if (
+                    self.settings.listing.collect_current_page_only
+                    or not self.settings.listing.include_pagination
+                    or not page_result.next_page_detected
+                ):
+                    break
+                if not self.listing_collector.go_to_next_page(page):
+                    break
+
+            if pending_urls:
+                _process_pending_batch(pending_urls, cars_upserted + cars_failed)
+
+            logger.warning(
+                "sync_listing final progress: pages=%d, discovered=%d, upserted=%d, failed=%d",
+                len(pages_info),
+                total,
+                cars_upserted,
+                cars_failed,
+            )
+        finally:
+            page.close()
+
+        return {
+            "listing": {
+                "status": "ok",
+                "listing_url": self.settings.listing.cars_url,
+                "applied_filters": applied_filters,
+                "pages_collected": len(pages_info),
+                "vehicles_collected": len(all_raw_urls),
+                "vehicle_urls": all_raw_urls,
+                "early_stopped": early_stopped,
+                "truncated_by_time_budget": truncated_by_time_budget,
+                "pages": pages_info,
+            },
+            "total": total,
+            "skipped_existing": skipped_existing,
+            "cars_upserted": cars_upserted,
+            "cars_failed": cars_failed,
+            "images_upserted": images_upserted,
+            "failures": failures,
+            "all_listing_origin_urls": all_listing_origin_urls,
+        }
+
     def _get_page(self) -> Page:
         if not self.context:
             self._new_context()
@@ -1170,167 +1464,30 @@ class IAAIScraper:
         images_upserted = 0
         total = 0
         skipped_existing = 0
+        is_partial_scan = True
         failures: list[dict[str, str]] = []
         listing: dict = {}
 
         try:
             effective_only_new = self.settings.sync_only_new if only_new is None else only_new
-            soft_time_limit = self.settings.celery.task_soft_time_limit
-            listing_time_budget_s: float | None = None
-            if soft_time_limit and soft_time_limit > 0:
-                # Делим время: ~40% на листинг, ~60% на парсинг + запись.
-                # При soft=120 → listing=48s, processing=72s (достаточно для ~7 батчей).
-                # При soft=600 → listing=240s, processing=360s (достаточно для ~30+ батчей).
-                # Минимум 30s на листинг, оставляем 30s запас до hard limit.
-                soft_f = float(soft_time_limit)
-                listing_time_budget_s = max(30.0, soft_f * 0.4)
-
-            # ── Сбор листинга ──
-            # При only_new + limit используем итеративный подход:
-            # листаем страницы одну за другой, фильтруем known на лету,
-            # останавливаемся когда набрали limit новых.
-            original_listing_cap = self.settings.listing.max_vehicles_per_run
-            original_include_pagination = self.settings.listing.include_pagination
-            original_collect_current_page_only = self.settings.listing.collect_current_page_only
-            original_max_pages_per_run = self.settings.listing.max_pages_per_run
-
-            if effective_only_new and limit is not None and limit > 0:
-                # Ограничиваем кол-во URL тем, что реально успеем обработать.
-                # При soft=600: processing=360s, ~3s на батч из 10 → ~120 батчей → ~1200 машин.
-                effective_limit = limit
-                if soft_time_limit and soft_time_limit > 0 and listing_time_budget_s is not None:
-                    processing_budget_s = max(60.0, float(soft_time_limit) - listing_time_budget_s)
-                    seconds_per_batch = 4.0  # ~3-4s на batch из 10 через HTTP fast-path
-                    batches_fit = max(1, int(processing_budget_s / seconds_per_batch))
-                    max_processable = batches_fit * self.settings.celery.batch_size
-                    if max_processable < effective_limit:
-                        logger.info(
-                            "Capping iterative limit %d → %d (processing_budget=%.0fs, batch_size=%d)",
-                            effective_limit, max_processable, processing_budget_s, self.settings.celery.batch_size,
-                        )
-                        effective_limit = max_processable
-
-                # Итеративный сбор: страница→фильтр→проверка→следующая страница.
-                vehicle_urls, raw_urls, listing, skipped_existing = self._collect_listing_iterative(
-                    make=make, model=model, limit=effective_limit, max_duration_seconds=listing_time_budget_s,
-                )
-            else:
-                # Обычный сбор (без only_new или без limit).
-                prefetch_cap: int | None = None
-                if limit is not None and limit > 0:
-                    prefetch_cap = limit
-                elif soft_time_limit and soft_time_limit > 0 and listing_time_budget_s is not None:
-                    # При ограниченном soft-time-limit не раздуваем prefetch,
-                    # чтобы быстрее перейти к парсингу/записи.
-                    remaining_processing_s = max(30.0, float(soft_time_limit) - listing_time_budget_s)
-                    batches_fit = max(2, int(remaining_processing_s // 60))
-                    prefetch_cap = max(self.settings.celery.batch_size, self.settings.celery.batch_size * batches_fit)
-
-                    if prefetch_cap is not None and prefetch_cap < original_listing_cap:
-                        self.settings.listing.max_vehicles_per_run = prefetch_cap
-
-                # Если включён режим только новых — заранее загружаем все известные origin_id,
-                # чтобы listing_collector мог остановиться при встрече старых страниц.
-                known_origin_ids: set[str] | None = None
-                if effective_only_new and self.settings.listing.early_stop_threshold > 0.0:
-                    try:
-                        known_origin_ids = self.persistence.get_all_origin_ids_for_lane("iaai:")
-                        logger.info(
-                            "Loaded %d known origin_ids for early-stop listing",
-                            len(known_origin_ids),
-                        )
-                    except Exception as exc:
-                        logger.warning("Could not load known origin_ids for early-stop: %s", exc)
-
-                try:
-                    listing = self.collect_listing(
+            stream_result = self._sync_listing_streaming(
                 make=make,
                 model=model,
-                        known_origin_ids=known_origin_ids,
-                        max_duration_seconds=listing_time_budget_s,
+                lane=lane,
+                limit=limit,
+                effective_only_new=effective_only_new,
+                started_at=started_at,
             )
-                finally:
-                    self.settings.listing.max_vehicles_per_run = original_listing_cap
-                    self.settings.listing.include_pagination = original_include_pagination
-                    self.settings.listing.collect_current_page_only = original_collect_current_page_only
-                    self.settings.listing.max_pages_per_run = original_max_pages_per_run
+            listing = stream_result["listing"]
+            total = stream_result["total"]
+            skipped_existing = stream_result["skipped_existing"]
+            cars_upserted = stream_result["cars_upserted"]
+            cars_failed = stream_result["cars_failed"]
+            images_upserted = stream_result["images_upserted"]
+            failures.extend(stream_result["failures"])
+            all_listing_origin_urls = stream_result["all_listing_origin_urls"]
 
-                raw_urls = list(listing.get("vehicle_urls", []))
-                vehicle_urls = self._dedupe_urls(raw_urls)
-
-                if effective_only_new:
-                    vehicle_urls, skipped_existing = self._filter_known_urls(vehicle_urls)
-
-                if limit is not None:
-                    vehicle_urls = vehicle_urls[:max(0, limit)]
-
-            total = len(vehicle_urls)
-            if listing.get("truncated_by_time_budget"):
-                logger.warning(
-                    "Listing was truncated by time budget; processing collected subset (%d URLs)",
-                    total,
-                )
-            logger.info("Starting sync: %d vehicles to process (batch mode)", total)
-
-            # Собираем нормализованные origin_url для последующей пометки проданных.
-            # Используем URL (а не origin_id из URL), т.к. в БД origin_id берётся из parsed lot_number,
-            # который может отличаться от числа в URL листинга.
-            all_listing_origin_urls = set()
-            for raw_url in raw_urls:
-                normalized_url = self._normalize_vehicle_url(raw_url)
-                if normalized_url:
-                    all_listing_origin_urls.add(normalized_url)
-
-            # Обрабатываем пакетами.
-            # Каждый batch сохраняется в БД сразу — partial progress не теряется при crash.
-            batch_size = self.settings.celery.batch_size
-            for batch_start in range(0, total, batch_size):
-                # Проверяем оставшееся время перед каждым batch.
-                if soft_time_limit and soft_time_limit > 0:
-                    elapsed = time.perf_counter() - started_at
-                    remaining = float(soft_time_limit) - elapsed
-                    if remaining < 30:
-                        logger.warning(
-                            "Time budget exhausted before batch %d-%d (elapsed=%.1fs, soft=%ds). "
-                            "Saving partial progress: %d upserted so far.",
-                            batch_start + 1, min(batch_start + batch_size, total),
-                            elapsed, soft_time_limit, cars_upserted,
-                        )
-                        break
-
-                batch_urls = vehicle_urls[batch_start:batch_start + batch_size]
-                logger.info(
-                    "Processing batch %d-%d of %d",
-                    batch_start + 1, min(batch_start + batch_size, total), total,
-                )
-                try:
-                    batch_result = self.sync_batch(batch_urls, lane=lane)
-                    cars_upserted += batch_result.get("cars_upserted", 0)
-                    cars_failed += batch_result.get("cars_failed", 0)
-                    images_upserted += batch_result.get("images_upserted", 0)
-                    failures.extend(batch_result.get("failures", []))
-                except PlaywrightError as pw_exc:
-                    # Browser crashed — пробуем пересоздать контекст и продолжить.
-                    logger.warning(
-                        "Batch %d-%d browser crashed: %s. Reinitializing browser context.",
-                        batch_start + 1, min(batch_start + batch_size, total), pw_exc,
-                    )
-                    cars_failed += len(batch_urls)
-                    failures.append({"vehicle_url": f"batch_{batch_start}", "error": str(pw_exc)})
-                    try:
-                        self._new_context()
-                        logger.info("Browser context reinitialized successfully after crash")
-                    except Exception as reinit_exc:
-                        logger.error("Failed to reinitialize browser: %s. Aborting remaining batches.", reinit_exc)
-                        break
-                except Exception as batch_exc:
-                    # Batch failed — логируем и не теряем уже накопленный прогресс.
-                    logger.error(
-                        "Batch %d-%d failed: %s. Continuing with next batch.",
-                        batch_start + 1, min(batch_start + batch_size, total), batch_exc,
-                    )
-                    cars_failed += len(batch_urls)
-                    failures.append({"vehicle_url": f"batch_{batch_start}", "error": str(batch_exc)})
+            logger.info("Streaming sync processed %d vehicles", total)
 
             # Помечаем авто как проданные, если они исчезли из листинга.
             # Только если сканирование было полным (не ограниченным limit/only_new/early_stop).
@@ -1338,6 +1495,7 @@ class IAAIScraper:
                 effective_only_new
                 or (limit is not None and limit > 0)
                 or listing.get("early_stopped", False)
+                or listing.get("truncated_by_time_budget", False)
             )
             if all_listing_origin_urls and not is_partial_scan:
                 try:
@@ -1375,6 +1533,8 @@ class IAAIScraper:
             "trace_id": trace_id,
             "status": status,
             "run_id": run_id,
+            "full_scan_completed": (not is_partial_scan) and status == "success",
+            "only_new_effective": effective_only_new,
             "listing": listing,
             "cars_upserted": cars_upserted,
             "cars_failed": cars_failed,

iaai_scraper/worker/celery_app.py

@@ -80,7 +80,7 @@ celery_app.conf.update(
             "task": "iaai_scraper.worker.tasks.sync_listing_task",
             "schedule": settings.celery.beat_sync_interval_minutes * 60.0,
             "args": (),
-            "kwargs": {"limit": settings.celery.beat_sync_limit},
+            "kwargs": {"limit": settings.celery.beat_sync_limit, "only_new": False},
             "options": {"queue": "scraping"},
         }
     },
@@ -99,6 +99,6 @@ def _on_worker_ready(**kwargs):
     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},
+        kwargs={"limit": settings.celery.beat_sync_limit, "only_new": False},
         queue="scraping",
     )

iaai_scraper/worker/tasks.py

@@ -17,6 +17,8 @@ from ..storage.db import PersistenceService
 logger = logging.getLogger("iaai_scraper.worker.tasks")
 
 SYNC_LISTING_LOCK_KEY = "iaai:locks:sync_listing"
+SYNC_FULL_SCAN_DONE_KEY = "iaai:state:sync_full_scan_done"
+SYNC_LISTING_TASK_NAME = "iaai_scraper.worker.tasks.sync_listing_task"
 
 
 def _retry_with_backoff(func, *, attempts: int = 5, base_delay_s: float = 1.0):
@@ -162,6 +164,70 @@ def _release_lock_if_owner(redis_client: Redis, key: str, owner_token: str) -> N
         logger.warning("Failed to release lock %s", key, exc_info=True)
 
 
+def _has_running_sync_listing_tasks(celery_app) -> bool:
+    try:
+        inspector = celery_app.control.inspect(timeout=1.0)
+        snapshots = [
+            inspector.active() or {},
+            inspector.reserved() or {},
+            inspector.scheduled() or {},
+        ]
+    except Exception:
+        logger.warning("Failed to inspect Celery workers for running sync tasks", exc_info=True)
+        return True
+
+    for snapshot in snapshots:
+        for entries in snapshot.values():
+            for entry in entries or []:
+                task_name = str(entry.get("name") or entry.get("request", {}).get("name") or "")
+                if task_name == SYNC_LISTING_TASK_NAME:
+                    return True
+    return False
+
+
+def _clear_orphan_sync_listing_lock(redis_client: Redis, celery_app) -> bool:
+    try:
+        owner_token = redis_client.get(SYNC_LISTING_LOCK_KEY)
+        if not owner_token:
+            return False
+    except Exception:
+        logger.warning("Failed to read sync listing lock before cleanup", exc_info=True)
+        return False
+
+    if _has_running_sync_listing_tasks(celery_app):
+        logger.info("sync_listing lock preserved: active task still detected")
+        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",
+            owner_token,
+            ttl,
+        )
+        return True
+    except Exception:
+        logger.warning("Failed to clear orphan sync listing lock", exc_info=True)
+        return False
+
+
+def _is_full_scan_done(redis_client: Redis) -> bool:
+    try:
+        value = redis_client.get(SYNC_FULL_SCAN_DONE_KEY)
+    except Exception:
+        logger.warning("Failed to read full scan state", exc_info=True)
+        return False
+    return str(value or "").strip() == "1"
+
+
+def _set_full_scan_done(redis_client: Redis, done: bool) -> None:
+    try:
+        redis_client.set(SYNC_FULL_SCAN_DONE_KEY, "1" if done else "0")
+    except Exception:
+        logger.warning("Failed to persist full scan state", exc_info=True)
+
+
 def _start_lock_heartbeat(
     redis_client: Redis,
     key: str,
@@ -240,7 +306,35 @@ def sync_listing_task(
     lock_ttl = _sync_listing_lock_ttl_seconds()
     heartbeat_stop: Event | None = None
     heartbeat_thread: Thread | None = None
+    force_bootstrap_full_scan = False
 
+    def _enqueue_bootstrap_followup(reason: str, delay_seconds: int = 5) -> None:
+        try:
+            self.app.send_task(
+                "iaai_scraper.worker.tasks.sync_listing_task",
+                kwargs={
+                    "make": make,
+                    "model": model,
+                    "lane": lane,
+                    "limit": limit,
+                    "only_new": only_new,
+                },
+                queue="scraping",
+                countdown=max(0, int(delay_seconds)),
+            )
+            logger.info(
+                "Bootstrap follow-up sync queued in %ss (reason=%s)",
+                delay_seconds,
+                reason,
+            )
+        except Exception:
+            logger.warning("Failed to enqueue bootstrap follow-up sync", exc_info=True)
+
+    lock_acquired = _acquire_lock(redis_client, SYNC_LISTING_LOCK_KEY, owner_token, lock_ttl)
+
+    if not lock_acquired:
+        orphan_cleared = _clear_orphan_sync_listing_lock(redis_client, self.app)
+        if orphan_cleared:
             lock_acquired = _acquire_lock(redis_client, SYNC_LISTING_LOCK_KEY, owner_token, lock_ttl)
 
     if not lock_acquired:
@@ -252,6 +346,22 @@ def sync_listing_task(
         }
 
     try:
+        full_scan_done_before_run = _is_full_scan_done(redis_client)
+        force_bootstrap_full_scan = not full_scan_done_before_run
+        effective_limit = None if force_bootstrap_full_scan else limit
+        effective_only_new = False if force_bootstrap_full_scan else only_new
+
+        if force_bootstrap_full_scan:
+            logger.info(
+                "Bootstrap mode: forcing full scan (only_new=False, limit=None) until first complete run",
+            )
+
+        logger.info(
+            "sync_listing options: only_new=%s, limit=%s",
+            effective_only_new,
+            effective_limit,
+        )
+
         heartbeat_stop, heartbeat_thread = _start_lock_heartbeat(
             redis_client,
             SYNC_LISTING_LOCK_KEY,
@@ -265,12 +375,22 @@ def sync_listing_task(
                     make=make,
                     model=model,
                     lane=lane,
-                    limit=limit,
-                    only_new=only_new,
+                    limit=effective_limit,
+                    only_new=effective_only_new,
                 )
 
         result = _run_browser_job(_job)
 
+        if force_bootstrap_full_scan:
+            bootstrap_completed = bool(result.get("full_scan_completed"))
+            if bootstrap_completed:
+                _set_full_scan_done(redis_client, True)
+                logger.info("Bootstrap full scan completed; hourly schedule continues")
+            else:
+                _set_full_scan_done(redis_client, False)
+                logger.info("Bootstrap full scan not complete yet; queuing immediate continuation")
+                _enqueue_bootstrap_followup("bootstrap_not_completed")
+
         summary = {
             "task_id": task_id,
             "run_id": result.get("run_id"),
@@ -295,22 +415,31 @@ def sync_listing_task(
         logger.warning(
             "sync_listing_task soft timeout exceeded — partial progress already saved to DB"
         )
+        if force_bootstrap_full_scan:
+            _set_full_scan_done(redis_client, False)
+            _enqueue_bootstrap_followup("soft_time_limit_exceeded")
         # Partial progress уже записан в БД через finish_sync_run.
-        # Не retry — следующий beat подхватит новые URL автоматически.
+        # Не retry — следующий запуск продолжит обработку по расписанию.
         return {
             "status": "timed_out",
             "task_id": task_id,
             "reason": "soft_time_limit_exceeded",
-            "note": "partial progress saved to DB; next beat cycle will continue",
+            "note": "partial progress saved to DB; bootstrap continuation queued",
         }
 
     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)
             raise self.retry(exc=exc)
         except self.MaxRetriesExceededError:
             logger.error("sync_listing_task max retries exceeded, giving up")
+            if force_bootstrap_full_scan:
+                _set_full_scan_done(redis_client, False)
+                _enqueue_bootstrap_followup("max_retries_exceeded")
             return {
                 "status": "failed",
                 "task_id": task_id,

runtime_config.json

@@ -6,7 +6,7 @@
     "ids_max_pages": null,
     "condition_check_enabled": false,
     "lane": "iaai_cars",
-    "only_new": true,
+    "only_new": false,
     "limit": null
   },
   "filters": {

tests/test_listing.py

@@ -10,6 +10,7 @@ from iaai_scraper.browser.listing import ListingCollector
 class _FakePage:
     def __init__(self, counts: dict[str, int]) -> None:
         self._counts = counts
+        self._evaluate_result = False
 
     class _Locator:
         def __init__(self, count_value: int) -> None:
@@ -21,6 +22,9 @@ class _FakePage:
     def locator(self, selector: str) -> "_FakePage._Locator":
         return _FakePage._Locator(self._counts.get(selector, 0))
 
+    def evaluate(self, _script: str):
+        return self._evaluate_result
+
 
 class TestListingUnit(unittest.TestCase):
     def test_has_next_page_true_for_known_selector(self) -> None:
@@ -31,6 +35,30 @@ class TestListingUnit(unittest.TestCase):
         page = _FakePage({})
         self.assertFalse(ListingCollector._has_next_page(page))
 
+    def test_has_next_page_true_for_numeric_pagination_fallback(self) -> None:
+        page = _FakePage({})
+        page._evaluate_result = True
+        self.assertTrue(ListingCollector._has_next_page(page))
+
+    def test_extract_vehicle_links_from_html_finds_detail_urls(self) -> None:
+        collector = ListingCollector(Settings(), HumanPacer(Settings()))
+        html = """
+        <div>
+            <h4><a href=\"/VehicleDetail/45184893~US\">Car 1</a></h4>
+            <script>window.__data = {\"href\":\"\\/VehicleDetail\\/45171480~US\"}</script>
+        </div>
+        """
+
+        links = collector._extract_vehicle_links_from_html(html)
+
+        self.assertEqual(
+            links,
+            [
+                ("https://www.iaai.com/VehicleDetail/45184893~US", "45184893"),
+                ("https://www.iaai.com/VehicleDetail/45171480~US", "45171480"),
+            ],
+        )
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_worker_tasks.py

@@ -58,6 +58,7 @@ class TestWorkerTaskLockHelpers(unittest.TestCase):
         with patch.object(tasks, "_get_persistence") as get_persistence, \
              patch.object(tasks, "_get_redis") as get_redis, \
              patch.object(tasks, "_acquire_lock", return_value=True), \
+             patch.object(tasks, "_is_full_scan_done", return_value=True), \
              patch.object(tasks, "_start_lock_heartbeat") as start_heartbeat, \
              patch.object(tasks, "_release_lock_if_owner") as release_lock, \
              patch.object(tasks, "_run_browser_job", return_value={
@@ -87,6 +88,74 @@ class TestWorkerTaskLockHelpers(unittest.TestCase):
         heartbeat_thread.join.assert_called_once()
         release_lock.assert_called_once()
 
+    def test_clear_orphan_sync_listing_lock_deletes_when_no_tasks_running(self) -> None:
+        redis_client = MagicMock()
+        redis_client.get.return_value = "owner-token"
+        redis_client.ttl.return_value = 120
+        celery_app = MagicMock()
+        inspector = MagicMock()
+        inspector.active.return_value = {"worker@node": []}
+        inspector.reserved.return_value = {"worker@node": []}
+        inspector.scheduled.return_value = {"worker@node": []}
+        celery_app.control.inspect.return_value = inspector
+
+        cleared = tasks._clear_orphan_sync_listing_lock(redis_client, celery_app)
+
+        self.assertTrue(cleared)
+        redis_client.delete.assert_called_once_with(tasks.SYNC_LISTING_LOCK_KEY)
+
+    def test_clear_orphan_sync_listing_lock_keeps_when_task_detected(self) -> None:
+        redis_client = MagicMock()
+        redis_client.get.return_value = "owner-token"
+        celery_app = MagicMock()
+        inspector = MagicMock()
+        inspector.active.return_value = {
+            "worker@node": [{"name": tasks.SYNC_LISTING_TASK_NAME}]
+        }
+        inspector.reserved.return_value = {"worker@node": []}
+        inspector.scheduled.return_value = {"worker@node": []}
+        celery_app.control.inspect.return_value = inspector
+
+        cleared = tasks._clear_orphan_sync_listing_lock(redis_client, celery_app)
+
+        self.assertFalse(cleared)
+        redis_client.delete.assert_not_called()
+
+    def test_sync_listing_task_recovers_orphan_lock_and_runs(self) -> None:
+        with patch.object(tasks, "_get_persistence") as get_persistence, \
+             patch.object(tasks, "_get_redis") as get_redis, \
+             patch.object(tasks, "_acquire_lock", side_effect=[False, True]) as acquire_lock, \
+             patch.object(tasks, "_clear_orphan_sync_listing_lock", return_value=True) as clear_orphan, \
+             patch.object(tasks, "_is_full_scan_done", return_value=True), \
+             patch.object(tasks, "_start_lock_heartbeat") as start_heartbeat, \
+             patch.object(tasks, "_release_lock_if_owner") as release_lock, \
+             patch.object(tasks, "_run_browser_job", return_value={
+                 "run_id": 9,
+                 "cars_upserted": 3,
+                 "cars_failed": 0,
+                 "images_upserted": 5,
+                 "skipped_existing": 0,
+                 "elapsed_seconds": 2.0,
+             }):
+            persistence = MagicMock()
+            get_persistence.return_value = persistence
+            redis_client = MagicMock()
+            get_redis.return_value = redis_client
+            stop_event = MagicMock()
+            heartbeat_thread = MagicMock()
+            start_heartbeat.return_value = (stop_event, heartbeat_thread)
+
+            tasks.sync_listing_task.push_request(id="task-456")
+            try:
+                result = tasks.sync_listing_task.run(make="Honda")
+            finally:
+                tasks.sync_listing_task.pop_request()
+
+        self.assertEqual(result["status"], "success")
+        clear_orphan.assert_called_once()
+        self.assertEqual(acquire_lock.call_count, 2)
+        release_lock.assert_called_once()
+
 
 if __name__ == "__main__":
     unittest.main()