Replace page-level checkpoint with segment-level resume

iaai_scraper/worker/tasks.py

@@ -1,7 +1,6 @@
 # Задачи Celery для синхронизации автомобилей и листинга IAAI.
 
 from concurrent.futures import ThreadPoolExecutor
-import json
 import logging
 from threading import Event, Thread
 import time
@@ -21,7 +20,6 @@ SYNC_LISTING_LOCK_KEY = "iaai:locks:sync_listing"
 SYNC_FULL_SCAN_DONE_KEY = "iaai:state:sync_full_scan_done"
 SYNC_LISTING_CHECKPOINT_KEY = "iaai:state:sync_listing_checkpoint"
 SYNC_LISTING_CHECKPOINT_TTL_SECONDS = 7 * 24 * 60 * 60
-SYNC_LISTING_CHECKPOINT_FAILURE_LIMIT = 2
 SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_KEY = "iaai:state:sync_listing_bootstrap_failure_streak"
 SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_LIMIT = 3
 SYNC_LISTING_BOOTSTRAP_FAILURE_STREAK_TTL_SECONDS = 24 * 60 * 60
@@ -248,55 +246,32 @@ def _set_full_scan_done(redis_client: Redis, done: bool) -> None:
         logger.warning("Failed to persist full scan state", exc_info=True)
 
 
-def _load_sync_checkpoint(redis_client: Redis) -> dict[str, object] | None:
+def _load_last_completed_segment(redis_client: Redis) -> int | None:
+    # Segment-only checkpoint: хранит индекс последнего ПОЛНОСТЬЮ пройденного сегмента.
     try:
         raw = redis_client.get(SYNC_LISTING_CHECKPOINT_KEY)
     except Exception:
         logger.warning("Failed to read sync listing checkpoint", exc_info=True)
         return None
-    if not raw:
-        return None
-    if isinstance(raw, str) and not raw.strip():
+    if raw is None:
         return None
     try:
-        data = json.loads(str(raw))
-    except Exception:
-        logger.warning("Failed to decode sync listing checkpoint", exc_info=True)
+        value = int(str(raw).strip())
+    except (TypeError, ValueError):
+        logger.warning("Invalid sync listing checkpoint value %r; clearing", raw)
         _clear_sync_checkpoint(redis_client)
         return None
-    if not isinstance(data, dict):
+    if value < 0:
         _clear_sync_checkpoint(redis_client)
         return None
-    return data
+    return value
 
 
-def _save_sync_checkpoint(
-    redis_client: Redis,
-    *,
-    task_id: str,
-    page_number: int,
-    make: str | None,
-    model: str | None,
-    lane: str,
-    segment_index: int | None = None,
-) -> None:
-    payload = {
-        "status": "in_progress",
-        "task_id": task_id,
-        # page_number=0 — sentinel: прошлый checkpoint признан stale,
-        # следующий запуск должен начать текущий scope заново с page 1.
-        "last_successful_page": max(0, int(page_number)),
-        "resume_failures": 0,
-        "make": make,
-        "model": model,
-        "lane": lane,
-        "segment_index": segment_index,
-        "updated_at": int(time.time()),
-    }
+def _save_last_completed_segment(redis_client: Redis, segment_index: int) -> None:
     try:
         redis_client.set(
             SYNC_LISTING_CHECKPOINT_KEY,
-            json.dumps(payload),
+            str(int(segment_index)),
             ex=SYNC_LISTING_CHECKPOINT_TTL_SECONDS,
         )
     except Exception:
@@ -310,45 +285,6 @@ def _clear_sync_checkpoint(redis_client: Redis) -> None:
         logger.warning("Failed to clear sync listing checkpoint", exc_info=True)
 
 
-def _bump_checkpoint_resume_failure(
-    redis_client: Redis,
-    checkpoint: dict[str, object] | None,
-    *,
-    reason: str,
-) -> tuple[int, bool]:
-    if not checkpoint:
-        return 0, False
-
-    failures = int(checkpoint.get("resume_failures") or 0) + 1
-    if failures >= SYNC_LISTING_CHECKPOINT_FAILURE_LIMIT:
-        logger.warning(
-            "Checkpoint resume failed %d times; deleting checkpoint and restarting from page 1 next run (reason=%s)",
-            failures,
-            reason,
-        )
-        _clear_sync_checkpoint(redis_client)
-        return failures, True
-
-    payload = dict(checkpoint)
-    payload["resume_failures"] = failures
-    payload["updated_at"] = int(time.time())
-    try:
-        redis_client.set(
-            SYNC_LISTING_CHECKPOINT_KEY,
-            json.dumps(payload),
-            ex=SYNC_LISTING_CHECKPOINT_TTL_SECONDS,
-        )
-    except Exception:
-        logger.warning("Failed to persist checkpoint resume failure counter", exc_info=True)
-    logger.warning(
-        "Checkpoint resume failure %d/%d recorded (reason=%s)",
-        failures,
-        SYNC_LISTING_CHECKPOINT_FAILURE_LIMIT,
-        reason,
-    )
-    return failures, False
-
-
 def _try_set_followup_pending(redis_client: Redis, *, ttl_seconds: int) -> bool:
     try:
         return bool(redis_client.set(SYNC_LISTING_FOLLOWUP_PENDING_KEY, "1", nx=True, ex=max(60, int(ttl_seconds))))
@@ -545,36 +481,15 @@ def sync_listing_task(
         force_bootstrap_full_scan = not full_scan_done_before_run
         effective_limit = None if force_bootstrap_full_scan else limit
         effective_only_new = False if force_bootstrap_full_scan else only_new
-        checkpoint = _load_sync_checkpoint(redis_client)
-        resume_from_page = 1
-        used_checkpoint_resume = False
 
-        if force_bootstrap_full_scan and checkpoint and str(checkpoint.get("status") or "") == "in_progress":
-            checkpoint_page = int(checkpoint.get("last_successful_page") or 0)
-            checkpoint_make = checkpoint.get("make")
-            checkpoint_model = checkpoint.get("model")
-            checkpoint_lane = checkpoint.get("lane")
-            checkpoint_segment = checkpoint.get("segment_index")
-            same_scope = (
-                checkpoint_make == make
-                and checkpoint_model == model
-                and checkpoint_lane == lane
-            )
-            # Для сегментированного режима: совпадение по lane + наличие segment_index
-            is_segmented_checkpoint = checkpoint_segment is not None and checkpoint_lane == lane
-            if checkpoint_page > 0 and (same_scope or is_segmented_checkpoint):
-                resume_from_page = checkpoint_page + 1
-                used_checkpoint_resume = True
-                logger.warning(
-                    "Resuming sync_listing from page %d (segment=%s) using checkpoint",
-                    resume_from_page,
-                    checkpoint_segment,
-                )
-            elif checkpoint_page > 0:
-                logger.info("Ignoring stale checkpoint due to different sync parameters")
-                _clear_sync_checkpoint(redis_client)
-        elif checkpoint:
-            logger.info("Ignoring leftover checkpoint because full scan is already complete; next run starts from page 1")
+        # Segment-level checkpoint: хранит индекс последнего ПОЛНОСТЬЮ пройденного сегмента.
+        # Используется только во время bootstrap для пропуска уже обработанных сегментов.
+        # Никаких page-level resume — внутри сегмента всегда стартуем с page 1.
+        last_completed_segment: int | None = None
+        if force_bootstrap_full_scan:
+            last_completed_segment = _load_last_completed_segment(redis_client)
+        else:
+            # После завершения bootstrap чекпоинт не нужен никогда.
             _clear_sync_checkpoint(redis_client)
 
         if force_bootstrap_full_scan:
@@ -604,11 +519,21 @@ def sync_listing_task(
         use_segmented = bool(segments) and make is None and model is None
 
         resume_from_segment = 0
-        if force_bootstrap_full_scan and use_segmented and checkpoint and str(checkpoint.get("status") or "") == "in_progress":
-            cp_segment = checkpoint.get("segment_index")
-            if cp_segment is not None and int(cp_segment) >= 0:
-                resume_from_segment = int(cp_segment)
-                # resume_from_page уже вычислен выше
+        if force_bootstrap_full_scan and use_segmented and last_completed_segment is not None:
+            resume_from_segment = max(0, last_completed_segment + 1)
+            if resume_from_segment >= len(segments):
+                # Все сегменты уже пройдены — чекпоинт устарел, начинаем заново.
+                logger.info(
+                    "Stored checkpoint segment=%d is beyond configured segments (%d); restarting bootstrap from segment 0",
+                    last_completed_segment, len(segments),
+                )
+                resume_from_segment = 0
+                _clear_sync_checkpoint(redis_client)
+            elif resume_from_segment > 0:
+                logger.warning(
+                    "Resuming segmented bootstrap from segment=%d (last completed=%d)",
+                    resume_from_segment, last_completed_segment,
+                )
 
         def _job():
             with IAAIScraper() as scraper:
@@ -618,17 +543,9 @@ def sync_listing_task(
                         lane=lane,
                         only_new=effective_only_new,
                         start_segment=resume_from_segment,
-                        start_page=resume_from_page,
+                        start_page=1,
                         progress_callback=(
-                            (lambda seg_idx, page_number: _save_sync_checkpoint(
-                                redis_client,
-                                task_id=task_id,
-                                page_number=page_number,
-                                make=None,
-                                model=None,
-                                lane=lane,
-                                segment_index=seg_idx,
-                            ))
+                            (lambda seg_idx: _save_last_completed_segment(redis_client, seg_idx))
                             if force_bootstrap_full_scan else None
                         ),
                     )
@@ -638,18 +555,6 @@ def sync_listing_task(
                     lane=lane,
                     limit=effective_limit,
                     only_new=effective_only_new,
-                    start_page=resume_from_page,
-                    progress_callback=(
-                        (lambda page_number: _save_sync_checkpoint(
-                            redis_client,
-                            task_id=task_id,
-                            page_number=page_number,
-                            make=make,
-                            model=model,
-                            lane=lane,
-                        ))
-                        if force_bootstrap_full_scan else None
-                    ),
                 )
 
         result = _run_browser_job(_job)
@@ -674,9 +579,7 @@ def sync_listing_task(
                     "bootstrap_not_completed",
                     count_as_failure=count_as_failure,
                 )
-        elif result.get("status") in ("success", "partial_success") and bool(result.get("full_scan_completed", False)):
-            _clear_sync_checkpoint(redis_client)
-        elif not force_bootstrap_full_scan:
+        else:
             _clear_sync_checkpoint(redis_client)
 
         summary = {
@@ -717,14 +620,6 @@ def sync_listing_task(
 
     except Exception as exc:
         logger.error("sync_listing_task failed: %s", exc, exc_info=True)
-        if force_bootstrap_full_scan and used_checkpoint_resume:
-            _, deleted = _bump_checkpoint_resume_failure(
-                redis_client,
-                checkpoint,
-                reason=str(exc),
-            )
-            if deleted:
-                checkpoint = None
         # Retry только на не-таймаутные ошибки (сеть, БД, браузер).
         try:
             if force_bootstrap_full_scan: