diff --git a/AGENTS.md b/AGENTS.md index 995d2e3..6192267 100644 --- a/AGENTS.md +++ b/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. --- diff --git a/README.md b/README.md index 12be847..7b0922f 100644 --- a/README.md +++ b/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_.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 diff --git a/src/modelscope_hub/__init__.py b/src/modelscope_hub/__init__.py index 4db8011..b6dc9c4 100644 --- a/src/modelscope_hub/__init__.py +++ b/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", diff --git a/src/modelscope_hub/_download.py b/src/modelscope_hub/_download.py index c1becfa..900e9ea 100644 --- a/src/modelscope_hub/_download.py +++ b/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, diff --git a/src/modelscope_hub/_download_control.py b/src/modelscope_hub/_download_control.py index f98b120..8557ce2 100644 --- a/src/modelscope_hub/_download_control.py +++ b/src/modelscope_hub/_download_control.py @@ -6,9 +6,9 @@ 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 @@ -16,6 +16,50 @@ 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) diff --git a/src/modelscope_hub/_download_state.py b/src/modelscope_hub/_download_state.py new file mode 100644 index 0000000..48bf953 --- /dev/null +++ b/src/modelscope_hub/_download_state.py @@ -0,0 +1,261 @@ +"""Durable report-mode state; kept outside downloaded repository contents.""" + +from __future__ import annotations + +import contextlib +import hashlib +import json +import math +import re +import tempfile +import threading +from dataclasses import asdict, replace +from pathlib import Path +from typing import TextIO +from urllib.parse import urlsplit, urlunsplit + +from .errors import InvalidParameter +from .types import DownloadFileResult, DownloadReport +from .utils.logger import get_logger + +logger = get_logger("download") + + +def safe_error(error: BaseException) -> str: + """Keep diagnostics without persisting credentials or presigned queries.""" + text = str(error) + text = re.sub( + r"https?://[^\s)]+", + lambda m: urlunsplit(urlsplit(m[0])._replace(query="", fragment="", netloc=urlsplit(m[0]).hostname or "")), + text, + ) + text = re.sub(r"(?i)(token|signature|authorization|cookie|password|secret)\s*[=:]\s*[^\s,;]+", r"\1=***", text) + return text[:2000] + + +class _DownloadState: + """Sparse JSON recovery hints, never an authority for file completeness. + + Persist exceptional files, failed pages, and the latest summary together. + Per-file results stay in an anonymous spool owned by the returned report; + no permanent JSONL or per-run summary files are created. Healthy transfers + do not rewrite the tracker; actual partial files suffice after a crash. + """ + + def __init__(self, path: Path, identity: dict[str, str]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + self.path = path + self.identity = dict(identity) + self.lock = threading.Lock() + self._exceptions: dict[str, DownloadFileResult] = {} + self._pages: dict[str, dict[int, str]] = {} + self._active: dict[str, DownloadFileResult] = {} + self._seen: set[bytes] = set() + self._run: str | None = None + self._spool: TextIO | None = tempfile.TemporaryFile(mode="w+t", encoding="utf-8") + self._counts: dict[str, int] = {} + self._summary: dict | None = None + self.warnings: list[str] = [] + try: + self._load() + if not path.exists(): + self._save() + except BaseException: + self.close() + raise + + def _warn(self, message: str) -> None: + if message not in self.warnings and len(self.warnings) < 10: + self.warnings.append(message) + logger.warning("%s: %s", message, self.path) + + def _load(self) -> None: + if not self.path.exists(): + return + try: + data = json.loads(self.path.read_text(encoding="utf-8")) + if not isinstance(data, dict) or data.get("version") != 1 or not isinstance(data.get("identity"), dict): + raise ValueError("Invalid tracker header") + if data["identity"] != self.identity: + raise InvalidParameter( + "Download tracker belongs to a different endpoint/repository/revision/output directory" + ) + files, pages = data.get("files", {}), data.get("pages", {}) + if not isinstance(files, dict) or not isinstance(pages, dict): + raise ValueError("Invalid recovery records") + for path, raw in files.items(): + try: + if not isinstance(raw, dict) or raw.get("path") != path: + raise ValueError("Invalid file record") + result = DownloadFileResult.from_dict(raw) + if result.status not in ("FAILED", "DEFERRED", "DOWNLOADING", "PENDING"): + raise ValueError("Not a recovery hint") + if ( + not isinstance(result.retries, int) + or result.retries < 0 + or not isinstance(result.recovery_seconds, (int, float)) + or not math.isfinite(result.recovery_seconds) + or result.recovery_seconds < 0 + ): + raise ValueError("Invalid recovery budget") + if result.status == "DOWNLOADING": + result.status = "DEFERRED" + result.error = "Previous process stopped during recovery" + result.error_type = "InterruptedDownload" + self._exceptions[path] = result + except (ValueError, TypeError): + self._warn("Ignoring malformed file recovery hints; actual files will be revalidated") + for root, records in pages.items(): + try: + if not isinstance(records, dict): + raise ValueError("Invalid page records") + self._pages[root] = { + int(page): error + for page, error in records.items() + if str(page).isdigit() and int(page) > 0 and isinstance(error, str) + } + except (ValueError, TypeError): + self._warn("Ignoring malformed listing recovery hints") + except InvalidParameter: + raise + except (ValueError, UnicodeError, OSError): + self._exceptions.clear() + self._pages.clear() + self._warn( + "Cannot load download recovery hints; continuing with actual-file validation and bounded retries" + ) + + def _save(self) -> None: + """Atomically replace sparse hints; write failures do not invalidate data.""" + try: + with tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + dir=self.path.parent, + prefix=self.path.name + ".", + suffix=".tmp", + delete=False, + ) as stream: + temporary = Path(stream.name) + json.dump( + { + "version": 1, + "identity": self.identity, + "files": {path: asdict(result) for path, result in self._exceptions.items()}, + "pages": self._pages, + "report": self._summary, + }, + stream, + ensure_ascii=False, + separators=(",", ":"), + ) + temporary.replace(self.path) + except OSError: + self._warn("Cannot save download recovery hints; restart isolation may be unavailable") + finally: + if "temporary" in locals(): + with contextlib.suppress(OSError): + temporary.unlink(missing_ok=True) + + def close(self) -> None: + if self._spool is not None: + self._spool.close() + self._spool = None + + def failed_pages(self, root: str, *, reset: bool = False) -> dict[int, str]: + with self.lock: + if reset and self._pages.pop(root, None) is not None: + self._save() + return dict(self._pages.get(root, {})) + + def fail_page(self, root: str, page: int, error: Exception) -> None: + with self.lock: + message = safe_error(error) + if self._pages.setdefault(root, {}).get(page) != message: + self._pages[root][page] = message + self._save() + + def select(self, run: str, path: str, size: int | None, sha256: str | None) -> DownloadFileResult | None: + with self.lock: + if self._run is None: + self._run = run + if self._run != run: + raise InvalidParameter("A download state instance belongs to one invocation") + key = hashlib.sha256(path.encode()).digest() + if key in self._seen: + return None + self._seen.add(key) + old = self._exceptions.get(path) + if old is not None and (old.size, old.sha256) == (size, sha256): + result = replace(old) + else: + result = DownloadFileResult(path=path, size=size, sha256=sha256) + if self._exceptions.pop(path, None) is not None: + self._save() + self._active[path] = result + return replace(result) + + def begin(self, path: str, *, reset: bool = False) -> DownloadFileResult: + with self.lock: + result = self._active[path] + if reset: + result.retries, result.recovery_seconds = 0, 0 + result.status = "DOWNLOADING" + result.error, result.error_type, result.retryable = None, None, False + return replace(result) + + def budget(self, path: str, retries: int, seconds: float) -> None: + with self.lock: + result = self._active[path] + result.retries, result.recovery_seconds = retries, seconds + self._exceptions[path] = replace(result) + self._save() + + def finish(self, path: str, status: str, error: BaseException | None = None) -> None: + with self.lock: + result = self._active.pop(path, None) + if result is None: + return + result.status = status + if error is not None: + result.error, result.error_type = safe_error(error), type(error).__name__ + result.retryable = bool(getattr(error, "retryable", False)) + elif status in ("VERIFIED", "PENDING"): + result.error, result.error_type, result.retryable = None, None, False + old = self._exceptions.get(path) + keep = status in ("FAILED", "DEFERRED") or (status == "PENDING" and result.retries > 0) + if keep: + self._exceptions[path] = replace(result) + else: + self._exceptions.pop(path, None) + if old != self._exceptions.get(path): + self._save() + assert self._spool is not None + self._spool.write(json.dumps(asdict(result), ensure_ascii=False) + "\n") + self._counts[status] = self._counts.get(status, 0) + 1 + + def report(self, run: str, report: DownloadReport) -> DownloadReport: + for path in list(self._active): + self.finish(path, "PENDING") + report.total = sum(self._counts.values()) + report.succeeded = self._counts.get("VERIFIED", 0) + report.failed = self._counts.get("FAILED", 0) + report.deferred = self._counts.get("DEFERRED", 0) + report.pending = self._counts.get("PENDING", 0) + report.state_warnings = list(self.warnings) + if report.status != "CANCELLED": + report.status = ( + "COMPLETE" + if report.listing_complete and not report.global_error and report.succeeded == report.total + else "PARTIAL" + ) + report.report_path = str(self.path) + report.results_path = "" # No persistent per-file manifest. + assert self._spool is not None + self._spool.flush() + report._attach_file_results(self._spool) + self._spool = None + self._summary = {**asdict(report), "run_id": run} + self._save() + report.state_warnings = list(self.warnings) + return report diff --git a/src/modelscope_hub/_legacy_api.py b/src/modelscope_hub/_legacy_api.py index b7844b5..369387e 100644 --- a/src/modelscope_hub/_legacy_api.py +++ b/src/modelscope_hub/_legacy_api.py @@ -479,10 +479,12 @@ def list_repo_files( ) return self._list_files_page(repo_id, repo_type, revision, recursive=False, root=root) + self._read_state.listing_incomplete = False entries = self._list_files_page(repo_id, repo_type, revision, recursive=recursive, root=root) if len(entries) < REPO_FILES_TRUNCATION_LIMIT: return entries if not recursive: + self._read_state.listing_incomplete = True logger.warning( "Directory %r of %s holds at least %d entries and the server returns no more " "than that for a single listing; the result is incomplete.", @@ -666,6 +668,7 @@ def one(subroot: str | None) -> list[dict]: break frontier = list(dict.fromkeys(next_frontier)) + self._read_state.listing_incomplete = budget_spent or bool(oversized) if budget_spent: logger.warning( "Repo %s: stopped walking the file tree after %d listings " @@ -747,6 +750,63 @@ def list_dataset_files_paginated( page_number += 1 return all_files + def iter_dataset_file_pages( + self, + repo_id: str, + revision: str = "master", + root_path: str = "/", + page_size: int = 200, + isolated_pages: dict[int, str] | None = None, + ) -> Iterator[tuple[int, list[dict], Exception | None]]: + """Yield pages and bounded listing failures without buffering the dataset. + + One failed page does not suppress later pages; two consecutive failures + stop this root to avoid probing an unknown infinite tail. Consumers must + report every failed page and mark enumeration incomplete. + """ + failures = 0 + logger.info("Streaming listing %s@%s Root=%s", repo_id, revision, root_path) + for page in range(1, REPO_TREE_MAX_REQUESTS + 1): + params = {"Revision": revision, "Recursive": "True", "PageNumber": page, "PageSize": page_size} + if root_path != "/": + params["Root"] = root_path + try: + if isolated_pages and page in isolated_pages: + raise NetworkError(f"Previously isolated page: {isolated_pages[page]}") + response = self._request_repo_tree( + f"datasets/{repo_id}/repo/tree", + params, + repo_id=repo_id, + authorized=self._tree_reads_ok, + max_attempts=2, + ) + payload = response.json() + if isinstance(payload, dict) and ( + payload.get("Success") is False + or payload.get("success") is False + or payload.get("Code", 200) not in (200, "200", None) + ): + raise NetworkError("File listing returned a business failure with HTTP 200") + data = payload.get("Data", payload) if isinstance(payload, dict) else payload + if not isinstance(data, (list, dict)): + raise NetworkError("Malformed file listing response") + files = data if isinstance(data, list) else (data.get("Files") or data.get("files") or []) + if not isinstance(files, list): + raise NetworkError("Malformed file listing entries") + except (NetworkError, ServerError, RateLimitError) as exc: + yield page, [], exc + failures += 1 + if failures >= 2: + return + continue + failures = 0 + yield page, files, None + if page % 10 == 0 or len(files) < page_size: + logger.info("Streaming listing %s Root=%s: page %d", repo_id, root_path, page) + if len(files) < page_size: + return + yield REPO_TREE_MAX_REQUESTS + 1, [], NetworkError("File listing request budget exhausted") + def _fetch_tree_page(self, path: str, params: dict[str, Any]) -> requests.Response: parent = getattr(self._read_state, "control", None) budget = float(REPO_TREE_PAGE_TIMEOUT_SECONDS) @@ -778,6 +838,7 @@ def _request_repo_tree( *, repo_id: str, authorized: bool, + max_attempts: int | None = None, ) -> Any: """Fetch one file-tree listing, retrying a spurious denial. @@ -794,7 +855,8 @@ def _request_repo_tree( fast and honest. """ last_error: NetworkError | ServerError | RateLimitError | PermissionDeniedError | None = None - for attempt in range(REPO_TREE_PAGE_MAX_ATTEMPTS): + attempts = REPO_TREE_PAGE_MAX_ATTEMPTS if max_attempts is None else max_attempts + for attempt in range(attempts): try: resp = self._fetch_tree_page(path, params) except (NetworkError, ServerError, RateLimitError, PermissionDeniedError) as error: @@ -803,7 +865,7 @@ def _request_repo_tree( if not error.retryable and not isinstance(error, PermissionDeniedError): raise last_error = error - if attempt < REPO_TREE_PAGE_MAX_ATTEMPTS - 1: + if attempt < attempts - 1: requested_wait = getattr(error, "retry_after", None) wait = min( max(0, requested_wait) if requested_wait is not None else 2**attempt, @@ -817,7 +879,7 @@ def _request_repo_tree( type(error).__name__, wait, attempt + 1, - REPO_TREE_PAGE_MAX_ATTEMPTS, + attempts, ) control = getattr(self._read_state, "control", None) if control is None: @@ -830,7 +892,7 @@ def _request_repo_tree( if isinstance(last_error, PermissionDeniedError): raise NetworkError( f"Repo {repo_id}: tree listing ({params.get('Root') or params.get('PageNumber') or '/'}) kept " - f"returning a denial after {REPO_TREE_PAGE_MAX_ATTEMPTS} attempts on an already-authorized " + f"returning a denial after {attempts} attempts on an already-authorized " f"repo: {last_error}" ) from last_error assert last_error is not None diff --git a/src/modelscope_hub/_resilient_download.py b/src/modelscope_hub/_resilient_download.py new file mode 100644 index 0000000..dbc9f36 --- /dev/null +++ b/src/modelscope_hub/_resilient_download.py @@ -0,0 +1,361 @@ +"""Streaming report-mode downloads with durable file failure isolation.""" + +from __future__ import annotations + +import errno +import hashlib +import json +import queue +import threading +import uuid +from collections.abc import Iterable +from concurrent.futures import FIRST_COMPLETED, Future, ThreadPoolExecutor, wait +from pathlib import Path, PurePosixPath +from typing import TYPE_CHECKING + +from filelock import FileLock, Timeout + +from ._download import _listing_roots, _matches_patterns +from ._download_control import _DownloadCancelledError, _DownloadControl, _DownloadDeferredError, _RecoveryBudget +from ._download_state import _DownloadState, safe_error +from .constants import DOWNLOAD_TOTAL_TIMEOUT, REPO_TREE_LIST_TIMEOUT_SECONDS +from .errors import AuthenticationError, FileIntegrityError, InvalidParameter, NotSupportedError +from .types import DownloadFileResult, DownloadReport, DownloadRetryPolicy +from .utils.logger import get_logger + +if TYPE_CHECKING: + from ._download import DownloadManager, ProgressCallback + +logger = get_logger("download") + + +def _validate_entry(entry: dict, output: Path) -> tuple[str, int | None, str | None]: + path = entry.get("Path") or entry.get("path") or entry.get("Name") + if not isinstance(path, str) or not path or "\\" in path or "\0" in path: + raise InvalidParameter("File entry has no safe repository-relative path") + parts = PurePosixPath(path) + if not parts.parts or parts.is_absolute() or ".." in parts.parts or path.endswith("/"): + raise InvalidParameter(f"Unsafe download path: {path}") + if not (output / path).resolve().is_relative_to(output.resolve()): + raise InvalidParameter(f"Download path escapes output directory: {path}") + raw_size = entry.get("Size", entry.get("size")) + size = int(raw_size) if raw_size is not None else None + if size is not None and size < 0: + raise InvalidParameter(f"Negative file size: {path}") + sha256 = entry.get("Sha256") or entry.get("sha256") + if sha256 is not None: + if not isinstance(sha256, str) or len(sha256) != 64 or any(c not in "0123456789abcdefABCDEF" for c in sha256): + raise InvalidParameter(f"Invalid SHA-256 metadata: {path}") + sha256 = sha256.lower() + if size is None and sha256 is None: + raise InvalidParameter(f"Missing size and SHA-256 metadata; refusing an unverifiable download: {path}") + return path, size, sha256 + + +def download_repo_report( + manager: DownloadManager, + 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: + """Ordinary file failures return PARTIAL after all known work has been processed. + + A tracker is scoped to endpoint, repo, revision and output, not include + patterns, so repeated ten-file batches share quarantine and retry budgets. + retry_failed is an explicit operator-authorized reset, never automatic. + """ + if not repo_id or max_workers < 1: + raise InvalidParameter("repo_id and a positive max_workers are required") + if repo_type not in ("model", "dataset"): + raise NotSupportedError("Report-mode downloads support model and dataset repositories") + revision = revision or "master" + revision_path = PurePosixPath(revision) + if revision_path.is_absolute() or ".." in revision_path.parts: + raise InvalidParameter("revision must not escape the snapshot cache directory") + output = local_dir or manager._find_legacy_repo_dir(repo_id, repo_type, cache_dir) + if output is None: + output = manager._repo_cache_dir_path(repo_id, repo_type, cache_dir) / "snapshots" / revision + output = output.absolute() + output.mkdir(parents=True, exist_ok=True) + identity = { + "endpoint": manager._client.endpoint, + "repo": repo_id, + "type": repo_type, + "revision": revision, + "output": str(output.resolve()), + "schema": "1", + } + key = hashlib.sha256(json.dumps(identity, sort_keys=True).encode()).hexdigest() + tracker = tracker_path or (cache_dir or manager._config.cache_dir) / ".download_state" / f".ms_download_{key}.json" + tracker = tracker.absolute() + if tracker.resolve().is_relative_to(output.resolve()): + raise InvalidParameter("Keep the tracker outside the download directory to avoid uploading state files") + tracker.parent.mkdir(parents=True, exist_ok=True) + task_lock = FileLock(str(tracker) + ".lock") + try: + task_lock.acquire(timeout=1) + except Timeout as exc: + raise InvalidParameter( + "Another downloader is using this tracker; use separate outputs for parallel tasks" + ) from exc + try: + state = _DownloadState(tracker, identity) + try: + return _run( + manager, + state, + output, + repo_id, + repo_type, + revision, + allow_patterns, + ignore_patterns, + max_workers, + file_list, + retry_failed, + retry_policy or DownloadRetryPolicy(), + user_agent, + progress_callbacks, + cache_dir, + ) + finally: + state.close() + finally: + task_lock.release() + + +def _run( + manager: DownloadManager, + state: _DownloadState, + output: Path, + repo_id: str, + repo_type: str, + revision: str, + include: list[str] | None, + exclude: list[str] | None, + workers: int, + file_list: Iterable[dict] | None, + retry_failed: bool, + policy: DownloadRetryPolicy, + user_agent: dict | str | None, + callbacks: list[type[ProgressCallback]] | None, + cache_dir: Path | None, +) -> DownloadReport: + run = uuid.uuid4().hex + report = DownloadReport(repo_id=repo_id, revision=revision, output_dir=str(output), tracker_path=str(state.path)) + stop = threading.Event() + incoming: queue.Queue[tuple[dict | None, str | None]] = queue.Queue(maxsize=workers * 4) + producer_done = threading.Event() + pool = ThreadPoolExecutor(max_workers=workers, thread_name_prefix="download-report") + futures: dict[Future[None], str] = {} + + def put(entry: dict | None, error: str | None = None) -> None: + while not stop.is_set(): + try: + incoming.put((entry, error), timeout=0.1) + return + except queue.Full: + pass + + def produce() -> None: + try: + if file_list is not None: + if retry_failed: + for root in _listing_roots(include): + state.failed_pages(root or "/", reset=True) + for entry in file_list: + if stop.is_set(): + break + put(entry) + else: + with _DownloadControl(total_timeout=REPO_TREE_LIST_TIMEOUT_SECONDS, cancel_event=stop) as control: + with manager._client.read_context(control): + for root in _listing_roots(include): + if repo_type == "dataset": + for page, entries, error in manager._client.iter_dataset_file_pages( + repo_id, + revision, + root or "/", + isolated_pages=state.failed_pages(root or "/", reset=retry_failed), + ): + if error is not None: + state.fail_page(root or "/", page, error) + put(None, f"Root={root or '/'} page={page}: {safe_error(error)}") + for entry in entries: + put(entry) + control.check() + else: + for entry in manager._client.list_repo_files(repo_id, repo_type, revision, root=root): + put(entry) + if getattr(manager._client._read_state, "listing_incomplete", False): + put(None, f"Root={root or '/'}: model file listing is truncated") + except _DownloadCancelledError: + pass + except Exception as exc: + if isinstance(exc, AuthenticationError): + report.global_error = safe_error(exc) + put(None, f"Enumeration stopped: {safe_error(exc)}") + finally: + producer_done.set() + + def transfer(result: DownloadFileResult) -> None: + try: + # State hints never override a file that passes current metadata checks. + _validate_entry({"Path": result.path, "Size": result.size, "Sha256": result.sha256}, output) + if result.status in ("FAILED", "DEFERRED") and not retry_failed: + target = output / result.path + with _DownloadControl(total_timeout=DOWNLOAD_TOTAL_TIMEOUT, cancel_event=stop) as control: + valid = target.is_file() and manager._cache_hit( + target, + result.sha256, + result.size, + control=control, + ) + if valid: + state.finish(result.path, "VERIFIED") + return + with state.lock: + report.skipped_failures += 1 + skipped = report.skipped_failures + if skipped <= 10 or skipped % 1000 == 0: + logger.warning( + "Skipping isolated file %s (%d skipped); use retry_failed=True after investigation", + result.path, + skipped, + ) + state.finish(result.path, result.status) + return + previous = state.begin(result.path, reset=retry_failed and result.status in ("FAILED", "DEFERRED")) + recovery = _RecoveryBudget( + policy, + previous.retries, + previous.recovery_seconds, + lambda count, seconds: state.budget(result.path, count, seconds), + ) + manager.download_file( + repo_id, + repo_type, + result.path, + revision=revision, + local_dir=output, + cache_dir=cache_dir, + file_size=result.size, + expected_sha256=result.sha256, + user_agent=user_agent, + progress_callbacks=callbacks, + _cancel_event=stop, + _recovery=recovery, + ) + except _DownloadCancelledError: + state.finish(result.path, result.status if result.status in ("FAILED", "DEFERRED") else "PENDING") + raise + except Exception as exc: + origin = exc.__cause__ if isinstance(exc, _DownloadDeferredError) and exc.__cause__ else exc + terminal = isinstance(origin, (FileIntegrityError, InvalidParameter)) or not getattr( + origin, "retryable", False + ) + state.finish(result.path, "FAILED" if terminal else "DEFERRED", exc) + if ( + isinstance(exc, AuthenticationError) + or isinstance(exc, OSError) + and exc.errno in (errno.ENOSPC, errno.EROFS) + ): + raise + logger.error("Isolated download failure %s; continuing remaining files: %s", result.path, safe_error(exc)) + else: + state.finish(result.path, "VERIFIED") + + producer = threading.Thread(target=produce, name="download-enumeration", daemon=True) + producer.start() + interrupted = False + entry_count = 0 + try: + while futures or not producer_done.is_set() or not incoming.empty(): + while len(futures) < workers * 2: + try: + entry, listing_error = incoming.get(timeout=0.05 if not futures else 0) + except queue.Empty: + break + if listing_error is not None: + report.listing_complete = False + if len(report.listing_errors) < 100: + report.listing_errors.append(listing_error) + logger.error("Incomplete listing; continuing known files: %s", listing_error) + continue + entry_count += 1 + try: + if not isinstance(entry, dict): + raise InvalidParameter("File entry must be a dictionary") + if (entry.get("Type") or entry.get("type")) == "tree": + continue + raw_path = entry.get("Path") or entry.get("path") or entry.get("Name") + if isinstance(raw_path, str): + if ( + include + and not _matches_patterns(raw_path, include) + or exclude + and _matches_patterns(raw_path, exclude) + ): + continue + path, size, sha256 = _validate_entry(entry, output) + result = state.select(run, path, size, sha256) + except (InvalidParameter, ValueError, TypeError, OverflowError) as exc: + path = ( + str(entry.get("Path", f"")) + if isinstance(entry, dict) + else f"" + ) + if state.select(run, path, None, None) is not None: + state.finish(path, "FAILED", exc) + logger.error("Invalid file metadata; continuing: %s", safe_error(exc)) + continue + if result is None: + continue + future = pool.submit(transfer, result) + futures[future] = path + if futures: + completed, _ = wait(futures, timeout=0.1, return_when=FIRST_COMPLETED) + for future in completed: + futures.pop(future) + future.result() + except KeyboardInterrupt: + interrupted = True + report.status = "CANCELLED" + stop.set() + except Exception as exc: + report.global_error = safe_error(exc) + report.listing_complete = False + stop.set() + finally: + if stop.is_set(): + for future, path in futures.items(): + if future.cancel(): + state.finish(path, "PENDING") + pool.shutdown(wait=True, cancel_futures=True) + stop.set() + producer.join() + state.report(run, report) + logger.info( + "Download %s: verified=%d failed=%d deferred=%d pending=%d; report=%s", + report.status, + report.succeeded, + report.failed, + report.deferred, + report.pending, + report.report_path, + ) + if interrupted: + raise KeyboardInterrupt + return report diff --git a/src/modelscope_hub/api.py b/src/modelscope_hub/api.py index 1453215..99035b0 100644 --- a/src/modelscope_hub/api.py +++ b/src/modelscope_hub/api.py @@ -60,6 +60,8 @@ AgentTokenRecord, CacheInfo, CacheVerification, + DownloadReport, + DownloadRetryPolicy, FileInfo, PagedResult, RepoInfo, @@ -1579,6 +1581,62 @@ def download_file( file_size=file_size, ) + def download_repo_report( + self, + repo_id: str, + repo_type: RepoTypeLike, + *, + revision: str | None = None, + cache_dir: str | Path | None = None, + local_dir: str | 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: str | 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: + """Download known files despite ordinary failures; never label partial data complete. + + Unlike download_repo, file failures return a PARTIAL report instead of + raising an aggregate exception. The default hidden JSON tracker is in + cache_dir/.download_state, outside repository contents. The single JSON + combines exceptional files, failed pages, and the latest call's summary, + scoped to endpoint/repo/revision/ + output so batches share recovery hints. Actual size/hash validation + overrides stale failure hints; otherwise retry_failed=True or changed + metadata resets them. Missing/corrupt hints trigger bounded fresh + recovery, not false success. Inspect report.state_warnings for issues. + + file_list accepts an iterable of legacy Path/Size/Sha256 entries. Without + it, datasets are enumerated page by page with bounded backpressure. + Use a pinned revision or an authoritative manifest for migration jobs. + Inspect report.complete and report.iter_files('VERIFIED') before using + the downloaded data. report.report_path equals report.tracker_path; + its JSON has the latest summary under 'report', without per-run files. + iter_files streams temporary results owned by the live report; call + report.close() after consumption. Persisted hints cannot prove success. + """ + return self.downloader.download_repo_report( + repo_id, + str(self._normalize_repo_type(repo_type)), + revision=revision, + cache_dir=Path(cache_dir) if cache_dir else None, + local_dir=Path(local_dir) if local_dir else None, + allow_patterns=allow_patterns, + ignore_patterns=ignore_patterns, + max_workers=max_workers, + file_list=file_list, + tracker_path=Path(tracker_path) if tracker_path else None, + retry_failed=retry_failed, + retry_policy=retry_policy, + user_agent=user_agent, + progress_callbacks=progress_callbacks, + ) + def download_repo( self, repo_id: str, diff --git a/src/modelscope_hub/cli/download.py b/src/modelscope_hub/cli/download.py index e041662..74dc750 100644 --- a/src/modelscope_hub/cli/download.py +++ b/src/modelscope_hub/cli/download.py @@ -8,6 +8,7 @@ from ..api import HubApi from ..constants import RepoType +from ..errors import InvalidParameter from .base import CLICommand, SubParsers, add_repo_type_arg, info, make_api, success, warn from .compat import ( PatternAction, @@ -85,6 +86,15 @@ def register(subparsers: SubParsers) -> None: help="Glob to exclude (snapshot mode). Repeatable.", ) p.add_argument("--force", action="store_true", help="Re-download even if cached.") + p.add_argument( + "--continue-on-error", action="store_true", help="Continue with a persistent report; exit 3 for PARTIAL." + ) + p.add_argument( + "--tracker-path", + default=None, + help="Single hidden JSON for recovery hints and summary, outside downloaded contents.", + ) + p.add_argument("--retry-failed", action="store_true", help="Explicitly reset isolated files' recovery budgets.") # Legacy compat (hidden from --help) add_legacy_download_args(p) @@ -95,6 +105,12 @@ def register(subparsers: SubParsers) -> None: def execute(self) -> None: normalize_download_args(self.args) + continuing = getattr(self.args, "continue_on_error", False) + if not continuing and (getattr(self.args, "tracker_path", None) or getattr(self.args, "retry_failed", False)): + raise InvalidParameter("--tracker-path and --retry-failed require --continue-on-error") + if continuing and (self.args.force or self.args.repo_type == "collection"): + raise InvalidParameter("--continue-on-error does not support --force or collection downloads") + # Handle collection download separately if self.args.repo_type == "collection": self._download_collection() @@ -104,6 +120,25 @@ def execute(self) -> None: cache_dir: Path | None = Path(self.args.cache_dir) if self.args.cache_dir else None local_dir: Path | None = Path(self.args.local_dir) if self.args.local_dir else None + if continuing: + report = api.download_repo_report( + self.args.repo_id, + self.args.repo_type, + revision=self.args.revision, + cache_dir=cache_dir, + local_dir=local_dir, + allow_patterns=self.args.files or normalize_patterns(self.args.allow_patterns), + ignore_patterns=normalize_patterns(self.args.ignore_patterns), + max_workers=self.args.max_workers, + tracker_path=self.args.tracker_path, + retry_failed=self.args.retry_failed, + ) + info(f"{report.status}: {report.succeeded}/{report.total} verified; report: {report.report_path}") + if not report.complete: + warn("Partial data is not a complete snapshot. Only VERIFIED files may be consumed.") + raise SystemExit(3 if report.global_error is None else 1) + return + if self.args.files: for file_path in self.args.files: local = api.download_file( diff --git a/src/modelscope_hub/compat/snapshot_download.py b/src/modelscope_hub/compat/snapshot_download.py index b06208d..1cf93ec 100644 --- a/src/modelscope_hub/compat/snapshot_download.py +++ b/src/modelscope_hub/compat/snapshot_download.py @@ -15,7 +15,8 @@ from ..api import HubApi from ..constants import RepoType -from ..errors import AuthenticationError, NotExistError, PermissionDeniedError +from ..errors import AuthenticationError, InvalidParameter, NotExistError, PermissionDeniedError +from ..types import DownloadRetryPolicy from ..utils import append_user_agent from ..utils.patterns import normalize_patterns from .file_download import _resolve_legacy_paths @@ -43,6 +44,10 @@ def snapshot_download( local_files_only: bool = False, user_agent: dict | str | None = None, progress_callbacks: list[type[ProgressCallback]] | None = None, + continue_on_error: bool = False, + tracker_path: str | None = None, + retry_failed: bool = False, + retry_policy: DownloadRetryPolicy | None = None, ) -> str: """Download a repo snapshot (legacy signature). @@ -51,7 +56,9 @@ def snapshot_download( ``allow_file_pattern``/``ignore_file_pattern`` aliases when both are set. ``progress_callbacks`` takes a list of :class:`ProgressCallback` subclasses (not instances); each is instantiated per file to report - download progress. + download progress. continue_on_error opts into durable failed-file isolation + and returns the output path even when the persisted report is PARTIAL. Use + HubApi.download_repo_report for direct access to verified-file results. """ effective_id = repo_id or model_id if not effective_id: @@ -86,6 +93,27 @@ def snapshot_download( local_dir, api, ) + if continue_on_error: + if local_files_only: + raise InvalidParameter("Report-mode downloads cannot use local_files_only") + report = api.download_repo_report( + effective_id, + effective_type, + revision=revision, + cache_dir=effective_cache, + local_dir=effective_local, + allow_patterns=include, + ignore_patterns=exclude, + max_workers=max_workers, + tracker_path=tracker_path, + retry_failed=retry_failed, + retry_policy=retry_policy, + user_agent=user_agent, + progress_callbacks=progress_callbacks, + ) + return report.output_dir + if tracker_path is not None or retry_failed or retry_policy is not None: + raise InvalidParameter("Tracker/recovery options require continue_on_error=True") try: result = api.download_repo( effective_id, @@ -121,8 +149,12 @@ def dataset_snapshot_download( endpoint: str | None = None, local_files_only: bool = False, user_agent: dict | str | None = None, + continue_on_error: bool = False, + tracker_path: str | None = None, + retry_failed: bool = False, + retry_policy: DownloadRetryPolicy | None = None, ) -> str: - """Download a dataset snapshot (legacy signature).""" + """Download a dataset snapshot; report-mode options match snapshot_download.""" if cookies is not None: warnings.warn( "The 'cookies' parameter is deprecated and ignored. Use 'token' instead.", @@ -146,6 +178,26 @@ def dataset_snapshot_download( local_dir, api, ) + if continue_on_error: + if local_files_only: + raise InvalidParameter("Report-mode downloads cannot use local_files_only") + report = api.download_repo_report( + dataset_id, + "dataset", + revision=revision, + cache_dir=effective_cache, + local_dir=effective_local, + allow_patterns=include, + ignore_patterns=exclude, + max_workers=max_workers, + tracker_path=tracker_path, + retry_failed=retry_failed, + retry_policy=retry_policy, + user_agent=user_agent, + ) + return report.output_dir + if tracker_path is not None or retry_failed or retry_policy is not None: + raise InvalidParameter("Tracker/recovery options require continue_on_error=True") try: result = api.download_repo( dataset_id, diff --git a/src/modelscope_hub/types.py b/src/modelscope_hub/types.py index 4cf8316..c6b90ac 100644 --- a/src/modelscope_hub/types.py +++ b/src/modelscope_hub/types.py @@ -7,11 +7,13 @@ from __future__ import annotations -from collections.abc import Mapping +import json +import threading +from collections.abc import Iterator, Mapping from dataclasses import dataclass, field, fields from datetime import datetime, timezone from enum import Enum -from typing import Any, ClassVar, Generic, TypedDict, TypeVar +from typing import Any, ClassVar, Generic, TextIO, TypedDict, TypeVar from .constants import RepoType, Visibility @@ -52,6 +54,129 @@ def from_dict(cls: type[_TDataclass], data: Mapping[str, Any] | None) -> _TDatac return cls(**kwargs) # type: ignore[arg-type] +@dataclass(frozen=True, slots=True) +class DownloadRetryPolicy: + """Short recovery limits for report-mode downloads, shared by all file parts. + + Healthy transfers keep the regular total file timeout. Only failed-request + inactivity and retry waits consume recovery_timeout. Longer Retry-After + values defer the file instead of occupying a worker or retrying early. + """ + + max_retries: int = 2 + recovery_timeout: float = 60 + idle_timeout: float = 15 + low_speed_window: float = 10 + max_retry_delay: float = 5 + + def __post_init__(self) -> None: + if self.max_retries < 0 or any( + value <= 0 + for value in ( + self.recovery_timeout, + self.idle_timeout, + self.low_speed_window, + self.max_retry_delay, + ) + ): + raise ValueError("Download recovery limits must be positive; max_retries may be zero") + + +@dataclass(slots=True) +class DownloadFileResult(_FromDictMixin): + """A file validation result or an auxiliary recovery hint.""" + + path: str = "" + status: str = "PENDING" + size: int | None = None + sha256: str | None = None + retries: int = 0 + recovery_seconds: float = 0 + error: str | None = None + error_type: str | None = None + retryable: bool = False + + +class _DownloadReportResults: + """Own an anonymous result spool without adding it to dataclass serialization.""" + + _file_results: TextIO | None = None + + def _attach_file_results(self, stream: TextIO) -> None: + self._file_results = stream + self._file_results_lock = threading.Lock() + + def close(self) -> None: + """Release this report's temporary per-file results after consumption.""" + if self._file_results is not None: + with self._file_results_lock: + if self._file_results is not None: + self._file_results.close() + self._file_results = None + + def _iter_file_results(self) -> Iterator[dict]: + if self._file_results is None: + raise ValueError( + "Per-file results are available only on an open live DownloadReport; " + "the persisted JSON contains summary and recovery hints, not a verified-file manifest. " + "Rerun download_repo_report to validate actual files." + ) + offset = 0 + while True: + with self._file_results_lock: + if self._file_results is None: + raise ValueError("DownloadReport results have been closed") + self._file_results.seek(offset) + line = self._file_results.readline() + offset = self._file_results.tell() + if not line: + break + yield json.loads(line) + + +@dataclass(slots=True) +class DownloadReport(_DownloadReportResults, _FromDictMixin): + """Report-mode completion is explicit: COMPLETE, PARTIAL, or CANCELLED. + + Counts cover this invocation's selected files, not every previous batch. + Per-file results use an anonymous temporary spool owned by this live object, + preserving earlier calls without permanent manifests or unbounded memory. + One hidden JSON holds the latest summary and sparse recovery hints only. + VERIFIED describes validation during this call, not a permanent guarantee: + downstream users must check changed files. close() releases the spool; + state_warnings reports unavailable or ignored auxiliary recovery hints. + """ + + repo_id: str = "" + revision: str = "" + output_dir: str = "" + tracker_path: str = "" + report_path: str = "" + results_path: str = "" + status: str = "PARTIAL" + total: int = 0 + succeeded: int = 0 + failed: int = 0 + deferred: int = 0 + pending: int = 0 + skipped_failures: int = 0 + listing_complete: bool = True + listing_errors: list[str] = field(default_factory=list) + global_error: str | None = None + state_warnings: list[str] = field(default_factory=list) + + @property + def complete(self) -> bool: + return self.status == "COMPLETE" + + def iter_files(self, status: str | None = None) -> Iterator[DownloadFileResult]: + """Stream this live call's results; persisted summaries cannot prove success.""" + for raw in self._iter_file_results(): + result = DownloadFileResult.from_dict(raw) + if status is None or result.status == status: + yield result + + # --------------------------------------------------------------------------- # User # --------------------------------------------------------------------------- @@ -606,6 +731,9 @@ class TokenSignPayload(TypedDict): __all__ = [ + "DownloadRetryPolicy", + "DownloadFileResult", + "DownloadReport", "AgentIdentity", "AgentIdentitySummary", "AgentIdConfiguration", diff --git a/src/modelscope_hub/version.py b/src/modelscope_hub/version.py index 8844a57..48b097a 100644 --- a/src/modelscope_hub/version.py +++ b/src/modelscope_hub/version.py @@ -1,3 +1,3 @@ """Version information for modelscope_hub.""" -__version__ = "0.5.0+main" +__version__ = "0.5.1+main" diff --git a/tests/download_corruption_harness.py b/tests/download_corruption_harness.py new file mode 100644 index 0000000..2e2e08b --- /dev/null +++ b/tests/download_corruption_harness.py @@ -0,0 +1,473 @@ +"""Deterministic HTTP corruption injection using the unmodified SDK download pipeline.""" + +from __future__ import annotations + +import contextlib +import hashlib +import json +import socket +import threading +import time +from collections.abc import Callable, Iterator +from dataclasses import asdict, dataclass +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from urllib.parse import parse_qs, quote, unquote, urlparse + +from modelscope_hub import DownloadRetryPolicy, HubApi +from modelscope_hub._download import ProgressCallback +from modelscope_hub._download_state import safe_error + +GROUPS = { + 1: "persistent_empty_200", + 2: "persistent_short_object", + 3: "persistent_short_chunked", + 4: "persistent_bit_flips", + 5: "persistent_wrong_object", + 6: "persistent_overlong_chunked", + 7: "transient_empty_200", + 8: "transient_hash_mismatch", + 9: "truncated_connection_then_resume", + 10: "wrong_range_start_then_restart", +} + + +@dataclass(frozen=True) +class FaultCase: + """One unique repository file, never an alias for another corruption case.""" + + path: str + content: bytes + group: int = 0 + lfs: bool = True + + @property + def sha256(self) -> str: + return hashlib.sha256(self.content).hexdigest() + + @property + def cut(self) -> int: + return len(self.content) // 3 + 13 + + @property + def metadata(self) -> dict: + return {"Path": self.path, "Type": "blob", "Size": len(self.content), "Sha256": self.sha256} + + +def synthetic_cases() -> list[FaultCase]: + """Provide ten distinct files per group plus one hundred normal controls.""" + cases = [] + for index in range(100): + group = index // 10 + 1 + path = f"group-{group:02}/file-{index:03}.bin" + content = hashlib.shake_256(path.encode()).digest(32768 + index * 17) + cases.append(FaultCase(path, content, group, lfs=group >= 9 or index % 2 == 0)) + path = f"controls/normal-{index:03}.bin" + content = hashlib.shake_256(path.encode()).digest(32768 + index * 19) + cases.append(FaultCase(path, content, lfs=index % 2 == 0)) + return cases + + +class FaultHTTPServer: + """Serve valid HTTP framing, truncated sockets, redirects and incorrect ranges.""" + + def __init__(self, cases: list[FaultCase]) -> None: + self.cases = {case.path: case for case in cases} + self.lock = threading.Lock() + self.attempts: dict[tuple[str, str], int] = {} + self.wire: list[dict] = [] + self.credential_requests = 0 + state = self + + class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, *args) -> None: + pass + + def do_GET(self) -> None: # noqa: N802 - http.server API + if self.headers.get("Authorization") or self.headers.get("Cookie"): + with state.lock: + state.credential_requests += 1 + self._send(403, b"Credentials must not be sent to the local fault server") + return + url = urlparse(self.path) + path = ( + unquote(url.path.removeprefix("/cdn/")) + if url.path.startswith("/cdn/") + else parse_qs(url.query).get("FilePath", [""])[0] + ) + case = state.cases.get(path) + if case is None: + self._send(404, b"Unknown test file") + return + stage = self.headers.get("X-Test-Stage", "first") + with state.lock: + state.wire.append({"stage": stage, "path": path, "cdn": url.path.startswith("/cdn/")}) + if case.lfs and not url.path.startswith("/cdn/"): + self._send(302, b"", {"Location": f"/cdn/{quote(path)}"}) + return + with state.lock: + key = (stage, path) + attempt = state.attempts.get(key, 0) + 1 + state.attempts[key] = attempt + inject = self.headers.get("X-Test-Inject") == "1" + active = inject and (case.group <= 6 or attempt == 1) + body = case.content + group = case.group if active else 0 + if group in (1, 7): + self._send(200, b"") + elif group == 2: + self._send(200, body[: case.cut]) + elif group == 3: + self._chunked(body[: case.cut]) + elif group in (4, 8): + damaged = bytearray(body) + for index in (0, len(body) // 2, len(body) - 1): + damaged[index] ^= 1 + self._send(200, bytes(damaged)) + elif group == 5: + self._send(200, bytes(byte ^ 0xA5 for byte in body)) + elif group == 6: + self._chunked(body + b"EXCESS-CONTENT" * 128) + elif group == 9: + self.send_response(200) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body[: case.cut]) + self.wfile.flush() + self.close_connection = True + self.connection.shutdown(socket.SHUT_WR) + elif group == 10: + requested = self.headers.get("Range", "bytes=0-") + start = int(requested.removeprefix("bytes=").partition("-")[0]) + 1 + self._send(206, body[start:], {"Content-Range": f"bytes {start}-{len(body) - 1}/{len(body)}"}) + else: + requested = self.headers.get("Range") + if requested and case.lfs: + start_text, _, end_text = requested.removeprefix("bytes=").partition("-") + start = int(start_text) + end = min(int(end_text), len(body) - 1) if end_text else len(body) - 1 + if start >= len(body): + self._send(416, b"", {"Content-Range": f"bytes */{len(body)}"}) + else: + self._send( + 206, body[start : end + 1], {"Content-Range": f"bytes {start}-{end}/{len(body)}"} + ) + elif case.lfs: + self._send(200, body, {"X-Linked-ETag": case.sha256}) + else: + self._chunked(body) + + def _send(self, status: int, body: bytes, headers: dict | None = None) -> None: + self.send_response(status) + for name, value in (headers or {}).items(): + self.send_header(name, value) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def _chunked(self, body: bytes) -> None: + self.send_response(200) + self.send_header("Transfer-Encoding", "chunked") + self.end_headers() + for start in range(0, len(body), 4096): + chunk = body[start : start + 4096] + self.wfile.write(f"{len(chunk):x}\r\n".encode() + chunk + b"\r\n") + self.wfile.write(b"0\r\n\r\n") + + class QuietServer(ThreadingHTTPServer): + daemon_threads = True + + def handle_error(self, request, client_address) -> None: + # Integrity rejection legitimately closes an unread response body. + pass + + self.server = QuietServer(("127.0.0.1", 0), Handler) + self.url = f"http://127.0.0.1:{self.server.server_port}" + self.thread = threading.Thread(target=self.server.serve_forever, kwargs={"poll_interval": 0.05}, daemon=True) + + def __enter__(self) -> FaultHTTPServer: + self.thread.start() + return self + + def __exit__(self, *args) -> None: + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=2) + + +class FaultRouter: + """Inject at the Response boundary, preserving all SDK orchestration and checks.""" + + def __init__(self, server: FaultHTTPServer, *, live: bool = False) -> None: + self.server = server + self.live = live + self.stage = "first" + self.inject = True + self.records: list[dict] = [] + self.lock = threading.Lock() + self.local_api = HubApi(endpoint=server.url, token="") + self.local_api.legacy._read_session.trust_env = False + + def attach(self, api: HubApi) -> None: + original = api.legacy.download_stream + + def stream(repo_id, repo_type, file_path, revision="master", headers=None): + case = self.server.cases[file_path] + use_local = not self.live or self.inject and case.group != 0 + record = { + "stage": self.stage, + "path": file_path, + "range": (headers or {}).get("Range"), + "transport": "fault_http" if use_local else "modelscope.cn", + } + with self.lock: + self.records.append(record) + started = time.monotonic() + try: + if use_local: + control = api.legacy._read_state.control + local_headers = { + **(headers or {}), + "X-Test-Stage": self.stage, + "X-Test-Inject": "1" if self.inject and case.group else "0", + } + # Never forward auth from the real API facade to localhost. + local_headers = { + key: value + for key, value in local_headers.items() + if key.lower() not in ("authorization", "cookie") + } + with self.local_api.legacy.read_context(control): + response = self.local_api.legacy.download_stream( + repo_id, repo_type, file_path, revision, headers=local_headers + ) + else: + response = original(repo_id, repo_type, file_path, revision, headers=headers) + record.update( + status=response.status_code, + redirect_hops=len(response.history), + content_length=response.headers.get("Content-Length"), + content_range=response.headers.get("Content-Range"), + ) + return response + except Exception as exc: + record["error"] = safe_error(exc) + raise + finally: + record["header_seconds"] = round(time.monotonic() - started, 4) + + api.legacy.download_stream = stream + + def for_stage(self, stage: str) -> list[dict]: + return [record for record in self.records if record["stage"] == stage] + + def close(self) -> None: + self.local_api.legacy._read_session.close() + self.local_api.legacy._session.close() + + +def verify_file(target: Path, metadata: dict) -> None: + assert target.is_file(), str(target) + assert target.stat().st_size == int(metadata["Size"]), str(target) + digest = hashlib.sha256() + with target.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(chunk) + assert digest.hexdigest() == metadata["Sha256"].lower(), str(target) + + +def assert_single_json(root: Path, cases: list[FaultCase]) -> None: + data_paths = {root / "files" / case.path for case in cases} + generated = {path for path in root.rglob("*") if path.is_file() and path not in data_paths} + assert {path for path in generated if path.suffix in (".json", ".jsonl")} == {root / ".ms_download_state.json"} + assert not any(path.name.endswith((".tmp", ".incomplete", ".parallel_tmp")) for path in generated) + + +def exercise_stages( + root: Path, + cases: list[FaultCase], + api_factory: Callable[[], HubApi], + router: FaultRouter, + *, + fresh_manifest: Callable[[], list[dict]] | None = None, + announce: Callable[[str], None] = lambda message: None, +) -> dict: + """Exercise first failure, restart, no-reset recovery, explicit repair and reuse.""" + assert len(cases) == len({case.path for case in cases}) == 200 + assert sum(case.group > 0 for case in cases) == 100 + assert all(sum(case.group == group for case in cases) == 10 for group in GROUPS) + output = root / "files" + tracker = root / ".ms_download_state.json" + assert not tracker.exists() and not output.exists(), "The corruption experiment needs a fresh isolated directory" + persistent = {case.path for case in cases if 1 <= case.group <= 6} + metadata = [case.metadata for case in cases] + for case in cases: + if case.group == 10: + target = output / case.path + target.parent.mkdir(parents=True, exist_ok=True) + target.with_suffix(target.suffix + ".incomplete").write_bytes(case.content[: case.cut]) + policy = DownloadRetryPolicy() + evidence: dict = {"policy": asdict(policy), "fault_files": 100, "control_files": 100, "stages": {}, "cases": []} + report = None + try: + for stage in ("first", "restart", "source_repaired_without_reset", "explicit_repair", "reuse"): + router.stage = stage + router.inject = stage in ("first", "restart") + api = api_factory() + router.attach(api) + started = time.monotonic() + transfer_times: dict[str, list[float]] = {} + timing_lock = threading.Lock() + + class CaseProgress(ProgressCallback): + def __init__(self, filename: str, file_size: int): + super().__init__(filename, file_size) + with timing_lock: + transfer_times.setdefault(filename, [time.monotonic(), time.monotonic()]) + + def end(self) -> None: + with timing_lock: + transfer_times[self.filename][1] = time.monotonic() + + try: + report = api.download_repo_report( + "wangxingjun778/test_upload_large_2g", + "dataset", + local_dir=output, + tracker_path=tracker, + file_list=iter(metadata), + max_workers=4, + retry_policy=policy, + retry_failed=stage == "explicit_repair", + progress_callbacks=[CaseProgress], + ) + rows = {row.path: row for row in report.iter_files()} + requests = router.for_stage(stage) + evidence["stages"][stage] = { + "summary": asdict(report), + "seconds": round(time.monotonic() - started, 4), + "logical_file_requests": len(requests), + "remote_file_requests": sum(request["transport"] == "modelscope.cn" for request in requests), + "redirect_hops": sum(request.get("redirect_hops", 0) for request in requests), + } + announce( + f"{stage}: {report.status}，成功 {report.succeeded}，失败 {report.failed}，请求 {len(requests)}" + ) + assert report.total == 200 and len(rows) == 200 + assert report.global_error is None and report.listing_complete and not report.state_warnings + if stage in ("first", "restart", "source_repaired_without_reset"): + assert (report.status, report.succeeded, report.failed, report.deferred, report.pending) == ( + "PARTIAL", + 140, + 60, + 0, + 0, + ), evidence["stages"][stage] + assert {row.path for row in rows.values() if row.status == "FAILED"} == persistent + assert all(not (output / path).exists() for path in persistent) + else: + assert report.complete and report.succeeded == 200 + hints = json.loads(tracker.read_text(encoding="utf-8")) + assert hints["files"] == {} and hints["pages"] == {} + if stage == "first": + assert len(requests) == 300 + for case in cases: + selected = [request for request in requests if request["path"] == case.path] + row = rows[case.path] + assert len(selected) == (2 if case.group else 1), (case.path, selected) + if not case.group: + continue + assert row.retries == 1, (case.path, row) + if case.group == 9: + assert [request["range"] for request in selected] == [None, f"bytes={case.cut}-"] + assert selected[1]["status"] == 206 + if case.group == 10: + assert [request["range"] for request in selected] == [f"bytes={case.cut}-", None] + assert [request["status"] for request in selected] == [206, 200] + if case.group <= 6: + expected_error = "FileIntegrityError" if case.group in (4, 5) else "FileSizeMismatchError" + assert row.error_type == expected_error, row + group_index = sum(item["group"] == case.group for item in evidence["cases"]) + 1 + evidence["cases"].append( + { + "id": f"G{case.group}-{group_index:02}", + "group": case.group, + "fault": GROUPS[case.group], + "path": case.path, + "size": len(case.content), + "sha256": case.sha256, + "lfs": case.lfs, + "result": asdict(row), + "requests": selected, + "passed": True, + "seconds": round(transfer_times[case.path][1] - transfer_times[case.path][0], 4), + } + ) + assert (output / cases[-1].path).is_file(), "The tail control file must complete" + evidence["tail_control_completed"] = cases[-1].path + for case in cases: + if case.path not in persistent: + verify_file(output / case.path, case.metadata) + elif stage in ("restart", "source_repaired_without_reset"): + assert report.skipped_failures == 60 and requests == [] + elif stage == "explicit_repair": + assert len(requests) == 60 and {request["path"] for request in requests} == persistent + if fresh_manifest is not None: + current = {entry["Path"]: entry for entry in fresh_manifest() if entry.get("Type") != "tree"} + for entry in metadata: + remote = current[entry["Path"]] + assert (int(remote["Size"]), remote["Sha256"].lower()) == (entry["Size"], entry["Sha256"]) + for entry in metadata: + verify_file(output / entry["Path"], entry) + evidence["independent_size_sha256_verified"] = 200 + evidence["verified_bytes"] = sum(entry["Size"] for entry in metadata) + elif stage == "reuse": + assert requests == [] + assert_single_json(root, cases) + assert router.server.credential_requests == 0 + finally: + if report is not None: + report.close() + report = None + api.legacy._read_session.close() + api.legacy._session.close() + evidence["passed"] = True + evidence["passed_fault_cases"] = len(evidence["cases"]) + evidence["group_results"] = { + f"G{group}": { + "fault": name, + "cases": 10, + "passed": 10, + "first_status": "FAILED" if group <= 6 else "VERIFIED", + } + for group, name in GROUPS.items() + } + return evidence + except BaseException as exc: + evidence["passed"] = False + evidence["error"] = safe_error(exc) + evidence["observed_requests"] = router.records + raise + finally: + # Keep both success and failure evidence in the task's only JSON file. + if tracker.exists(): + from tempfile import NamedTemporaryFile + + data = json.loads(tracker.read_text(encoding="utf-8")) + data["corruption_regression"] = evidence + with NamedTemporaryFile(mode="w", encoding="utf-8", dir=root, suffix=".tmp", delete=False) as stream: + json.dump(data, stream, ensure_ascii=False, separators=(",", ":")) + temporary = Path(stream.name) + temporary.replace(tracker) + + +@contextlib.contextmanager +def local_experiment(cases: list[FaultCase]) -> Iterator[tuple[FaultHTTPServer, FaultRouter]]: + with FaultHTTPServer(cases) as server: + router = FaultRouter(server) + try: + yield server, router + finally: + router.close() diff --git a/tests/test_download_corruption.py b/tests/test_download_corruption.py new file mode 100644 index 0000000..6420542 --- /dev/null +++ b/tests/test_download_corruption.py @@ -0,0 +1,38 @@ +"""One hundred corruption cases, real HTTP framing, isolation and explicit repair.""" + +from __future__ import annotations + +import json +from pathlib import Path + +from modelscope_hub import HubApi +from tests.download_corruption_harness import exercise_stages, local_experiment, synthetic_cases + + +def test_hundred_corrupt_files_continue_isolate_repair_and_reuse(tmp_path, monkeypatch): + monkeypatch.setenv("NO_PROXY", "127.0.0.1,localhost") + monkeypatch.setenv("MODELSCOPE_DOWNLOAD_INTRA_CLOUD", "false") + monkeypatch.delenv("MODELSCOPE_DOWNLOAD_INTRA_CLOUD_REGION", raising=False) + monkeypatch.delenv("MODELSCOPE_DOWNLOAD_INTER_CLOUD_REGIONS", raising=False) + cases = synthetic_cases() + root = tmp_path / "hundred-corrupt-files" + with local_experiment(cases) as (server, router): + evidence = exercise_stages(root, cases, lambda: HubApi(endpoint=server.url, token=""), router) + assert server.credential_requests == 0 + assert evidence["passed_fault_cases"] == 100 + assert all(group["passed"] == 10 for group in evidence["group_results"].values()) + assert evidence["independent_size_sha256_verified"] == 200 + stages = evidence["stages"] + assert [stage["logical_file_requests"] for stage in stages.values()] == [300, 0, 0, 60, 0] + assert stages["first"]["summary"]["succeeded"] == 140 + assert stages["explicit_repair"]["summary"]["succeeded"] == 200 + ranges = [row for row in evidence["cases"] if row["group"] in (9, 10)] + assert len(ranges) == 20 + assert all(row["result"]["retries"] == 1 for row in evidence["cases"]) + tracker = root / ".ms_download_state.json" + saved = json.loads(tracker.read_text()) + assert saved["files"] == {} and saved["pages"] == {} + assert saved["corruption_regression"]["passed"] + assert len(list(root.rglob("*.json"))) == 1 + assert not list(root.rglob("*.jsonl")) + assert Path(stages["reuse"]["summary"]["report_path"]) == tracker diff --git a/tests/test_download_lock_path.py b/tests/test_download_lock_path.py index 1d70d0e..75664b1 100644 --- a/tests/test_download_lock_path.py +++ b/tests/test_download_lock_path.py @@ -83,6 +83,8 @@ def test_dataset_file_download_accepts_long_file_path(self, tmp_path, monkeypatc been acquired, and write the expected target file directly. """ + acquired_lock_files = [] + def fake_download_with_resume( self, repo_id: str, @@ -92,6 +94,7 @@ def fake_download_with_resume( target: Path, **kwargs, ) -> Path: + acquired_lock_files.extend((cache_dir / ".lock").glob("*.lock")) target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(b"image-bytes") return target @@ -108,6 +111,6 @@ def fake_download_with_resume( ) assert Path(result).read_bytes() == b"image-bytes" - lock_files = list((cache_dir / ".lock").glob("*.lock")) - assert lock_files - assert all(len(path.name.encode("utf-8")) < 255 for path in lock_files) + # New filelock releases remove Unix lock files, so inspect them while held. + assert acquired_lock_files + assert all(len(path.name.encode("utf-8")) < 255 for path in acquired_lock_files) diff --git a/tests/test_download_report.py b/tests/test_download_report.py new file mode 100644 index 0000000..efd31ae --- /dev/null +++ b/tests/test_download_report.py @@ -0,0 +1,526 @@ +"""Persistent isolation and bounded recovery across large download batches.""" + +from __future__ import annotations + +import importlib +import json +import threading +import time +from dataclasses import asdict +from pathlib import Path + +import pytest + +from modelscope_hub import DownloadReport, DownloadRetryPolicy, HubApi +from modelscope_hub.cli.main import run_cmd +from modelscope_hub.errors import InvalidParameter, NetworkError +from tests.test_download_integrity import REPO, _sha256 +from tests.test_download_integrity import api as api +from tests.test_download_integrity import dirs as dirs +from tests.test_download_integrity import hub as hub + + +def manifest(hub): + return [ + {"Path": path, "Size": len(file.content), "Sha256": _sha256(file.content)} for path, file in hub.files.items() + ] + + +def run(api, dirs, **kwargs): + return api.download_repo_report(REPO, "dataset", local_dir=dirs[0], cache_dir=dirs[1], **kwargs) + + +def test_failure_does_not_stop_files_after_the_first_batch(hub, api, dirs): + for i in range(25): + hub.add(f"shards/{i:02}.tar", b"data" * 1024, served=b"" if i == 3 else None) + report = run(api, dirs, max_workers=2) + assert report.status == "PARTIAL" + assert (report.total, report.succeeded, report.failed) == (25, 24, 1) + assert report.listing_complete + assert len(hub.seen("shards/03.tar", "cdn")) == 2 + assert (dirs[0] / "shards/24.tar").exists() + assert not (dirs[0] / "shards/03.tar").exists() + assert len(list(report.iter_files("VERIFIED"))) == 24 + assert json.loads(Path(report.report_path).read_text())["report"]["status"] == "PARTIAL" + + +def test_new_api_instances_and_batches_share_failure_isolation(hub, api, dirs): + hub.add("bad.bin", b"data", served=b"") + hub.add("next.bin", b"good") + first = run(api, dirs, allow_patterns=["bad.bin"]) + again = run(HubApi(endpoint=hub.url, token=""), dirs) + assert first.tracker_path == again.tracker_path + assert again.skipped_failures == 1 + assert len(hub.seen("bad.bin", "cdn")) == 2 + assert (dirs[0] / "next.bin").read_bytes() == b"good" + assert first.total == 1 # The first report remains immutable after later batches. + assert [file.path for file in first.iter_files()] == ["bad.bin"] + + +def test_failed_file_is_only_reopened_explicitly_or_on_metadata_change(hub, api, dirs): + hub.add("bad.bin", b"data", served=b"") + first = run(api, dirs) + hub.files["bad.bin"].served = None + second = run(api, dirs) + assert not second.complete + assert len(hub.seen("bad.bin", "cdn")) == 2 + recovered = run(api, dirs, retry_failed=True) + assert recovered.complete + assert len(hub.seen("bad.bin", "cdn")) == 3 + hub.files["bad.bin"].content = b"new version" + changed = run(api, dirs) + assert changed.complete + assert (dirs[0] / "bad.bin").read_bytes() == b"new version" + assert first.failed == 1 + + +def test_tracker_scope_rejects_another_output_or_revision(hub, api, dirs, tmp_path): + hub.add("x.bin", b"data") + ledger = tmp_path / ".ms_download_state.json" + run(api, dirs, tracker_path=ledger) + with pytest.raises(InvalidParameter): + run(api, (tmp_path / "other", dirs[1]), tracker_path=ledger) + with pytest.raises(InvalidParameter): + run(api, dirs, tracker_path=ledger, revision="v2") + with pytest.raises(InvalidParameter): + run(api, dirs, tracker_path=dirs[0] / ".ms_download_state.json") + + +def test_long_retry_after_defers_without_sleeping_or_occupying_a_worker(hub, api, dirs): + hub.add("busy.bin", b"data", status=429, retry_after="9999") + hub.add("next.bin", b"good") + started = time.monotonic() + report = run(api, dirs, max_workers=1) + assert time.monotonic() - started < 1.5 + assert report.deferred == 1 + assert report.succeeded == 1 + assert len(hub.seen("busy.bin", "cdn")) == 1 + run(api, dirs) + assert len(hub.seen("busy.bin", "cdn")) == 1 + + +def test_transport_and_hash_retries_share_one_file_budget(hub, api, dirs): + hub.add("bad.bin", b"good" * 2048, served=b"evil" * 2048, truncate_first=3000) + report = run(api, dirs, file_list=manifest(hub)) + result = next(report.iter_files()) + assert result.status == "FAILED" + assert result.retries == 2 + assert len(hub.seen("bad.bin", "cdn")) == 3 + assert not (dirs[0] / "bad.bin").exists() + + +def test_complete_but_corrupt_partial_is_rejected_before_network_and_repaired(hub, api, dirs): + content = hub.add("partial.bin", b"correct data" * 4096) + dirs[0].mkdir() + partial = dirs[0] / "partial.bin.incomplete" + partial.write_bytes(b"wrong bytes!" * 4096) + assert partial.stat().st_size == len(content) + report = run(api, dirs, file_list=manifest(hub)) + assert report.complete + assert next(report.iter_files()).retries == 1 + assert (dirs[0] / "partial.bin").read_bytes() == content + assert not partial.exists() + assert len(hub.seen("partial.bin", "cdn")) == 1 + report.close() + + +def test_same_size_corrupt_parallel_parts_are_rejected_and_do_not_stop_next_file(hub, api, dirs, monkeypatch): + from modelscope_hub import _download + + monkeypatch.setattr(_download, "DOWNLOAD_PARALLEL_THRESHOLD", 64 * 1024) + monkeypatch.setattr(_download, "DOWNLOAD_PART_SIZE", 64 * 1024) + monkeypatch.setattr(_download, "DOWNLOAD_PARALLELS", 4) + hub.add("bad.bin", b"good" * 75000, served=b"evil" * 75000) + good = hub.add("next.bin", b"correct tail") + report = run(api, dirs, file_list=manifest(hub), max_workers=1) + assert (report.failed, report.succeeded) == (1, 1) + failed = next(report.iter_files("FAILED")) + assert failed.error_type == "FileIntegrityError" and failed.retries == 1 + assert len(hub.seen("bad.bin", "cdn")) == 10 # Five ranges per attempt, two complete attempts. + assert all(request.range for request in hub.seen("bad.bin", "cdn")) + assert not list(dirs[0].glob("bad.bin*")) + assert (dirs[0] / "next.bin").read_bytes() == good + report.close() + + +def test_stalled_headers_use_short_policy_and_no_nested_retries(hub, api, dirs): + hub.add("stall.bin", b"x", header_delay=2) + hub.add("good.bin", b"good") + policy = DownloadRetryPolicy(max_retries=1, idle_timeout=0.1, recovery_timeout=0.4) + report = run(api, dirs, retry_policy=policy, max_workers=1) + assert report.deferred == 1 + assert report.succeeded == 1 + assert len(hub.seen("stall.bin", "cdn")) == 2 + assert next(report.iter_files("DEFERRED")).retries == 1 + + +def test_invalid_metadata_and_paths_are_isolated_without_network(hub, api, dirs, tmp_path): + hub.add("good.bin", b"data") + entries = [ + {"Path": "missing.bin"}, + {"Path": "negative.bin", "Size": -1}, + {"Path": "../escape.bin", "Size": 1}, + {"Path": "bad-hash.bin", "Sha256": "invalid"}, + *manifest(hub), + ] + report = run(api, dirs, file_list=iter(entries)) + assert (report.failed, report.succeeded) == (4, 1) + assert not (tmp_path / "escape.bin").exists() + assert {request.path for request in hub.requests} == {"good.bin"} + + +def test_pagination_downloads_page_one_before_page_two_finishes(hub, api, dirs, monkeypatch): + hub.add("a.bin", b"a") + hub.add("b.bin", b"b") + entries = manifest(hub) + + def pages(*args, **kwargs): + yield 1, entries[:1], None + limit = time.monotonic() + 2 + while not (dirs[0] / "a.bin").exists() and time.monotonic() < limit: + time.sleep(0.01) + assert (dirs[0] / "a.bin").exists() + yield 2, [], NetworkError("page missing") + yield 3, entries[1:], None + + monkeypatch.setattr(api.legacy, "iter_dataset_file_pages", pages) + report = run(api, dirs) + assert report.succeeded == 2 + assert not report.listing_complete + assert "page=2" in report.listing_errors[0] + + +def test_manifest_iterable_is_bounded_and_not_materialized(hub, api, dirs): + hub.add("x.bin", b"x") + + def entries(): + yield manifest(hub)[0] + limit = time.monotonic() + 2 + while not (dirs[0] / "x.bin").exists() and time.monotonic() < limit: + time.sleep(0.01) + assert (dirs[0] / "x.bin").exists() + + assert run(api, dirs, file_list=entries()).complete + + +def test_cancelled_run_has_durable_pending_state_and_is_resumable(hub, api, dirs, monkeypatch): + import modelscope_hub._resilient_download as module + + hub.add("x.bin", b"x" * 100, header_delay=2) + entered = threading.Event() + original_wait = module.wait + + def interrupt(*args, **kwargs): + if not entered.is_set(): + entered.set() + time.sleep(0.05) + raise KeyboardInterrupt + return original_wait(*args, **kwargs) + + monkeypatch.setattr(module, "wait", interrupt) + with pytest.raises(KeyboardInterrupt): + run(api, dirs) + reports = list((dirs[1] / ".download_state").glob("*.json")) + assert len(reports) == 1 + assert json.loads(reports[0].read_text())["report"]["status"] == "CANCELLED" + hub.files["x.bin"].header_delay = 0 + assert run(api, dirs).complete + + +def test_crashed_recovery_does_not_receive_a_fresh_automatic_budget(hub, api, dirs): + hub.add("x.bin", b"x", served=b"") + first = run(api, dirs) + ledger = Path(first.tracker_path) + data = json.loads(ledger.read_text()) + data["files"]["x.bin"]["status"] = "DOWNLOADING" + ledger.write_text(json.dumps(data)) + again = run(api, dirs) + assert again.skipped_failures == 1 + assert again.deferred == 1 + assert len(hub.seen("x.bin", "cdn")) == 2 + + +def test_cli_partial_exit_three_and_strict_mode_unchanged(hub, dirs, monkeypatch): + monkeypatch.setenv("MODELSCOPE_API_TOKEN", "") + hub.add("bad.bin", b"data", served=b"") + hub.add("good.bin", b"good") + args = [ + "--endpoint", + hub.url, + "download", + REPO, + "--repo-type", + "dataset", + "--local-dir", + str(dirs[0]), + "--cache-dir", + str(dirs[1]), + ] + assert run_cmd([*args, "--continue-on-error"]) == 3 + assert (dirs[0] / "good.bin").exists() + assert run_cmd([*args, "--continue-on-error"]) == 3 + assert len(hub.seen("bad.bin", "cdn")) == 2 + assert run_cmd(args) == 1 + + +@pytest.mark.parametrize("name", ["snapshot_download", "dataset_snapshot_download"]) +def test_compat_continue_mode_is_explicit_and_survives_new_instances(hub, dirs, name): + module = importlib.import_module("modelscope_hub.compat.snapshot_download") + function = getattr(module, name) + hub.add("bad.bin", b"data", served=b"") + hub.add("good.bin", b"good") + kwargs = dict(endpoint=hub.url, token="", local_dir=str(dirs[0]), cache_dir=str(dirs[1]), continue_on_error=True) + if name == "snapshot_download": + kwargs["repo_type"] = "dataset" + assert function(REPO, **kwargs) == str(dirs[0]) + assert function(REPO, **kwargs) == str(dirs[0]) + assert len(hub.seen("bad.bin", "cdn")) == 2 + + +def test_bad_listing_page_is_durably_skipped_and_does_not_hide_later_pages(hub, api, dirs, monkeypatch): + hub.add("a.bin", b"a") + hub.add("b.bin", b"b") + entries = manifest(hub) + calls = [] + + def request(path, params, **kwargs): + page = params["PageNumber"] + calls.append(page) + if page == 2: + raise NetworkError("missing page") + response = type( + "Response", + (), + {"json": lambda self: {"Data": {"Files": [entries[0]] * 200 if page == 1 else [entries[1]]}}}, + )() + return response + + monkeypatch.setattr(api.legacy, "_request_repo_tree", request) + first = run(api, dirs) + assert first.succeeded == 2 + assert not first.listing_complete + assert calls == [1, 2, 3] + calls.clear() + again = run(api, dirs) + assert not again.listing_complete + assert calls == [1, 3] + + +def test_report_header_trickle_is_stopped_without_waiting_for_whole_file_deadline(hub, api, dirs): + hub.add("slow.bin", b"data", header_trickle=True) + hub.add("next.bin", b"next") + started = time.monotonic() + report = run(api, dirs, retry_policy=DownloadRetryPolicy(max_retries=0, idle_timeout=0.15)) + assert time.monotonic() - started < 1 + assert report.deferred == 1 + assert report.succeeded == 1 + + +def test_duplicates_cannot_reset_an_active_files_metadata(hub, api, dirs): + hub.add("x.bin", b"data") + entry = manifest(hub)[0] + report = run(api, dirs, file_list=[entry, dict(entry, Size=999), entry]) + assert report.total == 1 + assert report.succeeded == 1 + assert len(hub.seen("x.bin", "cdn")) == 1 + + +def test_business_listing_failure_is_not_an_empty_complete_dataset(hub, api, dirs, monkeypatch): + from unittest.mock import MagicMock + + response = MagicMock() + response.json.return_value = {"Success": False, "Code": 500} + monkeypatch.setattr(api.legacy, "_request_repo_tree", lambda *args, **kwargs: response) + report = run(api, dirs) + assert not report.complete + assert not report.listing_complete + assert len(report.listing_errors) == 2 + assert report.total == 0 + + +def test_healthy_files_do_not_write_persistent_records_per_file(hub, api, dirs, monkeypatch): + from modelscope_hub._download_state import _DownloadState + + writes = [] + original_save = _DownloadState._save + + def saved(state): + writes.append(1) + original_save(state) + + monkeypatch.setattr(_DownloadState, "_save", saved) + for i in range(30): + hub.add(f"file-{i}.bin", b"good") + report = run(api, dirs) + assert report.complete + assert writes == [1, 1] # One scope header and one final summary, independent of file count. + data = json.loads(Path(report.tracker_path).read_text()) + assert data["files"] == {} + assert Path(report.tracker_path).name.startswith(".") + assert len(list(report.iter_files("VERIFIED"))) == 30 + run(api, dirs) + assert writes == [1, 1, 1] # A rerun only replaces the latest summary. + + +def test_repaired_file_overrides_stale_failure_hint_without_network(hub, api, dirs): + content = hub.add("bad.bin", b"valid data", served=b"") + first = run(api, dirs) + assert first.failed == 1 + (dirs[0] / "bad.bin").write_bytes(content) + repaired = run(api, dirs) + assert repaired.complete + assert repaired.skipped_failures == 0 + assert len(hub.seen("bad.bin", "cdn")) == 2 + assert json.loads(Path(repaired.tracker_path).read_text())["files"] == {} + + +def test_deleted_state_does_not_redownload_valid_files(hub, api, dirs): + hub.add("good.bin", b"valid data") + first = run(api, dirs) + Path(first.tracker_path).unlink() + again = run(api, dirs) + assert again.complete + assert len(hub.seen("good.bin", "cdn")) == 1 + + +@pytest.mark.parametrize("damaged", ['{"version":', '{"version": 1, "identity": null}', "\ufffd"]) +def test_corrupt_state_warns_and_revalidates_actual_files(hub, api, dirs, damaged): + hub.add("good.bin", b"valid data") + first = run(api, dirs) + Path(first.tracker_path).write_text(damaged) + again = run(api, dirs) + assert again.complete + assert again.state_warnings + assert len(hub.seen("good.bin", "cdn")) == 1 + + +def test_success_hint_cannot_hide_same_size_corruption(hub, api, dirs): + hub.add("good.bin", b"good") + first = run(api, dirs) + (dirs[0] / "good.bin").write_bytes(b"evil") + data = json.loads(Path(first.tracker_path).read_text()) + data["files"]["good.bin"] = asdict(next(first.iter_files())) + Path(first.tracker_path).write_text(json.dumps(data)) + again = run(api, dirs) + assert again.complete + assert again.state_warnings # Ignore VERIFIED claims in auxiliary state. + assert (dirs[0] / "good.bin").read_bytes() == b"good" + assert len(hub.seen("good.bin", "cdn")) == 2 + + +def test_cannot_save_hints_does_not_block_downloads(hub, api, dirs, monkeypatch): + hub.add("bad.bin", b"data", served=b"") + hub.add("good.bin", b"good") + original = Path.replace + + def fail_state_replace(path, target): + if str(target).endswith(".json") and ".report." not in str(target): + raise PermissionError("hint file cannot be replaced") + return original(path, target) + + monkeypatch.setattr(Path, "replace", fail_state_replace) + report = run(api, dirs) + assert report.failed == 1 + assert report.succeeded == 1 + assert report.state_warnings + assert (dirs[0] / "good.bin").read_bytes() == b"good" + + +def test_missing_state_does_not_promote_a_partial_file_to_verified(hub, api, dirs): + hub.add("good.bin", b"good") + dirs[0].mkdir() + (dirs[0] / "good.bin").write_bytes(b"go") + report = run(api, dirs) + assert report.complete + assert (dirs[0] / "good.bin").read_bytes() == b"good" + + +def test_import_works_when_embedded_database_modules_are_unavailable(): + import subprocess + import sys + + code = """ +import importlib.abc +import sys +class BlockDatabase(importlib.abc.MetaPathFinder): + def find_spec(self, fullname, path=None, target=None): + if fullname.split(".")[0] in {"sqlite3", "duckdb", "_sqlite3"}: + raise ImportError("Embedded databases are unavailable") +sys.meta_path.insert(0, BlockDatabase()) +import modelscope_hub._download_state +import modelscope_hub._resilient_download +""" + subprocess.run([sys.executable, "-c", code], check=True, capture_output=True, text=True) + + +def test_single_json_is_reused_across_failure_restart_and_repair(hub, api, dirs, tmp_path): + hub.add("bad.bin", b"data", served=b"") + hub.add("good.bin", b"good") + task = tmp_path / "task" + tracker = task / ".ms_download_state.json" + first = run(api, dirs, tracker_path=tracker) + second = run(api, dirs, tracker_path=tracker) + repaired = run(api, dirs, tracker_path=tracker, retry_failed=True) + assert first.failed == second.failed == repaired.failed == 1 + hub.files["bad.bin"].served = None + final = run(api, dirs, tracker_path=tracker, retry_failed=True) + assert final.complete + assert all(report.report_path == report.tracker_path == str(tracker) for report in (first, second, final)) + assert final.results_path == "" + assert list(task.glob("*.json")) == [tracker] + assert not list(task.glob("*.jsonl")) + assert not list(task.glob("*.tmp")) + data = json.loads(tracker.read_text()) + assert data["files"] == {} and data["pages"] == {} + assert data["report"]["succeeded"] == 2 + assert data["report"]["status"] == "COMPLETE" + assert [result.status for result in first.iter_files("FAILED")] == ["FAILED"] + assert len(list(final.iter_files("VERIFIED"))) == 2 + assert "_file_results" not in asdict(final) + + +@pytest.mark.parametrize("reset", [False, True]) +def test_explicit_manifest_recovery_resets_selected_page_hints(hub, api, dirs, reset): + hub.add("good.bin", b"good") + first = run(api, dirs) + tracker = Path(first.tracker_path) + data = json.loads(tracker.read_text()) + data["pages"] = {"/": {"13": "Previous page timeout"}} + tracker.write_text(json.dumps(data)) + recovered = run(api, dirs, file_list=manifest(hub), retry_failed=reset) + assert recovered.complete + assert json.loads(tracker.read_text())["pages"] == ({} if reset else data["pages"]) + assert len(hub.seen("good.bin", "cdn")) == 1 + + +def test_result_iterators_are_independent_and_close_releases_spool(hub, api, dirs): + for i in range(3): + hub.add(f"{i}.bin", b"good") + report = run(api, dirs) + left, right = report.iter_files(), report.iter_files() + assert next(left).path == next(right).path + assert [entry.path for entry in left] == [entry.path for entry in right] + assert len(list(report.iter_files())) == 3 + spool = report._file_results + report.close() + report.close() + assert spool.closed + with pytest.raises(ValueError, match="open live DownloadReport"): + list(report.iter_files()) + + +def test_persisted_summary_cannot_be_used_as_verified_file_evidence(hub, api, dirs): + hub.add("good.bin", b"good") + report = run(api, dirs) + summary = json.loads(Path(report.report_path).read_text())["report"] + restored = DownloadReport.from_dict(summary) + assert restored.complete + with pytest.raises(ValueError, match="not a verified-file manifest"): + list(restored.iter_files("VERIFIED")) + + +def test_report_types_and_policy_are_public(): + assert not DownloadReport().complete + assert asdict(DownloadRetryPolicy())["max_retries"] == 2 + with pytest.raises(ValueError): + DownloadRetryPolicy(idle_timeout=0)