diff --git a/iaai_scraper/worker/self_heal.py b/iaai_scraper/worker/self_heal.py index 8641706..762359f 100644 --- a/iaai_scraper/worker/self_heal.py +++ b/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: diff --git a/iaai_scraper/worker/tasks.py b/iaai_scraper/worker/tasks.py index 8dea22f..2dd5b46 100644 --- a/iaai_scraper/worker/tasks.py +++ b/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" diff --git a/tests/test_self_heal.py b/tests/test_self_heal.py index f5c74e1..24eb4ce 100644 --- a/tests/test_self_heal.py +++ b/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") diff --git a/tests/test_worker_tasks.py b/tests/test_worker_tasks.py index b7e625b..0a95bd2 100644 --- a/tests/test_worker_tasks.py +++ b/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)