Continuous Downloads & Lightweight Recovery

AGENTS.md

@@ -14,6 +14,7 @@ This document defines repository-level engineering conventions that apply to eve
   4. Type checks: `mypy src/modelscope_hub/`
 - Never commit secrets. `~/.modelscope/credentials/` is private (it contains a pickled cookie jar and `m_session_id`); never expose its contents in logs or reports.
 - When changing the default value of a configuration constant, update the corresponding tests for its default value and registration table, as well as the environment-variable table in `README.md`.
+- Local SDK state must use lightweight files, without embedded database modules or packages. Keep recovery hints sparse and advisory: verify actual files against available size/hash metadata before trusting completion or stale failure records. Missing, corrupt, or unwritable hints must not imply success or prevent valid files from being reused. Each download task must reuse one hidden JSON for sparse recovery hints and its latest summary, without persistent per-call report or result files; healthy files must not trigger per-file state rewrites.
 
 ---
 

README.md

@@ -213,6 +213,40 @@ path = api.download_file("Qwen/Qwen3-0.6B", "model", "config.json", local_files_
 
 Downloads validate available size and SHA-256 metadata before replacing files, with safe resume, bounded retries, timeouts, and cancellation. Incomplete snapshots raise `FileDownloadError` (CLI exit code `1`), preserving completed files for subsequent retries.
 
+### Continuous dataset downloads
+
+For migration jobs, `download_repo_report` continues past isolated file failures and returns an explicit `COMPLETE` or `PARTIAL` report. The original `download_repo` and `snapshot_download` remain strict by default.
+
+```python
+from modelscope_hub import HubApi, DownloadRetryPolicy
+
+api = HubApi(endpoint="https://www.modelscope.ai")
+report = api.download_repo_report(
+    "owner/dataset", "dataset", revision="master", local_dir="./data",
+    tracker_path="./.ms_download_state.json", max_workers=8,
+    retry_policy=DownloadRetryPolicy(),
+)
+for entry in report.iter_files("VERIFIED"):
+    consume_verified_file(entry.path)
+print(report.status, report.succeeded, report.failed, report.deferred, report.report_path)
+report.close()  # Release temporary per-file results after consumption.
+```
+
+- Recovery defaults: at most **2 extra retries per file**, shared by range parts and integrity recovery; failed-request inactivity and waits have a **60-second cumulative budget**. Idle/header waits are limited to **15 seconds**, low-speed checks use a **10-second window**, and cooldowns longer than **5 seconds** defer the file instead of tying up a worker. Size/hash mismatches get at most one full re-download. Healthy transfer time still uses `MODELSCOPE_DOWNLOAD_TOTAL_TIMEOUT`; large files are not capped at 60 seconds.
+- Each task uses **one hidden JSON file** outside downloaded contents (`cache_dir/.download_state/.ms_download_<key>.json` by default), with no embedded database dependency. It combines sparse failures, recovery budgets, failed pages, and the latest call's summary under `report`; `report.report_path` equals `report.tracker_path`. Reruns and batches replace the same file, without extra report JSON or JSONL files. Successful files have no persistent per-file records; a healthy run writes only its scope header and final summary, not an update per file. Its scope is endpoint/repository/revision/output, independent of batch filters. Standard reusable lock files remain separate from JSON metadata.
+- State is advisory, never the sole evidence of success. Every rerun validates actual files against current size/hash metadata, even if hints say `FAILED` or `DEFERRED`; a repaired valid file clears its stale hint without network IO. Otherwise failures remain isolated until metadata changes or `retry_failed=True`. Missing or corrupt hints allow bounded fresh recovery; load/write problems are logged and exposed in `report.state_warnings`, not treated as data corruption or success. A crash during recovery preserves its budget if hints were saved; healthy in-flight files need no markers and resume from actual partial files. Old database-format trackers are not read or migrated: select a new JSON path; existing correct data is reused through validation.
+- Dataset pages are streamed with bounded queues. Failed pages are recorded and isolated across calls; later pages/roots can continue. Two consecutive failed pages stop that root, and `listing_complete=False` prevents false success. The next run can use `retry_failed=True` or an authoritative `file_list` iterable to recover incomplete enumeration. Model listings that hit an unrecoverable server cap also return a partial report.
+- Counts and `report.iter_files()` cover **this call's selection**. Per-file results stream from an anonymous temporary spool owned by the live report, so earlier report objects remain independent of later batches without permanent manifests or an in-memory list of every file. Call `report.close()` after consumption; garbage collection also releases its spool. `results_path` is empty. The single JSON retains only the latest summary, not historical reports or a verified-file manifest: reconstructing a report from it cannot provide per-file results; rerun actual-file validation instead. Aggregate batch counts for the whole migration. Only `VERIFIED` entries may be uploaded; changed files must be checked again before use, and partial reports must not authorize deleting source data. A pinned revision or authoritative manifest is recommended. Reuse the same tracker and output across ten-file batches; a tracker cannot be shared by simultaneous writers.
+- Credential loss or a full/read-only disk pauses further work and sets `global_error`; Ctrl+C writes a `CANCELLED` report and remains interruptible. DNS and blocked filesystem calls remain OS-constrained. Caller-provided iterables must also be responsive to cancellation.
+
+```bash
+ms-hub download owner/dataset --repo-type dataset --local-dir ./data --continue-on-error
+# Explicit repair run after the source is fixed
+ms-hub download owner/dataset --repo-type dataset --local-dir ./data --continue-on-error --retry-failed
+```
+
+CLI exit codes: `0` complete, `3` partial after remaining known files have been processed, `1` global failure, `130` interrupted. Schedulers must treat `3` as partial completion, not an instruction to retry the whole task indefinitely. The SDK's `modelscope_hub.compat.snapshot_download(..., continue_on_error=True, tracker_path=...)` preserves a string return path and logs the report location; use `HubApi.download_repo_report` to read results directly. Older `modelscope.hub.snapshot_download` wrappers may not accept the new options: import the compat function from `modelscope_hub` directly until the wrapper is updated.
+
 ### Upload
 
 ```bash

src/modelscope_hub/__init__.py

@@ -55,6 +55,9 @@
     CacheInfo,
     CacheVerification,
     CommitInfo,
+    DownloadFileResult,
+    DownloadReport,
+    DownloadRetryPolicy,
     FileInfo,
     PagedResult,
     RepoInfo,
@@ -97,6 +100,9 @@
     "CacheVerification",
     "CachedRepoInfo",
     "CommitInfo",
+    "DownloadFileResult",
+    "DownloadReport",
+    "DownloadRetryPolicy",
     "FileInfo",
     "PagedResult",
     "RepoInfo",

src/modelscope_hub/_download.py

@@ -27,7 +27,7 @@
 import threading
 import time
 import uuid
-from collections.abc import Callable
+from collections.abc import Callable, Iterable
 from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, as_completed, wait
 from pathlib import Path
 from typing import TYPE_CHECKING
@@ -44,6 +44,7 @@
     _DownloadDeadlineError,
     _network_context,
     _ReadHTTPAdapter,
+    _RecoveryBudget,
 )
 from .constants import (
     DOWNLOAD_CHUNK_SIZE,
@@ -82,6 +83,7 @@
 if TYPE_CHECKING:
     from ._legacy_api import LegacyClient
     from .config import HubConfig
+    from .types import DownloadReport, DownloadRetryPolicy
 
 logger = get_logger("download")
 
@@ -175,8 +177,11 @@ def _optional_file_lock(
     while True:
         if control is not None:
             control.check()
-            if time.monotonic() - started >= DOWNLOAD_FILE_LOCK_TIMEOUT_SECONDS:
-                raise _DownloadDeadlineError(f"等待下载文件锁超过 {DOWNLOAD_FILE_LOCK_TIMEOUT_SECONDS}s：{lock_path}")
+            lock_limit = float(DOWNLOAD_FILE_LOCK_TIMEOUT_SECONDS)
+            if control.recovery is not None:
+                lock_limit = min(lock_limit, control.recovery.policy.idle_timeout)
+            if time.monotonic() - started >= lock_limit:
+                raise _DownloadDeadlineError(f"Download file lock timed out after {lock_limit}s: {lock_path}")
         try:
             lock.acquire(timeout=default_interval)
         except Timeout:
@@ -342,6 +347,8 @@ def _incomplete_path(target: Path) -> Path:
 
 def _wait_download_retry(retry: Retry, exc: Exception, path: str, control: _DownloadControl) -> Retry:
     control.check()
+    if isinstance(exc, OSError) and exc.errno in (errno.ENOSPC, errno.EROFS, errno.EACCES):
+        raise exc
     if isinstance(exc, HubError) and not exc.retryable:
         raise exc
     try:
@@ -351,6 +358,15 @@ def _wait_download_retry(retry: Retry, exc: Exception, path: str, control: _Down
             raise exc
         raise NetworkError(f"下载 {path} 在 {DOWNLOAD_RETRY_TIMES} 次重试后失败：{type(exc).__name__}") from exc
     requested = getattr(exc, "retry_after", None)
+    if control.recovery is not None:
+        delay = max(0, requested) if requested is not None else min(retry.get_backoff_time(), 2)
+        try:
+            control.recovery.retry(delay)
+        except Exception as budget_error:
+            raise budget_error from exc
+        logger.warning("Download recovery for %s (%s), retrying in %ss", path, type(exc).__name__, delay)
+        control.sleep(delay)
+        return retry
     delay = min(
         max(0, requested) if requested is not None else retry.get_backoff_time(), DOWNLOAD_RETRY_MAX_DELAY_SECONDS
     )
@@ -855,13 +871,17 @@ def download_file(
         user_agent: dict | str | None = None,
         progress_callbacks: list[type[ProgressCallback]] | None = None,
         _cancel_event: threading.Event | None = None,
+        _recovery: _RecoveryBudget | None = None,
     ) -> Path:
         """Download one file with cache-first lookup and a shared budget for locks, probes, and retries."""
         with _DownloadControl(
             total_timeout=DOWNLOAD_TOTAL_TIMEOUT,
             cancel_event=_cancel_event,
             min_rate=DOWNLOAD_MIN_BYTES_PER_SECOND,
-            speed_window=DOWNLOAD_LOW_SPEED_WINDOW_SECONDS,
+            speed_window=_recovery.policy.low_speed_window
+            if _recovery is not None
+            else DOWNLOAD_LOW_SPEED_WINDOW_SECONDS,
+            recovery=_recovery,
         ) as control:
             with self._client.read_context(control):
                 return self._download_file(
@@ -992,7 +1012,12 @@ def _download_file(
             # the server consistently served the wrong object, so it only gets
             # one more attempt, over the default route.
             failures = {"hash": 0, "size": 0}
-            limits = {"hash": DOWNLOAD_HASH_RETRY_TIMES, "size": DOWNLOAD_SIZE_RETRY_TIMES}
+            limits = {
+                "hash": min(DOWNLOAD_HASH_RETRY_TIMES, 2)
+                if _control.recovery is not None
+                else DOWNLOAD_HASH_RETRY_TIMES,
+                "size": DOWNLOAD_SIZE_RETRY_TIMES,
+            }
             prefer_default_route = False
             while True:
                 _control.check()
@@ -1032,10 +1057,54 @@ def _download_file(
                         limits[kind],
                         exc.message,
                     )
+                    if _control.recovery is not None:
+                        try:
+                            _control.recovery.retry(delay)
+                        except Exception as budget_error:
+                            raise budget_error from exc
                     _control.sleep(delay)
 
         return target
 
+    def download_repo_report(
+        self,
+        repo_id: str,
+        repo_type: str,
+        *,
+        revision: str | None = None,
+        cache_dir: Path | None = None,
+        local_dir: Path | None = None,
+        allow_patterns: list[str] | None = None,
+        ignore_patterns: list[str] | None = None,
+        max_workers: int = 4,
+        file_list: Iterable[dict] | None = None,
+        tracker_path: Path | None = None,
+        retry_failed: bool = False,
+        retry_policy: DownloadRetryPolicy | None = None,
+        user_agent: dict | str | None = None,
+        progress_callbacks: list[type[ProgressCallback]] | None = None,
+    ) -> DownloadReport:
+        """Continue after isolated file failures and return a durable completion report."""
+        from ._resilient_download import download_repo_report
+
+        return download_repo_report(
+            self,
+            repo_id,
+            repo_type,
+            revision=revision,
+            cache_dir=cache_dir,
+            local_dir=local_dir,
+            allow_patterns=allow_patterns,
+            ignore_patterns=ignore_patterns,
+            max_workers=max_workers,
+            file_list=file_list,
+            tracker_path=tracker_path,
+            retry_failed=retry_failed,
+            retry_policy=retry_policy,
+            user_agent=user_agent,
+            progress_callbacks=progress_callbacks,
+        )
+
     def download_repo(
         self,
         repo_id: str,

src/modelscope_hub/_download_control.py

@@ -6,16 +6,60 @@
 import socket
 import threading
 import time
-from collections.abc import Iterator
+from collections.abc import Callable, Iterator
 from dataclasses import dataclass
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import requests
 from requests.adapters import HTTPAdapter
 from urllib3.exceptions import HTTPError as Urllib3HTTPError
 
 from .errors import NetworkError, RequestTimeoutError
 
+if TYPE_CHECKING:
+    from .types import DownloadRetryPolicy
+
+
+class _RecoveryBudget:
+    """One durable recovery budget per file, shared by range and integrity retries."""
+
+    def __init__(
+        self,
+        policy: DownloadRetryPolicy,
+        retries: int,
+        seconds: float,
+        persist: Callable[[int, float], None],
+    ) -> None:
+        self.policy = policy
+        self.retries = retries
+        self.seconds = seconds
+        self.persist = persist
+        self.lock = threading.Lock()
+        self.activity = threading.local()
+
+    def progress(self) -> None:
+        self.activity.last = time.monotonic()
+
+    def retry(self, delay: float) -> None:
+        with self.lock:
+            elapsed = max(0, time.monotonic() - getattr(self.activity, "last", time.monotonic()))
+            self.seconds += elapsed
+            self.persist(self.retries, self.seconds)
+            if delay > self.policy.max_retry_delay:
+                raise _DownloadDeferredError("Server cooldown exceeds the short recovery budget")
+            if self.retries >= self.policy.max_retries or self.seconds + delay >= self.policy.recovery_timeout:
+                raise _DownloadDeferredError("File recovery budget exhausted; automatic retry is disabled")
+            self.retries += 1
+            self.seconds += delay
+            self.persist(self.retries, self.seconds)
+            self.progress()
+
+
+class _DownloadDeferredError(NetworkError):
+    """Recovery is exhausted or a server cooldown requires a later explicit run."""
+
+    retryable = False
+
 
 class _DownloadCancelledError(Exception):
     """The parent download was cancelled; do not retry or report a remote failure."""
@@ -38,6 +82,7 @@ class _WatchedResponse:
     received: int = 0
     error: Exception | None = None
     monitor_speed: bool = True
+    header_deadline: float | None = None
 
 
 def _response_socket(response: requests.Response) -> Any:
@@ -62,11 +107,13 @@ def __init__(
         cancel_event: threading.Event | None = None,
         min_rate: int = 0,
         speed_window: float = 60,
+        recovery: _RecoveryBudget | None = None,
     ) -> None:
         self.cancel_event = cancel_event if cancel_event is not None else threading.Event()
         self.deadline = time.monotonic() + total_timeout
         self.min_rate = min_rate
         self.speed_window = speed_window
+        self.recovery = recovery
         self._responses: dict[requests.Response, _WatchedResponse] = {}
         self._transports: dict[Any, _WatchedResponse] = {}
         self._lock = threading.Lock()
@@ -95,6 +142,12 @@ def check(self) -> None:
     def timeouts(self, connect: float, read: float) -> tuple[float, float]:
         self.check()
         remaining = max(0.001, self.deadline - time.monotonic())
+        if self.recovery is not None:
+            self.recovery.progress()
+            with self.recovery.lock:
+                remaining = min(remaining, max(0.001, self.recovery.policy.recovery_timeout - self.recovery.seconds))
+            connect = min(connect, self.recovery.policy.idle_timeout)
+            read = min(read, self.recovery.policy.idle_timeout)
         return min(connect, remaining), min(read, remaining)
 
     def sleep(self, seconds: float) -> None:
@@ -118,6 +171,9 @@ def transport(self, sock: Any) -> Iterator[None]:
         """Interrupt sockets while reading headers so trickling headers cannot bypass deadlines."""
         self.check()
         state = _WatchedResponse(sock, time.monotonic(), monitor_speed=False)
+        if self.recovery is not None:
+            started = getattr(self.recovery.activity, "last", time.monotonic())
+            state.header_deadline = min(self.deadline, started + self.recovery.policy.idle_timeout)
         with self._lock:
             self._transports[sock] = state
             self._start_monitor()
@@ -148,6 +204,8 @@ def _check_response(self, response: requests.Response, received: int = 0) -> Non
             state = self._responses.get(response)
             if state is not None:
                 state.received += received
+                if received and self.recovery is not None:
+                    self.recovery.progress()
                 if state.error is not None:
                     raise state.error
 
@@ -165,6 +223,8 @@ def _watch(self) -> None:
                         state.error = _DownloadCancelledError("下载已取消")
                     elif now >= self.deadline:
                         state.error = _DownloadDeadlineError("下载总时限已耗尽")
+                    elif state.header_deadline is not None and now >= state.header_deadline:
+                        state.error = RequestTimeoutError("Response headers exceeded the short recovery time limit")
                     elif state.monitor_speed and self.min_rate and now - state.started >= self.speed_window:
                         rate = state.received / (now - state.started)
                         if rate < self.min_rate:
@@ -270,6 +330,26 @@ def _configure(manager: Any) -> None:
             )
         manager.pool_classes_by_scheme = classes
 
+    def build_response(self, req: Any, resp: Any) -> requests.Response:
+        response = super().build_response(req, resp)
+        control = getattr(_network_state, "control", None)
+        if control is not None:
+            # Requests consumes intermediate redirect bodies outside chunks().
+            # Track them here, and release before their connection can be reused.
+            original_close = response.close
+
+            def close() -> None:
+                control.release(response)
+                original_close()
+
+            response.close = close  # type: ignore[method-assign]
+            try:
+                control.track(response)
+            except BaseException:
+                response.close()
+                raise
+        return response
+
     def init_poolmanager(self, connections: int, maxsize: int, block: bool = False, **kwargs: Any) -> None:
         super().init_poolmanager(connections, maxsize, block=block, **kwargs)
         self._configure(self.poolmanager)