Add tools

.gitignore

@@ -19,4 +19,6 @@ build/
 artifacts/
 celerybeat-schedule*
 tokens_data/
-tokens.json
+tokens.json
+iaai_deploy.tar.gz
+.tmp_iaai_fast_ref/

Dockerfile

@@ -21,7 +21,7 @@ COPY . .
 RUN pip install --no-cache-dir -e .
 RUN python -m compileall -q iaai_scraper
 RUN chmod +x entrypoint.sh
-RUN mkdir -p /data && chown -R app:app /app /data
+RUN chown -R app:app /app
 
 STOPSIGNAL SIGINT
 

check_vps.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+import os
+import paramiko
+
+HOST = "2.26.123.84"
+USER = "root"
+PASSWORD = os.environ["VPS_PASSWORD"]
+
+cmds = [
+    "cd /root/iaai-parser && docker compose ps",
+    "docker logs --since 3m iaai-parser-worker-1 2>&1 | tail -n 120",
+    "docker exec -i iaai-postgres psql -U iaai -d iaai_scraper -c \"SELECT now() AS ts, count(*) AS cars FROM iaai_cars; SELECT count(*) AS runs FROM iaai_sync_runs;\"",
+]
+
+client = paramiko.SSHClient()
+client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+client.connect(
+    HOST,
+    username=USER,
+    password=PASSWORD,
+    look_for_keys=False,
+    allow_agent=False,
+    timeout=30,
+    banner_timeout=30,
+    auth_timeout=30,
+)
+try:
+    for cmd in cmds:
+        print(f"REMOTE_RUN {cmd}", flush=True)
+        stdin, stdout, stderr = client.exec_command(cmd, timeout=120)
+        code = stdout.channel.recv_exit_status()
+        print(stdout.read().decode("utf-8", "replace"), end="")
+        print(stderr.read().decode("utf-8", "replace"), end="")
+        print(f"EXIT {code}", flush=True)
+finally:
+    client.close()

clean_vps_db.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+import os
+import paramiko
+
+HOST = "2.26.123.84"
+USER = "root"
+PASSWORD = os.environ["VPS_PASSWORD"]
+
+commands = [
+    "cd /root/iaai-parser && docker exec -i iaai-postgres psql -U iaai -d iaai_scraper -c 'TRUNCATE TABLE iaai_images, iaai_cars, iaai_sync_runs RESTART IDENTITY CASCADE;'",
+    "docker exec -i iaai-redis redis-cli FLUSHALL",
+    "docker exec -i iaai-postgres psql -U iaai -d iaai_scraper -c 'SELECT count(*) AS cars FROM iaai_cars; SELECT count(*) AS runs FROM iaai_sync_runs;'",
+    "cd /root/iaai-parser && docker compose ps",
+    "docker logs --since 2m iaai-parser-worker-1 2>&1 | tail -n 100",
+]
+
+client = paramiko.SSHClient()
+client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+client.connect(
+    HOST,
+    username=USER,
+    password=PASSWORD,
+    look_for_keys=False,
+    allow_agent=False,
+    timeout=30,
+    banner_timeout=30,
+    auth_timeout=30,
+)
+try:
+    for command in commands:
+        print(f"REMOTE_RUN {command}", flush=True)
+        stdin, stdout, stderr = client.exec_command(command, timeout=180)
+        exit_code = stdout.channel.recv_exit_status()
+        print(stdout.read().decode("utf-8", "replace"), end="")
+        print(stderr.read().decode("utf-8", "replace"), end="")
+        print(f"EXIT {exit_code}", flush=True)
+        if exit_code != 0:
+            raise SystemExit(exit_code)
+finally:
+    client.close()

deploy_vps.py

@@ -0,0 +1,135 @@
+from __future__ import annotations
+
+import os
+import posixpath
+import stat
+import tarfile
+import time
+import socket
+from pathlib import Path
+
+import paramiko
+
+HOST = "2.26.123.84"
+USER = "root"
+PASSWORD = os.environ["VPS_PASSWORD"]
+LOCAL_ROOT = Path(__file__).resolve().parent
+ARCHIVE = LOCAL_ROOT / "iaai_deploy.tar.gz"
+REMOTE_DIR = "/root/iaai-parser"
+REMOTE_ARCHIVE = "/root/iaai_deploy.tar.gz"
+
+EXCLUDE_DIRS = {
+    ".git",
+    ".venv",
+    "__pycache__",
+    ".pytest_cache",
+    ".mypy_cache",
+    ".ruff_cache",
+    "iaai_scraper.egg-info",
+}
+EXCLUDE_FILES = {"iaai_deploy.tar.gz", "deploy_vps.py"}
+
+
+def _include(path: Path) -> bool:
+    rel = path.relative_to(LOCAL_ROOT)
+    parts = set(rel.parts)
+    if parts & EXCLUDE_DIRS:
+        return False
+    if path.name in EXCLUDE_FILES:
+        return False
+    if path.suffix in {".pyc", ".pyo"}:
+        return False
+    return True
+
+
+def make_archive() -> None:
+    if ARCHIVE.exists():
+        ARCHIVE.unlink()
+    with tarfile.open(ARCHIVE, "w:gz") as tf:
+        for path in LOCAL_ROOT.rglob("*"):
+            if not _include(path):
+                continue
+            tf.add(path, arcname=str(path.relative_to(LOCAL_ROOT)))
+    print(f"ARCHIVE_READY {ARCHIVE} {ARCHIVE.stat().st_size} bytes", flush=True)
+
+
+def connect() -> paramiko.SSHClient:
+    last_exc: BaseException | None = None
+    for attempt in range(1, 6):
+        client = paramiko.SSHClient()
+        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+        try:
+            print(f"SSH_CONNECT attempt={attempt}", flush=True)
+            client.connect(
+                HOST,
+                username=USER,
+                password=PASSWORD,
+                look_for_keys=False,
+                allow_agent=False,
+                timeout=45,
+                banner_timeout=60,
+                auth_timeout=45,
+            )
+            return client
+        except (paramiko.SSHException, socket.timeout, OSError) as exc:
+            last_exc = exc
+            client.close()
+            print(f"SSH_CONNECT_RETRY attempt={attempt} error={type(exc).__name__}: {exc}", flush=True)
+            time.sleep(3 * attempt)
+    raise SystemExit(f"SSH_CONNECT_FAILED: {last_exc}")
+
+
+def run(client: paramiko.SSHClient, cmd: str, timeout: int | None = None) -> None:
+    print(f"REMOTE_RUN {cmd}", flush=True)
+    stdin, stdout, stderr = client.exec_command(cmd, timeout=timeout)
+    exit_code = stdout.channel.recv_exit_status()
+    out = stdout.read().decode("utf-8", "replace")
+    err = stderr.read().decode("utf-8", "replace")
+    if out:
+        print(out, end="", flush=True)
+    if err:
+        print(err, end="", flush=True)
+    if exit_code != 0:
+        raise SystemExit(f"REMOTE_FAILED code={exit_code}: {cmd}")
+
+
+def upload_file(sftp: paramiko.SFTPClient, local: Path, remote: str) -> None:
+    total = local.stat().st_size
+    last = 0.0
+
+    def cb(done: int, _total: int) -> None:
+        nonlocal last
+        now = time.time()
+        if now - last >= 2 or done == total:
+            print(f"UPLOAD {done}/{total}", flush=True)
+            last = now
+
+    sftp.put(str(local), remote, callback=cb)
+
+
+def main() -> None:
+    make_archive()
+    client = connect()
+    try:
+        run(client, "echo VPS_OK && hostname && docker --version && docker compose version")
+        sftp = client.open_sftp()
+        try:
+            upload_file(sftp, ARCHIVE, REMOTE_ARCHIVE)
+        finally:
+            sftp.close()
+        run(
+            client,
+            f"mkdir -p {REMOTE_DIR} && cd {REMOTE_DIR} && docker compose down || true && "
+            f"find {REMOTE_DIR} -mindepth 1 -maxdepth 1 ! -name '.env' -exec rm -rf {{}} + && "
+            f"tar -xzf {REMOTE_ARCHIVE} -C {REMOTE_DIR} && rm -f {REMOTE_ARCHIVE}",
+            timeout=300,
+        )
+        run(client, f"cd {REMOTE_DIR} && docker compose up -d --build", timeout=1800)
+        run(client, f"cd {REMOTE_DIR} && docker compose ps", timeout=120)
+        run(client, "docker logs --since 2m iaai-parser-worker-1 2>&1 | tail -n 120 || docker compose -f /root/iaai-parser/docker-compose.yml logs --since 2m worker | tail -n 120", timeout=120)
+    finally:
+        client.close()
+
+
+if __name__ == "__main__":
+    main()

docker-compose.yml

@@ -5,40 +5,43 @@ x-app-env: &app-env
   CELERY_BROKER_URL: ${CELERY_BROKER_URL:-redis://redis:6379/0}
   CELERY_RESULT_BACKEND: ${CELERY_RESULT_BACKEND:-redis://redis:6379/0}
   IAAI_DATABASE_POOL_RECYCLE_SECONDS: ${IAAI_DATABASE_POOL_RECYCLE_SECONDS:-1800}
-  IAAI_DATABASE_POOL_SIZE: ${IAAI_DATABASE_POOL_SIZE:-20}
+  # Для больших Search-выборок (десятки тысяч лотов) run должен жить дольше одного батча.
-  IAAI_DATABASE_MAX_OVERFLOW: ${IAAI_DATABASE_MAX_OVERFLOW:-40}
+  CELERY_TASK_SOFT_TIME_LIMIT: ${CELERY_TASK_SOFT_TIME_LIMIT:-7200}
-  CELERY_TASK_SOFT_TIME_LIMIT: ${CELERY_TASK_SOFT_TIME_LIMIT:-900}
+  CELERY_TASK_TIME_LIMIT: ${CELERY_TASK_TIME_LIMIT:-7500}
-  CELERY_TASK_TIME_LIMIT: ${CELERY_TASK_TIME_LIMIT:-1200}
+  CELERY_BROKER_VISIBILITY_TIMEOUT: ${CELERY_BROKER_VISIBILITY_TIMEOUT:-10800}
-  CELERY_BROKER_VISIBILITY_TIMEOUT: ${CELERY_BROKER_VISIBILITY_TIMEOUT:-2400}
-  IAAI_PROFILE: ${IAAI_PROFILE:-fast}
-  IAAI_SCRAPING_PROFILE: ${IAAI_SCRAPING_PROFILE:-${IAAI_PROFILE:-fast}}
-  IAAI_HTTP_FIRST: ${IAAI_HTTP_FIRST:-true}
-  IAAI_BROWSER_FALLBACK_ENABLED: ${IAAI_BROWSER_FALLBACK_ENABLED:-false}
-  IAAI_ANONYMOUS_BOOTSTRAP_ENABLED: ${IAAI_ANONYMOUS_BOOTSTRAP_ENABLED:-false}
-  IAAI_CHALLENGE_REFRESH_ATTEMPTS: ${IAAI_CHALLENGE_REFRESH_ATTEMPTS:-1}
-  IAAI_LISTING_POST_ATTEMPTS: ${IAAI_LISTING_POST_ATTEMPTS:-2}
-  IAAI_REQUEST_JITTER_MAX_S: ${IAAI_REQUEST_JITTER_MAX_S:-0}
-  IAAI_DETAIL_RETRIES: ${IAAI_DETAIL_RETRIES:-1}
-  IAAI_LISTING_RETRIES: ${IAAI_LISTING_RETRIES:-1}
   # 0 = без лимита.
   CELERY_BEAT_SYNC_LIMIT: ${CELERY_BEAT_SYNC_LIMIT:-0}
-  CELERY_WORKER_CONCURRENCY: ${CELERY_WORKER_CONCURRENCY:-4}
+  # Следующий плановый прогон — не раньше чем через час после предыдущего tick beat.
-  CELERY_WORKER_POOL: ${CELERY_WORKER_POOL:-prefork}
+  CELERY_BEAT_SYNC_INTERVAL_MINUTES: ${CELERY_BEAT_SYNC_INTERVAL_MINUTES:-60}
+  # Любой внутренний follow-up после неполного bootstrap тоже не стартует сразу.
+  IAAI_SYNC_FOLLOWUP_MIN_DELAY_SECONDS: ${IAAI_SYNC_FOLLOWUP_MIN_DELAY_SECONDS:-3600}
   CELERY_WORKER_MAX_TASKS_PER_CHILD: ${CELERY_WORKER_MAX_TASKS_PER_CHILD:-5}
-  CELERY_BATCH_SIZE: ${CELERY_BATCH_SIZE:-1000}
+  CELERY_BATCH_SIZE: ${CELERY_BATCH_SIZE:-1500}
-  IAAI_INTER_BATCH_DELAY_SECONDS: ${IAAI_INTER_BATCH_DELAY_SECONDS:-0}
+  IAAI_INTER_BATCH_DELAY_SECONDS: ${IAAI_INTER_BATCH_DELAY_SECONDS:-0.3}
+  # Стабильный fast-профиль: не душим IAAI слишком большим числом HTTPS detail-соединений.
+  IAAI_SCRAPING_PROFILE: fast
+  IAAI_FETCH_CONCURRENCY: ${IAAI_FETCH_CONCURRENCY:-96}
+  IAAI_MAX_RETRIES: ${IAAI_MAX_RETRIES:-2}
+  IAAI_RETRY_DELAY_SECONDS: ${IAAI_RETRY_DELAY_SECONDS:-0.35}
+  IAAI_CHALLENGE_REFRESH_ATTEMPTS: ${IAAI_CHALLENGE_REFRESH_ATTEMPTS:-1}
+  CELERY_PARALLEL_SEGMENTS: ${CELERY_PARALLEL_SEGMENTS:-false}
+  CELERY_WORKER_CONCURRENCY: 4
+  CELERY_WORKER_POOL: ${CELERY_WORKER_POOL:-prefork}
   IAAI_FAIL_RATE_THRESHOLD: ${IAAI_FAIL_RATE_THRESHOLD:-0.9}
   IAAI_PARALLEL_TABS: ${IAAI_PARALLEL_TABS:-8}
-  IAAI_FETCH_CONCURRENCY: ${IAAI_FETCH_CONCURRENCY:-32}
   IAAI_BLOCK_RESOURCES: ${IAAI_BLOCK_RESOURCES:-true}
-  IAAI_FILTERED_SEARCH_URL: ${IAAI_FILTERED_SEARCH_URL:-}
+  IAAI_FAST_PATH_TIMEOUT_MS: ${IAAI_FAST_PATH_TIMEOUT_MS:-10000}
-  IAAI_FILTERED_SEARCH_URLS: ${IAAI_FILTERED_SEARCH_URLS:-}
+  IAAI_LISTING_SEGMENTS: ${IAAI_LISTING_SEGMENTS:-[]}
-  IAAI_LISTING_SEGMENTS: ${IAAI_LISTING_SEGMENTS:-runtime}
+  IAAI_DISCOVERY_MODE: ${IAAI_DISCOVERY_MODE:-listing}
+  IAAI_HOURLY_MODE: ${IAAI_HOURLY_MODE:-rolling_refresh}
+  IAAI_HOURLY_REFRESH_BATCH_SIZE: ${IAAI_HOURLY_REFRESH_BATCH_SIZE:-500}
   IAAI_MAX_PAGES_PER_RUN: ${IAAI_MAX_PAGES_PER_RUN:-9999}
   IAAI_MAX_VEHICLES_PER_RUN: ${IAAI_MAX_VEHICLES_PER_RUN:-50000}
-  IAAI_ALWAYS_FULL_SCAN: ${IAAI_ALWAYS_FULL_SCAN:-true}
+  IAAI_ALWAYS_FULL_SCAN: ${IAAI_ALWAYS_FULL_SCAN:-false}
-  IAAI_HUMAN_PACE_ENABLED: ${IAAI_HUMAN_PACE_ENABLED:-true}
+  IAAI_HUMAN_PACE_ENABLED: ${IAAI_HUMAN_PACE_ENABLED:-false}
-  IAAI_TOKENS_FILE: ${IAAI_TOKENS_FILE:-/home/app/tokens.json}
+  # Первый sync после clean restart стартует сразу; следующий запуск — только через hourly guard/beat.
+  IAAI_STARTUP_SYNC_ENABLED: ${IAAI_STARTUP_SYNC_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: chromium
   # Self-heal для worker.
@@ -47,8 +50,6 @@ x-app-env: &app-env
   IAAI_SELF_HEAL_STALL_SECONDS: ${IAAI_SELF_HEAL_STALL_SECONDS:-900}
   IAAI_SELF_HEAL_STARTUP_GRACE_SECONDS: ${IAAI_SELF_HEAL_STARTUP_GRACE_SECONDS:-300}
   IAAI_SELF_HEAL_RESTART_COOLDOWN_SECONDS: ${IAAI_SELF_HEAL_RESTART_COOLDOWN_SECONDS:-300}
-  # Если в Postgres нет записей дольше 60 минут при активной работе — полный restart с segment 1.
-  IAAI_DB_IDLE_RESTART_SECONDS: ${IAAI_DB_IDLE_RESTART_SECONDS:-3600}
   TZ: ${TZ:-UTC}
 
 x-env-file: &env-file
@@ -168,7 +169,9 @@ services:
     stop_grace_period: 60s
     command: >
       celery -A iaai_scraper.worker.celery_app worker
-      --loglevel=info --concurrency=${CELERY_WORKER_CONCURRENCY:-4} --pool=${CELERY_WORKER_POOL:-prefork}
+      --loglevel=info
+      --concurrency=${CELERY_WORKER_CONCURRENCY:-4}
+      --pool=${CELERY_WORKER_POOL:-prefork}
       --pidfile=/tmp/celery-worker.pid
       -Q iaai_sync --max-tasks-per-child=${CELERY_WORKER_MAX_TASKS_PER_CHILD:-5}
     healthcheck:

iaai_scraper/browser/fast_client.py

@@ -519,7 +519,17 @@ class IAAIFastClient:
             if max_pages is not None and max_pages > 0:
                 total_pages = min(total_pages, max_pages)
             gbp_search_query = first_page.gbp_search_query
+            logger.info(
+                "Fast listing first page parsed: scope=%s result_count=%s page_size=%s total_pages=%s first_page_vehicles=%s",
+                scope_path,
+                first_page.result_count,
+                page_size,
+                total_pages,
+                len(first_page.vehicles),
+            )
             for page_number in range(2, total_pages + 1):
+                if page_number == 2 or page_number % 25 == 0 or page_number == total_pages:
+                    logger.info("Fast listing page fetch progress: page=%s/%s", page_number, total_pages)
                 page_html = self._fetch_listing_page(scope_path, gbp_search_query, page_number, page_size)
                 parsed_page = parse_listing_page(page_html)
                 gbp_search_query = parsed_page.gbp_search_query

iaai_scraper/core/logs.py

@@ -18,8 +18,9 @@ def set_trace_id(trace_id: str) -> None:
 
 
 def setup_logging(level: str = "INFO", log_file: str | None = None) -> None:
-    # stderr — Docker и Celery prefork корректно его подхватывают.
+    # stdout — основной поток для `docker logs`, Docker Desktop и compose logs.
-    handlers: list[logging.Handler] = [logging.StreamHandler(sys.stderr)]
+    # stderr в PowerShell часто выглядит как NativeCommandError, хотя это обычные логи.
+    handlers: list[logging.Handler] = [logging.StreamHandler(sys.stdout)]
     if log_file:
         handlers.append(logging.FileHandler(log_file, encoding="utf-8"))
     trace_filter = TraceIdFilter()

iaai_scraper/fast_sync.py

@@ -183,6 +183,7 @@ class FastSyncEngine:
         prepared_rows: list[CarRecord] = []
         db_processed = 0
         retry_candidates: list[FastListingVehicle] = []
+        transient_failure_log_count = 0
         started_details = time.perf_counter()
         with concurrent.futures.ThreadPoolExecutor(max_workers=min(self.fetch_concurrency, len(candidates))) as executor:
             future_to_vehicle = {
@@ -232,13 +233,26 @@ class FastSyncEngine:
                         stats.protection_events += 1
                     if self._is_transient_detail_error(exc):
                         retry_candidates.append(vehicle)
-                        logger.info("Fast detail transient failure queued for retry inventory_id=%s: %s", vehicle.inventory_id, exc)
+                        transient_failure_log_count += 1
+                        if transient_failure_log_count % 100 == 0:
+                            logger.warning(
+                                "Fast detail transient failures queued for retry: count=%d latest_inventory_id=%s",
+                                transient_failure_log_count,
+                                vehicle.inventory_id,
+                            )
                     else:
                         logger.exception("Fast detail parse failed inventory_id=%s: %s", vehicle.inventory_id, exc)
 
                 if index % 100 == 0 or index == len(candidates):
                     elapsed = max(0.001, time.perf_counter() - started_details)
-                    if self.client._settings.scraping_profile.verbose_progress_logs:
+                    verbose_progress_logs = bool(
+                        getattr(
+                            getattr(getattr(self.client, "_settings", None), "scraping_profile", None),
+                            "verbose_progress_logs",
+                            False,
+                        )
+                    )
+                    if verbose_progress_logs:
                         logger.info(
                             "Fast HTTP-first details progress: processed=%d/%d ok=%d failed=%d queued_db=%d rate=%.2f/s",
                             index,

iaai_scraper/scraper.py

@@ -2234,6 +2234,8 @@ class IAAIScraper:
             only_new = rc.only_new
         if lane == "iaai_cars" and rc.lane is not None:
             lane = rc.lane
+        if listing_url is None and rc.name:
+            listing_url = rc.name
 
         trace_id = self._new_trace_id("sync-listing")
         started_at = time.perf_counter()
@@ -2260,6 +2262,8 @@ class IAAIScraper:
                         year_min,
                         year_max,
                     )
+                if listing_url:
+                    logger.info("sync_listing uses explicit listing_url from runtime/args: %s", listing_url)
                 fast_engine = FastSyncEngine(
                     client=self.fast_client,
                     mapper=self.fast_mapper,

iaai_scraper/worker/celery_app.py

@@ -16,6 +16,9 @@ logger = logging.getLogger("iaai_scraper.worker.celery_app")
 STARTUP_SYNC_DISPATCH_KEY = "iaai:state:startup_sync_dispatched"
 IAAI_SYNC_QUEUE = "iaai_sync"
 PROGRESS_KEY_PREFIX = "iaai:state:task_progress:"
+SYNC_LISTING_LOCK_KEY = "iaai:locks:sync_listing"
+SYNC_FULL_SCAN_DONE_KEY = "iaai:state:sync_full_scan_done"
+SYNC_LAST_COMPLETED_AT_KEY = "iaai:state:sync_listing_last_completed_at"
 
 
 def _env_bool(name: str, default: bool) -> bool:
@@ -43,6 +46,26 @@ def _has_fresh_active_progress(redis_client: Redis, *, max_age_seconds: int = 18
     return False
 
 
+def _seconds_until_next_allowed_sync(redis_client: Redis) -> int:
+    """Запрещает новый автозапуск раньше чем через интервал beat после завершения полного run."""
+    min_interval = max(0, int(settings.celery.beat_sync_interval_minutes * 60))
+    if min_interval <= 0:
+        return 0
+    try:
+        full_done = str(redis_client.get(SYNC_FULL_SCAN_DONE_KEY) or "").strip().lower() in {"1", "true", "yes", "on"}
+        if not full_done:
+            return 0
+        completed_raw = redis_client.get(SYNC_LAST_COMPLETED_AT_KEY)
+        if not completed_raw:
+            return 0
+        completed_at = int(float(completed_raw))
+    except Exception:
+        logger.warning("Failed to inspect last sync completion timestamp", exc_info=True)
+        return 0
+    elapsed = int(time.time()) - completed_at
+    return max(0, min_interval - elapsed)
+
+
 @celery_setup_logging.connect
 def _configure_logging(loglevel=None, **kwargs):
     # Перехватываем логирование Celery и пишем только в stderr (Docker logs).
@@ -112,7 +135,7 @@ celery_app.conf.update(
             "task": "iaai.sync_cars_feed",
             "schedule": settings.celery.beat_sync_interval_minutes * 60.0,
             "args": (),
-            "kwargs": {"limit": settings.celery.beat_sync_limit, "only_new": True},
+            "kwargs": {"limit": settings.celery.beat_sync_limit, "only_new": False},
             "options": {
                 "queue": IAAI_SYNC_QUEUE,
                 "expires": settings.celery.beat_sync_interval_minutes * 60.0,
@@ -147,7 +170,15 @@ def _on_worker_ready(**kwargs):
         )
 
         has_fresh_progress = _has_fresh_active_progress(redis_client)
-        for stale_key in ("iaai:locks:sync_listing",):
+        next_allowed_delay = _seconds_until_next_allowed_sync(redis_client)
+        if next_allowed_delay > 0:
+            logger.info(
+                "Worker ready: last full sync finished recently; next auto sync allowed in %ss, skip startup dispatch",
+                next_allowed_delay,
+            )
+            return
+
+        for stale_key in (SYNC_LISTING_LOCK_KEY,):
             try:
                 ttl = redis_client.ttl(stale_key)
                 if ttl is not None and ttl != -2 and not has_fresh_progress:

iaai_scraper/worker/fast_sync.py

@@ -0,0 +1,479 @@
+from __future__ import annotations
+
+import concurrent.futures
+import logging
+import random
+import time
+from dataclasses import dataclass
+from typing import Any, Callable
+
+from .browser.fast_client import FastListingVehicle, IAAIFastClient
+from .core.runtime_config import RuntimeConfig
+from .parsing.fast_mapper import INACTIVE_STATUS_VALUES, FastCarMapper
+from .storage.db import PersistenceService
+from .storage.schemas import CarRecord
+
+logger = logging.getLogger("iaai_scraper.fast_sync")
+
+
+@dataclass(slots=True)
+class FastSyncStats:
+    ids_fetched: int = 0
+    cars_upserted: int = 0
+    cars_failed: int = 0
+    cars_filtered: int = 0
+    images_upserted: int = 0
+    skipped_existing: int = 0
+    protection_events: int = 0
+
+
+def passes_condition_check(row: FastListingVehicle) -> bool:
+    if row.timed_auction_closed:
+        return False
+    status = (row.inventory_status or "").strip().upper()
+    return status not in INACTIVE_STATUS_VALUES
+
+
+class FastSyncEngine:
+    """Full iaai-fast style sync pipeline adapted to this project's storage.
+
+    Flow: hidden listing payloads -> concurrent ProductDetailsVM HTTP fetch ->
+    CarRecord preparation in memory -> single batch DB upsert. Browser is used
+    only inside IAAIFastClient to refresh cookies when IAAI challenge appears.
+    """
+
+    def __init__(
+        self,
+        *,
+        client: IAAIFastClient,
+        mapper: FastCarMapper,
+        persistence: PersistenceService,
+        batch_size: int,
+        fetch_concurrency: int,
+        report_progress: Callable[[str, Any], None] | None = None,
+    ) -> None:
+        self.client = client
+        self.mapper = mapper
+        self.persistence = persistence
+        self.batch_size = max(1, int(batch_size))
+        self.fetch_concurrency = max(1, int(fetch_concurrency))
+        self.report_progress = report_progress
+
+    @staticmethod
+    def _is_transient_detail_error(exc: Exception) -> bool:
+        text = str(exc).lower()
+        return any(
+            marker in text
+            for marker in (
+                "sslerror",
+                "ssleoferror",
+                "unexpected_eof_while_reading",
+                "eof occurred in violation of protocol",
+                "max retries exceeded",
+                "read timed out",
+                "readtimeout",
+                "connection reset",
+                "connection aborted",
+                "connection closed",
+                "temporarily unavailable",
+                "too many requests",
+                "status=429",
+                "status=500",
+                "status=502",
+                "status=503",
+                "status=504",
+            )
+        )
+
+    def sync_listing(
+        self,
+        *,
+        runtime_config: RuntimeConfig,
+        make: str | None = None,
+        model: str | None = None,
+        lane: str = "iaai_cars",
+        limit: int | None = None,
+        only_new: bool = False,
+        listing_url: str | None = None,
+        max_pages: int | None = None,
+        skip_mark_sold: bool = False,
+    ) -> dict[str, Any]:
+        del lane
+        started_at = time.perf_counter()
+        stats = FastSyncStats()
+        errors: list[dict[str, str]] = []
+        selected: dict[str, FastListingVehicle] = {}
+        rows_seen = 0
+        rows_skipped_condition = 0
+
+        filters = runtime_config.filters
+        logger.info(
+            "Fast HTTP-first listing started: make=%s listing_url=%s max_pages=%s concurrency=%s batch_size=%s",
+            make or "ALL",
+            listing_url or "default",
+            max_pages,
+            self.fetch_concurrency,
+            self.batch_size,
+        )
+        for vehicle in self.client.iter_listing_vehicles(
+            listing_start_url=listing_url,
+            make=make,
+            max_pages=max_pages,
+        ):
+            rows_seen += 1
+            if runtime_config.sync.condition_check_enabled and not passes_condition_check(vehicle):
+                rows_skipped_condition += 1
+                continue
+            if vehicle.inventory_id in selected:
+                continue
+            selected[vehicle.inventory_id] = vehicle
+            if limit is not None and limit > 0 and len(selected) >= limit:
+                break
+
+        candidates = list(selected.values())
+        all_listing_origin_urls = {f"https://www.iaai.com/VehicleDetail/{v.inventory_id}" for v in candidates}
+        if only_new and candidates:
+            origin_urls = [f"https://www.iaai.com/VehicleDetail/{v.inventory_id}" for v in candidates]
+            origin_ids = [f"iaai:{v.inventory_id}" for v in candidates]
+            existing_urls, existing_ids = self.persistence.get_existing_urls_and_ids(origin_urls, origin_ids)
+            fresh: list[FastListingVehicle] = []
+            for vehicle in candidates:
+                if f"https://www.iaai.com/VehicleDetail/{vehicle.inventory_id}" in existing_urls or f"iaai:{vehicle.inventory_id}" in existing_ids:
+                    stats.skipped_existing += 1
+                    continue
+                fresh.append(vehicle)
+            candidates = fresh
+
+        self._progress(
+            "fast_listing_collected",
+            rows_seen=rows_seen,
+            rows_filtered_condition=rows_skipped_condition,
+            rows_selected=len(candidates),
+            skipped_existing=stats.skipped_existing,
+        )
+        logger.info(
+            "Fast HTTP-first listing collected: rows_seen=%d selected=%d skipped_existing=%d filtered_condition=%d only_new=%s",
+            rows_seen,
+            len(candidates),
+            stats.skipped_existing,
+            rows_skipped_condition,
+            only_new,
+        )
+
+        scan_completed = not (limit is not None and limit > 0) and not errors
+
+        if not candidates:
+            return self._result(
+                started_at=started_at,
+                stats=stats,
+                failures=errors,
+                listing={
+                    "mode": "fast_hidden_payload",
+                    "vehicles_collected": 0,
+                    "vehicle_urls": [],
+                    "early_stopped": False,
+                    "truncated_by_time_budget": False,
+                    "rows_seen": rows_seen,
+                    "rows_filtered_condition": rows_skipped_condition,
+                },
+                all_listing_origin_urls=all_listing_origin_urls,
+                full_scan_completed=scan_completed,
+            )
+
+        prepared_rows: list[CarRecord] = []
+        db_processed = 0
+        retry_candidates: list[FastListingVehicle] = []
+        transient_failure_log_count = 0
+        started_details = time.perf_counter()
+        with concurrent.futures.ThreadPoolExecutor(max_workers=min(self.fetch_concurrency, len(candidates))) as executor:
+            future_to_vehicle = {
+                executor.submit(self._fetch_detail_payload, vehicle.inventory_id): vehicle
+                for vehicle in candidates
+            }
+            for index, future in enumerate(concurrent.futures.as_completed(future_to_vehicle), start=1):
+                vehicle = future_to_vehicle[future]
+                try:
+                    payload = future.result()
+                    record = self.mapper.map_payload_to_record(
+                        detail_payload=payload,
+                        vehicle_url=f"https://www.iaai.com/VehicleDetail/{vehicle.inventory_id}",
+                        listing_vehicle=vehicle,
+                    )
+                    if model and model.casefold() not in record.model.casefold():
+                        stats.cars_filtered += 1
+                        continue
+                    if not filters.matches({
+                        "brand": record.brand,
+                        "model": record.model,
+                        "year": record.year,
+                        "body_type": record.body_type,
+                        "color": record.color,
+                        "drive": record.drive,
+                        "gearbox": record.gearbox,
+                        "price": record.price,
+                        "mileage": record.mileage,
+                    }):
+                        stats.cars_filtered += 1
+                        continue
+                    prepared_rows.append(record)
+                    stats.ids_fetched += 1
+                    if len(prepared_rows) >= self.batch_size:
+                        db_processed += self._flush_db_records(
+                            rows=prepared_rows,
+                            stats=stats,
+                            errors=errors,
+                            processed=db_processed + len(prepared_rows),
+                            total=len(candidates),
+                        )
+                        prepared_rows.clear()
+                except Exception as exc:
+                    stats.cars_failed += 1
+                    errors.append({"vehicle_url": f"https://www.iaai.com/VehicleDetail/{vehicle.inventory_id}", "error": str(exc)})
+                    if _looks_like_protection(exc):
+                        stats.protection_events += 1
+                    if self._is_transient_detail_error(exc):
+                        retry_candidates.append(vehicle)
+                        transient_failure_log_count += 1
+                        if transient_failure_log_count % 100 == 0:
+                            logger.warning(
+                                "Fast detail transient failures queued for retry: count=%d latest_inventory_id=%s",
+                                transient_failure_log_count,
+                                vehicle.inventory_id,
+                            )
+                    else:
+                        logger.exception("Fast detail parse failed inventory_id=%s: %s", vehicle.inventory_id, exc)
+
+                if index % 100 == 0 or index == len(candidates):
+                    elapsed = max(0.001, time.perf_counter() - started_details)
+                    if self.client._settings.scraping_profile.verbose_progress_logs:
+                        logger.info(
+                            "Fast HTTP-first details progress: processed=%d/%d ok=%d failed=%d queued_db=%d rate=%.2f/s",
+                            index,
+                            len(candidates),
+                            stats.ids_fetched,
+                            stats.cars_failed,
+                            len(prepared_rows),
+                            index / elapsed,
+                        )
+                    self._progress(
+                        "fast_detail_progress",
+                        processed=index,
+                        total=len(candidates),
+                        ids_fetched=stats.ids_fetched,
+                        cars_failed=stats.cars_failed,
+                        queued_for_db=len(prepared_rows),
+                        throughput=round(index / elapsed, 2),
+                    )
+
+        if retry_candidates:
+            retry_started = time.perf_counter()
+            retry_workers = max(1, min(8, self.fetch_concurrency // 2, len(retry_candidates)))
+            retry_errors: list[dict[str, str]] = []
+            logger.info(
+                "Fast HTTP-first retrying transient detail failures: total=%d workers=%d",
+                len(retry_candidates),
+                retry_workers,
+            )
+            with concurrent.futures.ThreadPoolExecutor(max_workers=retry_workers) as executor:
+                future_to_vehicle = {
+                    executor.submit(self._fetch_detail_payload, vehicle.inventory_id): vehicle
+                    for vehicle in retry_candidates
+                }
+                for retry_index, future in enumerate(concurrent.futures.as_completed(future_to_vehicle), start=1):
+                    vehicle = future_to_vehicle[future]
+                    try:
+                        payload = future.result()
+                        record = self.mapper.map_payload_to_record(
+                            detail_payload=payload,
+                            vehicle_url=f"https://www.iaai.com/VehicleDetail/{vehicle.inventory_id}",
+                            listing_vehicle=vehicle,
+                        )
+                        if model and model.casefold() not in record.model.casefold():
+                            stats.cars_filtered += 1
+                            continue
+                        if not filters.matches({
+                            "brand": record.brand,
+                            "model": record.model,
+                            "year": record.year,
+                            "body_type": record.body_type,
+                            "color": record.color,
+                            "drive": record.drive,
+                            "gearbox": record.gearbox,
+                            "price": record.price,
+                            "mileage": record.mileage,
+                        }):
+                            stats.cars_filtered += 1
+                            continue
+                        prepared_rows.append(record)
+                        stats.ids_fetched += 1
+                        stats.cars_failed = max(0, stats.cars_failed - 1)
+                        if len(prepared_rows) >= self.batch_size:
+                            db_processed += self._flush_db_records(
+                                rows=prepared_rows,
+                                stats=stats,
+                                errors=errors,
+                                processed=db_processed + len(prepared_rows),
+                                total=len(candidates),
+                            )
+                            prepared_rows.clear()
+                    except Exception as exc:
+                        retry_errors.append({
+                            "vehicle_url": f"https://www.iaai.com/VehicleDetail/{vehicle.inventory_id}",
+                            "error": str(exc),
+                        })
+                        if _looks_like_protection(exc):
+                            stats.protection_events += 1
+                    if retry_index % 100 == 0 or retry_index == len(retry_candidates):
+                        elapsed = max(0.001, time.perf_counter() - retry_started)
+                        logger.info(
+                            "Fast HTTP-first retry progress: processed=%d/%d recovered=%d remaining_failed=%d rate=%.2f/s",
+                            retry_index,
+                            len(retry_candidates),
+                            len(retry_candidates) - len(retry_errors),
+                            len(retry_errors),
+                            retry_index / elapsed,
+                        )
+            transient_urls = {f"https://www.iaai.com/VehicleDetail/{v.inventory_id}" for v in retry_candidates}
+            errors = [error for error in errors if error.get("vehicle_url") not in transient_urls]
+            errors.extend(retry_errors)
+
+        if prepared_rows:
+            db_processed += self._flush_db_records(
+                rows=prepared_rows,
+                stats=stats,
+                errors=errors,
+                processed=db_processed + len(prepared_rows),
+                total=len(candidates),
+            )
+            prepared_rows.clear()
+
+        self.client.persist_session_state()
+
+        scan_completed = not (limit is not None and limit > 0) and not errors
+        mark_sold_scope_partial = bool(
+            (limit is not None and limit > 0)
+            or make
+            or model
+            or listing_url
+            or only_new
+        )
+        if all_listing_origin_urls and not mark_sold_scope_partial and not skip_mark_sold and scan_completed:
+            try:
+                sold_count = self.persistence.mark_sold_not_in_listing_by_urls(all_listing_origin_urls, lane="iaai")
+            except Exception as exc:
+                sold_count = 0
+                logger.warning("Fast sold reconcile failed: %s", exc)
+        else:
+            sold_count = 0
+
+        listing = {
+            "mode": "fast_hidden_payload",
+            "vehicles_collected": len(candidates),
+            "vehicle_urls": [f"https://www.iaai.com/VehicleDetail/{v.inventory_id}" for v in candidates],
+            "early_stopped": False,
+            "truncated_by_time_budget": False,
+            "rows_seen": rows_seen,
+            "rows_filtered_condition": rows_skipped_condition,
+            "sold_marked": sold_count,
+        }
+        return self._result(
+            started_at=started_at,
+            stats=stats,
+            failures=errors,
+            listing=listing,
+            all_listing_origin_urls=all_listing_origin_urls,
+            full_scan_completed=scan_completed,
+        )
+
+    def _result(
+        self,
+        *,
+        started_at: float,
+        stats: FastSyncStats,
+        failures: list[dict[str, str]],
+        listing: dict[str, Any],
+        all_listing_origin_urls: set[str],
+        full_scan_completed: bool,
+    ) -> dict[str, Any]:
+        status = "success" if not failures else ("partial_success" if stats.cars_upserted else "failed")
+        total = int(listing.get("vehicles_collected") or 0)
+        fail_ratio = (stats.cars_failed / total) if total > 0 else 0.0
+        protection_ratio = (stats.protection_events / total) if total > 0 else 0.0
+        anti_bot_detected = total > 0 and ((stats.protection_events >= 30 and protection_ratio >= 0.10) or fail_ratio >= 0.30)
+        return {
+            "status": status,
+            "listing": listing,
+            "total": total,
+            "total_discovered": total,
+            "skipped_existing": stats.skipped_existing,
+            "cars_upserted": stats.cars_upserted,
+            "cars_failed": stats.cars_failed,
+            "cars_filtered": stats.cars_filtered,
+            "images_upserted": stats.images_upserted,
+            "protection_events": stats.protection_events,
+            "failures": failures,
+            "all_listing_origin_urls": all_listing_origin_urls,
+            "full_scan_completed": full_scan_completed and not anti_bot_detected,
+            "anti_bot_detected": anti_bot_detected,
+            "fail_ratio": round(fail_ratio, 4),
+            "protection_ratio": round(protection_ratio, 4),
+            "elapsed_seconds": round(time.perf_counter() - started_at, 3),
+        }
+
+    def _progress(self, stage: str, **meta: Any) -> None:
+        if self.report_progress is None:
+            return
+        try:
+            self.report_progress(stage, **meta)
+        except Exception:
+            pass
+
+    def _fetch_detail_payload(self, inventory_id: str) -> dict[str, Any]:
+        jitter = float(self.client._settings.scraping_profile.request_jitter_max_s)
+        if jitter > 0:
+            time.sleep(random.uniform(0.0, jitter))
+        return self.client.fetch_vehicle_detail_payload(inventory_id)
+
+    def _flush_db_records(
+        self,
+        *,
+        rows: list[CarRecord],
+        stats: FastSyncStats,
+        errors: list[dict[str, str]],
+        processed: int,
+        total: int,
+    ) -> int:
+        if not rows:
+            return 0
+        batch = list(rows)
+        try:
+            upsert = self.persistence.upsert_cars_batch(batch)
+            stats.cars_upserted += int(upsert.get("inserted", 0)) + int(upsert.get("updated", 0))
+            stats.images_upserted += int(upsert.get("images_upserted", 0))
+        except Exception as exc:
+            stats.cars_failed += len(batch)
+            errors.append({"vehicle_url": f"db_batch_{processed - len(batch)}", "error": str(exc)})
+            logger.exception("Fast DB apply failed processed=%s size=%s: %s", processed, len(batch), exc)
+        self._progress(
+            "fast_db_progress",
+            processed=processed,
+            total=total,
+            cars_upserted=stats.cars_upserted,
+            cars_failed=stats.cars_failed,
+            images_upserted=stats.images_upserted,
+        )
+        logger.info(
+            "Fast HTTP-first DB batch: processed=%d/%d batch=%d upserted=%d failed=%d images=%d",
+            processed,
+            total,
+            len(batch),
+            stats.cars_upserted,
+            stats.cars_failed,
+            stats.images_upserted,
+        )
+        return len(batch)
+
+
+def _looks_like_protection(exc: Exception) -> bool:
+    message = str(exc).lower()
+    return any(token in message for token in ("captcha", "antibot", "challenge", "blocked", "403", "429", "incapsula"))

iaai_scraper/worker/self_heal.py

@@ -19,6 +19,15 @@ SYNC_FULL_SCAN_DONE_KEY = "iaai:state:sync_full_scan_done"
 SYNC_LISTING_CHECKPOINT_KEY = "iaai:state:sync_listing_checkpoint"
 SYNC_LISTING_FOLLOWUP_PENDING_KEY = "iaai:state:sync_listing_followup_pending"
 SIGKILL_FALLBACK = getattr(signal, "SIGKILL", signal.SIGTERM)
+TERMINAL_PROGRESS_STAGES = {
+	"segment_done",
+	"segment_failed",
+	"segment_task_completed",
+	"segment_task_failed",
+	"segment_task_soft_timeout",
+	"sync_done",
+	"failed",
+}
 
 
 def _env_bool(name: str, default: bool) -> bool:
@@ -148,13 +157,43 @@ def _has_inflight_work(redis_client: Redis, queue_name: str) -> tuple[bool, dict
 	queue_len = _safe_int(redis_client.llen(queue_name), 0)
 	has_lock = 1 if redis_client.get(SYNC_LISTING_LOCK_KEY) else 0
 	has_task_progress = 0
-	for _ in redis_client.scan_iter(match="iaai:state:task_progress:*"):
+	progress_keys_seen = 0
-		has_task_progress = 1
+	stale_progress_deleted = 0
-		break
+	stale_seconds = max(180, min(_env_int("IAAI_SELF_HEAL_STALL_SECONDS", 720), 1800))
+	now_ts = int(time.time())
+	for key in redis_client.scan_iter(match="iaai:state:task_progress:*"):
+		progress_keys_seen += 1
+		if queue_len > 0 or has_lock == 1:
+			has_task_progress = 1
+			break
+		try:
+			payload = redis_client.get(key)
+			if not payload:
+				continue
+			data = json.loads(payload)
+			stage = str(data.get("stage") or "")
+			ts = _safe_int(data.get("ts"), 0)
+			is_terminal = stage in TERMINAL_PROGRESS_STAGES
+			is_stale = ts <= 0 or now_ts - ts > stale_seconds
+			if is_terminal or is_stale:
+				redis_client.delete(key)
+				stale_progress_deleted += 1
+				continue
+			has_task_progress = 1
+			break
+		except Exception:
+			# Битые progress payload не должны держать worker в вечном false-stall цикле.
+			try:
+				redis_client.delete(key)
+				stale_progress_deleted += 1
+			except Exception:
+				pass
 	flags = {
 		"queue_len": queue_len,
 		"has_lock": has_lock,
 		"has_task_progress": has_task_progress,
+		"progress_keys_seen": progress_keys_seen,
+		"stale_progress_deleted": stale_progress_deleted,
 	}
 	return (queue_len > 0 or has_lock == 1 or has_task_progress == 1), flags
 

iaai_scraper/worker/tasks.py

@@ -12,8 +12,7 @@ from billiard.exceptions import SoftTimeLimitExceeded
 from celery import shared_task
 from redis import Redis
 
-from ..core.config import Settings, build_fast_listing_segments_for_makes, build_listing_segments_for_makes, parse_listing_segments
+from ..core.config import Settings, parse_listing_segments
-from ..core.runtime_config import RuntimeConfig
 from ..scraper import IAAIScraper
 from ..storage.db import PersistenceService
 from ..discovery import SitemapDiscoveryError, discover_vehicle_urls_from_sitemap_with_stats
@@ -48,41 +47,23 @@ HOURLY_FAILURE_STREAK_LIMIT = 3
 HOURLY_FAILURE_STREAK_TTL_SECONDS = 6 * 60 * 60  # сброс через 6 часов
 SYNC_LISTING_FOLLOWUP_PENDING_KEY = "iaai:state:sync_listing_followup_pending"
 SYNC_LISTING_TASK_NAME = "iaai.sync_cars_feed"
+SYNC_LAST_COMPLETED_AT_KEY = "iaai:state:sync_listing_last_completed_at"
 SYNC_SEGMENT_LOCK_KEY_FMT = "iaai:locks:sync_segment:{idx}"
 SYNC_SEGMENTS_PROGRESS_KEY = "iaai:state:sync_segments_progress"
 SYNC_SEGMENTS_TOTAL_KEY = "iaai:state:sync_segments_total"
 SYNC_SEGMENTS_PROGRESS_TTL_SECONDS = 24 * 60 * 60
 TASK_PROGRESS_KEY_FMT = "iaai:state:task_progress:{task_id}"
 GLOBAL_PROGRESS_TS_KEY = "iaai:state:last_progress_ts"
-GLOBAL_DB_PROGRESS_TS_KEY = "iaai:state:last_db_progress_ts"
 SITEMAP_HOURLY_LAST_COUNT_KEY = "iaai:state:sitemap_hourly_last_count"
 SITEMAP_HOURLY_REFRESH_OFFSET_KEY = "iaai:state:sitemap_hourly_refresh_offset"
 STALL_WATCHDOG_NAVIGATION_STAGES = {
     "listing_next_page_started",
     "listing_resume_progress",
 }
-STALL_WATCHDOG_LONG_RUNNING_STAGES = {
-    "fast_listing_collected",
-    "fast_detail_progress",
-}
 STALL_WATCHDOG_NAVIGATION_GRACE_SECONDS = max(
     300,
     int(os.getenv("STALL_WATCHDOG_NAVIGATION_GRACE_SECONDS", "900")),
 )
-STALL_WATCHDOG_DETAIL_GRACE_SECONDS = max(
-    600,
-    int(os.getenv("STALL_WATCHDOG_DETAIL_GRACE_SECONDS", "1200")),
-)
-DB_IDLE_RESTART_SECONDS = max(60, int(os.getenv("IAAI_DB_IDLE_RESTART_SECONDS", "3600")))
-TERMINAL_PROGRESS_STAGES = {
-    "segment_done",
-    "segment_failed",
-    "segment_task_completed",
-    "segment_task_failed",
-    "segment_task_soft_timeout",
-    "sync_done",
-    "failed",
-}
 
 
 def _retry_with_backoff(func, *, attempts: int = 5, base_delay_s: float = 1.0):
@@ -143,19 +124,6 @@ def _update_task_progress(
 ) -> None:
     try:
         now_ts = int(time.time())
-        existing_task_started_ts: int | None = None
-        try:
-            existing_raw = redis_client.get(_task_progress_key(task_id))
-            if existing_raw:
-                existing_payload = json.loads(existing_raw)
-                existing_task_started_ts = _safe_int(existing_payload.get("task_started_ts"))
-        except Exception:
-            existing_task_started_ts = None
-        payload.setdefault("task_started_ts", existing_task_started_ts or now_ts)
-        if stage != "fast_db_progress" and "last_db_progress_ts" not in payload:
-            last_db_progress_ts = _safe_int(redis_client.get(GLOBAL_DB_PROGRESS_TS_KEY))
-            if last_db_progress_ts is not None:
-                payload["last_db_progress_ts"] = last_db_progress_ts
         data = {
             "task_id": task_id,
             "stage": stage,
@@ -173,8 +141,6 @@ def _update_task_progress(
         # Нужен, чтобы контейнер мог самовосстанавливаться при полном зависании воркера
         # (когда PID жив, но прогресс по задачам не двигается).
         pipe.set(GLOBAL_PROGRESS_TS_KEY, str(now_ts), ex=max(ttl, 7 * 24 * 60 * 60))
-        if stage == "fast_db_progress":
-            pipe.set(GLOBAL_DB_PROGRESS_TS_KEY, str(now_ts), ex=max(ttl, 7 * 24 * 60 * 60))
         pipe.execute()
     except Exception:
         logger.warning("Failed to update task progress for %s", task_id, exc_info=True)
@@ -190,8 +156,6 @@ def _clear_task_progress(redis_client: Redis, task_id: str) -> None:
 def _stall_timeout_for_progress(stage: str | None, default_timeout: int) -> int:
     if stage in STALL_WATCHDOG_NAVIGATION_STAGES:
         return max(int(default_timeout), STALL_WATCHDOG_NAVIGATION_GRACE_SECONDS)
-    if stage in STALL_WATCHDOG_LONG_RUNNING_STAGES:
-        return max(int(default_timeout), STALL_WATCHDOG_DETAIL_GRACE_SECONDS)
     return int(default_timeout)
 
 
@@ -444,7 +408,6 @@ def _start_stall_watchdog(
     stall_timeout_seconds: int,
     lock_key: str | None = None,
     lock_owner: str | None = None,
-    db_idle_restart_seconds: int | None = None,
 ) -> tuple[Event, Thread]:
     stop_event = Event()
     interval_seconds = max(5.0, min(30.0, stall_timeout_seconds / 3))
@@ -456,7 +419,6 @@ def _start_stall_watchdog(
         watchdog_born = time.monotonic()
         absolute_deadline = stall_timeout_seconds * 3
         while not stop_event.wait(interval_seconds):
-            db_idle_restart = False
             try:
                 raw = redis_client.get(key)
                 if not raw:
@@ -482,32 +444,18 @@ def _start_stall_watchdog(
                     last_ts = int(data.get("ts") or 0)
                     if not last_ts:
                         continue
-                    db_idle_restart = bool(
+                    effective_stall_timeout = _stall_timeout_for_progress(stage, stall_timeout_seconds)
-                        db_idle_restart_seconds
+                    age = int(time.time()) - last_ts
-                        and _should_restart_for_db_idle(data, db_idle_restart_seconds)
+                    if age < effective_stall_timeout:
+                        watchdog_born = time.monotonic()  # reset absolute deadline on real progress
+                        continue
+                    logger.error(
+                        "Task %s stalled for %ss at stage=%s payload=%s; cleaning up and restarting",
+                        task_id,
+                        age,
+                        stage,
+                        data,
                     )
-                    if db_idle_restart:
-                        logger.error(
-                            "Task %s has no DB writes for >%ss at segment=%s/%s stage=%s; full restart required",
-                            task_id,
-                            db_idle_restart_seconds,
-                            data.get("segment_index"),
-                            data.get("segments_total"),
-                            stage,
-                        )
-                    else:
-                        effective_stall_timeout = _stall_timeout_for_progress(stage, stall_timeout_seconds)
-                        age = int(time.time()) - last_ts
-                        if age < effective_stall_timeout:
-                            watchdog_born = time.monotonic()  # reset absolute deadline on real progress
-                            continue
-                        logger.error(
-                            "Task %s stalled for %ss at stage=%s payload=%s; cleaning up and restarting",
-                            task_id,
-                            age,
-                            stage,
-                            data,
-                        )
             except Exception:
                 logger.warning("Failed to inspect task progress for stall watchdog", exc_info=True)
                 # Если Redis тоже не отвечает дольше дедлайна — убиваем.
@@ -516,12 +464,6 @@ def _start_stall_watchdog(
                 else:
                     continue
 
-            if db_idle_restart:
-                _restart_bootstrap_from_first_segment(
-                    redis_client,
-                    reason=f"no DB writes for >{db_idle_restart_seconds}s",
-                )
-
             # ── Pre-SIGTERM cleanup: release lock so next task can run ──
             if lock_key and lock_owner:
                 try:
@@ -574,55 +516,6 @@ def _start_stall_watchdog(
     return stop_event, thread
 
 
-def _should_restart_for_db_idle(progress: dict, db_idle_restart_seconds: int) -> bool:
-    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)
-        return progress_ts > 0 and int(time.time()) - progress_ts >= timeout
-
-    segments_total = _safe_int(progress.get("segments_total"))
-    segment_index = _safe_int(progress.get("segment_index"))
-    if segments_total is None or segment_index is None:
-        return False
-    if segments_total <= 0 or segment_index >= segments_total - 1:
-        return False
-
-    now_ts = int(time.time())
-    if progress_ts <= 0:
-        return False
-
-    db_progress_ts = _safe_int(progress.get("last_db_progress_ts"))
-    if db_progress_ts is None:
-        db_progress_ts = _read_global_db_progress_ts()
-    task_started_ts = _safe_int(progress.get("task_started_ts")) or progress_ts
-    last_db_or_start_ts = max(db_progress_ts or 0, task_started_ts)
-    return now_ts - last_db_or_start_ts >= int(db_idle_restart_seconds)
-
-
-def _safe_int(value) -> int | None:
-    try:
-        return int(value)
-    except (TypeError, ValueError):
-        return None
-
-
-def _read_global_db_progress_ts() -> int | None:
-    try:
-        redis_client = _get_redis()
-        raw = redis_client.get(GLOBAL_DB_PROGRESS_TS_KEY)
-        return _safe_int(raw)
-    except Exception:
-        return None
-
-
 _persistence_instance: PersistenceService | None = None
 
 
@@ -830,21 +723,6 @@ def _save_last_completed_segment(redis_client: Redis, segment_index: int) -> Non
         logger.warning("Failed to save sync listing checkpoint", exc_info=True)
 
 
-def _restart_bootstrap_from_first_segment(redis_client: Redis, *, reason: str) -> None:
-    """Clear bootstrap checkpoint so the next full scan starts from segment 1."""
-    try:
-        pipe = redis_client.pipeline()
-        pipe.delete(SYNC_LISTING_CHECKPOINT_KEY)
-        pipe.delete(SYNC_LISTING_FOLLOWUP_PENDING_KEY)
-        pipe.delete(GLOBAL_PROGRESS_TS_KEY)
-        pipe.delete(GLOBAL_DB_PROGRESS_TS_KEY)
-        pipe.set(SYNC_FULL_SCAN_DONE_KEY, "0")
-        pipe.execute()
-        logger.error("Bootstrap restart requested from segment 1: %s", reason)
-    except Exception:
-        logger.warning("Failed to reset bootstrap checkpoint for DB-idle restart", exc_info=True)
-
-
 def _clear_sync_checkpoint(redis_client: Redis) -> None:
     try:
         redis_client.delete(SYNC_LISTING_CHECKPOINT_KEY)
@@ -867,6 +745,31 @@ def _clear_followup_pending(redis_client: Redis) -> None:
         logger.warning("Failed to clear follow-up pending flag", exc_info=True)
 
 
+def _mark_sync_completed(redis_client: Redis) -> None:
+    try:
+        redis_client.set(SYNC_LAST_COMPLETED_AT_KEY, str(int(time.time())), ex=7 * 24 * 60 * 60)
+    except Exception:
+        logger.warning("Failed to mark sync completion timestamp", exc_info=True)
+
+
+def _seconds_until_next_allowed_sync(redis_client: Redis, settings: Settings) -> int:
+    min_interval = max(0, int(settings.celery.beat_sync_interval_minutes * 60))
+    if min_interval <= 0:
+        return 0
+    try:
+        if not _is_full_scan_done(redis_client):
+            return 0
+        completed_raw = redis_client.get(SYNC_LAST_COMPLETED_AT_KEY)
+        if not completed_raw:
+            return 0
+        completed_at = int(float(completed_raw))
+    except Exception:
+        logger.warning("Failed to inspect next allowed sync time", exc_info=True)
+        return 0
+    elapsed = int(time.time()) - completed_at
+    return max(0, min_interval - elapsed)
+
+
 def _bump_bootstrap_failure_streak(
     redis_client: Redis,
     *,
@@ -975,7 +878,6 @@ def _start_lock_heartbeat(
     key: str,
     owner_token: str,
     ttl_seconds: int,
-    stop_on_lost: bool = True,
 ) -> tuple[Event, Thread]:
     stop_event = Event()
     interval_seconds = max(5.0, min(30.0, ttl_seconds / 3))
@@ -986,10 +888,7 @@ def _start_lock_heartbeat(
             refreshed = _refresh_lock_if_owner(redis_client, key, owner_token, ttl_seconds)
             if refreshed is False:
                 logger.warning("Lost sync_listing lock ownership for %s", owner_token)
-                if stop_on_lost:
+                return
-                    return
-                consecutive_failures += 1
-                continue
             if refreshed is None:
                 consecutive_failures += 1
                 if consecutive_failures >= 5:
@@ -1030,49 +929,6 @@ def _mark_segment_completed(redis_client: Redis, segment_index: int) -> tuple[in
         return 0, 0
 
 
-def _build_listing_segments(settings: Settings) -> list[dict[str, str | int | None]]:
-    """Возвращает сегменты листинга с учётом runtime_config.
-
-    При IAAI_LISTING_SEGMENTS=runtime сегменты строятся из filters.brands / filters.include.brands.
-    Остальные значения IAAI_LISTING_SEGMENTS сохраняют прежнее поведение: auto или JSON.
-    """
-    filtered_urls = settings.listing.filtered_search_urls
-    if len(filtered_urls) > 1:
-        return [
-            {"make": None, "year_min": None, "year_max": None, "listing_url": url}
-            for url in filtered_urls
-        ]
-    if len(filtered_urls) == 1:
-        return []
-
-    raw_segments = settings.listing.listing_segments_json.strip()
-    if raw_segments.casefold() != "runtime":
-        return parse_listing_segments(raw_segments)
-
-    parsed_override = parse_listing_segments(raw_segments)
-    if parsed_override:
-        return parsed_override
-
-    runtime_config = RuntimeConfig.from_file(settings.runtime_config_file)
-    brands = runtime_config.filters.include.brands
-
-    # 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))
-    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"
-        )
-    return segments
-
-
 @shared_task(
     name="iaai_scraper.worker.tasks.sync_segment_task",
     queue=IAAI_SYNC_QUEUE,
@@ -1108,10 +964,7 @@ def sync_segment_task(
     seg_make = segment.get("make")
     seg_year_min = segment.get("year_min")
     seg_year_max = segment.get("year_max")
-    seg_listing_url = segment.get("listing_url")
     seg_label = f"{seg_make or 'ALL'}"
-    if seg_listing_url:
-        seg_label = "FILTERED_SEARCH"
     if seg_year_min is not None or seg_year_max is not None:
         seg_label += f" ({seg_year_min}-{seg_year_max})"
 
@@ -1153,12 +1006,14 @@ def sync_segment_task(
                         **meta,
                     )
                 )
+                base_url = scraper.settings.listing.cars_url
+                seg_url = scraper._build_segment_listing_url(base_url, seg_make) if seg_make else None
                 return scraper.sync_listing(
-                    make=seg_make,
+                    make=None if seg_url else seg_make,
                     model=None,
                     lane=lane,
                     only_new=only_new,
-                    listing_url=str(seg_listing_url) if seg_listing_url else None,
+                    listing_url=seg_url,
                     year_min=seg_year_min,
                     year_max=seg_year_max,
                     skip_mark_sold=True,
@@ -1310,6 +1165,11 @@ def sync_listing_task(
     watchdog_stop: Event | None = None
     watchdog_thread: Thread | None = None
     force_bootstrap_full_scan = False
+    settings = Settings()
+    followup_min_delay_seconds = max(
+        0,
+        int(os.getenv("IAAI_SYNC_FOLLOWUP_MIN_DELAY_SECONDS", str(settings.celery.beat_sync_interval_minutes * 60))),
+    )
 
     def _enqueue_bootstrap_followup(
         reason: str,
@@ -1317,6 +1177,7 @@ def sync_listing_task(
         *,
         count_as_failure: bool = False,
     ) -> None:
+        delay_seconds = max(int(delay_seconds), followup_min_delay_seconds)
         flag_ttl = max(lock_ttl, delay_seconds + 300)
         if not _try_set_followup_pending(redis_client, ttl_seconds=flag_ttl):
             logger.info(
@@ -1391,7 +1252,18 @@ def sync_listing_task(
         }
 
     try:
-        settings = Settings()
+        next_allowed_delay = _seconds_until_next_allowed_sync(redis_client, settings)
+        if next_allowed_delay > 0:
+            logger.info(
+                "sync_listing_task skipped: previous full run finished recently; next run allowed in %ss",
+                next_allowed_delay,
+            )
+            return {
+                "status": "skipped",
+                "reason": "next_sync_not_due_yet",
+                "retry_after_seconds": next_allowed_delay,
+                "task_id": task_id,
+            }
 
         # Любой реально стартовавший sync_listing снимает pending-флаг followup,
         # чтобы watchdog/continuation могли корректно планировать следующий run
@@ -1405,10 +1277,16 @@ def sync_listing_task(
             SYNC_LISTING_LOCK_KEY,
             owner_token,
             lock_ttl,
-            stop_on_lost=False,
         )
         stall_timeout = max(120, int(settings.celery.task_stall_timeout_seconds))
         progress_ttl = max(lock_ttl + 120, stall_timeout + 120)
+        watchdog_stop, watchdog_thread = _start_stall_watchdog(
+            redis_client,
+            task_id=task_id,
+            stall_timeout_seconds=stall_timeout,
+            lock_key=SYNC_LISTING_LOCK_KEY,
+            lock_owner=owner_token,
+        )
         _update_task_progress(
             redis_client,
             task_id=task_id,
@@ -1418,17 +1296,7 @@ def sync_listing_task(
 
         full_scan_done_before_run = _is_full_scan_done(redis_client)
         always_full_scan = bool(settings.discovery.always_full_scan)
-        explicit_filtered_run = bool(
+        force_bootstrap_full_scan = always_full_scan or (not full_scan_done_before_run)
-            make is not None
-            or model is not None
-            or (limit is not None and int(limit or 0) > 0)
-            or only_new is True
-        )
-        force_bootstrap_full_scan = (always_full_scan or (not full_scan_done_before_run)) and not explicit_filtered_run
-        if explicit_filtered_run and not full_scan_done_before_run:
-            logger.info(
-                "Explicit sync_listing request detected; honoring make/model/limit/only_new before bootstrap full scan is complete",
-            )
         effective_limit = None if force_bootstrap_full_scan else limit
         effective_only_new = False if force_bootstrap_full_scan else only_new
         hourly_mode = settings.discovery.hourly_mode.strip().lower()
@@ -1445,31 +1313,13 @@ def sync_listing_task(
             and not effective_only_new
             and not always_full_scan
         )
-        # Определяем сегменты из конфига/env или из runtime_config при IAAI_LISTING_SEGMENTS=runtime.
+        use_hourly_sitemap_sync = (not always_full_scan) and full_scan_done_before_run and prefer_sitemap_mainline
-        filtered_listing_urls = settings.listing.filtered_search_urls
-        filtered_listing_url = filtered_listing_urls[0] if filtered_listing_urls else None
-        segments = _build_listing_segments(settings)
-
-        watchdog_stop, watchdog_thread = _start_stall_watchdog(
-            redis_client,
-            task_id=task_id,
-            stall_timeout_seconds=stall_timeout,
-            lock_key=SYNC_LISTING_LOCK_KEY,
-            lock_owner=owner_token,
-            db_idle_restart_seconds=(DB_IDLE_RESTART_SECONDS if segments else None),
-        )
-        use_hourly_sitemap_sync = (
-            (not always_full_scan)
-            and full_scan_done_before_run
-            and prefer_sitemap_mainline
-            and not segments
-        )
 
         # Segment-level checkpoint: хранит индекс последнего ПОЛНОСТЬЮ пройденного сегмента.
         # Используется только во время bootstrap для пропуска уже обработанных сегментов.
         # Никаких page-level resume — внутри сегмента всегда стартуем с page 1.
         last_completed_segment: int | None = None
-        if force_bootstrap_full_scan and (always_full_scan or not full_scan_done_before_run):
+        if force_bootstrap_full_scan:
             last_completed_segment = _load_last_completed_segment(redis_client)
         else:
             # После завершения bootstrap чекпоинт не нужен никогда.
@@ -1611,11 +1461,14 @@ def sync_listing_task(
 
         self.update_state(state="STARTED", meta={"stage": "sync_listing_started", "task_id": task_id})
 
+        # Определяем сегменты из конфига.
+        segments = parse_listing_segments(settings.listing.listing_segments_json)
         use_segmented = (
             bool(segments)
             and make is None
             and model is None
-            and not use_hourly_sitemap_sync
+            and not prefer_sitemap_mainline
+            and discovery_mode != "sitemap"
         )
 
         if prefer_sitemap_mainline:
@@ -1733,7 +1586,7 @@ def sync_listing_task(
                         start_page=1,
                         progress_callback=(
                             (lambda seg_idx: _save_last_completed_segment(redis_client, seg_idx))
-                            if force_bootstrap_full_scan and (always_full_scan or not full_scan_done_before_run) else None
+                            if force_bootstrap_full_scan else None
                         ),
                     )
                 return scraper.sync_listing(
@@ -1742,7 +1595,6 @@ def sync_listing_task(
                     lane=lane,
                     limit=effective_limit,
                     only_new=effective_only_new,
-                    listing_url=filtered_listing_url,
                 )
 
         result = _run_browser_job(_job)
@@ -1832,6 +1684,7 @@ def sync_listing_task(
             summary["cars_failed"],
             summary["failures_count"],
         )
+        _mark_sync_completed(redis_client)
         return summary
 
     except SoftTimeLimitExceeded:

runtime_config.json

@@ -1,6 +1,6 @@
 {
   "sync": {
-    "name": null,
+    "name": "https://www.iaai.com/Search?url=B%2bU066dM8%2flZtRvnzTwIEi8Pib9%2fM2fLpCTQDIgQRm4%3d",
     "ids_initial_size": null,
     "ids_next_size": null,
     "ids_max_pages": null,

status_vps.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+import os
+import paramiko
+
+HOST = "2.26.123.84"
+USER = "root"
+PASSWORD = os.environ["VPS_PASSWORD"]
+
+commands = [
+    "cd /root/iaai-parser && docker compose ps",
+    "cd /root/iaai-parser && docker compose exec -T worker celery -A iaai_scraper.worker.celery_app inspect active reserved scheduled",
+    "docker exec -i iaai-postgres psql -U iaai -d iaai_scraper -c \"SELECT now() AS ts, (SELECT count(*) FROM iaai_cars) AS cars, (SELECT count(*) FROM iaai_sync_runs) AS runs, (SELECT status FROM iaai_sync_runs ORDER BY id DESC LIMIT 1) AS last_status, (SELECT cars_upserted FROM iaai_sync_runs ORDER BY id DESC LIMIT 1) AS last_upserted, (SELECT cars_failed FROM iaai_sync_runs ORDER BY id DESC LIMIT 1) AS last_failed, (SELECT started_at FROM iaai_sync_runs ORDER BY id DESC LIMIT 1) AS last_started;\"",
+    "docker logs --since 10m iaai-parser-worker-1 2>&1 | tail -n 160",
+]
+
+client = paramiko.SSHClient()
+client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+client.connect(
+    HOST,
+    username=USER,
+    password=PASSWORD,
+    look_for_keys=False,
+    allow_agent=False,
+    timeout=30,
+    banner_timeout=30,
+    auth_timeout=30,
+)
+try:
+    for command in commands:
+        print(f"\n=== REMOTE_RUN {command} ===", flush=True)
+        stdin, stdout, stderr = client.exec_command(command, timeout=180)
+        exit_code = stdout.channel.recv_exit_status()
+        print(stdout.read().decode("utf-8", "replace"), end="")
+        print(stderr.read().decode("utf-8", "replace"), end="")
+        print(f"\n=== EXIT {exit_code} ===", flush=True)
+finally:
+    client.close()

tests/test_self_heal.py

@@ -72,6 +72,98 @@ class TestSelfHeal(unittest.TestCase):
 
 		self.assertTrue(active)
 
+	@patch("iaai_scraper.worker.self_heal.time.time", return_value=5000)
+	def test_has_inflight_work_deletes_stale_terminal_progress_without_work(self, _time_mock) -> None:
+		redis_client = MagicMock()
+		redis_client.llen.return_value = 0
+		redis_client.get.side_effect = lambda key: {
+			self_heal.SYNC_LISTING_LOCK_KEY: None,
+			"iaai:state:task_progress:old": json.dumps(
+				{
+					"task_id": "old",
+					"stage": "segment_done",
+					"ts": 1000,
+					"segment_full_scan_completed": True,
+				}
+			),
+		}.get(key)
+		redis_client.scan_iter.return_value = ["iaai:state:task_progress:old"]
+
+		has_inflight, flags = self_heal._has_inflight_work(redis_client, self_heal.IAAI_SYNC_QUEUE)
+
+		self.assertFalse(has_inflight)
+		self.assertEqual(flags["has_task_progress"], 0)
+		redis_client.delete.assert_called_once_with("iaai:state:task_progress:old")
+
+	@patch("iaai_scraper.worker.self_heal.time.time", return_value=1010)
+	def test_has_inflight_work_keeps_recent_non_terminal_progress(self, _time_mock) -> None:
+		redis_client = MagicMock()
+		redis_client.llen.return_value = 0
+		redis_client.get.side_effect = lambda key: {
+			self_heal.SYNC_LISTING_LOCK_KEY: None,
+			"iaai:state:task_progress:active": json.dumps(
+				{
+					"task_id": "active",
+					"stage": "fast_listing_collected",
+					"ts": 1000,
+				}
+			),
+		}.get(key)
+		redis_client.scan_iter.return_value = ["iaai:state:task_progress:active"]
+
+		has_inflight, flags = self_heal._has_inflight_work(redis_client, self_heal.IAAI_SYNC_QUEUE)
+
+		self.assertTrue(has_inflight)
+		self.assertEqual(flags["has_task_progress"], 1)
+		redis_client.delete.assert_not_called()
+
+	@patch("iaai_scraper.worker.self_heal.time.time", return_value=5000)
+	def test_has_inflight_work_deletes_stale_non_terminal_progress_without_work(self, _time_mock) -> None:
+		redis_client = MagicMock()
+		redis_client.llen.return_value = 0
+		redis_client.get.side_effect = lambda key: {
+			self_heal.SYNC_LISTING_LOCK_KEY: None,
+			"iaai:state:task_progress:stale": json.dumps(
+				{
+					"task_id": "stale",
+					"stage": "segment_started",
+					"ts": 1000,
+					"segment_index": 11,
+				}
+			),
+		}.get(key)
+		redis_client.scan_iter.return_value = ["iaai:state:task_progress:stale"]
+
+		has_inflight, flags = self_heal._has_inflight_work(redis_client, self_heal.IAAI_SYNC_QUEUE)
+
+		self.assertFalse(has_inflight)
+		self.assertEqual(flags["has_task_progress"], 0)
+		self.assertEqual(flags["stale_progress_deleted"], 1)
+		redis_client.delete.assert_called_once_with("iaai:state:task_progress:stale")
+
+	@patch("iaai_scraper.worker.self_heal.time.time", return_value=5000)
+	def test_has_inflight_work_keeps_stale_progress_when_lock_exists(self, _time_mock) -> None:
+		redis_client = MagicMock()
+		redis_client.llen.return_value = 0
+		redis_client.get.side_effect = lambda key: {
+			self_heal.SYNC_LISTING_LOCK_KEY: "owner",
+			"iaai:state:task_progress:stale": json.dumps(
+				{
+					"task_id": "stale",
+					"stage": "segment_started",
+					"ts": 1000,
+				}
+			),
+		}.get(key)
+		redis_client.scan_iter.return_value = ["iaai:state:task_progress:stale"]
+
+		has_inflight, flags = self_heal._has_inflight_work(redis_client, self_heal.IAAI_SYNC_QUEUE)
+
+		self.assertTrue(has_inflight)
+		self.assertEqual(flags["has_lock"], 1)
+		self.assertEqual(flags["has_task_progress"], 1)
+		redis_client.delete.assert_not_called()
+
 	@patch("iaai_scraper.worker.self_heal.time.sleep", return_value=None)
 	@patch("iaai_scraper.worker.self_heal.os.kill")
 	@patch("builtins.open")

vps_monitor.sh

@@ -0,0 +1,179 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+# VPS monitor for iaai-parser. Run on the VPS from any directory:
+#   bash /root/iaai-parser/vps_monitor.sh
+# Live mode:
+#   watch -n 30 bash /root/iaai-parser/vps_monitor.sh
+
+APP_DIR="${APP_DIR:-/root/iaai-parser}"
+cd "$APP_DIR"
+
+LINE='============================================================'
+SMALL='------------------------------------------------------------'
+
+section() {
+  echo
+  echo "$LINE"
+  echo "$1"
+  echo "$LINE"
+}
+
+subsection() {
+  echo
+  echo "$SMALL"
+  echo "$1"
+  echo "$SMALL"
+}
+
+run_psql() {
+  docker compose exec -T postgres psql -U iaai -d iaai_scraper "$@"
+}
+
+run_redis() {
+  docker compose exec -T redis redis-cli "$@"
+}
+
+now_utc="$(date -u '+%Y-%m-%d %H:%M:%S UTC')"
+echo "IAAI VPS monitor | $now_utc | app=$APP_DIR"
+
+section "1) Services / restarts"
+docker compose ps
+
+subsection "Container restart counters"
+for svc in api worker beat postgres redis; do
+  cid="$(docker compose ps -q "$svc" 2>/dev/null || true)"
+  if [[ -z "$cid" ]]; then
+    printf '%-10s %s\n' "$svc" "not found"
+    continue
+  fi
+  name="$(docker inspect --format '{{.Name}}' "$cid" | sed 's#^/##')"
+  restarts="$(docker inspect --format '{{.RestartCount}}' "$cid")"
+  started="$(docker inspect --format '{{.State.StartedAt}}' "$cid")"
+  status="$(docker inspect --format '{{.State.Status}}' "$cid")"
+  health="$(docker inspect --format '{{if .State.Health}}{{.State.Health.Status}}{{else}}n/a{{end}}' "$cid")"
+  printf '%-22s restarts=%-3s started=%s status=%s health=%s\n' "$name" "$restarts" "$started" "$status" "$health"
+done
+
+subsection "Startup / restart log markers"
+docker compose logs --since '24h' worker beat 2>/dev/null \
+  | grep -E 'Worker ready|startup sync|dispatching initial|Scheduler: Sending due task|warm shutdown|Restoring|mingle:|ready\.|ERROR|CRITICAL' \
+  | tail -80 || true
+
+section "2) Main DB result"
+run_psql -c "
+select
+  count(*) as total_cars,
+  count(*) filter (where is_sold = false) as active_cars,
+  count(*) filter (where is_sold = true) as sold_cars,
+  count(*) filter (where last_seen_at > now() - interval '5 minutes') as touched_5m,
+  count(*) filter (where last_seen_at > now() - interval '15 minutes') as touched_15m,
+  count(*) filter (where last_seen_at > now() - interval '60 minutes') as touched_60m,
+  max(last_seen_at) as last_seen
+from iaai_cars;
+"
+
+subsection "Writes from sync_runs"
+run_psql -c "
+select
+  coalesce(sum(cars_upserted) filter (where started_at > now() - interval '1 hour'), 0) as cars_upserted_1h,
+  coalesce(sum(cars_upserted) filter (where started_at > now() - interval '6 hours'), 0) as cars_upserted_6h,
+  coalesce(sum(cars_upserted) filter (where started_at::date = now()::date), 0) as cars_upserted_today,
+  count(*) filter (where status = 'running') as running_runs,
+  count(*) filter (where status = 'success' and started_at > now() - interval '24 hours') as success_runs_24h,
+  count(*) filter (where status = 'failed' and started_at > now() - interval '24 hours') as failed_runs_24h
+from iaai_sync_runs;
+"
+
+section "3) Segments / passes"
+subsection "Redis checkpoint / active progress"
+checkpoint="$(run_redis GET iaai:state:sync_listing_checkpoint || true)"
+segments_done="$(run_redis SCARD iaai:state:sync_segments_done 2>/dev/null || echo 0)"
+segments_total="$(run_redis GET iaai:state:sync_segments_total || true)"
+full_scan_done="$(run_redis GET iaai:state:sync_full_scan_done || true)"
+queue_len="$(run_redis LLEN iaai_sync || true)"
+lock_value="$(run_redis GET iaai:locks:sync_listing || true)"
+echo "checkpoint_last_completed_segment=${checkpoint:-empty}"
+echo "segments_done_set_count=${segments_done:-0}"
+echo "segments_total=${segments_total:-unknown}"
+echo "full_scan_done=${full_scan_done:-empty}"
+echo "queue_len=${queue_len:-unknown}"
+echo "sync_lock=${lock_value:-empty}"
+
+echo
+echo "Active/latest task progress keys:"
+progress_keys="$(run_redis --scan --pattern 'iaai:state:task_progress:*' || true)"
+if [[ -z "$progress_keys" ]]; then
+  echo "no task_progress keys"
+else
+  while IFS= read -r key; do
+    [[ -z "$key" ]] && continue
+    echo "$key"
+    run_redis GET "$key" || true
+    echo
+  done <<< "$progress_keys"
+fi
+
+subsection "Completed full passes by worker logs"
+passes_24h="$(docker compose logs --since '24h' worker 2>/dev/null | grep -Ec 'Segment 65/65 done|Segment 64/65 done: RIMAC|Bootstrap full scan completed|sync_listing_task completed: status=success' || true)"
+passes_6h="$(docker compose logs --since '6h' worker 2>/dev/null | grep -Ec 'Segment 65/65 done|Segment 64/65 done: RIMAC|Bootstrap full scan completed|sync_listing_task completed: status=success' || true)"
+echo "estimated_completed_passes_6h=$passes_6h"
+echo "estimated_completed_passes_24h=$passes_24h"
+
+subsection "Recent segment events"
+docker compose logs --tail=220 worker 2>/dev/null \
+  | grep -E 'Segment [0-9]+/[0-9]+|Fast HTTP-first listing collected|Fast HTTP-first DB batch|sync_listing_task completed|Bootstrap full scan completed|Bootstrap continuation stopped|sync_already_running' \
+  | tail -90 || true
+
+section "4) Errors / warnings"
+subsection "Failed sync_runs last 24h"
+run_psql -c "
+select id, status, cars_upserted, cars_failed, started_at, finished_at, left(coalesce(error_summary, ''), 180) as error_summary
+from iaai_sync_runs
+where status <> 'success' and started_at > now() - interval '24 hours'
+order by id desc
+limit 20;
+"
+
+subsection "Recent ERROR/WARNING logs"
+docker compose logs --since '6h' worker beat api 2>/dev/null \
+  | grep -E 'ERROR|WARNING|CRITICAL|Traceback|TimeLimitExceeded|SoftTimeLimitExceeded|Read timed out|anti-bot|circuit breaker' \
+  | tail -120 || true
+
+section "5) Schedule / hourly behavior"
+subsection "Relevant environment"
+docker compose exec -T worker sh -lc 'env | grep -E "IAAI_STARTUP_SYNC_ENABLED|CELERY_BEAT_SYNC_INTERVAL_MINUTES|IAAI_LISTING_SEGMENTS|IAAI_ALWAYS_FULL_SCAN|IAAI_FULL_SCAN_ON_STARTUP|IAAI_HOURLY_MODE|IAAI_HOURLY_REFRESH_BATCH_SIZE|IAAI_SELF_HEAL_STALL_SECONDS|IAAI_DB_IDLE_RESTART_SECONDS|CELERY_TASK_SOFT_TIME_LIMIT|CELERY_TASK_TIME_LIMIT|CELERY_BROKER_VISIBILITY_TIMEOUT" | sort' || true
+
+subsection "Beat sends / skipped duplicates"
+docker compose logs --since '12h' beat worker 2>/dev/null \
+  | grep -E 'Scheduler: Sending due task periodic-sync-listing|sync_already_running|sync_listing_task hourly|Hourly mode|Worker ready|startup sync dispatch disabled|dispatching initial' \
+  | tail -120 || true
+
+section "6) Sold cars"
+run_psql -c "
+select
+  count(*) filter (where is_sold = true) as sold_total,
+  count(*) filter (where is_sold = true and last_seen_at > now() - interval '24 hours') as sold_seen_24h,
+  min(last_seen_at) filter (where is_sold = true) as oldest_sold_seen,
+  max(last_seen_at) filter (where is_sold = true) as newest_sold_seen
+from iaai_cars;
+"
+
+subsection "Top sold brands"
+run_psql -c "
+select brand, count(*) as sold
+from iaai_cars
+where is_sold = true
+group by brand
+order by sold desc
+limit 15;
+"
+
+subsection "Sold marking logs"
+docker compose logs --since '24h' worker 2>/dev/null \
+  | grep -E 'Marked [0-9]+ cars as sold|sold_marked|Fast sold reconcile failed|mark_sold safety abort' \
+  | tail -80 || true
+
+section "7) Quick health hints"
+echo "OK if: worker/beat healthy, queue_len is 0 or small, one sync_lock exists during active run, checkpoint grows, failed_runs_24h is low."
+echo "If IAAI_STARTUP_SYNC_ENABLED is empty/true, worker restart can dispatch an immediate sync. Set it to false to rely only on hourly beat."