Merge branch 'main' into prod-preparation

- docker-compose: switch worker to --pool=solo --concurrency=1 --max-tasks-per-child=1
- tasks.py: remove ThreadPoolExecutor wrapper from _run_browser_job (greenlet crash)
- scraper.py: browser fallback runs sequentially instead of ThreadPoolExecutor
- scraper.py: add gc.collect() after scraper close to prevent memory leaks

.env.example

@@ -55,3 +55,6 @@ CELERY_BROKER_VISIBILITY_TIMEOUT=90000
 CELERY_WORKER_CONCURRENCY=1
 CELERY_BEAT_SYNC_INTERVAL_MINUTES=60
 CELERY_BEAT_SYNC_LIMIT=0
+IAAI_ALWAYS_FULL_SCAN=true
+SYNC_LISTING_BOOTSTRAP_CONTINUATION_STREAK_LIMIT=6
+SYNC_LISTING_BOOTSTRAP_CONTINUATION_STREAK_TTL_SECONDS=21600

Dockerfile

@@ -13,8 +13,8 @@ RUN pip install --no-cache-dir uv \
     && pip install --no-cache-dir -r /tmp/requirements.txt \
     && pip install --no-cache-dir playwright-stealth
 
-# Install both Chromium and Firefox. Chromium is the default engine in Docker
+# Ставим Chromium и Firefox.
-# because it works more reliably with the current IAAI listing page.
+# По умолчанию используем Chromium.
 RUN python -m playwright install chromium firefox
 
 COPY . .
@@ -27,6 +27,6 @@ STOPSIGNAL SIGINT
 
 USER app
 
-# По умолчанию  API, но worker/beat переопределяют CMD в docker-compose
+# По умолчанию запускаем API.
 ENTRYPOINT ["./entrypoint.sh"]
 CMD ["uvicorn", "iaai_scraper.api.app:app", "--host", "0.0.0.0", "--port", "8000"]

README.md

@@ -124,9 +124,10 @@ docker compose ps
 - `worker` имеет healthcheck через `celery inspect ping`
 - `beat` имеет healthcheck по файлу `celerybeat-schedule`
 
+
 ## API
 
-**Здоровье и статистика:**
+**Статистика:**
 - `GET /health` — статус сервиса и подключения к БД
 - `GET /api/v1/stats` — сколько машин и картинок в базе, топ брендов
 
@@ -242,11 +243,6 @@ pytest -q
 
 ## `pyproject.toml` + `uv.lock`
 
-Проект переведён на современную схему зависимостей:
-
-- `pyproject.toml` — декларация зависимостей и метаданных проекта
-- `uv.lock` — зафиксированные версии для воспроизводимых установок
-
 Локально можно использовать:
 
 ```bash
@@ -254,10 +250,6 @@ uv sync
 uv run pytest -q
 ```
 
-## Runtime-фильтры
-
-Файл `runtime_config.json` управляет runtime-поведением sync и фильтрацией автомобилей.
-
 ### Секция `sync`
 
 Поддерживаются поля:

alembic/env.py

@@ -14,6 +14,7 @@ from iaai_scraper.core.config import Settings
 from iaai_scraper.storage.models import Base
 
 config = context.config
+VERSION_TABLE = "iaai_parser_alembic_version"
 
 # Logging
 if config.config_file_name is not None:
@@ -32,6 +33,7 @@ def run_migrations_offline() -> None:
     context.configure(
         url=url,
         target_metadata=target_metadata,
+        version_table=VERSION_TABLE,
         literal_binds=True,
         dialect_opts={"paramstyle": "named"},
     )
@@ -47,7 +49,11 @@ def run_migrations_online() -> None:
         poolclass=pool.NullPool,
     )
     with connectable.connect() as connection:
-        context.configure(connection=connection, target_metadata=target_metadata)
+        context.configure(
+            connection=connection,
+            target_metadata=target_metadata,
+            version_table=VERSION_TABLE,
+        )
         with context.begin_transaction():
             context.run_migrations()
 

alembic/versions/001_initial.py

@@ -1,4 +1,4 @@
-# Начальная схема: cars, images, sync_runs
+# Начальная схема: iaai_cars, iaai_images, iaai_sync_runs
 from typing import Sequence, Union
 
 from alembic import op
@@ -10,70 +10,94 @@ branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None
 
 
+def _schema_name() -> str | None:
+    bind = op.get_bind()
+    return "public" if bind.dialect.name == "postgresql" else None
+
+
+def _table_exists(table_name: str) -> bool:
+    inspector = sa.inspect(op.get_bind())
+    return table_name in inspector.get_table_names(schema=_schema_name())
+
+
+def _index_exists(table_name: str, index_name: str) -> bool:
+    if not _table_exists(table_name):
+        return False
+    inspector = sa.inspect(op.get_bind())
+    indexes = inspector.get_indexes(table_name, schema=_schema_name())
+    return any(index.get("name") == index_name for index in indexes)
+
+
 def upgrade() -> None:
-    # Таблица cars — аукционные машины с IAAI
+    # Таблица iaai_cars — аукционные машины с IAAI
-    op.create_table(
+    if not _table_exists("iaai_cars"):
-        "cars",
+        op.create_table(
-        sa.Column("id", sa.BigInteger, primary_key=True, autoincrement=True),
+            "iaai_cars",
-        sa.Column("parser_id", sa.String(50), nullable=False, unique=True),
+            sa.Column("id", sa.BigInteger, primary_key=True, autoincrement=True),
-        sa.Column("brand", sa.String(50), nullable=False),
+            sa.Column("parser_id", sa.String(50), nullable=False, unique=True),
-        sa.Column("model", sa.String(50), nullable=False),
+            sa.Column("brand", sa.String(50), nullable=False),
-        sa.Column("year", sa.Integer, nullable=True),
+            sa.Column("model", sa.String(50), nullable=False),
-        sa.Column("price", sa.BigInteger, nullable=True),
+            sa.Column("year", sa.Integer, nullable=True),
-        sa.Column("currency", sa.String(10), nullable=False, server_default="USD"),
+            sa.Column("price", sa.BigInteger, nullable=True),
-        sa.Column("mileage", sa.Integer, nullable=False, server_default="0"),
+            sa.Column("currency", sa.String(10), nullable=False, server_default="USD"),
-        sa.Column("country", sa.String(10), nullable=False, server_default="NA"),
+            sa.Column("mileage", sa.Integer, nullable=False, server_default="0"),
-        sa.Column("is_sold", sa.Boolean, nullable=False, server_default=sa.text("false")),
+            sa.Column("country", sa.String(10), nullable=False, server_default="NA"),
-        sa.Column("color", sa.String, nullable=False, server_default="other"),
+            sa.Column("is_sold", sa.Boolean, nullable=False, server_default=sa.text("false")),
-        sa.Column("drive", sa.String(10), nullable=True),
+            sa.Column("color", sa.String, nullable=False, server_default="other"),
-        sa.Column("gearbox", sa.String(10), nullable=True),
+            sa.Column("drive", sa.String(10), nullable=True),
-        sa.Column("steering_wheel", sa.String(10), nullable=True),
+            sa.Column("gearbox", sa.String(10), nullable=True),
-        sa.Column("body_type", sa.String(20), nullable=False, server_default="OTHER"),
+            sa.Column("steering_wheel", sa.String(10), nullable=True),
-        sa.Column("engine_volume", sa.Integer, nullable=True),
+            sa.Column("body_type", sa.String(20), nullable=False, server_default="OTHER"),
-        sa.Column("selling_type", sa.String(20), nullable=False, server_default="NA"),
+            sa.Column("engine_volume", sa.Integer, nullable=True),
-        sa.Column("one_owner", sa.Boolean, nullable=False, server_default=sa.text("false")),
+            sa.Column("selling_type", sa.String(20), nullable=False, server_default="NA"),
-        sa.Column("new_car", sa.Boolean, nullable=False, server_default=sa.text("false")),
+            sa.Column("one_owner", sa.Boolean, nullable=False, server_default=sa.text("false")),
-        sa.Column("is_hidden", sa.Boolean, nullable=False, server_default=sa.text("false")),
+            sa.Column("new_car", sa.Boolean, nullable=False, server_default=sa.text("false")),
-        sa.Column("origin", sa.String(20), nullable=False, server_default="NA"),
+            sa.Column("is_hidden", sa.Boolean, nullable=False, server_default=sa.text("false")),
-        sa.Column("origin_url", sa.String, nullable=False),
+            sa.Column("origin", sa.String(20), nullable=False, server_default="NA"),
-        sa.Column("origin_id", sa.String, nullable=False, unique=True),
+            sa.Column("origin_url", sa.String, nullable=False),
-        sa.Column("is_damaged", sa.Boolean, nullable=False, server_default=sa.text("false")),
+            sa.Column("origin_id", sa.String, nullable=False, unique=True),
-        sa.Column("evaluation", sa.String, nullable=True),
+            sa.Column("is_damaged", sa.Boolean, nullable=False, server_default=sa.text("false")),
-        sa.Column("non_smoking", sa.Boolean, nullable=False, server_default=sa.text("true")),
+            sa.Column("evaluation", sa.String, nullable=True),
-        sa.Column("rental", sa.Boolean, nullable=False, server_default=sa.text("false")),
+            sa.Column("non_smoking", sa.Boolean, nullable=False, server_default=sa.text("true")),
-        sa.Column("repair_history", sa.Boolean, nullable=False, server_default=sa.text("false")),
+            sa.Column("rental", sa.Boolean, nullable=False, server_default=sa.text("false")),
-        sa.Column("slug", sa.String, nullable=False),
+            sa.Column("repair_history", sa.Boolean, nullable=False, server_default=sa.text("false")),
-        sa.Column("last_seen_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+            sa.Column("slug", sa.String, nullable=False),
-    )
+            sa.Column("last_seen_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
-    op.create_index("ix_cars_origin_url", "cars", ["origin_url"])
+        )
-    op.create_index("ix_cars_origin_id", "cars", ["origin_id"], unique=True)
 
-    # Таблица images — изображения машин
+    if not _index_exists("iaai_cars", "ix_iaai_cars_origin_url"):
-    op.create_table(
+        op.create_index("ix_iaai_cars_origin_url", "iaai_cars", ["origin_url"])
-        "images",
+    if not _index_exists("iaai_cars", "ix_iaai_cars_origin_id"):
-        sa.Column("id", sa.BigInteger, primary_key=True, autoincrement=True),
+        op.create_index("ix_iaai_cars_origin_id", "iaai_cars", ["origin_id"], unique=True)
-        sa.Column("fullres_image", sa.String, nullable=False),
+
-        sa.Column("preview_image", sa.String, nullable=False),
+    # Таблица iaai_images — изображения машин
-        sa.Column("order_index", sa.Integer, nullable=False),
+    if not _table_exists("iaai_images"):
-        sa.Column("car_id", sa.Integer, sa.ForeignKey("cars.id", ondelete="CASCADE"), nullable=False),
+        op.create_table(
-    )
+            "iaai_images",
+            sa.Column("id", sa.BigInteger, primary_key=True, autoincrement=True),
+            sa.Column("fullres_image", sa.String, nullable=False),
+            sa.Column("preview_image", sa.String, nullable=False),
+            sa.Column("order_index", sa.Integer, nullable=False),
+            sa.Column("car_id", sa.Integer, sa.ForeignKey("iaai_cars.id", ondelete="CASCADE"), nullable=False),
+        )
+
+    # Таблица iaai_sync_runs — логирование синхронизаций
+    if not _table_exists("iaai_sync_runs"):
+        op.create_table(
+            "iaai_sync_runs",
+            sa.Column("id", sa.BigInteger, primary_key=True, autoincrement=True),
+            sa.Column("started_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+            sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column("status", sa.Text, nullable=False),
+            sa.Column("lane", sa.Text, nullable=False),
+            sa.Column("ids_fetched", sa.Integer, nullable=False, server_default="0"),
+            sa.Column("cars_upserted", sa.Integer, nullable=False, server_default="0"),
+            sa.Column("cars_failed", sa.Integer, nullable=False, server_default="0"),
+            sa.Column("images_upserted", sa.Integer, nullable=False, server_default="0"),
+            sa.Column("error_summary", sa.Text, nullable=True),
+        )
 
-    # Таблица sync_runs — логирование синхронизаций
-    op.create_table(
-        "sync_runs",
-        sa.Column("id", sa.BigInteger, primary_key=True, autoincrement=True),
-        sa.Column("started_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
-        sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
-        sa.Column("status", sa.Text, nullable=False),
-        sa.Column("lane", sa.Text, nullable=False),
-        sa.Column("ids_fetched", sa.Integer, nullable=False, server_default="0"),
-        sa.Column("cars_upserted", sa.Integer, nullable=False, server_default="0"),
-        sa.Column("cars_failed", sa.Integer, nullable=False, server_default="0"),
-        sa.Column("images_upserted", sa.Integer, nullable=False, server_default="0"),
-        sa.Column("error_summary", sa.Text, nullable=True),
-    )
 
 def downgrade() -> None:
-    op.drop_table("sync_runs")
+    # Compatibility-only migration for shared production databases.
-    op.drop_table("images")
+    pass
-    op.drop_table("cars")

alembic/versions/002_add_indexes.py

@@ -10,20 +10,15 @@ depends_on: Union[str, Sequence[str], None] = None
 
 
 def upgrade() -> None:
-    op.create_index("ix_cars_brand", "cars", ["brand"])
+    op.execute("CREATE INDEX IF NOT EXISTS ix_iaai_cars_brand ON iaai_cars (brand)")
-    op.create_index("ix_cars_brand_model", "cars", ["brand", "model"])
+    op.execute("CREATE INDEX IF NOT EXISTS ix_iaai_cars_brand_model ON iaai_cars (brand, model)")
-    op.create_index("ix_cars_year", "cars", ["year"])
+    op.execute("CREATE INDEX IF NOT EXISTS ix_iaai_cars_year ON iaai_cars (year)")
-    op.create_index("ix_cars_is_sold", "cars", ["is_sold"])
+    op.execute("CREATE INDEX IF NOT EXISTS ix_iaai_cars_is_sold ON iaai_cars (is_sold)")
-    op.create_index("ix_cars_last_seen_at", "cars", ["last_seen_at"])
+    op.execute("CREATE INDEX IF NOT EXISTS ix_iaai_cars_last_seen_at ON iaai_cars (last_seen_at)")
-    op.create_index("ix_images_car_id", "images", ["car_id"])
+    op.execute("CREATE INDEX IF NOT EXISTS ix_iaai_images_car_id ON iaai_images (car_id)")
-    op.create_index("ix_sync_runs_status", "sync_runs", ["status"])
+    op.execute("CREATE INDEX IF NOT EXISTS ix_iaai_sync_runs_status ON iaai_sync_runs (status)")
 
 
 def downgrade() -> None:
-    op.drop_index("ix_sync_runs_status", table_name="sync_runs")
+    # Compatibility-only migration for shared production databases.
-    op.drop_index("ix_images_car_id", table_name="images")
+    pass
-    op.drop_index("ix_cars_last_seen_at", table_name="cars")
-    op.drop_index("ix_cars_is_sold", table_name="cars")
-    op.drop_index("ix_cars_year", table_name="cars")
-    op.drop_index("ix_cars_brand_model", table_name="cars")
-    op.drop_index("ix_cars_brand", table_name="cars")

alembic/versions/003_add_composite_indexes.py

@@ -13,26 +13,24 @@ def upgrade() -> None:
     # Partial index для активных машин IAAI (is_sold = FALSE)
     # Ускоряет поиск при mark_sold операциях
     op.execute(
-        "CREATE INDEX IF NOT EXISTS ix_cars_active_iaai "
+        "CREATE INDEX IF NOT EXISTS ix_iaai_cars_active_iaai "
-        "ON cars (origin_url) "
+        "ON iaai_cars (origin_url) "
         "WHERE is_sold = FALSE AND origin_id LIKE 'iaai:%'"
     )
 
     # Composite index для проверки дубликатов изображений
-    op.create_index(
+    op.execute(
-        "ix_images_car_id_fullres",
+        "CREATE INDEX IF NOT EXISTS ix_iaai_images_car_id_fullres "
-        "images",
+        "ON iaai_images (car_id, fullres_image)"
-        ["car_id", "fullres_image"],
     )
 
     # Index для быстрого поиска existing машин по origin_url
     op.execute(
-        "CREATE INDEX IF NOT EXISTS ix_cars_origin_url_id "
+        "CREATE INDEX IF NOT EXISTS ix_iaai_cars_origin_url_id "
-        "ON cars (origin_url, origin_id)"
+        "ON iaai_cars (origin_url, origin_id)"
     )
 
 
 def downgrade() -> None:
-    op.execute("DROP INDEX IF EXISTS ix_cars_origin_url_id")
+    # Compatibility-only migration for shared production databases.
-    op.drop_index("ix_images_car_id_fullres", table_name="images")
+    pass
-    op.execute("DROP INDEX IF EXISTS ix_cars_active_iaai")

alembic/versions/004_add_ingestion_tables.py

@@ -0,0 +1,17 @@
+# Placeholder migration (ingestion tables not yet needed)
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = "004_add_ingestion_tables"
+down_revision: Union[str, None] = "003_add_composite_indexes"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass

docker-compose.yml

@@ -1,27 +1,35 @@
 x-app-env: &app-env
-  # NB: hardcoded to Postgres — .env may contain sqlite for local CLI, don't let it leak here
+  # Всегда используем Postgres.
   IAAI_DATABASE_URL: postgresql+psycopg2://iaai:iaai@postgres:5432/iaai_scraper
   IAAI_REDIS_URL: ${IAAI_REDIS_URL:-redis://redis:6379/0}
   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}
-  CELERY_TASK_SOFT_TIME_LIMIT: ${CELERY_TASK_SOFT_TIME_LIMIT:-3300}
+  CELERY_TASK_SOFT_TIME_LIMIT: ${CELERY_TASK_SOFT_TIME_LIMIT:-900}
-  CELERY_TASK_TIME_LIMIT: ${CELERY_TASK_TIME_LIMIT:-3600}
+  CELERY_TASK_TIME_LIMIT: ${CELERY_TASK_TIME_LIMIT:-1200}
-  CELERY_BROKER_VISIBILITY_TIMEOUT: ${CELERY_BROKER_VISIBILITY_TIMEOUT:-7200}
+  CELERY_BROKER_VISIBILITY_TIMEOUT: ${CELERY_BROKER_VISIBILITY_TIMEOUT:-2400}
-  # 0 => без лимита (в коде интерпретируется как None).
+  # 0 = без лимита.
-  # После первичного полного прохода 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}
+  IAAI_INTER_BATCH_DELAY_SECONDS: ${IAAI_INTER_BATCH_DELAY_SECONDS:-0.3}
+  IAAI_FAIL_RATE_THRESHOLD: ${IAAI_FAIL_RATE_THRESHOLD:-0.9}
+  IAAI_PARALLEL_TABS: ${IAAI_PARALLEL_TABS:-8}
   IAAI_BLOCK_RESOURCES: ${IAAI_BLOCK_RESOURCES:-true}
   IAAI_LISTING_SEGMENTS: ${IAAI_LISTING_SEGMENTS:-auto}
   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_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: chromium
+  # Self-heal для worker.
+  IAAI_SELF_HEAL_ENABLED: ${IAAI_SELF_HEAL_ENABLED:-true}
+  IAAI_SELF_HEAL_CHECK_INTERVAL_SECONDS: ${IAAI_SELF_HEAL_CHECK_INTERVAL_SECONDS:-30}
+  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}
   TZ: ${TZ:-UTC}
 
 x-env-file: &env-file
@@ -42,7 +50,7 @@ x-worker-service: &worker-service
     - tokens_data:/data
 
 services:
-  # PostgreSQL
+  # PostgreSQL.
   postgres:
     image: postgres:16-alpine
     container_name: iaai-postgres
@@ -66,7 +74,7 @@ services:
         max-size: "10m"
         max-file: "5"
 
-  # Redis (Celery broker)
+  # Redis.
   redis:
     image: redis:7-alpine
     container_name: iaai-redis
@@ -90,7 +98,7 @@ services:
         max-size: "10m"
         max-file: "5"
 
-  # DB migrations
+  # Миграции.
   migrate:
     <<: *app-service
     container_name: iaai-migrate
@@ -100,7 +108,7 @@ services:
         condition: service_healthy
     command: alembic upgrade head
 
-  # FastAPI
+  # API.
   api:
     <<: *app-service
     container_name: iaai-api
@@ -128,10 +136,9 @@ services:
         max-size: "10m"
         max-file: "5"
 
-  # Celery Worker
+  # Worker.
   worker:
     <<: *worker-service
-    container_name: iaai-worker
     restart: unless-stopped
     init: true
     depends_on:
@@ -142,11 +149,12 @@ services:
     stop_grace_period: 60s
     command: >
       celery -A iaai_scraper.worker.celery_app worker
-      --loglevel=info --concurrency=${CELERY_WORKER_CONCURRENCY:-4} --pool=prefork
+      --loglevel=info --concurrency=1 --pool=solo
       --pidfile=/tmp/celery-worker.pid
       -Q scraping --max-tasks-per-child=${CELERY_WORKER_MAX_TASKS_PER_CHILD:-5}
     healthcheck:
-      test: ["CMD-SHELL", "test -f /tmp/celery-worker.pid && kill -0 $(cat /tmp/celery-worker.pid)"]
+      # Проверка worker.
+      test: ["CMD-SHELL", "test -f /tmp/celery-worker.pid && kill -0 $(cat /tmp/celery-worker.pid) && pgrep -f 'iaai_scraper.worker.self_heal' >/dev/null"]
       interval: 60s
       timeout: 10s
       retries: 3
@@ -157,7 +165,7 @@ services:
         max-size: "10m"
         max-file: "5"
 
-  # Celery Beat (периодический планировщик)
+  # Beat.
   beat:
     <<: *worker-service
     container_name: iaai-beat

entrypoint.sh

@@ -55,6 +55,35 @@ start_proxy_bridge_if_needed() {
   echo "[entrypoint] Proxy bridge started (PID ${BRIDGE_PID})"
 }
 
+start_self_heal_watchdog_if_worker() {
+  if ! is_worker_command "$@"; then
+    return 0
+  fi
+
+  # После reboot/restart контейнера файловая система контейнера сохраняется,
+  # поэтому stale pidfile может остаться от прошлого запуска и блокировать старт Celery.
+  # Удаляем его перед запуском worker-процесса.
+  rm -f /tmp/celery-worker.pid
+
+  if [ "${IAAI_SELF_HEAL_ENABLED:-true}" = "false" ]; then
+    echo "[entrypoint] Self-heal watchdog disabled"
+    return 0
+  fi
+
+  echo "[entrypoint] Starting self-heal watchdog for worker..."
+  python -m iaai_scraper.worker.self_heal &
+  SELF_HEAL_PID=$!
+  sleep 1
+
+  if ! kill -0 "${SELF_HEAL_PID}" 2>/dev/null; then
+    echo "[entrypoint] ERROR: self-heal watchdog failed to start"
+    exit 1
+  fi
+
+  echo "[entrypoint] Self-heal watchdog started (PID ${SELF_HEAL_PID})"
+}
+
 start_proxy_bridge_if_needed "$@"
+start_self_heal_watchdog_if_worker "$@"
 
 exec "$@"

iaai_scraper/browser/factory.py

@@ -15,7 +15,7 @@ logger = logging.getLogger("iaai_scraper.browser")
 
 
 def _build_init_script() -> str:
-    # Патч признаков автоматизации.
+    # Маскировка браузера.
     hardware_concurrency = random.choice([4, 8, 12, 16])
     device_memory = random.choice([4, 8, 16])
     languages = ["en-US", "en"]
@@ -69,11 +69,10 @@ class BrowserFactory:
         self.settings = settings
 
     def _resolve_engine(self) -> str:
-        # Выбор движка браузера.
+        # Выбор движка.
         engine = self.settings.browser_engine.strip().lower()
         if engine == "auto":
-            # Для IAAI в headless-режиме Chromium со stealth-скриптами
+            # Для IAAI стабильнее Chromium.
-            # значительно стабильнее Firefox по anti-bot.
             return "chromium"
         if engine in ("firefox", "chromium"):
             return engine
@@ -90,11 +89,11 @@ class BrowserFactory:
             if proxy_dict:
                 launch_kwargs["proxy"] = proxy_dict
                 logger.info("Using proxy: %s", self.settings.proxy.server)
-            # Параметры Firefox для headless-режима.
+            # Настройки Firefox.
             launch_kwargs["firefox_user_prefs"] = {
                 "dom.webdriver.enabled": False,
                 "useAutomationExtension": False,
-                # Базовые оптимизации для VPS.
+                # Базовые оптимизации.
                 "media.autoplay.default": 5,
                 "media.volume_scale": "0.0",
                 "media.audio.playback.standalone": False,
@@ -111,7 +110,7 @@ class BrowserFactory:
             logger.info("Launching Firefox (headless=%s)", self.settings.headless)
             return playwright.firefox.launch(**launch_kwargs)
 
-        # Путь запуска Chromium.
+        # Запуск Chromium.
         args = [
             "--disable-blink-features=AutomationControlled",
             "--no-default-browser-check",
@@ -154,7 +153,7 @@ class BrowserFactory:
         }
 
         if is_firefox:
-            # Заголовки и user-agent для Firefox.
+            # Заголовки Firefox.
             ctx_kwargs["user_agent"] = (
                 "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:128.0) "
                 "Gecko/20100101 Firefox/128.0"
@@ -183,7 +182,7 @@ class BrowserFactory:
         context.set_default_navigation_timeout(self.settings.default_timeout_ms)
 
         if not is_firefox:
-            # Патчи маскировки для Chromium.
+            # Маскировка Chromium.
             context.add_init_script(_build_init_script())
             if stealth_sync:
                 context.on("page", lambda page: stealth_sync(page))
@@ -193,7 +192,7 @@ class BrowserFactory:
 
     @staticmethod
     def enable_resource_blocking(page) -> None:
-        # Блокируем тяжёлые ресурсы для ускорения загрузки страниц.
+        # Блокируем тяжёлые ресурсы.
         BLOCKED_TYPES = {"image", "stylesheet", "font", "media"}
         BLOCKED_URL_PATTERNS = (
             "google-analytics", "googletagmanager", "facebook.net",

iaai_scraper/browser/listing.py

@@ -133,6 +133,28 @@ class ListingCollector:
 
 		return not old_first_href
 
+	@staticmethod
+	def has_page_number(page: Page, target_page_number: int) -> bool:
+		try:
+			return bool(page.evaluate(
+				"""
+				(targetPageNumber) => {
+					const visible = (el) => !!(el && (el.offsetWidth || el.offsetHeight || el.getClientRects().length));
+					const controls = Array.from(document.querySelectorAll('a,button,[role="button"],span,div'));
+					return controls.some((el) => {
+						const text = (el.textContent || '').trim();
+						const cls = (el.getAttribute('class') || '').toLowerCase();
+						const ariaDisabled = (el.getAttribute('aria-disabled') || '').toLowerCase();
+						const disabled = el.hasAttribute('disabled') || ariaDisabled === 'true' || cls.includes('disabled');
+						return visible(el) && !disabled && /^\d+$/.test(text) && parseInt(text, 10) === targetPageNumber;
+					});
+				}
+				""",
+				target_page_number,
+			))
+		except Exception:
+			return False
+
 	def open_cars_listing(self, page: Page, *, url_override: str | None = None) -> None:
 		url = url_override or self.settings.listing.cars_url
 		logger.info("Opening cars listing page: %s", url)
@@ -625,23 +647,64 @@ class ListingCollector:
 	@staticmethod
 	def _has_next_page(page: Page) -> bool:
 		for selector in ListingCollector._NEXT_PAGE_SELECTORS:
-			if page.locator(selector).count() > 0:
+			locator = page.locator(selector)
-				return True
+			count = locator.count()
+			if count == 0:
+				continue
+			# Проверяем, что хотя бы один элемент не disabled.
+			# Disabled "Next" на последней странице не означает наличия следующей.
+			for i in range(min(count, 3)):
+				try:
+					el = locator.nth(i)
+					disabled_attr = el.get_attribute("disabled", timeout=300)
+					aria_disabled = el.get_attribute("aria-disabled", timeout=300)
+					cls = (el.get_attribute("class", timeout=300) or "").lower()
+					if disabled_attr is None and aria_disabled != "true" and "disabled" not in cls:
+						return True
+				except Exception:
+					continue
 		try:
 			return bool(page.evaluate(
 				"""
-				() => Array.from(document.querySelectorAll('a,button,[role="button"]')).some((el) => {
+				() => {
-					const text = (el.textContent || '').trim().toLowerCase();
+					const visible = (el) => !!(el && (el.offsetWidth || el.offsetHeight || el.getClientRects().length));
-					const aria = (el.getAttribute('aria-label') || '').trim().toLowerCase();
+					const controls = Array.from(document.querySelectorAll('a,button,[role="button"]')).filter((el) => {
-					const title = (el.getAttribute('title') || '').trim().toLowerCase();
+						const cls = (el.getAttribute('class') || '').trim().toLowerCase();
-					const rel = (el.getAttribute('rel') || '').trim().toLowerCase();
+						const disabled = el.hasAttribute('disabled') || el.getAttribute('aria-disabled') === 'true' || cls.includes('disabled');
-					const cls = (el.getAttribute('class') || '').trim().toLowerCase();
+						return !disabled && visible(el);
-					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 hasExplicitNext = controls.some((el) => {
-					const looksNumericNext = /^\d+$/.test(text);
+						const text = (el.textContent || '').trim().toLowerCase();
-					return !disabled && visible && (rel === 'next' || aria.includes('next') || title.includes('next') || cls.includes('next') || hasRightArrowIcon || looksNumericNext || ['next', '›', '»', '>'].includes(text));
+						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 rel === 'next' || aria.includes('next') || title.includes('next') || cls.includes('next') || hasRightArrowIcon || ['next', '›', '»', '>'].includes(text);
+					});
+
+					if (hasExplicitNext) {
+						return true;
+					}
+
+					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) {
+						return false;
+					}
+
+					const currentPage = parseInt((current.textContent || '').trim(), 10);
+					return controls.some((el) => {
+						const text = (el.textContent || '').trim();
+						return /^\d+$/.test(text) && parseInt(text, 10) === currentPage + 1;
+					});
+				}
 				"""
 			))
 		except Exception:

iaai_scraper/core/config.py

@@ -204,6 +204,19 @@ class DatabaseConfig:
 @dataclass(slots=True)
 class RedisConfig:
     url: str = _env_str("IAAI_REDIS_URL", "redis://localhost:6379/0")
+    socket_timeout_seconds: float = _env_float("IAAI_REDIS_SOCKET_TIMEOUT_SECONDS", 10.0)
+    socket_connect_timeout_seconds: float = _env_float("IAAI_REDIS_SOCKET_CONNECT_TIMEOUT_SECONDS", 5.0)
+    health_check_interval_seconds: int = _env_int("IAAI_REDIS_HEALTH_CHECK_INTERVAL_SECONDS", 30)
+
+
+# --- Конфиг Discovery (режим обнаружения, hourly batch) ---
+
+@dataclass(slots=True)
+class DiscoveryConfig:
+    mode: str = _env_str("IAAI_DISCOVERY_MODE", "sitemap")
+    hourly_mode: str = _env_str("IAAI_HOURLY_MODE", "rolling_refresh")
+    hourly_refresh_batch_size: int = _env_int("IAAI_HOURLY_REFRESH_BATCH_SIZE", 500)
+    always_full_scan: bool = _env_bool("IAAI_ALWAYS_FULL_SCAN", True)
 
 
 # --- Конфиг Celery (лимиты задач, concurrency, beat-расписание) ---
@@ -214,6 +227,7 @@ class CeleryConfig:
     result_backend: str = _env_str("CELERY_RESULT_BACKEND", "")
     task_soft_time_limit: int = _env_int("CELERY_TASK_SOFT_TIME_LIMIT", 3300)
     task_time_limit: int = _env_int("CELERY_TASK_TIME_LIMIT", 3600)
+    task_stall_timeout_seconds: int = _env_int("CELERY_TASK_STALL_TIMEOUT_SECONDS", 600)
     worker_concurrency: int = _env_int("CELERY_WORKER_CONCURRENCY", 4)
     worker_max_tasks_per_child: int = _env_int("CELERY_WORKER_MAX_TASKS_PER_CHILD", 5)
     broker_visibility_timeout: int = _env_int("CELERY_BROKER_VISIBILITY_TIMEOUT", 7200)
@@ -221,6 +235,7 @@ class CeleryConfig:
     beat_sync_limit: int | None = _env_int("CELERY_BEAT_SYNC_LIMIT", 0) or None
     batch_size: int = _env_int("CELERY_BATCH_SIZE", 50)
     parallel_tabs: int = _env_int("IAAI_PARALLEL_TABS", 8)
+    parallel_segments: bool = _env_bool("CELERY_PARALLEL_SEGMENTS", False)
     block_resources: bool = _env_bool("IAAI_BLOCK_RESOURCES", True)
 
 
@@ -279,6 +294,7 @@ class Settings:
     redis: RedisConfig = field(default_factory=RedisConfig)
     celery: CeleryConfig = field(default_factory=CeleryConfig)
     proxy: ProxyConfig = field(default_factory=ProxyConfig)
+    discovery: DiscoveryConfig = field(default_factory=DiscoveryConfig)
 
     @property
     def parallel_tabs(self) -> int:

iaai_scraper/core/exceptions.py

@@ -8,3 +8,7 @@ class AntiBotDetectedError(ScraperError):
 
 class SiteStructureChangedError(ScraperError):
     """Вызывается, когда структура страницы изменилась и данных не хватает."""
+
+
+class ListingResumeError(ScraperError):
+    """Вызывается, когда resume по checkpoint больше недостижим."""

iaai_scraper/discovery/__init__.py

@@ -0,0 +1,15 @@
+from .sitemap import (
+    SitemapDiscoveryError,
+    SitemapDiscoveryResult,
+    SitemapDiscoveryStats,
+    discover_vehicle_urls_from_sitemap,
+    discover_vehicle_urls_from_sitemap_with_stats,
+)
+
+__all__ = [
+    "SitemapDiscoveryError",
+    "SitemapDiscoveryResult",
+    "SitemapDiscoveryStats",
+    "discover_vehicle_urls_from_sitemap",
+    "discover_vehicle_urls_from_sitemap_with_stats",
+]

iaai_scraper/discovery/sitemap.py

@@ -0,0 +1,210 @@
+from __future__ import annotations
+
+import gzip
+import io
+import logging
+import re
+from dataclasses import dataclass, field
+from typing import Iterable
+from urllib.parse import urlsplit, urlunsplit
+from urllib.request import Request, urlopen, ProxyHandler, build_opener
+import xml.etree.ElementTree as ET
+
+logger = logging.getLogger("iaai_scraper.discovery.sitemap")
+
+DEFAULT_SITEMAP_INDEX_URL = "https://www.iaai.com/Xj9rDOVMEi0hc38S/sitemap_index.xml"
+_SITEMAP_TIMEOUT_SECONDS = 30
+_LOC_TAG_RE = re.compile(rb"<loc>\s*(.*?)\s*</loc>", re.IGNORECASE | re.DOTALL)
+
+
+class SitemapDiscoveryError(RuntimeError):
+    pass
+
+
+def _is_vehicle_sitemap_url(url: str) -> bool:
+    lowered = url.strip().lower()
+    # Берём только sitemap с авто.
+    if "sitemapbranches" in lowered or "sitemapauctions" in lowered:
+        return False
+    return lowered.endswith(".xml") or lowered.endswith(".xml.gz")
+
+
+def _normalize_vehicle_url(url: str) -> str:
+    parts = urlsplit(url.strip())
+    return urlunsplit((parts.scheme, parts.netloc, parts.path, "", ""))
+
+
+def _download_bytes(url: str, proxy_url: str | None = None) -> bytes:
+    request = Request(
+        url,
+        headers={
+            "User-Agent": (
+                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+                "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/135.0.0.0 Safari/537.36"
+            ),
+            "Accept": "application/xml,text/xml,application/xhtml+xml,text/html;q=0.9,*/*;q=0.8",
+            "Accept-Encoding": "gzip",
+        },
+    )
+    if proxy_url:
+        handler = ProxyHandler({"http": proxy_url, "https": proxy_url})
+        opener = build_opener(handler)
+        response = opener.open(request, timeout=_SITEMAP_TIMEOUT_SECONDS)
+    else:
+        response = urlopen(request, timeout=_SITEMAP_TIMEOUT_SECONDS)
+    with response:
+        payload = response.read()
+        encoding = str(response.headers.get("Content-Encoding") or "").lower()
+        if encoding == "gzip" or url.lower().endswith(".gz"):
+            return gzip.GzipFile(fileobj=io.BytesIO(payload)).read()
+        return payload
+
+
+def _local_name(tag: str) -> str:
+    if "}" in tag:
+        return tag.rsplit("}", 1)[1]
+    return tag
+
+
+def _iter_loc_values_fallback(xml_bytes: bytes) -> Iterable[str]:
+    for match in _LOC_TAG_RE.finditer(xml_bytes):
+        try:
+            value = match.group(1).decode("utf-8", errors="ignore").strip()
+        except Exception:
+            continue
+        if value:
+            yield value
+
+
+def _iter_loc_values(xml_bytes: bytes) -> Iterable[str]:
+    try:
+        root = ET.fromstring(xml_bytes)
+    except ET.ParseError as exc:
+        fallback_values = list(_iter_loc_values_fallback(xml_bytes))
+        if fallback_values:
+            logger.warning(
+                "Falling back to regex sitemap loc extraction after XML parse error: %s",
+                exc,
+            )
+            yield from fallback_values
+            return
+        raise SitemapDiscoveryError(f"Invalid sitemap XML: {exc}") from exc
+
+    for element in root.iter():
+        if _local_name(element.tag) != "loc":
+            continue
+        if not element.text:
+            continue
+        value = element.text.strip()
+        if value:
+            yield value
+
+
+def _filter_vehicle_urls(urls: Iterable[str]) -> list[str]:
+    result: list[str] = []
+    seen: set[str] = set()
+    for url in urls:
+        normalized = _normalize_vehicle_url(url)
+        if "/VehicleDetail/" not in normalized and "/vehicledetail/" not in normalized:
+            continue
+        if normalized in seen:
+            continue
+        seen.add(normalized)
+        result.append(normalized)
+    return result
+
+
+def _looks_like_vehicle_detail_sitemap(urls: list[str]) -> bool:
+    return any("/vehicledetail/" in url.lower() for url in urls)
+
+
+def discover_vehicle_urls_from_sitemap(index_url: str = DEFAULT_SITEMAP_INDEX_URL, proxy_url: str | None = None) -> list[str]:
+    logger.info("Downloading sitemap index: %s", index_url)
+    index_xml = _download_bytes(index_url, proxy_url=proxy_url)
+    sitemap_urls = [url for url in _iter_loc_values(index_xml) if _is_vehicle_sitemap_url(url)]
+    if not sitemap_urls:
+        raise SitemapDiscoveryError("Sitemap index returned no sitemap URLs")
+
+    all_vehicle_urls: list[str] = []
+    for sitemap_url in sitemap_urls:
+        logger.info("Downloading sitemap: %s", sitemap_url)
+        try:
+            sitemap_xml = _download_bytes(sitemap_url, proxy_url=proxy_url)
+            raw_urls = list(_iter_loc_values(sitemap_xml))
+        except SitemapDiscoveryError as exc:
+            logger.warning("Skipping malformed sitemap %s: %s", sitemap_url, exc)
+            continue
+
+        vehicle_urls = _filter_vehicle_urls(raw_urls)
+        if raw_urls and not vehicle_urls and not _looks_like_vehicle_detail_sitemap(raw_urls):
+            logger.info("Skipping non-vehicle sitemap %s", sitemap_url)
+            continue
+        logger.info("Sitemap %s yielded %d vehicle URLs", sitemap_url, len(vehicle_urls))
+        all_vehicle_urls.extend(vehicle_urls)
+
+    deduped = _filter_vehicle_urls(all_vehicle_urls)
+    if not deduped:
+        raise SitemapDiscoveryError("No vehicle detail URLs discovered from vehicle sitemaps")
+    logger.info("Sitemap discovery done: %d vehicle URLs", len(deduped))
+    return deduped
+
+
+@dataclass
+class SitemapDiscoveryStats:
+    transport: str = "http"
+    fetched_sitemaps: int = 0
+    blocked_sitemaps: int = 0
+    malformed_sitemaps: int = 0
+    direct_probe_hits: int = 0
+    direct_probe_misses: int = 0
+
+
+@dataclass
+class SitemapDiscoveryResult:
+    vehicle_urls: list[str] = field(default_factory=list)
+    stats: SitemapDiscoveryStats = field(default_factory=SitemapDiscoveryStats)
+
+
+def discover_vehicle_urls_from_sitemap_with_stats(
+    settings=None,
+    index_url: str = DEFAULT_SITEMAP_INDEX_URL,
+) -> SitemapDiscoveryResult:
+    """Возвращает URL и статистику."""
+    proxy_url = None
+    if settings is not None and hasattr(settings, 'proxy') and settings.proxy.server:
+        proxy_url = settings.proxy.server
+    stats = SitemapDiscoveryStats(transport="http")
+    logger.info("Downloading sitemap index: %s", index_url)
+    index_xml = _download_bytes(index_url, proxy_url=proxy_url)
+    sitemap_urls = [url for url in _iter_loc_values(index_xml) if _is_vehicle_sitemap_url(url)]
+    if not sitemap_urls:
+        raise SitemapDiscoveryError("Sitemap index returned no sitemap URLs")
+
+    all_vehicle_urls: list[str] = []
+    for sitemap_url in sitemap_urls:
+        logger.info("Downloading sitemap: %s", sitemap_url)
+        try:
+            sitemap_xml = _download_bytes(sitemap_url, proxy_url=proxy_url)
+            raw_urls = list(_iter_loc_values(sitemap_xml))
+            stats.fetched_sitemaps += 1
+        except SitemapDiscoveryError as exc:
+            logger.warning("Skipping malformed sitemap %s: %s", sitemap_url, exc)
+            stats.malformed_sitemaps += 1
+            continue
+        except Exception as exc:
+            logger.warning("Blocked/failed sitemap %s: %s", sitemap_url, exc)
+            stats.blocked_sitemaps += 1
+            continue
+
+        vehicle_urls = _filter_vehicle_urls(raw_urls)
+        if raw_urls and not vehicle_urls and not _looks_like_vehicle_detail_sitemap(raw_urls):
+            logger.info("Skipping non-vehicle sitemap %s", sitemap_url)
+            continue
+        logger.info("Sitemap %s yielded %d vehicle URLs", sitemap_url, len(vehicle_urls))
+        all_vehicle_urls.extend(vehicle_urls)
+
+    deduped = _filter_vehicle_urls(all_vehicle_urls)
+    if not deduped:
+        raise SitemapDiscoveryError("No vehicle detail URLs discovered from vehicle sitemaps")
+    logger.info("Sitemap discovery done: %d vehicle URLs", len(deduped))
+    return SitemapDiscoveryResult(vehicle_urls=deduped, stats=stats)

iaai_scraper/parsing/mapper.py

@@ -20,7 +20,7 @@ from ..storage.schemas import CarRecord, ImageRecord
 
 
 class CarMapper:
-	# Преобразование данных IAAI в CarRecord.
+	# Маппер IAAI в CarRecord.
 
 	BODY_MAP = {
 		"sedan": "SEDAN", "coupe": "COUPE", "hatchback": "HATCHBACK", "sport utility": "SUV",
@@ -48,7 +48,7 @@ class CarMapper:
 	NO_DAMAGE_MARKERS = {"normal wear", "normal wear & tear", "normal wear and tear", "n/a", "na", "none", "no damage", "minor dents/scratches"}
 
 	def map_to_car_record(self, vehicle_url: str, vehicle_summary: dict[str, Any], payload_insights: dict[str, Any]) -> CarRecord:
-		# Собираем нормализованную DB-модель.
+		# Собираем DB-модель.
 		vehicle_summary = vehicle_summary or {}
 		payload_insights = payload_insights or {}
 		core = payload_insights.get("vehicle_core", {})
@@ -77,7 +77,7 @@ class CarMapper:
 		)
 		color = self._normalize_color(first_non_empty([core.get("color"), vehicle_summary.get("color"), "other"]))
 		drive = self._normalize_drive(first_non_empty([core.get("drive"), vehicle_summary.get("drive")]))
-		# Пытаемся определить привод из строки двигателя.
+		# Пробуем взять привод из двигателя.
 		if not drive or drive == "NA":
 			engine_text = self._as_str(first_non_empty([core.get("engine"), vehicle_summary.get("engine")]))
 			if engine_text:
@@ -144,7 +144,7 @@ class CarMapper:
 
 	@classmethod
 	def _to_money_int(cls, value: Any) -> int | None:
-		# Нормализация стоимости из разных форматов.
+		# Нормализация цены.
 		if value is None:
 			return None
 		if isinstance(value, bool):
@@ -168,14 +168,14 @@ class CarMapper:
 		for number in numbers:
 			clean = number.replace(" ", "")
 			if "," in clean and "." in clean:
-					# Поддержка 1,234.56 и 1.234,56.
+					# Поддержка двух форматов.
 				if clean.rfind(",") > clean.rfind("."):
 					clean = clean.replace(".", "").replace(",", ".")
 				else:
 					clean = clean.replace(",", "")
 			elif "," in clean:
 				parts = clean.split(",")
-				# 123,45 -> 123.45, иначе разделитель тысяч.
+				# Десятичный или тысячный разделитель.
 				if len(parts[-1]) in {1, 2} and len(parts) == 2:
 					clean = clean.replace(",", ".")
 				else:
@@ -214,7 +214,7 @@ class CarMapper:
 
 	@staticmethod
 	def _parse_odometer(value: Any) -> int:
-		# Разбор пробега из строк IAAI.
+		# Разбор пробега.
 		if value is None:
 			return 0
 		text = str(value).strip()
@@ -223,7 +223,7 @@ class CarMapper:
 		lowered = text.lower()
 		if any(skip in lowered for skip in ["not required", "exempt", "n/a", "na", "unknown", "tbd"]):
 			return 0
-		# Извлекаем число из строкового формата одометра.
+		# Ищем число.
 		numbers = re.findall(r"[\d,]+", text)
 		for num_str in numbers:
 			clean = num_str.replace(",", "")
@@ -294,7 +294,7 @@ class CarMapper:
 		empty_default: str | None,
 		fallback: str | None,
 	) -> str | None:
-		# Общий helper для enum-нормализации.
+		# Общая нормализация enum.
 		text = self._as_str(value).lower()
 		if not text:
 			return empty_default
@@ -329,7 +329,7 @@ class CarMapper:
 		return any(token in self._as_str(auction.get("sale_status")).lower() for token in ["sold", "closed", "ended"])
 
 	def _build_images(self, urls: list[Any]) -> list[ImageRecord]:
-		# Для imageKeys берем самый большой размер.
+		# Для imageKeys берём самый большой размер.
 		best_by_key: dict[str, str] = {}
 		key_order: list[str] = []
 		non_keyed: list[str] = []
@@ -375,11 +375,11 @@ class CarMapper:
 
 	@classmethod
 	def _generate_parser_id(cls, origin_id: str) -> str:
-		# Стабильный parser_id по origin_id.
+		# Стабильный parser_id.
 		digest = hashlib.sha256(origin_id.encode()).digest()
 		alphabet = cls._PARSER_ID_ALPHABET
 		base = len(alphabet)
-		num = int.from_bytes(digest[:17], "big")  # 17 байт = 136 бит, хватает на 22 символа
+		num = int.from_bytes(digest[:17], "big")  # Хватает на 22 символа.
 		chars: list[str] = []
 		for _ in range(22):
 			num, idx = divmod(num, base)
@@ -387,7 +387,7 @@ class CarMapper:
 		return "car-" + "".join(chars)
 
 	def _build_origin_id(self, vehicle_url: str, vehicle_summary: dict[str, Any], core: dict[str, Any]) -> str:
-		# Формат: iaai:{lot_number} (аналог copart:94323985).
+		# Формат: iaai:{lot_number}.
 		for value in [core.get("lot_number"), vehicle_summary.get("lot_number")]:
 			text = self._as_str(value)
 			if text:

iaai_scraper/parsing/parser.py

@@ -10,9 +10,9 @@ logger = logging.getLogger("iaai_scraper.parsers")
 
 
 class VehicleParser:
-	# Парсер данных страницы автомобиля.
+	# Парсер страницы авто.
 
-	# Регулярные выражения для парсинга и детекции защиты.
+	# Регулярки парсинга и защиты.
 	_BUY_NOW_RE = re.compile(r"Buy\s+Now[:\s]*\$\s*([\d,]+(?:\.\d{1,2})?)", re.IGNORECASE)
 	_CAPTCHA_TOKENS = frozenset(["captcha", "verify you are human", "i am human", "recaptcha", "cloudflare"])
 	_ANTIBOT_TOKENS = frozenset(["incapsula", "access denied", "request unsuccessful", "bot detection"])
@@ -101,11 +101,11 @@ class VehicleParser:
 		max_fields = len(set(self.DOM_LABEL_MAP.values()))
 
 		for i, line in enumerate(lines):
-			# Ранний выход, когда уже нашли все поля.
+			# Ранний выход.
 			if len(result) >= max_fields:
 				break
 
-			# Сценарий 1: "метка: значение" в одной строке.
+			# Метка и значение в одной строке.
 			colon_pos = line.find(":")
 			if colon_pos > 0:
 				label_part = line[:colon_pos].strip().lower()
@@ -116,7 +116,7 @@ class VehicleParser:
 						result[field_name] = value_part
 						continue
 
-			# Сценарий 2: метка и значение на соседних строках.
+			# Метка и значение в соседних строках.
 			clean = line.rstrip(":").strip().lower()
 			clean_alt = clean.rstrip("#").strip()
 			matched_label = None
@@ -164,7 +164,7 @@ class VehicleParser:
 			page_title = title_match.group(1).strip()
 		title_parsed = self._parse_title_for_year_make_model(page_title, dom_text)
 
-		# Один проход по payload для всех полей.
+		# Один проход по payload.
 		all_found = deep_find_all_keys(payloads, self.SUMMARY_KEY_MAP)
 
 		summary: dict[str, Any] = {"source_url": vehicle_url}
@@ -199,7 +199,7 @@ class VehicleParser:
 			p = item.get("payload")
 			if isinstance(p, (dict, list)):
 				payloads.append(p)
-				# Дополнительный проход по встроенному JSON.
+				# Доп. проход по JSON.
 				extra = deep_find_all_keys([p], self.SUMMARY_KEY_MAP)
 				for field, vals in extra.items():
 					if not summary.get(field):
@@ -207,7 +207,7 @@ class VehicleParser:
 						if v:
 							summary[field] = v
 
-		# Передаём image_urls без повторного извлечения.
+		# Передаём готовые image_urls.
 		image_urls = summary.get("image_urls") or []
 		return {
 			"vehicle_summary": summary,
@@ -325,7 +325,7 @@ class VehicleParser:
 		for script_text in scripts:
 			if "{" not in script_text and "[" not in script_text:
 				continue
-			# Пропускаем слишком большие минифицированные блоки.
+			# Пропускаем большие блоки.
 			if len(script_text) > 51_200:
 				continue
 			try:

iaai_scraper/storage/db.py

@@ -20,6 +20,7 @@ CAR_DB_FIELDS = {
 }
 
 _IN_CHUNK_SIZE = 5000
+CAR_TABLE_NAME = Car.__tablename__
 
 
 class PersistenceService:
@@ -35,11 +36,16 @@ class PersistenceService:
 			engine_kwargs["max_overflow"] = settings.database.max_overflow
 			engine_kwargs["pool_pre_ping"] = True
 			engine_kwargs["pool_recycle"] = settings.database.pool_recycle_seconds
+			engine_kwargs["pool_timeout"] = 30
+			# Таймауты запросов и блокировок.
+			engine_kwargs["connect_args"] = {
+				"options": "-c statement_timeout=120000 -c lock_timeout=30000"
+			}
 		self.engine = create_engine(settings.database.url, **engine_kwargs)
 		self.session_factory = sessionmaker(bind=self.engine, expire_on_commit=False, future=True)
 
 	def create_tables(self) -> None:
-		# В тестах/локально на SQLite разрешаем create_all; для non-SQLite в проде — только через миграции.
+		# Для SQLite можно create_all.
 		is_sqlite = self.settings.database.url.startswith("sqlite")
 		if not is_sqlite and not self.settings.database.auto_create_tables:
 			return
@@ -105,7 +111,7 @@ class PersistenceService:
 	def get_existing_urls_and_ids(
 		self, origin_urls: list[str], origin_ids: list[str],
 	) -> tuple[set[str], set[str]]:
-		# Загрузка существующих URL и origin_id.
+		# Загрузка URL и origin_id.
 		if not origin_urls and not origin_ids:
 			return set(), set()
 		urls: set[str] = set()
@@ -314,7 +320,7 @@ class PersistenceService:
 		return {"inserted": inserted, "updated": updated, "images_upserted": images_total}
 
 	def upsert_car(self, record: CarRecord):
-		# Вставка или обновление автомобиля по origin_id/origin_url.
+		# Вставка или обновление авто.
 		payload = self._car_payload(record)
 		images = [image.model_dump(mode="python") for image in record.images]
 		with self.session_scope() as session:
@@ -375,7 +381,7 @@ class PersistenceService:
 		- Один DELETE по car_id IN (...) вместо удаления по одному.
 		- Fallback на по-одному upsert если batch commit упал.
 		"""
-		# ── Дедупликация записей внутри батча ──
+		# Дедупликация батча.
 		seen_ids: dict[str, int] = {}
 		unique_records: list[CarRecord] = []
 		for idx, r in enumerate(records):
@@ -406,20 +412,20 @@ class PersistenceService:
 		images_total = 0
 
 		with self.session_scope() as session:
-			# Получаем существующие записи chunked-запросами.
+			# Загружаем существующие записи.
 			origin_ids = [r.origin_id for r in records if r.origin_id]
 			origin_urls = [r.origin_url for r in records if r.origin_url]
 
 			existing_by_id, existing_by_url = self._load_existing_cars(session, origin_ids, origin_urls)
 
-			# Предзагружаем ВСЕ изображения для обновляемых машин одним запросом.
+			# Предзагружаем изображения.
 			existing_car_ids = set()
 			for record in records:
 				car = existing_by_id.get(record.origin_id) or existing_by_url.get(record.origin_url)
 				if car is not None:
 					existing_car_ids.add(int(car.id))
 
-			# Строим маппинг car_id → set(image_urls) для сравнения.
+			# Готовим map car_id -> image_urls.
 			existing_images_map = self._load_existing_image_urls(session, existing_car_ids)
 
 			new_cars: list[tuple[Car, list[dict]]] = []
@@ -441,7 +447,7 @@ class PersistenceService:
 					car.last_seen_at = record.last_seen_at
 					updated += 1
 
-					# Проверяем, изменились ли изображения.
+					# Проверяем изменения картинок.
 					new_image_urls = {img.get("fullres_image", "") for img in images}
 					old_image_urls = existing_images_map.get(int(car.id), set())
 					if new_image_urls != old_image_urls:
@@ -449,15 +455,15 @@ class PersistenceService:
 					else:
 						images_total += len(old_image_urls)
 
-			# Один flush для всех вставок.
+			# Один flush.
 			session.flush()
 
-			# Добавляем изображения для новых автомобилей.
+			# Добавляем картинки новым авто.
 			for car, images in new_cars:
 				self._add_images(session, int(car.id), images)
 				images_total += len(images)
 
-			# Массово обновляем изображения только для машин с изменёнными картинками.
+			# Обновляем только изменённые картинки.
 			if update_cars_needing_images:
 				update_ids = [int(car.id) for car, _ in update_cars_needing_images]
 				for i in range(0, len(update_ids), _IN_CHUNK_SIZE):
@@ -470,7 +476,7 @@ class PersistenceService:
 		return {"inserted": inserted, "updated": updated, "images_upserted": images_total}
 
 	def _upsert_cars_individually(self, records: list[CarRecord]) -> dict[str, int]:
-		# Fallback на поштучный upsert.
+		# Запасной поштучный upsert.
 		inserted = 0
 		updated = 0
 		images_total = 0
@@ -510,55 +516,107 @@ class PersistenceService:
 
 		Для PostgreSQL использует временную таблицу + LEFT JOIN вместо NOT IN,
 		что кардинально быстрее при больших объёмах (100K+ URLs).
+		Встроенная защита: нормализация URL (тильда/дефис) + safety-check на аномальный процент.
 		"""
 		if not active_origin_urls:
 			return 0
 
+		# Нормализация URL.
+		def _norm(url: str) -> str:
+			return url.replace("~", "-")
+
+		normalized_urls = {_norm(u) for u in active_origin_urls}
+
 		is_postgres = "postgresql" in self.settings.database.url
 
 		with self.session_scope() as session:
 			if is_postgres:
-				# Создаём временную таблицу с активными URL.
+				# Считаем кандидатов на sold.
+				total_active = session.execute(
+					text(f"SELECT count(*) FROM {CAR_TABLE_NAME} WHERE is_sold = FALSE AND origin_id LIKE 'iaai:%%'")
+				).scalar() or 0
+
+				if total_active == 0:
+					return 0
+
+				# Временная таблица URL.
 				session.execute(text("CREATE TEMP TABLE IF NOT EXISTS _active_urls (url TEXT NOT NULL) ON COMMIT DROP"))
 				session.execute(text("TRUNCATE _active_urls"))
 
-				# Вставляем активные URL чанками.
+				# Вставляем URL чанками.
-				url_list = list(active_origin_urls)
+				url_list = list(normalized_urls)
 				for i in range(0, len(url_list), _IN_CHUNK_SIZE):
 					chunk = url_list[i:i + _IN_CHUNK_SIZE]
 					values = ",".join(f"(:{f'u{j}'})" for j in range(len(chunk)))
 					params = {f"u{j}": url for j, url in enumerate(chunk)}
 					session.execute(text(f"INSERT INTO _active_urls (url) VALUES {values}"), params)
 
-				# Создаём индекс на временной таблице для ускорения JOIN.
+				# Индекс для JOIN.
 				session.execute(text("CREATE INDEX IF NOT EXISTS _ix_active_urls ON _active_urls (url)"))
 
-				# Массовая пометка проданных в PostgreSQL.
+				# Считаем будущие sold.
+				would_mark = session.execute(text("""
+					SELECT count(*)
+					FROM {car_table} c
+					LEFT JOIN _active_urls a ON replace(c.origin_url, '~', '-') = a.url
+					WHERE a.url IS NULL
+					AND c.is_sold = FALSE
+					AND c.origin_id LIKE 'iaai:%%'
+				""".format(car_table=CAR_TABLE_NAME))).scalar() or 0
+
+				# Защита от аномалии.
+				if total_active > 100 and would_mark > total_active * 0.8:
+					logger.error(
+						"mark_sold safety abort: would mark %d/%d (%.0f%%) as sold — likely URL format mismatch",
+						would_mark, total_active, would_mark / total_active * 100,
+					)
+					return 0
+
+				# Массовая пометка sold.
 				result = session.execute(text("""
-					UPDATE cars
+					UPDATE {car_table}
 					SET is_sold = TRUE
 					FROM (
 						SELECT c.id
-						FROM cars c
+						FROM {car_table} c
-						LEFT JOIN _active_urls a ON c.origin_url = a.url
+						LEFT JOIN _active_urls a ON replace(c.origin_url, '~', '-') = a.url
 						WHERE a.url IS NULL
 						AND c.is_sold = FALSE
 						AND c.origin_id LIKE 'iaai:%%'
 					) sub
-					WHERE cars.id = sub.id
+					WHERE {car_table}.id = sub.id
-				"""))
+				""".format(car_table=CAR_TABLE_NAME)))
 				count = result.rowcount or 0
 			else:
 				# Упрощённый путь для SQLite.
 				stmt = (
 					update(Car)
-					.where(Car.origin_url.notin_(active_origin_urls))
 					.where(Car.is_sold == False)  # noqa: E712
 					.where(Car.origin_id.like("iaai:%"))
 					.values(is_sold=True)
 				)
-				result = session.execute(stmt)
+				# Загружаем active URL.
-				count = result.rowcount or 0
+				all_active = session.execute(
+					select(Car.id, Car.origin_url).where(
+						Car.is_sold == False, Car.origin_id.like("iaai:%")  # noqa: E712
+					)
+				).all()
+				mark_ids = [row[0] for row in all_active if _norm(row[1]) not in normalized_urls]
+
+				if not mark_ids:
+					return 0
+				total_active = len(all_active)
+				if total_active > 100 and len(mark_ids) > total_active * 0.8:
+					logger.error(
+						"mark_sold safety abort: would mark %d/%d (%.0f%%) as sold — likely URL format mismatch",
+						len(mark_ids), total_active, len(mark_ids) / total_active * 100,
+					)
+					return 0
+
+				for i in range(0, len(mark_ids), _IN_CHUNK_SIZE):
+					chunk = mark_ids[i:i + _IN_CHUNK_SIZE]
+					session.execute(update(Car).where(Car.id.in_(chunk)).values(is_sold=True))
+				count = len(mark_ids)
 
 			if count:
 				logger.info("Marked %d cars as sold by URL (no longer in listing)", count)
@@ -573,4 +631,46 @@ class PersistenceService:
 			result = session.execute(
 				select(Car.origin_id).where(Car.origin_id.like(f"{prefix}%")).execution_options(yield_per=10000)
 			)
-			return {str(row[0]) for row in result if row and row[0]}
+			return {str(row[0]) for row in result if row and row[0]}
+
+	def get_all_active_origin_urls_for_lane(self, prefix: str = "iaai:") -> set[str]:
+		"""Возвращает origin_url всех активных (не проданных) авто для заданного lane-префикса."""
+		with self.session_scope() as session:
+			result = session.execute(
+				select(Car.origin_url).where(
+					Car.origin_id.like(f"{prefix}%"),
+					Car.is_sold == False,  # noqa: E712
+				).execution_options(yield_per=10000)
+			)
+			return {str(row[0]) for row in result if row and row[0]}
+
+	def count_active_cars_for_lane(self, prefix: str = "iaai:") -> int:
+		"""Возвращает количество активных (не проданных) авто для заданного lane-префикса."""
+		from sqlalchemy import func as sa_func
+		with self.session_scope() as session:
+			result = session.execute(
+				select(sa_func.count()).select_from(Car).where(
+					Car.origin_id.like(f"{prefix}%"),
+					Car.is_sold == False,  # noqa: E712
+				)
+			)
+			return int(result.scalar() or 0)
+
+	def get_active_origin_urls_batch_for_refresh(
+		self,
+		prefix: str = "iaai:",
+		offset: int = 0,
+		limit: int = 500,
+	) -> list[str]:
+		"""Возвращает батч origin_url активных авто для rolling refresh.
+
+		Сортировка по last_seen_at ASC — давно не обновлённые идут первыми.
+		"""
+		with self.session_scope() as session:
+			result = session.execute(
+				select(Car.origin_url).where(
+					Car.origin_id.like(f"{prefix}%"),
+					Car.is_sold == False,  # noqa: E712
+				).order_by(Car.last_seen_at.asc()).offset(offset).limit(limit)
+			)
+			return [str(row[0]) for row in result if row and row[0]]

iaai_scraper/storage/models.py

@@ -20,10 +20,10 @@ class Base(DeclarativeBase):
 
 
 class Car(Base):
-	__tablename__ = "cars"
+	__tablename__ = "iaai_cars"
 	__table_args__ = (
-		Index("ix_cars_brand_model", "brand", "model"),
+		Index("ix_iaai_cars_brand_model", "brand", "model"),
-		Index("ix_cars_origin_id_not_sold", "origin_id", "is_sold"),
+		Index("ix_iaai_cars_origin_id_not_sold", "origin_id", "is_sold"),
 	)
 	id: Mapped[int] = mapped_column(BigInteger().with_variant(Integer, "sqlite"), primary_key=True, autoincrement=True)
 	parser_id: Mapped[str] = mapped_column(String(50), nullable=False, unique=True)
@@ -59,17 +59,17 @@ class Car(Base):
 
 
 class Image(Base):
-	__tablename__ = "images"
+	__tablename__ = "iaai_images"
 	id: Mapped[int] = mapped_column(BigInteger().with_variant(Integer, "sqlite"), primary_key=True, autoincrement=True)
 	fullres_image: Mapped[str] = mapped_column(String(), nullable=False)
 	preview_image: Mapped[str] = mapped_column(String(), nullable=False)
 	order_index: Mapped[int] = mapped_column(Integer, nullable=False)
-	car_id: Mapped[int] = mapped_column(Integer, ForeignKey("cars.id", ondelete="CASCADE"), nullable=False, index=True)
+	car_id: Mapped[int] = mapped_column(Integer, ForeignKey("iaai_cars.id", ondelete="CASCADE"), nullable=False, index=True)
 	car: Mapped[Car] = relationship("Car", back_populates="images")
 
 
 class SyncRun(Base):
-	__tablename__ = "sync_runs"
+	__tablename__ = "iaai_sync_runs"
 	id: Mapped[int] = mapped_column(BigInteger().with_variant(Integer, "sqlite"), primary_key=True, autoincrement=True)
 	started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=func.now())
 	finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)

iaai_scraper/worker/celery_app.py

@@ -1,14 +1,23 @@
 # Инициализация Celery-приложения и периодических задач.
 
 import logging
+import os
 
 from celery import Celery
 from celery.signals import worker_process_init, worker_ready, setup_logging as celery_setup_logging
+from redis import Redis
 
 from ..core.config import settings
 from ..core.logs import setup_logging
 
 logger = logging.getLogger("iaai_scraper.worker.celery_app")
+STARTUP_SYNC_DISPATCH_KEY = "iaai:state:startup_sync_dispatched"
+IAAI_SYNC_QUEUE = "iaai_sync"
+
+
+def _env_bool(name: str, default: bool) -> bool:
+    raw = os.getenv(name, "true" if default else "false").strip().lower()
+    return raw in {"1", "true", "yes", "on"}
 
 
 @celery_setup_logging.connect
@@ -66,7 +75,7 @@ celery_app.conf.update(
     task_track_started=True,
     worker_concurrency=settings.celery.worker_concurrency,
     worker_max_tasks_per_child=settings.celery.worker_max_tasks_per_child,
-    worker_pool="prefork",
+    worker_pool="solo",
     worker_prefetch_multiplier=1,
     broker_connection_retry_on_startup=True,
     broker_transport_options={
@@ -77,15 +86,19 @@ celery_app.conf.update(
     worker_hijack_root_logger=False,
     beat_schedule={
         "periodic-sync-listing": {
-            "task": "iaai_scraper.worker.tasks.sync_listing_task",
+            "task": "iaai.sync_cars_feed",
             "schedule": settings.celery.beat_sync_interval_minutes * 60.0,
             "args": (),
-            "kwargs": {"limit": settings.celery.beat_sync_limit, "only_new": False},
+            "kwargs": {"limit": settings.celery.beat_sync_limit, "only_new": True},
-            "options": {"queue": "scraping"},
+            "options": {
+                "queue": IAAI_SYNC_QUEUE,
+                "expires": settings.celery.beat_sync_interval_minutes * 60.0,
+            },
         }
     },
     task_routes={
-        "iaai_scraper.worker.tasks.*": {"queue": "scraping"}
+        "iaai.sync_cars_feed": {"queue": IAAI_SYNC_QUEUE},
+        "iaai_scraper.worker.tasks.*": {"queue": IAAI_SYNC_QUEUE},
     },
 )
 
@@ -94,11 +107,58 @@ celery_app.autodiscover_tasks(["iaai_scraper.worker"])
 
 @worker_ready.connect
 def _on_worker_ready(**kwargs):
-    """Сразу при старте worker отправляем первую задачу sync_listing,
+    """При старте worker отправляем первый sync_listing, если очередь пуста."""
-    чтобы не ждать час до первого beat-цикла."""
+    if not _env_bool("IAAI_STARTUP_SYNC_ENABLED", True):
+        logger.info("Worker ready: startup sync dispatch disabled by IAAI_STARTUP_SYNC_ENABLED")
+        return
+
+    redis_client = None
+    try:
+        redis_client = Redis.from_url(
+            settings.redis.url,
+            decode_responses=True,
+            socket_connect_timeout=settings.redis.socket_connect_timeout_seconds,
+            socket_timeout=settings.redis.socket_timeout_seconds,
+            health_check_interval=settings.redis.health_check_interval_seconds,
+            retry_on_timeout=True,
+        )
+
+        for stale_key in ("iaai:locks:sync_listing",):
+            try:
+                ttl = redis_client.ttl(stale_key)
+                if ttl is not None and ttl != -2:
+                    redis_client.delete(stale_key)
+                    logger.warning("Cleared stale lock on startup: %s (ttl was %s)", stale_key, ttl)
+            except Exception:
+                logger.warning("Failed to clear stale lock %s on startup", stale_key, exc_info=True)
+
+        try:
+            queue_len = int(redis_client.llen(IAAI_SYNC_QUEUE) or 0)
+        except Exception:
+            queue_len = 0
+        if queue_len > 0:
+            logger.info("Worker ready: iaai_sync queue already has %d task(s); skip startup dispatch", queue_len)
+            return
+
+        should_dispatch = bool(redis_client.set(STARTUP_SYNC_DISPATCH_KEY, "1", nx=True, ex=600))
+    except Exception:
+        logger.warning("Worker ready startup sync dedupe check failed; skipping immediate dispatch", exc_info=True)
+        return
+    finally:
+        if redis_client is not None:
+            try:
+                redis_client.close()
+            except Exception:
+                pass
+
+    if not should_dispatch:
+        logger.info("Worker ready immediate sync already dispatched recently; skipping duplicate enqueue")
+        return
+
     logger.info("Worker ready — dispatching initial sync_listing task")
     celery_app.send_task(
-        "iaai_scraper.worker.tasks.sync_listing_task",
+        "iaai.sync_cars_feed",
         kwargs={"limit": settings.celery.beat_sync_limit, "only_new": False},
-        queue="scraping",
+        queue=IAAI_SYNC_QUEUE,
-    )
+        expires=settings.celery.beat_sync_interval_minutes * 60.0,
+    )

iaai_scraper/worker/self_heal.py

@@ -0,0 +1,219 @@
+import json
+import logging
+import os
+import random
+import signal
+import time
+
+from redis import Redis
+
+
+logger = logging.getLogger("iaai_scraper.worker.self_heal")
+
+GLOBAL_PROGRESS_TS_KEY = "iaai:state:last_progress_ts"
+SELF_HEAL_RESTART_LOCK_KEY = "iaai:state:self_heal_restart_in_progress"
+SYNC_LISTING_LOCK_KEY = "iaai:locks:sync_listing"
+SIGKILL_FALLBACK = getattr(signal, "SIGKILL", signal.SIGTERM)
+
+
+def _env_bool(name: str, default: bool) -> bool:
+	value = os.getenv(name)
+	if value is None:
+		return default
+	return value.strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _env_int(name: str, default: int) -> int:
+	value = os.getenv(name)
+	if value is None:
+		return default
+	try:
+		return int(value.strip())
+	except Exception:
+		return default
+
+
+def _get_redis() -> Redis:
+	url = os.getenv("IAAI_REDIS_URL", "redis://redis:6379/0")
+	return Redis.from_url(
+		url,
+		decode_responses=True,
+		socket_connect_timeout=5.0,
+		socket_timeout=10.0,
+		health_check_interval=30,
+		retry_on_timeout=True,
+	)
+
+
+def _safe_int(value: str | None, default: int = 0) -> int:
+	if value is None:
+		return default
+	try:
+		return int(str(value).strip())
+	except Exception:
+		return default
+
+
+def _read_last_progress_ts(redis_client: Redis) -> int | None:
+	raw = redis_client.get(GLOBAL_PROGRESS_TS_KEY)
+	if raw:
+		ts = _safe_int(raw)
+		if ts > 0:
+			return ts
+
+	# Fallback: если глобальный ключ не найден, берём max(ts) из task_progress:*.
+	# Это дороже, но выполняется только при отсутствии основного маркера.
+	max_ts = 0
+	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)
+			ts = _safe_int(data.get("ts"), 0)
+			if ts > max_ts:
+				max_ts = ts
+		except Exception:
+			continue
+	return max_ts or None
+
+
+def _has_inflight_work(redis_client: Redis, queue_name: str) -> tuple[bool, dict[str, int]]:
+	"""Есть ли признаки активной/зависшей работы, даже если очередь пуста."""
+	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:*"):
+		has_task_progress = 1
+		break
+	flags = {
+		"queue_len": queue_len,
+		"has_lock": has_lock,
+		"has_task_progress": has_task_progress,
+	}
+	return (queue_len > 0 or has_lock == 1 or has_task_progress == 1), flags
+
+
+def _kill_worker_process() -> None:
+	pid_file = "/tmp/celery-worker.pid"
+	pid: int | None = None
+	try:
+		with open(pid_file, "r", encoding="utf-8") as f:
+			pid = int(f.read().strip())
+	except Exception:
+		pid = None
+
+	if not pid:
+		logger.error("Self-heal: failed to read worker pid from %s", pid_file)
+		return
+
+	logger.error("Self-heal: terminating stuck worker process pid=%s", pid)
+	try:
+		os.kill(pid, signal.SIGTERM)
+	except Exception:
+		logger.exception("Self-heal: failed to send SIGTERM to pid=%s", pid)
+		return
+
+	time.sleep(20)
+	try:
+		# Если процесс ещё жив — принудительно убиваем.
+		os.kill(pid, 0)
+		logger.error("Self-heal: worker pid=%s did not stop after SIGTERM; sending SIGKILL", pid)
+		os.kill(pid, SIGKILL_FALLBACK)
+	except ProcessLookupError:
+		pass
+	except Exception:
+		logger.exception("Self-heal: failed to send SIGKILL to pid=%s", pid)
+
+
+def main() -> None:
+	if not _env_bool("IAAI_SELF_HEAL_ENABLED", True):
+		logger.info("Self-heal watchdog disabled via IAAI_SELF_HEAL_ENABLED")
+		return
+
+	queue_name = os.getenv("IAAI_CELERY_QUEUE", "scraping")
+	check_interval = max(5, _env_int("IAAI_SELF_HEAL_CHECK_INTERVAL_SECONDS", 30))
+	stall_seconds = max(180, _env_int("IAAI_SELF_HEAL_STALL_SECONDS", 720))
+	startup_grace = max(30, _env_int("IAAI_SELF_HEAL_STARTUP_GRACE_SECONDS", 300))
+	restart_cooldown = max(60, _env_int("IAAI_SELF_HEAL_RESTART_COOLDOWN_SECONDS", 300))
+
+	logger.warning(
+		"Self-heal watchdog enabled: queue=%s check_interval=%ss stall=%ss startup_grace=%ss cooldown=%ss",
+		queue_name,
+		check_interval,
+		stall_seconds,
+		startup_grace,
+		restart_cooldown,
+	)
+
+	started_at = time.time()
+	redis_client: Redis | None = None
+
+	while True:
+		try:
+			if redis_client is None:
+				redis_client = _get_redis()
+				redis_client.ping()
+
+			has_inflight, inflight = _has_inflight_work(redis_client, queue_name)
+			if not has_inflight:
+				time.sleep(check_interval)
+				continue
+
+			last_progress_ts = _read_last_progress_ts(redis_client)
+			now_ts = int(time.time())
+			age = None if last_progress_ts is None else max(0, now_ts - int(last_progress_ts))
+
+			if age is None:
+				if now_ts - int(started_at) < startup_grace:
+					time.sleep(check_interval)
+					continue
+				logger.warning(
+					"Self-heal: inflight=%s but no progress timestamp found after startup grace",
+					inflight,
+				)
+				age = stall_seconds + 1
+
+			if age <= stall_seconds:
+				time.sleep(check_interval)
+				continue
+
+			# Глобальный anti-storm lock: чтобы много воркеров не рестартились одновременно.
+			acquired = bool(
+				redis_client.set(
+					SELF_HEAL_RESTART_LOCK_KEY,
+					str(now_ts),
+					nx=True,
+					ex=restart_cooldown,
+				)
+			)
+			if not acquired:
+				time.sleep(check_interval)
+				continue
+
+			logger.error(
+				"Self-heal: detected global stall (inflight=%s, progress_age=%ss > %ss). Restarting worker process...",
+				inflight,
+				age,
+				stall_seconds,
+			)
+			# Небольшой джиттер, чтобы при одинаковом событии у разных контейнеров
+			# перезапуск был не строго одновременно.
+			time.sleep(random.uniform(0.3, 2.0))
+			_kill_worker_process()
+			# После kill pid1 контейнер будет перезапущен Docker restart-policy.
+			# На случай неуспеха не молотим цикл.
+			time.sleep(check_interval)
+
+		except Exception:
+			logger.exception("Self-heal watchdog iteration failed")
+			redis_client = None
+			time.sleep(check_interval)
+
+
+if __name__ == "__main__":
+	logging.basicConfig(
+		level=os.getenv("IAAI_LOG_LEVEL", "INFO"),
+		format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
+	)
+	main()

tests/test_listing.py

@@ -35,7 +35,7 @@ class TestListingUnit(unittest.TestCase):
         self.assertTrue(ListingCollector._has_next_page(_FakePage({"a[aria-label*='Next']":  1})))
         # Нет селекторов → False.
         self.assertFalse(ListingCollector._has_next_page(_FakePage({})))
-        # Числовая пагинация через JS → True.
+        # Числовая пагинация через JS → True только если evaluate явно нашёл next.
         page = _FakePage({})
         page._evaluate_result = True
         self.assertTrue(ListingCollector._has_next_page(page))

tests/test_resilience.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+import unittest
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from iaai_scraper.scraper import IAAIScraper
+from iaai_scraper.core.config import Settings
+from iaai_scraper.worker import tasks
+
+
+class TestResilience(unittest.TestCase):
+	def _make_scraper(self) -> IAAIScraper:
+		s = Settings()
+		s.log_level = "CRITICAL"
+		s.database.url = "sqlite://"
+		return IAAIScraper(s)
+
+	def test_update_task_progress_updates_global_marker(self) -> None:
+		redis_client = MagicMock()
+		pipe = MagicMock()
+		redis_client.pipeline.return_value = pipe
+
+		tasks._update_task_progress(
+			redis_client,
+			task_id="task-abc",
+			stage="batch_upserted",
+			ttl_seconds=180,
+			cars_upserted=10,
+		)
+
+		redis_client.pipeline.assert_called_once()
+		self.assertEqual(pipe.set.call_count, 2)
+		first_call = pipe.set.call_args_list[0]
+		second_call = pipe.set.call_args_list[1]
+
+		self.assertEqual(first_call.args[0], tasks._task_progress_key("task-abc"))
+		self.assertEqual(second_call.args[0], tasks.GLOBAL_PROGRESS_TS_KEY)
+		pipe.execute.assert_called_once()
+
+	def test_streaming_sync_stops_cleanly_when_page_stays_empty(self) -> None:
+		scraper = self._make_scraper()
+		scraper.settings.celery.batch_size = 50
+
+		page = MagicMock()
+		empty_page_result = SimpleNamespace(
+			page_number=1,
+			vehicle_links=[],
+			next_page_detected=True,
+		)
+
+		scraper._open_listing_for_stream = MagicMock(return_value=(
+			page,
+			{"make": None, "model": None, "year_min": None, "year_max": None},
+		))
+		scraper.listing_collector.collect_current_page = MagicMock(return_value=empty_page_result)
+		scraper._recover_empty_listing_page = MagicMock(return_value=(empty_page_result, []))
+		scraper.sync_batch = MagicMock()
+
+		result = scraper._sync_listing_streaming(
+			make=None,
+			model=None,
+			lane="iaai_cars",
+			limit=None,
+			effective_only_new=False,
+			started_at=0.0,
+			listing_url="https://www.iaai.com/Vehiclelisting/Cars?Make=TEST",
+		)
+
+		self.assertEqual(result["total"], 0)
+		self.assertEqual(result["cars_upserted"], 0)
+		self.assertEqual(result["cars_failed"], 0)
+		self.assertEqual(result["listing"]["pages_collected"], 1)
+		scraper._recover_empty_listing_page.assert_called_once()
+		scraper.sync_batch.assert_not_called()
+
+
+if __name__ == "__main__":
+	unittest.main()

tests/test_scraper.py

@@ -1,8 +1,9 @@
 from __future__ import annotations
 
 import unittest
+from concurrent.futures import TimeoutError as FuturesTimeoutError
 from types import SimpleNamespace
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 
 from iaai_scraper.core.config import Settings
 from iaai_scraper.core.exceptions import AntiBotDetectedError, SiteStructureChangedError
@@ -107,20 +108,52 @@ class TestScraperSync(unittest.TestCase):
             {"make": "HONDA", "year_min": None, "year_max": None},
         ]
 
-        # Resume: пропускаем TOYOTA, начинаем с FORD на стр. 5.
+        # Resume: пропускаем TOYOTA, начинаем сразу с FORD.
         result = scraper.sync_listing_segmented(
-            segments=segments, start_segment=1, start_page=5,
+            segments=segments, start_segment=1,
         )
 
         self.assertEqual(result["segments_completed"], 2)  # FORD + HONDA
         self.assertEqual(result["cars_upserted"], 10)
-        # FORD: start_page=5, HONDA: start_page=1.
-        self.assertEqual(call_args_log[0]["start_page"], 5)
-        self.assertEqual(call_args_log[1]["start_page"], 1)
         # URL содержит бренд.
         self.assertIn("FORD", call_args_log[0]["listing_url"])
         self.assertIn("HONDA", call_args_log[1]["listing_url"])
 
+    def test_segmented_sync_does_not_mark_full_scan_completed_when_segment_failed(self) -> None:
+        scraper = self._make_scraper()
+        scraper.sync_listing = MagicMock(side_effect=[
+            {
+                "status": "failed",
+                "full_scan_completed": False,
+                "cars_upserted": 0,
+                "cars_failed": 0,
+                "images_upserted": 0,
+                "skipped_existing": 0,
+                "listing": {"vehicles_collected": 0},
+                "failures": [{"vehicle_url": "segment", "error": "resume failed"}],
+            },
+            {
+                "status": "success",
+                "full_scan_completed": True,
+                "cars_upserted": 1,
+                "cars_failed": 0,
+                "images_upserted": 0,
+                "skipped_existing": 0,
+                "listing": {"vehicles_collected": 10},
+                "failures": [],
+            },
+        ])
+
+        result = scraper.sync_listing_segmented(
+            segments=[
+                {"make": "EAGLE", "year_min": None, "year_max": None},
+                {"make": "FORD", "year_min": None, "year_max": None},
+            ]
+        )
+
+        self.assertFalse(result["full_scan_completed"])
+        self.assertEqual(result["status"], "partial_success")
+
     def test_build_segment_listing_url(self) -> None:
         base = "https://www.iaai.com/Vehiclelisting/Cars"
         self.assertEqual(
@@ -188,6 +221,99 @@ class TestScraperSync(unittest.TestCase):
         scraper.listing_collector.open_cars_listing.assert_called_once()
         page.reload.assert_not_called()
 
+    def test_sync_listing_always_starts_from_page_one(self) -> None:
+        scraper = self._make_scraper()
+        scraper.settings.celery.batch_size = 1
+
+        page = MagicMock()
+        page_result = SimpleNamespace(
+            page_number=1,
+            vehicle_links=[SimpleNamespace(href="https://www.iaai.com/VehicleDetail/999~US", lot_number="999")],
+            next_page_detected=False,
+        )
+
+        scraper._get_page_with_warmup = MagicMock(return_value=page)
+        # Pagination-resume убран: _reopen_listing_and_resume не должен вызываться.
+        scraper._reopen_listing_and_resume = MagicMock(
+            side_effect=AssertionError("pagination resume must not be used")
+        )
+        scraper.listing_collector.open_cars_listing = MagicMock()
+        scraper.listing_collector.apply_filters = MagicMock(return_value={
+            "make": None,
+            "model": None,
+            "year_min": None,
+            "year_max": None,
+        })
+        scraper.listing_collector.collect_current_page = MagicMock(return_value=page_result)
+        scraper._extract_page_urls = MagicMock(return_value=["https://www.iaai.com/VehicleDetail/999~US"])
+        scraper.sync_batch = MagicMock(return_value={
+            "cars_upserted": 1,
+            "cars_failed": 0,
+            "images_upserted": 0,
+            "failures": [],
+        })
+
+        result = scraper._sync_listing_streaming(
+            make=None,
+            model=None,
+            lane="iaai_cars",
+            limit=None,
+            effective_only_new=False,
+            started_at=0.0,
+            listing_url="https://www.iaai.com/Vehiclelisting/Cars?Make=EAGLE",
+        )
+
+        scraper.listing_collector.open_cars_listing.assert_called_once()
+        scraper._reopen_listing_and_resume.assert_not_called()
+        scraper.sync_batch.assert_called_once()
+        self.assertEqual(result["cars_upserted"], 1)
+        self.assertEqual(result["total"], 1)
+
+    def test_sync_batch_timeout_does_not_duplicate_already_processed_urls(self) -> None:
+        scraper = self._make_scraper()
+        scraper.persistence.upsert_cars_batch = MagicMock(return_value={
+            "inserted": 1,
+            "updated": 0,
+            "images_upserted": 0,
+        })
+
+        urls = [
+            "https://www.iaai.com/VehicleDetail/111~US",
+            "https://www.iaai.com/VehicleDetail/222~US",
+            "https://www.iaai.com/VehicleDetail/333~US",
+        ]
+        first_record = CarRecord.model_validate(make_db_record("111"))
+
+        class _FakeExecutor:
+            def __init__(self, *args, **kwargs):
+                pass
+
+            def __enter__(self):
+                return self
+
+            def __exit__(self, exc_type, exc, tb):
+                return False
+
+            def map(self, fn, iterable, timeout=None):  # noqa: ARG002
+                yield 0, first_record
+                raise FuturesTimeoutError()
+
+        with patch("iaai_scraper.core.runtime_config.RuntimeFiltersConfig.is_empty", return_value=True), \
+             patch("iaai_scraper.scraper.ThreadPoolExecutor", _FakeExecutor), \
+             patch.object(scraper, "_browser_fallback_parallel", return_value={
+                 "records": [],
+                 "failures": [],
+                 "cars_failed": 0,
+                 "protection_events": 0,
+             }) as fallback_mock:
+            result = scraper.sync_batch(urls)
+
+        self.assertEqual(result["cars_upserted"], 1)
+        fallback_urls = fallback_mock.call_args.args[0]
+        self.assertEqual(len(fallback_urls), 2)
+        self.assertEqual({u for u, _ in fallback_urls}, {urls[1], urls[2]})
+        self.assertNotIn(urls[0], {u for u, _ in fallback_urls})
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_self_heal.py

@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+import unittest
+from unittest.mock import MagicMock, patch
+
+from iaai_scraper.worker import self_heal
+
+
+class TestSelfHeal(unittest.TestCase):
+	def test_read_last_progress_ts_uses_global_key(self) -> None:
+		redis_client = MagicMock()
+		redis_client.get.return_value = "1776800000"
+
+		ts = self_heal._read_last_progress_ts(redis_client)
+
+		self.assertEqual(ts, 1776800000)
+		redis_client.scan_iter.assert_not_called()
+
+	def test_read_last_progress_ts_fallbacks_to_task_progress_keys(self) -> None:
+		redis_client = MagicMock()
+		redis_client.get.side_effect = lambda key: {
+			self_heal.GLOBAL_PROGRESS_TS_KEY: None,
+			"iaai:state:task_progress:a": '{"ts": 100}',
+			"iaai:state:task_progress:b": '{"ts": 250}',
+			"iaai:state:task_progress:c": '{"ts": 150}',
+		}.get(key)
+		redis_client.scan_iter.return_value = [
+			"iaai:state:task_progress:a",
+			"iaai:state:task_progress:b",
+			"iaai:state:task_progress:c",
+		]
+
+		ts = self_heal._read_last_progress_ts(redis_client)
+
+		self.assertEqual(ts, 250)
+
+	def test_read_last_progress_ts_ignores_broken_payloads(self) -> None:
+		redis_client = MagicMock()
+		redis_client.get.side_effect = lambda key: {
+			self_heal.GLOBAL_PROGRESS_TS_KEY: None,
+			"iaai:state:task_progress:a": "{bad-json}",
+			"iaai:state:task_progress:b": '{"foo": "bar"}',
+		}.get(key)
+		redis_client.scan_iter.return_value = [
+			"iaai:state:task_progress:a",
+			"iaai:state:task_progress:b",
+		]
+
+		ts = self_heal._read_last_progress_ts(redis_client)
+
+		self.assertIsNone(ts)
+
+	@patch("iaai_scraper.worker.self_heal.time.sleep", return_value=None)
+	@patch("iaai_scraper.worker.self_heal.os.kill")
+	@patch("builtins.open")
+	def test_kill_worker_process_sends_term_and_kill(self, open_mock, kill_mock, _sleep_mock) -> None:
+		open_mock.return_value.__enter__.return_value.read.return_value = "123"
+		# SIGTERM -> process alive check (pid,0) -> SIGKILL
+		kill_mock.side_effect = [None, None, None]
+
+		self_heal._kill_worker_process()
+
+		self.assertEqual(kill_mock.call_args_list[0].args[0], 123)
+		self.assertEqual(kill_mock.call_args_list[1].args, (123, 0))
+		self.assertEqual(kill_mock.call_args_list[2].args[0], 123)
+
+	@patch("iaai_scraper.worker.self_heal.time.sleep", return_value=None)
+	@patch("iaai_scraper.worker.self_heal.os.kill")
+	@patch("builtins.open")
+	def test_kill_worker_process_skips_sigkill_when_already_exited(self, open_mock, kill_mock, _sleep_mock) -> None:
+		open_mock.return_value.__enter__.return_value.read.return_value = "123"
+		kill_mock.side_effect = [None, ProcessLookupError()]
+
+		self_heal._kill_worker_process()
+
+		# Только SIGTERM и проверка существования процесса.
+		self.assertEqual(len(kill_mock.call_args_list), 2)
+
+
+if __name__ == "__main__":
+	unittest.main()

tests/test_worker_tasks.py

@@ -1,10 +1,12 @@
 from __future__ import annotations
 
 import json
+from dataclasses import replace
 import unittest
 from unittest.mock import MagicMock, patch
 
 from iaai_scraper.worker import tasks
+from iaai_scraper.core.config import settings as base_settings
 
 
 class TestWorkerTaskLockHelpers(unittest.TestCase):
@@ -146,22 +148,84 @@ class TestWorkerTaskLockHelpers(unittest.TestCase):
         self.assertEqual(acquire_lock.call_count, 2)
         release_lock.assert_called_once()
 
-    def test_sync_listing_checkpoint_save_load_and_resume(self) -> None:
+    def test_checkpoint_save_load_roundtrip(self) -> None:
-        # Roundtrip: save → load.
-        redis_client = MagicMock()
         storage: dict[str, str] = {}
-        redis_client.set.side_effect = lambda key, value: storage.__setitem__(key, value)
+
+        def _fake_set(key, value, ex=None):
+            storage[key] = value
+            return True
+
+        redis_client = MagicMock()
+        redis_client.set.side_effect = _fake_set
         redis_client.get.side_effect = lambda key: storage.get(key)
 
-        tasks._save_sync_checkpoint(
+        tasks._save_last_completed_segment(redis_client, 12)
-            redis_client, task_id="task-1", page_number=12,
-            make=None, model=None, lane="iaai_cars",
-        )
-        checkpoint = tasks._load_sync_checkpoint(redis_client)
-        self.assertIsNotNone(checkpoint)
-        self.assertEqual(checkpoint["last_successful_page"], 12)
 
-        # Resume: задача стартует со страницы checkpoint + 1.
+        self.assertEqual(tasks._load_last_completed_segment(redis_client), 12)
+        self.assertEqual(redis_client.set.call_args.kwargs["ex"], tasks.SYNC_LISTING_CHECKPOINT_TTL_SECONDS)
+
+    def test_load_checkpoint_clears_invalid_payload(self) -> None:
+        redis_client = MagicMock()
+        redis_client.get.return_value = "{not-a-number"
+
+        self.assertIsNone(tasks._load_last_completed_segment(redis_client))
+        redis_client.delete.assert_called_once_with(tasks.SYNC_LISTING_CHECKPOINT_KEY)
+
+    def test_load_checkpoint_empty_when_missing(self) -> None:
+        redis_client = MagicMock()
+        redis_client.get.return_value = None
+
+        self.assertIsNone(tasks._load_last_completed_segment(redis_client))
+        redis_client.delete.assert_not_called()
+
+    def test_sync_listing_resumes_from_next_segment_during_bootstrap(self) -> None:
+        segments = [
+            {"make": "ACURA"},
+            {"make": "AUDI"},
+            {"make": "BMW"},
+            {"make": "EAGLE"},
+        ]
+        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=False), \
+             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", side_effect=lambda fn: fn()), \
+             patch.object(tasks.sync_listing_task, "update_state"), \
+             patch("iaai_scraper.worker.tasks.parse_listing_segments", return_value=segments):
+            get_persistence.return_value = MagicMock()
+            redis_client = MagicMock()
+            redis_client.get.side_effect = lambda key: (
+                "1" if key == tasks.SYNC_LISTING_CHECKPOINT_KEY else None
+            )
+            get_redis.return_value = redis_client
+            start_heartbeat.return_value = (MagicMock(), MagicMock())
+
+            sync_segmented_mock = MagicMock(return_value={
+                "run_id": 11, "status": "success", "full_scan_completed": True,
+                "cars_upserted": 1, "cars_failed": 0, "images_upserted": 0,
+                "skipped_existing": 0, "elapsed_seconds": 1.0, "failures": [],
+            })
+            scraper_ctx = MagicMock()
+            scraper_ctx.__enter__.return_value.sync_listing_segmented = sync_segmented_mock
+            scraper_ctx.__exit__.return_value = None
+
+            with patch.object(tasks, "IAAIScraper", return_value=scraper_ctx):
+                tasks.sync_listing_task.push_request(id="task-resume-seg")
+                try:
+                    result = tasks.sync_listing_task.run()
+                finally:
+                    tasks.sync_listing_task.pop_request()
+
+        self.assertEqual(result["status"], "success")
+        # last_completed=1 → start_segment=2 (AUDI завершён, возобновляем с BMW).
+        self.assertEqual(sync_segmented_mock.call_args.kwargs["start_segment"], 2)
+        self.assertEqual(sync_segmented_mock.call_args.kwargs["start_page"], 1)
+        release_lock.assert_called_once()
+
+    def test_sync_listing_ignores_checkpoint_after_full_scan_completed(self) -> None:
+        segments = [{"make": "ACURA"}, {"make": "AUDI"}]
         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), \
@@ -171,39 +235,362 @@ class TestWorkerTaskLockHelpers(unittest.TestCase):
              patch.object(tasks, "_clear_sync_checkpoint") as clear_checkpoint, \
              patch.object(tasks, "_run_browser_job", side_effect=lambda fn: fn()), \
              patch.object(tasks.sync_listing_task, "update_state"), \
-             patch("iaai_scraper.worker.tasks.parse_listing_segments", return_value=[]):
+             patch("iaai_scraper.worker.tasks.parse_listing_segments", return_value=segments):
             get_persistence.return_value = MagicMock()
-            redis_client2 = MagicMock()
+            redis_client = MagicMock()
-            redis_client2.get.side_effect = lambda key: json.dumps({
+            # Оставшийся чекпоинт не должен использоваться.
-                "status": "in_progress", "last_successful_page": 9,
+            redis_client.get.side_effect = lambda key: (
-                "make": None, "model": None, "lane": "iaai_cars",
+                "0" if key == tasks.SYNC_LISTING_CHECKPOINT_KEY else None
-            }) if key == tasks.SYNC_LISTING_CHECKPOINT_KEY else None
+            )
-            get_redis.return_value = redis_client2
+            get_redis.return_value = redis_client
-            stop_event = MagicMock()
+            start_heartbeat.return_value = (MagicMock(), MagicMock())
-            heartbeat_thread = MagicMock()
-            start_heartbeat.return_value = (stop_event, heartbeat_thread)
 
-            sync_listing_mock = MagicMock(return_value={
+            sync_segmented_mock = MagicMock(return_value={
-                "run_id": 11, "status": "success", "full_scan_completed": True,
+                "run_id": 14, "status": "success", "full_scan_completed": True,
                 "cars_upserted": 1, "cars_failed": 0, "images_upserted": 0,
                 "skipped_existing": 0, "elapsed_seconds": 1.0, "failures": [],
             })
             scraper_ctx = MagicMock()
-            scraper_ctx.__enter__.return_value.sync_listing = sync_listing_mock
+            scraper_ctx.__enter__.return_value.sync_listing_segmented = sync_segmented_mock
             scraper_ctx.__exit__.return_value = None
 
             with patch.object(tasks, "IAAIScraper", return_value=scraper_ctx):
-                tasks.sync_listing_task.push_request(id="task-789")
+                tasks.sync_listing_task.push_request(id="task-792")
                 try:
                     result = tasks.sync_listing_task.run()
                 finally:
                     tasks.sync_listing_task.pop_request()
 
         self.assertEqual(result["status"], "success")
-        self.assertEqual(sync_listing_mock.call_args.kwargs["start_page"], 10)
+        self.assertEqual(sync_segmented_mock.call_args.kwargs["start_segment"], 0)
+        self.assertIsNone(sync_segmented_mock.call_args.kwargs["progress_callback"])
         clear_checkpoint.assert_called()
         release_lock.assert_called_once()
 
+    def test_sync_listing_checkpoint_beyond_segments_restarts_from_zero(self) -> None:
+        segments = [{"make": "ACURA"}, {"make": "AUDI"}]
+        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=False), \
+             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", side_effect=lambda fn: fn()), \
+             patch.object(tasks.sync_listing_task, "update_state"), \
+             patch("iaai_scraper.worker.tasks.parse_listing_segments", return_value=segments):
+            get_persistence.return_value = MagicMock()
+            redis_client = MagicMock()
+            redis_client.get.side_effect = lambda key: (
+                "99" if key == tasks.SYNC_LISTING_CHECKPOINT_KEY else None
+            )
+            get_redis.return_value = redis_client
+            start_heartbeat.return_value = (MagicMock(), MagicMock())
+
+            sync_segmented_mock = MagicMock(return_value={
+                "run_id": 15, "status": "success", "full_scan_completed": True,
+                "cars_upserted": 0, "cars_failed": 0, "images_upserted": 0,
+                "skipped_existing": 0, "elapsed_seconds": 1.0, "failures": [],
+            })
+            scraper_ctx = MagicMock()
+            scraper_ctx.__enter__.return_value.sync_listing_segmented = sync_segmented_mock
+            scraper_ctx.__exit__.return_value = None
+
+            with patch.object(tasks, "IAAIScraper", return_value=scraper_ctx):
+                tasks.sync_listing_task.push_request(id="task-beyond")
+                try:
+                    result = tasks.sync_listing_task.run()
+                finally:
+                    tasks.sync_listing_task.pop_request()
+
+        self.assertEqual(result["status"], "success")
+        self.assertEqual(sync_segmented_mock.call_args.kwargs["start_segment"], 0)
+        release_lock.assert_called_once()
+
+    def test_sync_listing_task_clears_checkpoint_on_hourly_run(self) -> None:
+        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, "_clear_sync_checkpoint") as clear_checkpoint, \
+             patch.object(tasks, "_run_browser_job", return_value={
+                 "run_id": 13,
+                 "status": "partial_success",
+                 "full_scan_completed": True,
+                 "cars_upserted": 2,
+                 "cars_failed": 1,
+                 "images_upserted": 3,
+                 "skipped_existing": 0,
+                 "elapsed_seconds": 4.0,
+                 "failures": [{"vehicle_url": "v", "error": "e"}],
+             }), \
+             patch.object(tasks.sync_listing_task, "update_state"):
+            get_persistence.return_value = MagicMock()
+            redis_client = MagicMock()
+            redis_client.get.return_value = None
+            get_redis.return_value = redis_client
+            start_heartbeat.return_value = (MagicMock(), MagicMock())
+
+            tasks.sync_listing_task.push_request(id="task-791")
+            try:
+                result = tasks.sync_listing_task.run(make="Toyota")
+            finally:
+                tasks.sync_listing_task.pop_request()
+
+        self.assertEqual(result["status"], "partial_success")
+        # Hourly-ветка (full_scan_done=True) всегда удаляет оставшийся чекпоинт
+        # до браузерного job + после. Главное — вызов произошёл.
+        clear_checkpoint.assert_called()
+        release_lock.assert_called_once()
+
+    def test_try_set_followup_pending_deduplicates(self) -> None:
+        redis_client = MagicMock()
+        redis_client.set.side_effect = [True, False]
+
+        self.assertTrue(tasks._try_set_followup_pending(redis_client, ttl_seconds=120))
+        self.assertFalse(tasks._try_set_followup_pending(redis_client, ttl_seconds=120))
+
+    def test_bump_bootstrap_failure_streak_opens_circuit_breaker(self) -> None:
+        redis_client = MagicMock()
+        redis_client.incr.return_value = tasks.SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_LIMIT
+
+        streak, should_enqueue = tasks._bump_bootstrap_failure_streak(
+            redis_client,
+            reason="max_retries_exceeded",
+        )
+
+        self.assertEqual(streak, tasks.SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_LIMIT)
+        self.assertFalse(should_enqueue)
+        redis_client.expire.assert_called_once_with(
+            tasks.SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_KEY,
+            tasks.SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_TTL_SECONDS,
+        )
+
+    def test_bump_bootstrap_failure_streak_allows_retry_before_limit(self) -> None:
+        redis_client = MagicMock()
+        redis_client.incr.return_value = 1
+
+        streak, should_enqueue = tasks._bump_bootstrap_failure_streak(
+            redis_client,
+            reason="bootstrap_not_completed",
+        )
+
+        self.assertEqual(streak, 1)
+        self.assertTrue(should_enqueue)
+
+    def test_sync_listing_task_does_not_enqueue_followup_after_bootstrap_error_limit(self) -> None:
+        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=False), \
+             patch.object(tasks, "_start_lock_heartbeat") as start_heartbeat, \
+             patch.object(tasks, "_release_lock_if_owner") as release_lock, \
+             patch.object(tasks, "_try_set_followup_pending", return_value=True), \
+             patch.object(tasks, "_bump_bootstrap_failure_streak", return_value=(tasks.SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_LIMIT, False)) as bump_streak, \
+             patch.object(tasks, "_clear_followup_pending") as clear_pending, \
+             patch.object(tasks.sync_listing_task, "update_state"), \
+             patch.object(tasks, "_run_browser_job", return_value={
+                 "run_id": 99,
+                 "status": "failed",
+                 "full_scan_completed": False,
+                 "cars_upserted": 0,
+                 "cars_failed": 0,
+                 "images_upserted": 0,
+                 "skipped_existing": 0,
+                 "elapsed_seconds": 1.0,
+                 "failures": [{"vehicle_url": "listing", "error": "bad resume"}],
+                 "listing": {"vehicles_collected": 0},
+             }):
+            get_persistence.return_value = MagicMock()
+            redis_client = MagicMock()
+            redis_client.get.return_value = None
+            get_redis.return_value = redis_client
+            start_heartbeat.return_value = (MagicMock(), MagicMock())
+
+            with patch.object(tasks.sync_listing_task, "app", new=MagicMock()) as task_app:
+                tasks.sync_listing_task.push_request(id="task-900")
+                try:
+                    result = tasks.sync_listing_task.run()
+                finally:
+                    tasks.sync_listing_task.pop_request()
+
+        self.assertEqual(result["status"], "failed")
+        bump_streak.assert_called_once()
+        clear_pending.assert_called()
+        task_app.send_task.assert_not_called()
+
+    def test_sync_listing_task_soft_timeout_deduplicates_continuation(self) -> None:
+        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", return_value=(MagicMock(), MagicMock())), \
+             patch.object(tasks, "_release_lock_if_owner"), \
+             patch.object(tasks, "_run_browser_job", side_effect=tasks.SoftTimeLimitExceeded()), \
+             patch.object(tasks, "_try_set_followup_pending", return_value=False) as set_pending, \
+             patch.object(tasks.sync_listing_task, "update_state"):
+            get_persistence.return_value = MagicMock()
+            redis_client = MagicMock()
+            redis_client.get.return_value = None
+            get_redis.return_value = redis_client
+
+            with patch.object(tasks.sync_listing_task, "app", new=MagicMock()) as task_app:
+                tasks.sync_listing_task.push_request(id="task-soft-timeout")
+                try:
+                    result = tasks.sync_listing_task.run(make="Toyota")
+                finally:
+                    tasks.sync_listing_task.pop_request()
+
+        self.assertEqual(result["status"], "timed_out")
+        set_pending.assert_called_once()
+        task_app.send_task.assert_not_called()
+
+    def test_sync_listing_task_stops_immediate_bootstrap_continuation_after_limit(self) -> None:
+        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=False), \
+             patch.object(tasks, "_start_lock_heartbeat", return_value=(MagicMock(), MagicMock())), \
+             patch.object(tasks, "_release_lock_if_owner"), \
+             patch.object(tasks, "_try_set_followup_pending", return_value=True), \
+             patch.object(tasks, "_bump_bootstrap_continuation_streak", return_value=(999, False)), \
+             patch.object(tasks, "_set_full_scan_done") as set_full_scan_done, \
+             patch.object(tasks, "_clear_sync_checkpoint") as clear_checkpoint, \
+             patch.object(tasks.sync_listing_task, "update_state"), \
+             patch.object(tasks, "_run_browser_job", return_value={
+                 "run_id": 101,
+                 "status": "partial_success",
+                 "full_scan_completed": False,
+                 "cars_upserted": 2,
+                 "cars_failed": 0,
+                 "images_upserted": 1,
+                 "skipped_existing": 0,
+                 "elapsed_seconds": 1.0,
+                 "failures": [],
+                 "listing": {"vehicles_collected": 2},
+             }):
+            get_persistence.return_value = MagicMock()
+            redis_client = MagicMock()
+            redis_client.get.return_value = None
+            get_redis.return_value = redis_client
+
+            with patch.object(tasks.sync_listing_task, "app", new=MagicMock()) as task_app:
+                tasks.sync_listing_task.push_request(id="task-breaker-stop")
+                try:
+                    result = tasks.sync_listing_task.run()
+                finally:
+                    tasks.sync_listing_task.pop_request()
+
+        self.assertEqual(result["status"], "partial_success")
+        # Сначала bootstrap помечается незавершённым, затем breaker переключает на hourly.
+        self.assertTrue(any(call.args == (redis_client, False) for call in set_full_scan_done.call_args_list))
+        self.assertTrue(any(call.args == (redis_client, True) for call in set_full_scan_done.call_args_list))
+        clear_checkpoint.assert_called_once()
+        task_app.send_task.assert_not_called()
+
+    def test_sync_listing_task_always_full_scan_forces_bootstrap_even_after_done(self) -> None:
+        with patch.object(tasks, "_get_persistence") as get_persistence, \
+             patch.object(tasks, "_get_redis") as get_redis, \
+             patch.object(
+                 tasks,
+                 "Settings",
+                 return_value=replace(
+                     base_settings,
+                     discovery=replace(base_settings.discovery, always_full_scan=True),
+                 ),
+             ), \
+             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, "_set_full_scan_done") as set_full_scan_done, \
+             patch.object(tasks.sync_listing_task, "update_state"), \
+             patch.object(tasks, "_run_browser_job") as run_job, \
+             patch("iaai_scraper.worker.tasks.parse_listing_segments", return_value=[]):
+            get_persistence.return_value = MagicMock()
+            redis_client = MagicMock()
+            redis_client.get.return_value = None
+            get_redis.return_value = redis_client
+            start_heartbeat.return_value = (MagicMock(), MagicMock())
+
+            run_job.return_value = {
+                "run_id": 17,
+                "status": "success",
+                "full_scan_completed": True,
+                "cars_upserted": 1,
+                "cars_failed": 0,
+                "images_upserted": 0,
+                "skipped_existing": 0,
+                "elapsed_seconds": 1.0,
+                "failures": [],
+            }
+
+            tasks.sync_listing_task.push_request(id="task-always-full")
+            try:
+                result = tasks.sync_listing_task.run()
+            finally:
+                tasks.sync_listing_task.pop_request()
+
+        self.assertEqual(result["status"], "success")
+        set_full_scan_done.assert_called_with(redis_client, False)
+        release_lock.assert_called_once()
+
+    def test_sync_listing_task_always_full_scan_uses_segmented_resume_path(self) -> None:
+        segments = [{"make": "TOYOTA"}, {"make": "FORD"}, {"make": "HONDA"}]
+        with patch.object(tasks, "_get_persistence") as get_persistence, \
+             patch.object(tasks, "_get_redis") as get_redis, \
+             patch.object(
+                 tasks,
+                 "Settings",
+                 return_value=replace(
+                     base_settings,
+                     discovery=replace(base_settings.discovery, always_full_scan=True),
+                 ),
+             ), \
+             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", side_effect=lambda fn: fn()), \
+             patch.object(tasks.sync_listing_task, "update_state"), \
+             patch("iaai_scraper.worker.tasks.parse_listing_segments", return_value=segments):
+            get_persistence.return_value = MagicMock()
+            redis_client = MagicMock()
+            redis_client.get.side_effect = lambda key: (
+                "0" if key == tasks.SYNC_LISTING_CHECKPOINT_KEY else None
+            )
+            get_redis.return_value = redis_client
+            start_heartbeat.return_value = (MagicMock(), MagicMock())
+
+            sync_segmented_mock = MagicMock(return_value={
+                "run_id": 18,
+                "status": "success",
+                "full_scan_completed": True,
+                "cars_upserted": 1,
+                "cars_failed": 0,
+                "images_upserted": 0,
+                "skipped_existing": 0,
+                "elapsed_seconds": 1.0,
+                "failures": [],
+            })
+            scraper_ctx = MagicMock()
+            scraper_ctx.__enter__.return_value.sync_listing_segmented = sync_segmented_mock
+            scraper_ctx.__enter__.return_value.sync_listing = MagicMock()
+            scraper_ctx.__exit__.return_value = None
+
+            with patch.object(tasks, "IAAIScraper", return_value=scraper_ctx):
+                tasks.sync_listing_task.push_request(id="task-always-full-resume")
+                try:
+                    result = tasks.sync_listing_task.run()
+                finally:
+                    tasks.sync_listing_task.pop_request()
+
+        self.assertEqual(result["status"], "success")
+        self.assertEqual(sync_segmented_mock.call_args.kwargs["start_segment"], 1)
+        self.assertEqual(sync_segmented_mock.call_args.kwargs["start_page"], 1)
+        release_lock.assert_called_once()
+
 
 if __name__ == "__main__":
     unittest.main()