Pause & Resume with Checkpoints

@dataclass
class CheckpointData:
    """Container for checkpoint state."""

    requests: List["Request"] = field(default_factory=list)
    seen: Set[bytes] = field(default_factory=set)

async def save(self, data: CheckpointData) -> None:
    """Save checkpoint data to disk atomically."""
    await self.crawldir.mkdir(parents=True, exist_ok=True)

    temp_path = self._checkpoint_path.with_suffix(".tmp")

    try:
        serialized = pickle.dumps(data, protocol=pickle.HIGHEST_PROTOCOL)
        async with await anyio.open_file(temp_path, "wb") as f:
            await f.write(serialized)

        await temp_path.rename(self._checkpoint_path)

        log.info(f"Checkpoint saved: {len(data.requests)} requests, {len(data.seen)} seen URLs")
    except Exception as e:
        # Clean up temp file if it exists
        if await temp_path.exists():
            await temp_path.unlink()
        log.error(f"Failed to save checkpoint: {e}")
        raise

async def load(self) -> Optional[CheckpointData]:
    """Load checkpoint data from disk.

    Returns None if no checkpoint exists or if loading fails.
    """
    if not await self.has_checkpoint():
        return None

    try:
        async with await anyio.open_file(self._checkpoint_path, "rb") as f:
            content = await f.read()
            data: CheckpointData = pickle.loads(content)

        log.info(f"Checkpoint loaded: {len(data.requests)} requests, {len(data.seen)} seen URLs")
        return data

    except Exception as e:
        log.error(f"Failed to load checkpoint (starting fresh): {e}")
        return None

async def crawl(self) -> CrawlStats:
    """Run the spider and return CrawlStats."""
    self._running = True
    self._items.clear()
    self.paused = False
    self._pause_requested = False
    self._force_stop = False
    self.stats = CrawlStats(start_time=anyio.current_time())

    # Check for existing checkpoint
    resuming = (await self._restore_from_checkpoint()) if self._checkpoint_system_enabled else False
    self._last_checkpoint_time = anyio.current_time()

async def _restore_from_checkpoint(self) -> bool:
    """Attempt to restore state from checkpoint.

    Returns True if successfully restored, False otherwise.
    """
    if not self._checkpoint_system_enabled:
        raise

    data = await self._checkpoint_manager.load()
    if data is None:
        return False

    self.scheduler.restore(data)

    # Restore callbacks from spider after scheduler restore
    for request in data.requests:
        request._restore_callback(self.spider)

    return True

def _restore_callback(self, spider: "Spider") -> None:
    """Restore callback from spider after unpickling.

    :param spider: Spider instance to look up callback method on
    """
    if hasattr(self, "_callback_name") and self._callback_name:
        self.callback = getattr(spider, self._callback_name, None) or spider.parse
        del self._callback_name
    elif hasattr(self, "_callback_name"):
        del self._callback_name

def _is_checkpoint_time(self) -> bool:
    """Check if it's time for the periodic checkpoint."""
    if not self._checkpoint_system_enabled:
        return False

    if self._checkpoint_manager.interval == 0:
        return False

    current_time = anyio.current_time()
    return (current_time - self._last_checkpoint_time) >= self._checkpoint_manager.interval

if self._checkpoint_system_enabled and self._is_checkpoint_time():
    await self._save_checkpoint()

if self._pause_requested:
    if self._active_tasks == 0 or self._force_stop:
        if self._force_stop:
            log.warning(f"Force stopping with {self._active_tasks} active tasks")
            tg.cancel_scope.cancel()

        # Only save checkpoint if checkpoint system is enabled
        if self._checkpoint_system_enabled:
            await self._save_checkpoint()
            self.paused = True
            log.info("Spider paused, checkpoint saved")
        else:
            log.info("Spider stopped gracefully")

        self._running = False
        break

    # Wait briefly and check again
    await anyio.sleep(0.05)
    continue

def snapshot(self) -> Tuple[List[Request], Set[bytes]]:
    """Create a snapshot of the current state for checkpoints."""
    sorted_items = sorted(self._pending.values(), key=lambda x: (x[0], x[1]))  # Maintain queue order
    requests = [item[2] for item in sorted_items]
    return requests, self._seen.copy()

def restore(self, data: "CheckpointData") -> None:
    """Restore scheduler state from checkpoint data.

    :param data: CheckpointData containing requests and seen set
    """
    self._seen = data.seen.copy()

    # Restore pending requests in order (they're already sorted by priority)
    for request in data.requests:
        counter = next(self._counter)
        item = (-request.priority, counter, request)
        self._pending[counter] = item
        self._queue.put_nowait(item)

    log.info(f"Scheduler restored: {len(data.requests)} requests, {len(data.seen)} seen")

finally:
    await self.spider.on_close()
    # Clean up checkpoint files on successful completion (not paused)
    if not self.paused and self._checkpoint_system_enabled:
        await self._checkpoint_manager.cleanup()