fix self heal

iaai_scraper/worker/self_heal.py

@@ -108,6 +108,31 @@ def _read_last_db_progress_ts(redis_client: Redis) -> int | None:
 	return max_ts or None
 
 
+def _has_active_recent_progress(redis_client: Redis, now_ts: int, stall_seconds: int) -> bool:
+	"""Return True when a task is still reporting non-DB progress.
+
+	Hourly only_new runs can legitimately skip every listed vehicle as existing.
+	Those runs may not emit fast_db_progress for a long time, but they still emit
+	regular listing/segment progress. Treating old DB timestamps as fatal caused
+	the watchdog to kill healthy production workers during such scans.
+	"""
+	for key in redis_client.scan_iter(match="iaai:state:task_progress:*"):
+		try:
+			payload = redis_client.get(key)
+			if not payload:
+				continue
+			data = json.loads(payload)
+			stage = str(data.get("stage") or "")
+			if stage in {"failed", "sync_done", "segment_task_failed", "segment_task_soft_timeout"}:
+				continue
+			ts = _safe_int(data.get("ts"), 0)
+			if ts > 0 and now_ts - ts <= max(60, int(stall_seconds)):
+				return True
+		except Exception:
+			continue
+	return False
+
+
 def _reset_bootstrap_checkpoint_for_db_idle(redis_client: Redis) -> None:
 	pipe = redis_client.pipeline()
 	pipe.delete(SYNC_LISTING_CHECKPOINT_KEY)
@@ -219,6 +244,12 @@ def main() -> None:
 			restart_reason = f"progress_age={age}s > {stall_seconds}s"
 			db_idle_restart = False
 			if age <= stall_seconds:
+				# If the task is actively reporting progress, do not require DB writes.
+				# Hourly only_new listing scans often skip already-known cars and can
+				# legitimately have no DB writes while still moving through segments.
+				if _has_active_recent_progress(redis_client, now_ts, stall_seconds):
+					time.sleep(check_interval)
+					continue
 				last_db_ts = _read_last_db_progress_ts(redis_client)
 				db_age = None if last_db_ts is None else max(0, now_ts - int(last_db_ts))
 				if db_age is None:

iaai_scraper/worker/tasks.py

@@ -578,9 +578,14 @@ def _should_restart_for_db_idle(progress: dict, db_idle_restart_seconds: int) ->
     stage = str(progress.get("stage") or "")
     if stage in TERMINAL_PROGRESS_STAGES:
         return False
+    # Listing/segment progress means the task is alive even if it writes no new
+    # cars. This is normal for hourly only_new runs when all vehicles are already
+    # present in DB. Do not kill healthy scans just because DB progress is idle.
+    progress_ts = _safe_int(progress.get("ts")) or 0
+    if progress_ts > 0 and int(time.time()) - progress_ts < int(db_idle_restart_seconds):
+        return False
     if stage in STALL_WATCHDOG_LONG_RUNNING_STAGES:
         timeout = _stall_timeout_for_progress(stage, db_idle_restart_seconds)
-        progress_ts = _safe_int(progress.get("ts")) or 0
         return progress_ts > 0 and int(time.time()) - progress_ts >= timeout
 
     segments_total = _safe_int(progress.get("segments_total"))
@@ -591,7 +596,6 @@ def _should_restart_for_db_idle(progress: dict, db_idle_restart_seconds: int) ->
         return False
 
     now_ts = int(time.time())
-    progress_ts = _safe_int(progress.get("ts")) or 0
     if progress_ts <= 0:
         return False
 
@@ -1051,10 +1055,17 @@ def _build_listing_segments(settings: Settings) -> list[dict[str, str | int | No
 
     runtime_config = RuntimeConfig.from_file(settings.runtime_config_file)
     brands = runtime_config.filters.include.brands
-    if settings.scraping_profile.http_first and settings.listing.fast_segment_year_splits:
+
+    # Fast HTTP-first режим не использует UI year-фильтры IAAI.
+    # Значит runtime-сегменты должны быть "один бренд = один сегмент"
+    # без доп. year split, иначе получаем много лишних долгих сегментов,
+    # которые fast-профиль всё равно игнорирует.
+    if settings.scraping_profile.http_first:
         segments = build_fast_listing_segments_for_makes(list(brands))
-    else:
+    elif settings.listing.fast_segment_year_splits:
         segments = build_listing_segments_for_makes(list(brands))
+    else:
+        segments = build_fast_listing_segments_for_makes(list(brands))
     if not segments:
         logger.warning(
             "IAAI_LISTING_SEGMENTS=runtime, but runtime_config filters.brands is empty; segmented listing disabled"

tests/test_self_heal.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 import unittest
 from unittest.mock import MagicMock, patch
 
@@ -50,6 +51,27 @@ class TestSelfHeal(unittest.TestCase):
 
 		self.assertIsNone(ts)
 
+	def test_active_recent_progress_prevents_db_idle_restart(self) -> None:
+		redis_client = MagicMock()
+		redis_client.scan_iter.return_value = ["iaai:state:task_progress:hourly"]
+		redis_client.get.return_value = json.dumps(
+			{
+				"task_id": "hourly",
+				"stage": "fast_listing_collected",
+				"ts": 1000,
+				"last_db_progress_ts": 1,
+				"skipped_existing": 2417,
+			}
+		)
+
+		active = self_heal._has_active_recent_progress(
+			redis_client,
+			now_ts=1010,
+			stall_seconds=900,
+		)
+
+		self.assertTrue(active)
+
 	@patch("iaai_scraper.worker.self_heal.time.sleep", return_value=None)
 	@patch("iaai_scraper.worker.self_heal.os.kill")
 	@patch("builtins.open")

tests/test_worker_tasks.py

@@ -19,6 +19,8 @@ class TestWorkerTaskLockHelpers(unittest.TestCase):
 
         settings = Settings(runtime_config_file=runtime_config_file)
         settings.listing.listing_segments_json = "runtime"
+        settings.scraping_profile.http_first = False
+        settings.listing.fast_segment_year_splits = True
 
         segs = tasks._build_listing_segments(settings)