diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..2b94ee3 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,140 @@ +# AGENTS.md — modelscope_hub Engineering Conventions + +This document defines repository-level engineering conventions that apply to every automated or manual change in this repository. In addition to the general requirements, it **strictly governs changes to the `upload_folder` upload path**: every change that touches this path must perform the specified scenario-based simulation review and include an Upload Simulation Regression Report. + +--- + +## 1. General Engineering Requirements + +- This repository uses a src-layout. Always run code and tests against local sources: `PYTHONPATH=src` (a package installed with `pip install` may be outdated). +- Before committing, run the following gates in order; all must pass: + 1. Targeted unit tests: `PYTHONPATH=src pytest -q ` + 2. Full non-remote test suite: `PYTHONPATH=src pytest -q tests/ -k 'not remote' --ignore=tests/integration` + 3. Style checks: `ruff check src/ tests/` and `ruff format --check src/ tests/` + 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`. + +--- + +## 2. Controlled Scope: `upload_folder` Path + +Changing any of the following files or symbols constitutes a change to the upload path and requires the mandatory simulation regression in Section 5: + +- `src/modelscope_hub/_upload.py` + - Batching: `_calculate_adaptive_batch_size`, `_plan_commit_batches`, `_estimate_commit_operation_bytes`, `_plan_result_batches` + - Routing: `_is_lfs`, `_is_inline_metadata`, `_upload_mode` + - Main flow and recovery: `upload_folder`, `_retry_failed_commits`, `_retry_failed_files_react`, `_retry_failed_simple`, `_commit_with_retry` + - Failure classification: `classify_error`, `_ErrorCategory`, `_is_retryable_commit_error` + - Preflight validation: `_warn_advisory_upload_limits`, `_prepare_upload_folder` + - State tracking: `UploadTracker` (`begin_attempt` / `mark_*`) +- `src/modelscope_hub/errors.py`: `raise_for_status`, `_COMMIT_RETRYABLE_BUSINESS_CODES`, and HTTP/business-code mappings +- `src/modelscope_hub/_legacy_api.py`: `create_commit`, including the HTTP 200 / `Success:false` check +- `src/modelscope_hub/constants.py`: upload-related constants such as `UPLOAD_*` and `COMMIT_MAX_ACTIONS_PER_REQUEST` + +--- + +## 3. Commit Batching Semantics That Must Be Preserved + +Only files that are **pending** (not `COMMITTED`) for the current run are planned into batches. This prevents a resumed upload from turning a small number of remaining files into multiple tiny commits. + +Apply the following three constraints; start a new batch as soon as any constraint is reached: + +1. **Operation target (primary constraint):** `UPLOAD_COMMIT_BATCH_MAX_OPERATIONS`, default `256`. `_calculate_adaptive_batch_size` must return `min(target, 2000, pending_count)`—for fewer files, it converges to the pending count and by default never exceeds `256`. +2. **Request-body budget (secondary constraint):** `UPLOAD_COMMIT_MAX_INLINE_BYTES`, default `8 MiB`. Count only normal files, using base64 expansion (`×4/3`) plus JSON overhead. For LFS files, count only pointer metadata (about `370 B/op`); **never include blob payload bytes**. +3. **Server hard limit:** `COMMIT_MAX_ACTIONS_PER_REQUEST=2000`. Clamp to this limit in every case; it must never be exceeded. + +The following behavior must also be preserved: + +- **Tail-batch rebalancing:** when `last_batch × 4 < previous_batch` and both rebalanced halves meet all constraints, rebalance the final two batches (for example, `256/32 → 144/144`). Do not add commits, and preserve deterministic path order. +- **Oversized inline fallback:** if one inline file alone exceeds the budget, place it in its own batch and emit a warning; the upload must not stall. +- **Recovery uses the same planner:** `_retry_failed_commits`, ReAct recovery, and simple retry must all use `_plan_result_batches` (and therefore the same `_plan_commit_batches`). +- **`delete_files` batches independently:** split only at the hard limit of `2000`, not at `256`. +- **`upload_file`:** one file and one commit per call; no batching and no tracker. + +### Routing Rules: LFS vs. Inline + +- Files matched by `_is_inline_metadata` (such as `README.md`, `.gitattributes`, `.gitignore`, and `config*.json`) are **always inline**. +- Otherwise, `size > UPLOAD_LFS_FORCE_THRESHOLD_BYTES` (default **64 KiB**) routes to LFS. +- Otherwise, files matching `DATASET_LFS_SUFFIX` or `MODEL_LFS_SUFFIX` route to LFS. +- All other files are normal inline files. + +--- + +## 4. Failure and Retry Model That Must Be Preserved + +- Failures have three categories, and **all of them must be included in final statistics**: + - `retry_failed_files`: transport failure → ReAct recovery (parallel or serial retries, backoff, then single-file handling). + - `retry_commit_batches`: retryable commit failure → **commit-only recovery** (do not re-upload blobs). + - `terminal_failures`: non-retryable in the current run; record as `FAILED` in the tracker and include in the final `Failed` count. +- **Retryability in commit context** (`_is_retryable_commit_error` / `classify_error(commit_context=True)`): + - Business code `10030000001` ("branch changed, please retry" / "commit rejected by repository policy") is retryable only at a `/commit/` endpoint; `raise_for_status` maps it to `ServerError`. + - During commit-only recovery, policy-related rejections recursively split batches in half by file count, down to individual files. + - Genuine parameter errors (for example, an invalid-path E3021, a 400 without a policy code, or 401/403/404) remain **non-retryable**. +- **HTTP 200 with `Success:false`:** `LegacyClient.create_commit` must raise `APIError` and enter the unified failure path; false successes are prohibited. +- **Abort after consecutive failures:** after `UPLOAD_COMMIT_MAX_CONSECUTIVE_FAILED_BATCHES` (default `3`) batches fail completely in succession, the main flow must raise `RuntimeError`. The tracker must already be persisted so that the upload can be resumed. +- **Manual rerun semantics (mandatory):** regardless of the prior error, skip only `COMMITTED` entries. `FAILED`, `UPLOADED`, and missing entries must re-enter the full upload-and-commit path; `begin_attempt` clears the previous `error_type`. A terminal failure must raise `StorageError` and suppress synchronous deletion. +- **Advisory preflight limits:** capacity-related thresholds (total file count, files per directory, individual file size, and total normal-file bytes) must only warn and never block. Structural errors (an empty `repo_id`, empty input, a nonexistent or out-of-bound path, invalid `repo_type`, or a file modified during upload), together with the server's hard limit of `2000`, remain enforced constraints. + +--- + +## 5. Mandatory Simulation Regression for Upload-Path Changes + +> Rule: **Before committing any change within the scope in Section 2, perform a scenario-based simulation review of every case below and include an Upload Simulation Regression Report in the delivery notes.** A simulation is a logical derivation from the code's current strategy and must match the implementation. If a conclusion changes because of the current change, explicitly document the before/after difference and its cause. + +### 5.1 Required Dataset-Scale Scenarios + +| ID | Dataset characteristics | Required simulated conclusions | +|---|---|---| +| D1 | 1 TB / 2 million files, from tens of KiB to tens of MiB | Routing distribution, blob PUT order of magnitude, presign request count, commit-batch order of magnitude, rate-limit risk, memory and hashing overhead | +| D2 | 50 GB / 20 thousand files | Nearly all LFS, presign count, commit-batch count (including tail-batch rebalancing), dominant costs | +| D3 | 5 GB / 2 thousand files | Commit-batch count and tail behavior (including whether rebalancing applies), blob-dominated cost | +| D4 | 200 GB / 10 files | All LFS, **one commit**, blob-transfer dominated, no capacity warning | +| DX | 1,000 files × 1 GB | All LFS, the 8 MiB budget does not apply, commit batches (`256×3+232=4`), approximately 1,000 PUTs | + +For every scenario, derive at least: the LFS/normal routing result, the `_calculate_adaptive_batch_size` target, `_plan_commit_batches` count plus operations and estimated body size per batch, blob PUT count, presign request count, advisory warnings triggered, and resume behavior. + +### 5.2 Required Error-Type Scenarios + +For each error below, describe its classification, whether it is automatically retried in the current run, its handling path, and its final outcome: + +- Transport-layer network failure or timeout (`E1020` / `E1001`, blob stage) +- Commit 5xx (`E1002`) +- 429 rate limiting (`E1021`), including `Retry-After` and the configured limit +- 409 branch changed / 400 repository policy (business code `10030000001`, `/commit/`) +- HTTP 200 + `Success:false` +- Genuine parameter error (invalid-path E3021 without a policy code) +- Authentication, authorization, or nonexistent resource (`E3001` / `E3002` / `E3020`) +- File modified during upload (size mismatch) +- Shared blob owner failure (replicas are deferred) +- Three consecutive batches failing completely (protective abort and resumable rerun) + +### 5.3 Upload Simulation Regression Report Template + +``` +# upload_folder Simulation Regression Report +- Change summary / affected Section 2 symbols: +- Baseline configuration snapshot: LFS threshold / 256 / 8 MiB / 2000 / recovery switches / consecutive-failure threshold +- Dataset scenario derivations: D1 / D2 / D3 / D4 / DX (routing, blob PUTs, presigning, commit batches with operations and body sizes, warnings, resume behavior) +- Error-type behavior: each item (classification / automatic retry / handling path / final outcome) +- Difference from the previous version: behavior changed by this modification and the reason (write "None" if there is no difference) +- Automated validation: pytest / ruff / mypy results +- Conclusion: compliance with the semantics in Sections 3 and 4; risks and tuning recommendations +``` + +### 5.4 Required Automated Validation (Run in Parallel with the Simulation) + +- `PYTHONPATH=src pytest -q tests/test_upload_batching.py tests/test_commit_action_limit.py tests/test_errors_openapi_codes.py tests/test_upload_config.py tests/test_upload_lfs_gate.py tests/test_upload_blob_dedup.py` +- If error mappings changed, additionally run `tests/test_login_failure_paths.py`. +- The full non-remote suite, ruff, and mypy must pass. + +--- + +## 6. Change Checklist + +After changing the upload path, check whether the following must be updated as well: + +- `tests/`: assertions for batching, errors, configuration, deduplication, LFS gating, and related behavior—especially assertions for default values. +- The upload environment-variable table and semantic documentation in `README.md`. +- `__all__` exports and compatibility aliases in `src/modelscope_hub/compat/` and `constants.py`. +- This file, if new constants, error codes, or scenarios were introduced. diff --git a/README.md b/README.md index 18294d0..45b922d 100644 --- a/README.md +++ b/README.md @@ -931,13 +931,21 @@ Token is persisted locally after `ms-hub login` and auto-loaded in subsequent se | Variable | Default | Description | |----------|---------|-------------| | `MODELSCOPE_UPLOAD_MAX_CONCURRENT_WORKERS` | `min(8, cpu+4)` | Default parallel worker threads | -| `MODELSCOPE_UPLOAD_CACHE_ENABLED` | `true` | Enable resumable upload cache | +| `MODELSCOPE_UPLOAD_CACHE_ENABLED` | `true` | Enable resumable upload cache; only committed files are skipped on a later run | | `MODELSCOPE_UPLOAD_IGNORE_FILE_PATTERN` | — | File pattern excluded by legacy `push_to_hub` uploads | -| `MODELSCOPE_UPLOAD_MAX_FILE_SIZE_MB` | `102400` | Max single file size (MB, default 100 GB) | -| `MODELSCOPE_UPLOAD_MAX_FILE_COUNT` | `100000` | Max total files per upload | +| `MODELSCOPE_UPLOAD_MAX_FILE_SIZE_MB` | `102400` | Advisory single-file warning threshold (MB); uploads continue above it | +| `MODELSCOPE_UPLOAD_MAX_FILE_COUNT` | `100000` | Advisory total file-count warning threshold; uploads continue above it | +| `MODELSCOPE_UPLOAD_MAX_FILES_PER_DIRECTORY` | `50000` | Advisory per-directory file-count warning threshold | +| `MODELSCOPE_UPLOAD_NORMAL_FILES_TOTAL_SIZE_MB` | `500` | Advisory total inline-file size warning threshold (MB) | +| `MODELSCOPE_UPLOAD_LFS_FORCE_THRESHOLD` | `1MiB` | Route larger non-metadata files through LFS; accepts byte-unit suffixes | +| `MODELSCOPE_UPLOAD_COMMIT_BATCH_MAX_OPERATIONS` | `256` | Target actions per commit, clamped to the server hard ceiling | +| `MODELSCOPE_UPLOAD_COMMIT_MAX_INLINE_BYTES` | `8MiB` | Estimated commit request-body budget used as the secondary batch constraint | +| `MODELSCOPE_UPLOAD_COMMIT_MAX_ATTEMPTS` | `5` | Maximum attempts for one transient commit failure | | `MODELSCOPE_UPLOAD_BLOB_CONNECT_TIMEOUT_SECONDS` | `30` | Blob upload connect timeout (seconds) | | `MODELSCOPE_UPLOAD_BLOB_READ_TIMEOUT_SECONDS` | `3600` | Blob upload read timeout (seconds) | +Capacity thresholds are advisory and emit warnings without blocking upload. Structural errors (invalid paths, missing inputs, changed files) still fail immediately, while the server's per-commit action ceiling is always enforced by splitting. A manual rerun retries every file not marked committed, including files whose previous run ended with a non-retryable error. + **Logging:** | Variable | Default | Description | diff --git a/src/modelscope_hub/_legacy_api.py b/src/modelscope_hub/_legacy_api.py index aa3d217..211614d 100644 --- a/src/modelscope_hub/_legacy_api.py +++ b/src/modelscope_hub/_legacy_api.py @@ -40,6 +40,7 @@ RepoType, ) from .errors import ( + APIError, InvalidParameter, NetworkError, PermissionDeniedError, @@ -815,7 +816,19 @@ def create_commit( f"repos/{segment}/{repo_id}/commit/{revision}", json_body=payload, ) - return self._json_data(resp) + body = resp.json() + if isinstance(body, dict) and (body.get("Success") is False or body.get("success") is False): + message = body.get("Message") or body.get("message") or "Commit was rejected by the server" + request_id = body.get("RequestId") or body.get("requestId") or body.get("request_id") + raise APIError( + str(message), + status_code=resp.status_code, + request_id=request_id, + response_body=body, + url=resp.url, + method="POST", + ) + return body.get("Data", body) if isinstance(body, dict) else body # ------------------------------------------------------------------ # Blob Upload (LFS) diff --git a/src/modelscope_hub/_upload.py b/src/modelscope_hub/_upload.py index 0b4f0de..a97cced 100644 --- a/src/modelscope_hub/_upload.py +++ b/src/modelscope_hub/_upload.py @@ -178,15 +178,11 @@ def _upload_mode(path: str | Path, size: int, repo_type: str) -> str: def _calculate_adaptive_batch_size(total_files: int, max_operations: int) -> int: - """Commit batch size from the file count alone, capped by *max_operations*. + """Return the operation target, always below the server's hard ceiling. - Fewer, fuller commits are strictly better: the Hub throttles commits per - repository, so commit count -- not commit size -- is the scarce resource. - The only reason to stop growing a batch is the operation cap, or the inlined - content limit that :func:`_plan_commit_batches` applies on top of this. - - The cap is itself clamped to :data:`COMMIT_MAX_ACTIONS_PER_REQUEST`, which - the server rejects outright rather than truncating. + File count is the primary constraint because repository publishing is more + sensitive to action count than to blob volume. Request bytes are enforced + independently by :func:`_plan_commit_batches`. """ if total_files <= 0: return 1 @@ -195,6 +191,21 @@ def _calculate_adaptive_batch_size(total_files: int, max_operations: int) -> int return max(1, min(cap, ceiling, total_files)) +_COMMIT_ENVELOPE_ESTIMATED_BYTES = 512 +_COMMIT_OPERATION_ESTIMATED_BYTES = 192 +_LFS_POINTER_ESTIMATED_BYTES = 128 + + +def _estimate_commit_operation_bytes(path_in_repo: str, size: int, repo_type: str) -> int: + """Estimate the JSON bytes one create operation contributes to a commit.""" + path_bytes = len(path_in_repo.encode("utf-8")) + scalar_bytes = len(str(max(0, size))) + if _is_lfs(path_in_repo, size, repo_type): + return _COMMIT_OPERATION_ESTIMATED_BYTES + _LFS_POINTER_ESTIMATED_BYTES + path_bytes + scalar_bytes + encoded_bytes = (max(0, size) + 2) // 3 * 4 + return _COMMIT_OPERATION_ESTIMATED_BYTES + path_bytes + scalar_bytes + encoded_bytes + + def _plan_commit_batches( files: list[tuple[str, str]], repo_type: str, @@ -203,35 +214,60 @@ def _plan_commit_batches( max_inline_bytes: int, sizes: dict[str, int] | None = None, ) -> list[int]: - """Split *files* into commit batches, returning each batch's file count. - - Two limits close a batch, whichever is reached first: ``max_operations`` - files, or ``max_inline_bytes`` of content that will ride *inside* the commit - request. Only non-LFS files contribute to the byte total -- an LFS file adds - a fixed-size pointer -- and base64 expansion is accounted for, because the - request carries the encoded form. A batch always holds at least one file, so - a single oversized inline file still makes progress instead of deadlocking. + """Split files by action count first and estimated request bytes second. + + Normal content is counted after base64 expansion. LFS blob bytes are not in + the commit request, but their pointer/action JSON still consumes a small + amount of space. A single oversized inline operation is allowed to make + progress and emits a warning. A tiny final count-only batch is rebalanced + with its predecessor when both balanced batches still fit all constraints. """ if not files: return [] cap = max_operations if max_operations > 0 else len(files) cap = max(1, min(cap, max(1, COMMIT_MAX_ACTIONS_PER_REQUEST))) - - batches: list[int] = [] - count = 0 - inline_bytes = 0 + weights: list[int] = [] for path_in_repo, file_path in files: size = sizes.get(file_path, 0) if sizes is not None else _safe_size(file_path) - encoded = 0 if _is_lfs(path_in_repo, size, repo_type) else (size + 2) // 3 * 4 - if count > 0 and (count >= cap or (max_inline_bytes > 0 and inline_bytes + encoded > max_inline_bytes)): - batches.append(count) - count = 0 - inline_bytes = 0 - count += 1 - inline_bytes += encoded - if count: - batches.append(count) - return batches + is_lfs = _is_lfs(path_in_repo, size, repo_type) + weight = _estimate_commit_operation_bytes(path_in_repo, size, repo_type) + weights.append(weight) + if not is_lfs and max_inline_bytes > 0 and _COMMIT_ENVELOPE_ESTIMATED_BYTES + weight > max_inline_bytes: + logger.warning( + "Inline file %s alone exceeds the advisory commit request budget (%d > %d bytes); " + "sending it in a dedicated commit.", + path_in_repo, + _COMMIT_ENVELOPE_ESTIMATED_BYTES + weight, + max_inline_bytes, + ) + + batch_indexes: list[list[int]] = [] + current: list[int] = [] + request_bytes = _COMMIT_ENVELOPE_ESTIMATED_BYTES + for index, weight in enumerate(weights): + exceeds_bytes = max_inline_bytes > 0 and request_bytes + weight > max_inline_bytes + if current and (len(current) >= cap or exceeds_bytes): + batch_indexes.append(current) + current = [] + request_bytes = _COMMIT_ENVELOPE_ESTIMATED_BYTES + current.append(index) + request_bytes += weight + if current: + batch_indexes.append(current) + + if len(batch_indexes) >= 2 and len(batch_indexes[-1]) * 4 < len(batch_indexes[-2]): + combined = batch_indexes[-2] + batch_indexes[-1] + midpoint = (len(combined) + 1) // 2 + candidates = (combined[:midpoint], combined[midpoint:]) + + def _fits(candidate: list[int]) -> bool: + estimated = _COMMIT_ENVELOPE_ESTIMATED_BYTES + sum(weights[i] for i in candidate) + return len(candidate) <= cap and (max_inline_bytes <= 0 or estimated <= max_inline_bytes) + + if all(_fits(candidate) for candidate in candidates): + batch_indexes[-2:] = [list(candidates[0]), list(candidates[1])] + + return [len(batch) for batch in batch_indexes] def _safe_size(file_path: str) -> int: @@ -241,6 +277,61 @@ def _safe_size(file_path: str) -> int: return 0 +def _warn_advisory_upload_limits( + files: list[tuple[str, str]], + sizes: dict[str, int], + repo_type: str, +) -> None: + """Warn about advisory scale limits without refusing a valid upload.""" + if len(files) > UPLOAD_MAX_FILE_COUNT: + logger.warning( + "Upload contains %d files, above the advisory limit of %d; continuing.", + len(files), + UPLOAD_MAX_FILE_COUNT, + ) + + directory_counts: dict[str, int] = {} + oversized: list[tuple[str, int]] = [] + normal_size = 0 + for path_in_repo, file_path in files: + directory = posixpath.dirname(path_in_repo) + directory_counts[directory] = directory_counts.get(directory, 0) + 1 + size = sizes.get(file_path, 0) + if size > UPLOAD_MAX_FILE_SIZE_BYTES: + oversized.append((path_in_repo, size)) + if not _is_lfs(path_in_repo, size, repo_type): + normal_size += size + + crowded = [ + (path or "/", count) for path, count in directory_counts.items() if count > UPLOAD_MAX_FILES_PER_DIRECTORY + ] + if crowded: + preview = ", ".join(f"{path} ({count})" for path, count in crowded[:3]) + suffix = f" and {len(crowded) - 3} more" if len(crowded) > 3 else "" + logger.warning( + "Upload has directories above the advisory %d-file limit: %s%s; continuing.", + UPLOAD_MAX_FILES_PER_DIRECTORY, + preview, + suffix, + ) + if oversized: + preview = ", ".join(f"{path} ({size} bytes)" for path, size in oversized[:3]) + suffix = f" and {len(oversized) - 3} more" if len(oversized) > 3 else "" + logger.warning( + "Upload has files above the advisory %d-byte single-file limit: %s%s; continuing.", + UPLOAD_MAX_FILE_SIZE_BYTES, + preview, + suffix, + ) + if normal_size > UPLOAD_NORMAL_FILES_TOTAL_SIZE_BYTES: + logger.warning( + "Total normal (non-LFS) content is %d bytes, above the advisory limit of %d; continuing. " + "Consider lowering MODELSCOPE_UPLOAD_LFS_FORCE_THRESHOLD.", + normal_size, + UPLOAD_NORMAL_FILES_TOTAL_SIZE_BYTES, + ) + + def _normalize_path_in_repo(path_in_repo: str | None) -> str: """Collapse a repo destination prefix to a clean, root-relative form. @@ -349,13 +440,38 @@ def is_retryable(cls, category: str) -> bool: } -def classify_error(error: Exception) -> str: - """Classify an exception for retry strategy using the SDK error hierarchy. +_RETRYABLE_COMMIT_BUSINESS_CODES = frozenset({10030000001}) +_RETRYABLE_COMMIT_MESSAGE_MARKERS = ( + "branch changed", + "could not update refs", + "please retry", + "try again", + "commit rejected by repository policy", +) + - For :class:`HubError` instances, classification is driven by - :attr:`~HubError.error_code` via a lookup table — no ``isinstance`` - chains or string-matching heuristics needed for known error types. - """ +def _is_retryable_commit_error(error: Exception) -> bool: + """Return whether a commit failure is transient in commit context.""" + if isinstance(error, HubError) and error.retryable: + return True + body = getattr(error, "response_body", None) + if isinstance(body, dict): + raw_code = body.get("Code") if body.get("Code") is not None else body.get("code") + try: + if raw_code is not None and int(raw_code) in _RETRYABLE_COMMIT_BUSINESS_CODES: + return True + except (TypeError, ValueError): + pass + message = getattr(error, "message", None) or str(error) + lowered = str(message).lower() + status_code = getattr(error, "status_code", None) + return status_code in (400, 409) and any(marker in lowered for marker in _RETRYABLE_COMMIT_MESSAGE_MARKERS) + + +def classify_error(error: Exception, *, commit_context: bool = False) -> str: + """Classify an exception for retry strategy using the SDK error hierarchy.""" + if commit_context and _is_retryable_commit_error(error): + return _ErrorCategory.TRANSIENT_SERVER if isinstance(error, HubError): code = getattr(error, "error_code", None) if code and code in _CATEGORY_BY_ERROR_CODE: @@ -438,11 +554,24 @@ def get_status(self, rel_path: str, mtime: float, size: int) -> str | None: entry = self._files.get(key) return entry.get("status") if entry else None + def begin_attempt(self, rel_path: str, mtime: float, size: int) -> None: + """Clear stale failure metadata when a later run retries a file.""" + key = self._make_key(rel_path, mtime, size) + with self._lock: + entry = self._files.get(key) + if entry is None or entry.get("status") == FileStatus.COMMITTED: + return + if entry.get("status") == FileStatus.FAILED: + entry.pop("status", None) + entry.pop("error_type", None) + self._dirty = True + def mark_uploaded(self, rel_path: str, mtime: float, size: int) -> None: key = self._make_key(rel_path, mtime, size) with self._lock: if key in self._files: self._files[key]["status"] = FileStatus.UPLOADED + self._files[key].pop("error_type", None) self._dirty = True def mark_committed_batch(self, file_keys: list[tuple[str, float, int]]) -> None: @@ -451,6 +580,7 @@ def mark_committed_batch(self, file_keys: list[tuple[str, float, int]]) -> None: key = self._make_key(rel_path, mtime, size) if key in self._files: self._files[key]["status"] = FileStatus.COMMITTED + self._files[key].pop("error_type", None) self._dirty = True def mark_failed(self, rel_path: str, mtime: float, size: int, error_type: str = "") -> None: @@ -574,6 +704,9 @@ def is_committed(self, rel_path: str, mtime: float, size: int) -> bool: def get_status(self, rel_path: str, mtime: float, size: int) -> None: return None + def begin_attempt(self, rel_path: str, mtime: float, size: int) -> None: + pass + def mark_uploaded(self, rel_path: str, mtime: float, size: int) -> None: pass @@ -779,6 +912,12 @@ def upload_file( hash_info = _compute_file_hash(path_or_fileobj, buffer_size_mb) file_hash = hash_info["file_hash"] file_size = hash_info["file_size"] + warning_source = str(path_or_fileobj) if isinstance(path_or_fileobj, (str, Path)) else path_in_repo + _warn_advisory_upload_limits( + [(path_in_repo, warning_source)], + {warning_source: file_size}, + repo_type, + ) # If BinaryIO was consumed, _compute_file_hash returns the bytes if not isinstance(path_or_fileobj, (str, Path, bytes)): path_or_fileobj = hash_info["file_path_or_obj"] @@ -971,6 +1110,7 @@ def upload_folder( repo_type=repo_type, allow_patterns=None, ignore_patterns=None, + warn_limits=False, ) all_local_paths_in_repo = {p for p, _ in all_local_files_in_repo} else: @@ -984,61 +1124,73 @@ def upload_folder( if self._create_repo_fn is not None: self._create_repo_fn(repo_id, repo_type) - # Sort for deterministic batch assignment + # Sort for deterministic assignment, then remove committed files before + # planning. Otherwise a resume with four pending files in four old batch + # ranges needlessly creates four one-file commits. sorted_files = sorted(sorted_files, key=lambda x: x[0]) - # Plan commit batches. The operation cap bounds the file count; the - # inline-content cap bounds how many bytes a commit body carries, which - # only non-LFS files add to. - max_operations = ( - _calculate_adaptive_batch_size(len(sorted_files), UPLOAD_COMMIT_BATCH_MAX_OPERATIONS) - if UPLOAD_ADAPTIVE_BATCHING_ENABLED - else (UPLOAD_COMMIT_BATCH_MAX_OPERATIONS if UPLOAD_COMMIT_BATCH_MAX_OPERATIONS > 0 else len(sorted_files)) - ) - batch_plan = _plan_commit_batches( - sorted_files, - repo_type, - max_operations=max_operations, - max_inline_bytes=UPLOAD_COMMIT_MAX_INLINE_BYTES, - sizes=file_sizes, - ) - logger.info( - "Commit plan: %d batch(es) for %d file(s) (max %d ops, max %d inline bytes per commit).", - len(batch_plan), - len(sorted_files), - max_operations, - UPLOAD_COMMIT_MAX_INLINE_BYTES, - ) - - # Initialize tracker. The cache normally lives in the uploaded folder, - # but a caller that stages files into a throwaway tree (a link tree, a - # per-chunk directory) must be able to keep it outside, or every run - # rediscovers hashes and re-commits what was already committed. + # The cache normally lives in the uploaded folder, but a caller that + # stages files into a throwaway tree can keep it outside. folder_path_resolved = Path(folder_path).resolve() if use_cache: cache_path = Path(tracker_path).expanduser() if tracker_path else folder_path_resolved / UPLOAD_CACHE_FILE tracker: UploadTracker | NullTracker = UploadTracker(cache_path, repo_id=repo_id) else: tracker = NullTracker() - batch_tracker = BatchTracker(len(sorted_files), batch_plan) - # Skip individually committed files - files_to_upload: list[tuple[int, tuple[str, str]]] = [] - skipped_indices: set[int] = set() - for file_idx, (file_path_in_repo, file_path) in enumerate(sorted_files): + pending_files: list[tuple[str, str]] = [] + skipped_count = 0 + for file_path_in_repo, file_path in sorted_files: try: st = os.stat(file_path) if tracker.is_committed(file_path_in_repo, st.st_mtime, st.st_size): - skipped_indices.add(file_idx) - batch_tracker.mark_file_skipped(file_idx) + skipped_count += 1 continue + tracker.begin_attempt(file_path_in_repo, st.st_mtime, st.st_size) except OSError as e: logger.warning( "Cannot stat file %s, will re-upload: %s", file_path_in_repo, e, ) - files_to_upload.append((file_idx, (file_path_in_repo, file_path))) + pending_files.append((file_path_in_repo, file_path)) + tracker.save() + + max_operations = ( + _calculate_adaptive_batch_size(len(pending_files), UPLOAD_COMMIT_BATCH_MAX_OPERATIONS) + if UPLOAD_ADAPTIVE_BATCHING_ENABLED + else (UPLOAD_COMMIT_BATCH_MAX_OPERATIONS if UPLOAD_COMMIT_BATCH_MAX_OPERATIONS > 0 else len(pending_files)) + ) + batch_plan = _plan_commit_batches( + pending_files, + repo_type, + max_operations=max_operations, + max_inline_bytes=UPLOAD_COMMIT_MAX_INLINE_BYTES, + sizes=file_sizes, + ) + batch_tracker = BatchTracker(len(pending_files), batch_plan) + files_to_upload = list(enumerate(pending_files)) + + plan_details: list[str] = [] + offset = 0 + for count in batch_plan: + batch_files = pending_files[offset : offset + count] + estimated = _COMMIT_ENVELOPE_ESTIMATED_BYTES + sum( + _estimate_commit_operation_bytes(path, file_sizes.get(local, 0), repo_type) + for path, local in batch_files + ) + lfs_count = sum(1 for path, local in batch_files if _is_lfs(path, file_sizes.get(local, 0), repo_type)) + plan_details.append(f"{count} ops/{estimated} bytes/{lfs_count} LFS") + offset += count + logger.info( + "Commit plan: %d batch(es), %d pending, %d skipped (target %d ops, request budget %d bytes): %s", + len(batch_plan), + len(pending_files), + skipped_count, + max_operations, + UPLOAD_COMMIT_MAX_INLINE_BYTES, + "; ".join(plan_details[:8]) + (f"; ... {len(plan_details) - 8} more" if len(plan_details) > 8 else ""), + ) # Batch pre-validation for every LFS candidate. # @@ -1135,7 +1287,6 @@ def upload_folder( deduped_bytes, ) - skipped_count = len(skipped_indices) if skipped_count > 0: logger.info("%d file(s) already committed, skipping.", skipped_count) @@ -1189,7 +1340,9 @@ def _upload_worker(file_idx: int, file_info: tuple, pre_validated: Any = None) - # Pipeline: consume batches in order commit_infos: list[dict] = [] all_results: list[dict] = [] - total_failed_files: list[tuple] = [] + retry_failed_files: list[tuple] = [] + terminal_failures: list[tuple] = [] + retry_commit_batches: list[tuple[list[dict], Exception]] = [] num_batches = batch_tracker.num_batches committed_files = 0 committed_bytes = 0 @@ -1345,15 +1498,6 @@ def _flush_wire() -> None: total=num_batches, disable=disable_tqdm, ): - batch_start, batch_end = batch_tracker.batch_range(batch_idx) - if all(i in skipped_indices for i in range(batch_start, batch_end)): - logger.info( - "Batch %d/%d fully committed, skipping.", - batch_idx + 1, - num_batches, - ) - continue - results, failures = batch_tracker.wait_for_batch(batch_idx) # Every file of this batch has finished its transfer by now, # so publish the bytes the throttle may still be holding @@ -1361,7 +1505,7 @@ def _flush_wire() -> None: _flush_wire() if failures: - total_failed_files.extend(failures) + retry_failed_files.extend(failures) for item, err in failures: logger.error(" Failed: %s - %s", item[0], err) @@ -1393,7 +1537,7 @@ def _flush_wire() -> None: num_batches, len(orphaned), ) - total_failed_files.extend( + retry_failed_files.extend( ( (item_r["file_path_in_repo"], item_r["file_path"]), StorageError("shared blob upload failed"), @@ -1454,7 +1598,7 @@ def _flush_wire() -> None: sum(r["file_size_on_disk"] for r in results), error=str(e), ) - category = classify_error(e) + category = classify_error(e, commit_context=True) if not _ErrorCategory.is_retryable(category): for r in results: tracker.mark_failed( @@ -1463,30 +1607,26 @@ def _flush_wire() -> None: r["file_size_on_disk"], error_type="commit_" + category, ) + terminal_failures.append(((r["file_path_in_repo"], r["file_path"]), e)) logger.error( - "Batch %d/%d: permanent failure (%s), %d file(s) will not be retried.", + "Batch %d/%d: terminal failure for this run (%s); %d file(s) will not be retried " + "automatically. Rerun upload to retry them.", batch_idx + 1, num_batches, category, len(results), ) - consecutive_failures += 1 else: - for r in results: - total_failed_files.append( - ( - (r["file_path_in_repo"], r["file_path"]), - e, - ) - ) + retry_commit_batches.append((list(results), e)) logger.warning( - "Batch %d/%d: %d file(s) recovered to retry queue (error_category=%s).", + "Batch %d/%d: retained %d uploaded file(s) for commit-only recovery " + "(error_category=%s).", batch_idx + 1, num_batches, len(results), category, ) - consecutive_failures += 1 + consecutive_failures += 1 if consecutive_failures >= UPLOAD_COMMIT_MAX_CONSECUTIVE_FAILED_BATCHES: raise RuntimeError( @@ -1530,9 +1670,23 @@ def _report_recovery(results: list[dict], label: str) -> None: } ) - if total_failed_files and UPLOAD_RECOVERY_ENABLED: - total_failed_files, react_commits, react_results = self._retry_failed_files_react( - failed_files=total_failed_files, + if retry_commit_batches: + commit_failures, recovered_commits, recovered_results = self._retry_failed_commits( + failed_batches=retry_commit_batches, + tracker=tracker, + repo_id=repo_id, + repo_type=repo_type, + commit_message=commit_message, + revision=revision, + on_committed=_report_recovery, + ) + terminal_failures.extend(commit_failures) + commit_infos.extend(recovered_commits) + all_results.extend(recovered_results) + + if retry_failed_files and UPLOAD_RECOVERY_ENABLED: + retry_failed_files, react_commits, react_results = self._retry_failed_files_react( + failed_files=retry_failed_files, tracker=tracker, repo_id=repo_id, repo_type=repo_type, @@ -1545,9 +1699,9 @@ def _report_recovery(results: list[dict], label: str) -> None: ) commit_infos.extend(react_commits) all_results.extend(react_results) - elif total_failed_files: - total_failed_files = self._retry_failed_simple( - failed_files=total_failed_files, + elif retry_failed_files: + retry_failed_files = self._retry_failed_simple( + failed_files=retry_failed_files, tracker=tracker, repo_id=repo_id, repo_type=repo_type, @@ -1562,10 +1716,11 @@ def _report_recovery(results: list[dict], label: str) -> None: ) tracker.save() + all_failures = terminal_failures + retry_failed_files - # Sync: delete remote orphan files + # Sync: delete remote orphan files only after every upload commit landed. deleted_count = 0 - if sync_remote_repo and not total_failed_files: + if sync_remote_repo and not all_failures: prefix = path_in_repo.strip("/") if path_in_repo else "" orphans = self._compute_remote_orphans( repo_id=repo_id, @@ -1597,7 +1752,7 @@ def _report_recovery(results: list[dict], label: str) -> None: # Upload report elapsed = time.time() - start_time total_files = len(sorted_files) - failed_count = len(total_failed_files) + failed_count = len(all_failures) lfs_reused_count = sum(1 for r in all_results if r.get("upload_mode") == "lfs" and r.get("is_reused")) lfs_uploaded_count = sum(1 for r in all_results if r.get("is_blob_uploaded")) normal_count = sum(1 for r in all_results if r.get("upload_mode") == "normal") @@ -1617,8 +1772,8 @@ def _report_recovery(results: list[dict], label: str) -> None: print(f" Elapsed : {elapsed:.1f}s") print("=" * 60) - if total_failed_files: - for (path_in_repo_f, _), err in total_failed_files: + if all_failures: + for (path_in_repo_f, _), err in all_failures: logger.error(" - %s: %s: %s", path_in_repo_f, type(err).__name__, err) succeeded = total_files - failed_count raise StorageError( @@ -2043,11 +2198,8 @@ def _commit_with_retry( revision=revision, ) except HubError as e: - if not e.retryable: - error_str = str(e) - retryable_patterns = ["Could not update refs", "try again"] - if not any(p in error_str for p in retryable_patterns): - raise + if not _is_retryable_commit_error(e): + raise last_error = e except (ConnectionError, TimeoutError) as e: last_error = e @@ -2182,6 +2334,118 @@ def _track_committed_batch( ) tracker.save() + def _plan_result_batches(self, results: list[dict], repo_type: str) -> list[list[dict]]: + """Apply the normal commit planner to already-uploaded results.""" + if not results: + return [] + files = [(r["file_path_in_repo"], r["file_path"]) for r in results] + sizes = {r["file_path"]: r["file_size_on_disk"] for r in results} + max_operations = _calculate_adaptive_batch_size(len(files), UPLOAD_COMMIT_BATCH_MAX_OPERATIONS) + counts = _plan_commit_batches( + files, + repo_type, + max_operations=max_operations, + max_inline_bytes=UPLOAD_COMMIT_MAX_INLINE_BYTES, + sizes=sizes, + ) + batches: list[list[dict]] = [] + offset = 0 + for count in counts: + batches.append(results[offset : offset + count]) + offset += count + return batches + + @staticmethod + def _should_split_commit_failure(error: Exception) -> bool: + body = getattr(error, "response_body", None) + raw_code = body.get("Code") if isinstance(body, dict) else None + if raw_code is None and isinstance(body, dict): + raw_code = body.get("code") + try: + policy_code = raw_code is not None and int(raw_code) in _RETRYABLE_COMMIT_BUSINESS_CODES + except (TypeError, ValueError): + policy_code = False + message = str(getattr(error, "message", None) or error).lower() + return "repository policy" in message or (policy_code and getattr(error, "status_code", None) == 400) + + def _retry_failed_commits( + self, + *, + failed_batches: list[tuple[list[dict], Exception]], + tracker: UploadTracker | NullTracker, + repo_id: str, + repo_type: str, + commit_message: str, + revision: str, + on_committed: Any = None, + ) -> tuple[list[tuple], list[dict], list[dict]]: + """Retry commits without re-uploading blobs, splitting policy rejects.""" + failures: list[tuple] = [] + commit_infos: list[dict] = [] + committed_results: list[dict] = [] + + def commit_batch( + batch: list[dict], + label: str, + prior_error: Exception, + *, + split_before_attempt: bool = False, + ) -> None: + if split_before_attempt and self._should_split_commit_failure(prior_error) and len(batch) > 1: + midpoint = (len(batch) + 1) // 2 + logger.warning( + "Commit-only recovery is splitting a policy-rejected batch of %d into %d and %d files.", + len(batch), + midpoint, + len(batch) - midpoint, + ) + commit_batch(batch[:midpoint], label + "a", prior_error) + commit_batch(batch[midpoint:], label + "b", prior_error) + return + + try: + commit_info = self._commit_with_retry( + repo_id=repo_id, + repo_type=repo_type, + operations=self._build_batch_operations(batch, repo_type), + commit_message=f"{commit_message} (commit recovery {label})", + revision=revision, + ) + except Exception as error: + if self._should_split_commit_failure(error) and len(batch) > 1: + midpoint = (len(batch) + 1) // 2 + commit_batch(batch[:midpoint], label + "a", error) + commit_batch(batch[midpoint:], label + "b", error) + return + category = classify_error(error, commit_context=True) + for result in batch: + tracker.mark_failed( + result["file_path_in_repo"], + result["file_mtime"], + result["file_size_on_disk"], + error_type="commit_" + category, + ) + failures.append(((result["file_path_in_repo"], result["file_path"]), error)) + logger.error("Commit-only recovery %s failed for %d file(s): %s", label, len(batch), error) + return + + commit_infos.append(commit_info) + committed_results.extend(batch) + self._track_committed_batch(tracker, batch) + if on_committed is not None: + on_committed(batch, f"commit recovery {label}") + logger.info("Commit-only recovery %s committed %d file(s).", label, len(batch)) + + for batch_index, (results, error) in enumerate(failed_batches, start=1): + for part_index, batch in enumerate(self._plan_result_batches(results, repo_type), start=1): + commit_batch( + batch, + f"{batch_index}.{part_index}", + error, + split_before_attempt=True, + ) + return failures, commit_infos, committed_results + # ------------------------------------------------------------------ # Internal: file collection # ------------------------------------------------------------------ @@ -2193,56 +2457,14 @@ def _prepare_upload_folder( allow_patterns: list[str] | None = None, ignore_patterns: list[str] | None = None, sizes_out: dict[str, int] | None = None, + warn_limits: bool = True, ) -> list[tuple[str, str]]: folder = Path(folder_path).expanduser().resolve() if not folder.is_dir(): raise InvalidParameter(f"Provided path: '{folder}' is not a directory") all_files = sorted(path for path in folder.glob("**/*") if path.is_file()) - - if len(all_files) > UPLOAD_MAX_FILE_COUNT: - raise InvalidParameter(f"Too many files ({len(all_files)}) in folder, max allowed: {UPLOAD_MAX_FILE_COUNT}") - - # Per-directory file count check - dir_counts: dict[str, int] = {} - for path in all_files: - parent = str(path.parent) - dir_counts[parent] = dir_counts.get(parent, 0) + 1 - for dir_path, count in dir_counts.items(): - if count > UPLOAD_MAX_FILES_PER_DIRECTORY: - raise InvalidParameter( - f"Too many files ({count}) in directory {dir_path}, " - f"max allowed per directory: {UPLOAD_MAX_FILES_PER_DIRECTORY}" - ) - - # File size checks. Sizes are handed back through ``sizes_out`` because - # batch planning needs them next; re-stating a large tree costs one - # syscall per file for no new information. - total_size = 0 - normal_size = 0 - for path in all_files: - fsize = path.stat().st_size - if fsize > UPLOAD_MAX_FILE_SIZE_BYTES: - raise InvalidParameter( - f"File too large: {path} ({fsize / 1024 / 1024:.1f} MB), " - f"max allowed: {UPLOAD_MAX_FILE_SIZE_BYTES / 1024 / 1024:.0f} MB" - ) - total_size += fsize - if not _is_lfs(str(path), fsize, repo_type): - normal_size += fsize - if sizes_out is not None: - sizes_out[str(path)] = fsize - - if normal_size > UPLOAD_NORMAL_FILES_TOTAL_SIZE_BYTES: - logger.warning( - "Total normal (non-LFS) file size %d bytes exceeds soft " - "limit %d bytes. Consider using LFS for large files.", - normal_size, - UPLOAD_NORMAL_FILES_TOTAL_SIZE_BYTES, - ) - relpath_to_abspath = {path.relative_to(folder).as_posix(): str(path) for path in all_files} - filtered_keys = _filter_repo_objects( list(relpath_to_abspath.keys()), allow_patterns=allow_patterns, @@ -2250,14 +2472,19 @@ def _prepare_upload_folder( ) # ``path_in_repo`` is a destination prefix, and "." / "./" / "" / "/" - # all mean the repo root. Collapsing them is not cosmetic: a literal - # value like "." otherwise rides into every commit action's ``path`` as - # a "./" prefix, which the Hub rejects wholesale with E3021 "invalid - # commit action". + # all mean the repository root. norm_prefix = _normalize_path_in_repo(path_in_repo) prefix = f"{norm_prefix}/" if norm_prefix else "" prepared = [(prefix + relpath, relpath_to_abspath[relpath]) for relpath in filtered_keys] + selected_sizes: dict[str, int] = {} + for _path_in_repo, file_path in prepared: + selected_sizes[file_path] = Path(file_path).stat().st_size + if sizes_out is not None: + sizes_out.update(selected_sizes) + if warn_limits: + _warn_advisory_upload_limits(prepared, selected_sizes, repo_type) + logger.info("Prepared %d files for upload.", len(prepared)) return prepared @@ -2314,21 +2541,18 @@ def _retry_failed_files_react( "name": "Round 1 (parallel)", "parallel": True, "workers": max(1, max_workers // 2), - "batch_size": 16, "delay": 0, }, { "name": "Round 2 (serial+backoff)", "parallel": False, "workers": 1, - "batch_size": 8, "delay": UPLOAD_RECOVERY_SERIAL_BACKOFF_BASE_SECONDS, }, { "name": "Round 3 (single-file)", "parallel": False, "workers": 1, - "batch_size": 1, "delay": UPLOAD_RECOVERY_SINGLE_FILE_DELAY_SECONDS, }, ] @@ -2406,13 +2630,8 @@ def _retry_failed_files_react( ) round_failures.append(((path_in_repo_r, file_path_r), e)) - all_successes.extend(round_successes) - - batch_size = min(cfg["batch_size"], max(1, len(round_successes))) - for batch_start in range(0, len(round_successes), batch_size): - batch = round_successes[batch_start : batch_start + batch_size] - self._track_uploaded_batch(tracker, batch) - + self._track_uploaded_batch(tracker, round_successes) + for batch in self._plan_result_batches(round_successes, repo_type): operations = self._build_batch_operations(batch, repo_type) if not operations: continue @@ -2425,6 +2644,7 @@ def _retry_failed_files_react( revision=revision, ) commit_infos.append(commit_info) + all_successes.extend(batch) self._track_committed_batch(tracker, batch) if on_committed is not None: on_committed(batch, round_name) @@ -2435,18 +2655,18 @@ def _retry_failed_files_react( ) except Exception as e: logger.error("[ReAct] %s commit failed: %s", round_name, e) - category = classify_error(e) - if not _ErrorCategory.is_retryable(category): - for r in batch: - tracker.mark_failed( - r["file_path_in_repo"], - r["file_mtime"], - r["file_size_on_disk"], - error_type="commit_" + category, - ) - else: - for r in batch: - round_failures.append(((r["file_path_in_repo"], r["file_path"]), e)) + commit_failures, recovered_commits, recovered_results = self._retry_failed_commits( + failed_batches=[(batch, e)], + tracker=tracker, + repo_id=repo_id, + repo_type=repo_type, + commit_message=commit_message, + revision=revision, + on_committed=on_committed, + ) + permanent_failures.extend(commit_failures) + commit_infos.extend(recovered_commits) + all_successes.extend(recovered_results) new_retryable = [] for item_err in round_failures: @@ -2531,6 +2751,7 @@ def _retry_failed_simple( on_committed: Any = None, ) -> list[tuple]: total_failed_files = list(failed_files) + terminal_failures: list[tuple] = [] for retry_round in range(UPLOAD_FAILED_FILE_MAX_RETRY_ROUNDS): if not total_failed_files: break @@ -2560,8 +2781,10 @@ def _retry_failed_simple( retry_failures.append(((path_in_repo_r, file_path_r), e)) if retry_successes: self._track_uploaded_batch(tracker, retry_successes) - operations = self._build_batch_operations(retry_successes, repo_type) - if operations: + for batch in self._plan_result_batches(retry_successes, repo_type): + operations = self._build_batch_operations(batch, repo_type) + if not operations: + continue try: commit_info = self._commit_with_retry( repo_id=repo_id, @@ -2571,14 +2794,14 @@ def _retry_failed_simple( revision=revision, ) commit_infos.append(commit_info) - all_results.extend(retry_successes) - self._track_committed_batch(tracker, retry_successes) + all_results.extend(batch) + self._track_committed_batch(tracker, batch) if on_committed is not None: - on_committed(retry_successes, f"retry round {retry_round + 1}") + on_committed(batch, f"retry round {retry_round + 1}") logger.info( " Retry round %d: committed %d file(s).", retry_round + 1, - len(retry_successes), + len(batch), ) except Exception as e: logger.error( @@ -2586,25 +2809,17 @@ def _retry_failed_simple( retry_round + 1, e, ) - category = classify_error(e) - if not _ErrorCategory.is_retryable(category): - for result in retry_successes: - tracker.mark_failed( - result["file_path_in_repo"], - result["file_mtime"], - result["file_size_on_disk"], - error_type="commit_" + category, - ) - else: - for result in retry_successes: - retry_failures.append( - ( - ( - result["file_path_in_repo"], - result.get("file_path", ""), - ), - e, - ) - ) + commit_failures, recovered_commits, recovered_results = self._retry_failed_commits( + failed_batches=[(batch, e)], + tracker=tracker, + repo_id=repo_id, + repo_type=repo_type, + commit_message=commit_message, + revision=revision, + on_committed=on_committed, + ) + terminal_failures.extend(commit_failures) + commit_infos.extend(recovered_commits) + all_results.extend(recovered_results) total_failed_files = retry_failures - return total_failed_files + return terminal_failures + total_failed_files diff --git a/src/modelscope_hub/constants.py b/src/modelscope_hub/constants.py index cf1b76b..e317f99 100644 --- a/src/modelscope_hub/constants.py +++ b/src/modelscope_hub/constants.py @@ -839,17 +839,15 @@ def _env_register( UPLOAD_COMMIT_MAX_INLINE_BYTES: int = _env_bytes( "MODELSCOPE_UPLOAD_COMMIT_MAX_INLINE_BYTES", 8 * 1024 * 1024, - "Maximum inlined (non-LFS) content carried by one commit request", + "Advisory estimated request-body budget for one upload commit", "Upload", ) -"""Byte ceiling on the base64 content a single commit may carry. - -Non-LFS files travel *inside* the commit request body, so a batch sized purely -by file count can produce a request tens of megabytes large -- big enough for -the server to time out mid-read, which surfaces on the client as an unrelated -write timeout. Batching therefore closes a batch on whichever limit is reached -first, this one or :data:`UPLOAD_COMMIT_BATCH_MAX_OPERATIONS`. LFS files -contribute only a pointer, so they do not count against it. +"""Advisory byte budget for an estimated commit request body. + +Normal files contribute their base64-expanded content plus operation JSON; LFS +files contribute only pointer/action metadata. Batching closes on this budget or +the operation target, whichever comes first. A single oversized inline file is +still sent alone so this advisory budget never blocks an upload. """ UPLOAD_COMMIT_MAX_PER_HOUR: int = _env_int( "MODELSCOPE_UPLOAD_COMMIT_MAX_PER_HOUR", @@ -1012,7 +1010,7 @@ def get_upload_ignore_file_pattern() -> str | None: # Upload: limits UPLOAD_LFS_FORCE_THRESHOLD_BYTES: int = _env_bytes( "MODELSCOPE_UPLOAD_LFS_FORCE_THRESHOLD", - 1024 * 1024, + 64 * 1024, "File size above which LFS mode is forced (bytes; accepts a unit suffix, 0 forces LFS for every file)", "Upload", deprecated_mb_names=("MODELSCOPE_UPLOAD_LFS_FORCE_THRESHOLD_MB",), @@ -1049,7 +1047,7 @@ def get_upload_ignore_file_pattern() -> str | None: UPLOAD_MAX_FILE_SIZE_BYTES: int = _env_int_mb_with_deprecated_units( "MODELSCOPE_UPLOAD_MAX_FILE_SIZE_MB", 100 * 1024, - "Maximum single upload file size (MB, default 100 GB)", + "Advisory single upload file size warning threshold (MB, default 100 GB)", "Upload", deprecated_mb_names=("UPLOAD_MAX_FILE_SIZE_MB",), deprecated_byte_names=("UPLOAD_MAX_FILE_SIZE",), @@ -1057,21 +1055,21 @@ def get_upload_ignore_file_pattern() -> str | None: UPLOAD_MAX_FILE_COUNT: int = _env_int( "MODELSCOPE_UPLOAD_MAX_FILE_COUNT", 100_000, - "Maximum total files per upload", + "Advisory total file-count warning threshold per upload", "Upload", "UPLOAD_MAX_FILE_COUNT", ) UPLOAD_MAX_FILES_PER_DIRECTORY: int = _env_int( "MODELSCOPE_UPLOAD_MAX_FILES_PER_DIRECTORY", 50_000, - "Maximum files in one uploaded directory", + "Advisory per-directory file-count warning threshold", "Upload", "UPLOAD_MAX_FILE_COUNT_IN_DIR", ) UPLOAD_NORMAL_FILES_TOTAL_SIZE_BYTES: int = _env_int_mb_with_deprecated_units( "MODELSCOPE_UPLOAD_NORMAL_FILES_TOTAL_SIZE_MB", 500, - "Maximum total size of normal (non-LFS) files (MB)", + "Advisory total normal-file size warning threshold (MB)", "Upload", deprecated_byte_names=("UPLOAD_NORMAL_FILE_SIZE_TOTAL_LIMIT",), ) diff --git a/src/modelscope_hub/errors.py b/src/modelscope_hub/errors.py index d3a3799..c2e1623 100644 --- a/src/modelscope_hub/errors.py +++ b/src/modelscope_hub/errors.py @@ -440,6 +440,12 @@ class NotSupportedError(HubError): 10010202004: AlreadyExistsError, # 国际站 - 名称已被使用 } +# The legacy commit endpoint overloads 400/409 for optimistic branch conflicts +# and repository-side commit policy throttling. The numeric code is scoped to +# that endpoint so it must not globally turn unrelated invalid requests into +# retryable server failures. +_COMMIT_RETRYABLE_BUSINESS_CODES: frozenset[int] = frozenset({10030000001}) + # --------------------------------------------------------------------------- # OpenAPI string error-code -> exception mapping @@ -583,7 +589,13 @@ def raise_for_status(response: Response) -> None: business_cls: type[APIError] | None = None if status < 500: code = _business_code(body) - business_cls = _BUSINESS_CODE_MAP.get(code) if code is not None else None + is_commit_request = isinstance(url, str) and "/commit/" in urlparse(url).path + if code in _COMMIT_RETRYABLE_BUSINESS_CODES and is_commit_request: + # A branch race or repository-side policy gate is transient even + # though the endpoint reports it with a client-error HTTP status. + business_cls = ServerError + else: + business_cls = _BUSINESS_CODE_MAP.get(code) if code is not None else None if business_cls is None: # The OpenAPI surface publishes a string code instead of a numeric one. openapi_code = _openapi_code(body) diff --git a/tests/test_errors_openapi_codes.py b/tests/test_errors_openapi_codes.py index fdfd14e..c80ef38 100644 --- a/tests/test_errors_openapi_codes.py +++ b/tests/test_errors_openapi_codes.py @@ -15,6 +15,7 @@ import pytest import requests +from modelscope_hub._legacy_api import LegacyClient from modelscope_hub._openapi import _RETRYABLE_EXC from modelscope_hub.errors import ( AlreadyExistsError, @@ -128,6 +129,57 @@ def test_413_maps_to_invalid_parameter(): raise_for_status(_plain_error(413)) +def test_commit_policy_business_code_is_retryable_only_on_commit_endpoint(): + body = { + "Code": 10030000001, + "Message": "commit rejected by repository policy", + "Success": False, + } + commit_response = requests.Response() + commit_response.status_code = 400 + commit_response._content = json.dumps(body).encode() + commit_response.headers["Content-Type"] = "application/json" + commit_response.url = "https://modelscope.cn/api/v1/repos/datasets/owner/repo/commit/master" + + with pytest.raises(ServerError) as excinfo: + raise_for_status(commit_response) + assert excinfo.value.retryable is True + assert excinfo.value.status_code == 400 + + non_commit_response = requests.Response() + non_commit_response.status_code = 400 + non_commit_response._content = json.dumps(body).encode() + non_commit_response.headers["Content-Type"] = "application/json" + non_commit_response.url = "https://modelscope.cn/api/v1/login" + with pytest.raises(InvalidParameter): + raise_for_status(non_commit_response) + + +def test_legacy_commit_rejects_http_200_success_false(monkeypatch): + response = requests.Response() + response.status_code = 200 + response._content = json.dumps( + { + "Code": 10030000001, + "Message": "commit rejected by repository policy", + "Success": False, + } + ).encode() + response.headers["Content-Type"] = "application/json" + response.url = "https://modelscope.cn/api/v1/repos/datasets/owner/repo/commit/master" + client = LegacyClient(endpoint="https://modelscope.cn", token="test") + monkeypatch.setattr(client, "_request", lambda *args, **kwargs: response) + + with pytest.raises(APIError, match="repository policy") as excinfo: + client.create_commit( + repo_id="owner/repo", + repo_type="dataset", + operations=[{"action": "create", "path": "a.txt"}], + commit_message="test", + ) + assert excinfo.value.response_body["Success"] is False + + # --------------------------------------------------------------------------- # Retry policy # --------------------------------------------------------------------------- diff --git a/tests/test_upload_batching.py b/tests/test_upload_batching.py index 157a458..1c42c81 100644 --- a/tests/test_upload_batching.py +++ b/tests/test_upload_batching.py @@ -14,13 +14,21 @@ from modelscope_hub._upload import ( BatchTracker, UploadManager, + UploadTracker, _calculate_adaptive_batch_size, _is_inline_metadata, _normalize_path_in_repo, _plan_commit_batches, _upload_mode, ) -from modelscope_hub.errors import InvalidParameter, NetworkError, RateLimitError +from modelscope_hub.errors import ( + AlreadyExistsError, + APIError, + InvalidParameter, + NetworkError, + RateLimitError, + StorageError, +) def _make_manager() -> tuple[UploadManager, MagicMock]: @@ -68,9 +76,11 @@ def test_adaptive_batch_size_is_monotonic_and_never_exceeds_cap() -> None: assert max(sizes) == 256 -def test_batch_plan_closes_on_inline_bytes_before_operation_cap(tmp_path: Path) -> None: +def test_batch_plan_closes_on_inline_bytes_before_operation_cap(tmp_path: Path, monkeypatch) -> None: # 8 inline files of 100 KiB: the operation cap would take all 8 in one - # commit, but their base64 form exceeds a 512 KiB inline budget. + # commit, but their base64 form exceeds a 512 KiB inline budget. Pin the LFS + # threshold above their size so they really travel inline under any default. + monkeypatch.setattr(upload_module, "UPLOAD_LFS_FORCE_THRESHOLD_BYTES", 2 * 1024 * 1024) files = [] sizes = {} for index in range(8): @@ -114,9 +124,72 @@ def test_batch_plan_ignores_lfs_bytes_in_the_inline_budget(tmp_path: Path) -> No assert plan == [8] -def test_batch_plan_keeps_one_oversized_inline_file_per_batch(tmp_path: Path) -> None: - # Below the 1 MiB LFS threshold, so these really do travel inline; each one +def test_four_25gb_lfs_files_share_one_commit_plan() -> None: + files = [(f"shard-{index}.bin", f"/virtual/shard-{index}.bin") for index in range(4)] + sizes = {local: 25 * 1024**3 for _, local in files} + + plan = _plan_commit_batches( + files, + "dataset", + max_operations=256, + max_inline_bytes=8 * 1024 * 1024, + sizes=sizes, + ) + + assert plan == [4] + + +def test_four_oversized_inline_metadata_files_require_four_commits() -> None: + files = [(f"part-{index}/README.md", f"/virtual/part-{index}/README.md") for index in range(4)] + sizes = {local: 10 * 1024**2 for _, local in files} + + plan = _plan_commit_batches( + files, + "dataset", + max_operations=256, + max_inline_bytes=8 * 1024 * 1024, + sizes=sizes, + ) + + assert plan == [1, 1, 1, 1] + + +def test_forty_thousand_small_files_use_count_target() -> None: + files = [(f"part-{index}.txt", f"/virtual/part-{index}.txt") for index in range(40_000)] + sizes = dict.fromkeys((local for _, local in files), 1) + + plan = _plan_commit_batches( + files, + "dataset", + max_operations=256, + max_inline_bytes=8 * 1024 * 1024, + sizes=sizes, + ) + + assert len(plan) == 157 + assert sum(plan) == 40_000 + assert max(plan) == 256 + + +def test_tiny_tail_batch_is_rebalanced_without_adding_commits() -> None: + files = [(f"part-{index}.txt", f"/virtual/part-{index}.txt") for index in range(257)] + sizes = dict.fromkeys((local for _, local in files), 1) + + plan = _plan_commit_batches( + files, + "dataset", + max_operations=256, + max_inline_bytes=8 * 1024 * 1024, + sizes=sizes, + ) + + assert plan == [129, 128] + + +def test_batch_plan_keeps_one_oversized_inline_file_per_batch(tmp_path: Path, monkeypatch) -> None: + # Pinned above the LFS threshold, so these really do travel inline; each one # alone blows the inline budget, which must not stall the plan. + monkeypatch.setattr(upload_module, "UPLOAD_LFS_FORCE_THRESHOLD_BYTES", 2 * 1024 * 1024) files = [] sizes = {} for index in range(3): @@ -481,6 +554,77 @@ def test_external_tracker_lets_a_second_run_skip_committed_files(tmp_path: Path) assert client.create_commit.call_count == 1 +def test_resume_replans_only_pending_files_into_compact_batches(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setattr(upload_module, "UPLOAD_COMMIT_BATCH_MAX_OPERATIONS", 4) + monkeypatch.setattr(upload_module, "UPLOAD_ADAPTIVE_BATCHING_ENABLED", False) + manager, client = _make_manager() + cache_path = tmp_path / ".ms_upload_cache" + tracker = UploadTracker(cache_path, repo_id="owner/repo") + + pending = [] + for index in range(8): + path = tmp_path / f"file-{index}.txt" + path.write_bytes(bytes([index])) + stat = path.stat() + tracker.put_hash( + path.name, + stat.st_mtime, + stat.st_size, + {"file_hash": f"{index:064x}", "file_size": stat.st_size}, + ) + if index % 2 == 0: + tracker.mark_committed_batch([(path.name, stat.st_mtime, stat.st_size)]) + else: + pending.append(path.name) + tracker.mark_failed(path.name, stat.st_mtime, stat.st_size, error_type="commit_permanent") + tracker.save() + + manager.upload_folder( + repo_id="owner/repo", + repo_type="dataset", + folder_path=tmp_path, + max_workers=1, + use_cache=True, + disable_tqdm=True, + ) + + client.create_commit.assert_called_once() + assert [op["path"] for op in client.create_commit.call_args.kwargs["operations"]] == pending + saved = json.loads(cache_path.read_text(encoding="utf-8")) + assert all(entry.get("status") == "c" for entry in saved["files"].values()) + assert all("error_type" not in entry for entry in saved["files"].values()) + + +def test_terminal_commit_failure_is_reported_and_manual_rerun_retries(tmp_path: Path) -> None: + manager, client = _make_manager() + (tmp_path / "README.md").write_bytes(b"hello") + client.create_commit.side_effect = [InvalidParameter("path is not allowed"), {"ok": True}] + + with pytest.raises(StorageError, match="1 file"): + manager.upload_folder( + repo_id="owner/repo", + repo_type="model", + folder_path=tmp_path, + max_workers=1, + use_cache=True, + disable_tqdm=True, + sync_remote_repo=True, + ) + + client.list_repo_files.assert_not_called() + result = manager.upload_folder( + repo_id="owner/repo", + repo_type="model", + folder_path=tmp_path, + max_workers=1, + use_cache=True, + disable_tqdm=True, + ) + + assert result == {"ok": True} + assert client.create_commit.call_count == 2 + + # ------------------------------------------------------------ progress events @@ -848,6 +992,84 @@ def test_upload_file_honors_retry_after(monkeypatch) -> None: assert slept == [7.0] +def test_upload_file_manual_rerun_retries_a_permanent_failure() -> None: + manager, client = _make_manager() + client.create_commit.side_effect = [InvalidParameter("path is not allowed"), {"ok": True}] + + with pytest.raises(InvalidParameter): + manager.upload_file( + repo_id="owner/repo", + repo_type="model", + path_or_fileobj=b"hello", + path_in_repo="README.md", + disable_tqdm=True, + ) + + assert manager.upload_file( + repo_id="owner/repo", + repo_type="model", + path_or_fileobj=b"hello", + path_in_repo="README.md", + disable_tqdm=True, + ) == {"ok": True} + assert client.create_commit.call_count == 2 + + +def test_branch_changed_409_is_retried_in_commit_context(monkeypatch) -> None: + manager, client = _make_manager() + monkeypatch.setattr(upload_module.time, "sleep", lambda _seconds: None) + client.create_commit.side_effect = [ + AlreadyExistsError( + "branch changed, please retry", + status_code=409, + response_body={"Code": 10030000001, "Message": "branch changed, please retry"}, + ), + {"ok": True}, + ] + + assert manager._commit_with_retry( + repo_id="owner/repo", + repo_type="dataset", + operations=[{"action": "create", "path": "a.txt"}], + commit_message="retry branch race", + ) == {"ok": True} + assert client.create_commit.call_count == 2 + + +def test_policy_rejected_commit_splits_without_reuploading_blobs(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setattr(upload_module, "UPLOAD_LFS_FORCE_THRESHOLD_BYTES", 0) + monkeypatch.setattr(upload_module, "UPLOAD_COMMIT_MAX_ATTEMPTS", 2) + monkeypatch.setattr(upload_module.time, "sleep", lambda _seconds: None) + manager, client = _make_manager() + for index in range(4): + (tmp_path / f"blob-{index}.dat").write_bytes(bytes([index + 1]) * 32) + + def create_commit(**kwargs): + operations = kwargs["operations"] + if len(operations) == 4: + raise APIError( + "commit rejected by repository policy", + status_code=400, + response_body={"Code": 10030000001}, + ) + return {"ok": True} + + client.create_commit.side_effect = create_commit + manager.upload_folder( + repo_id="owner/repo", + repo_type="dataset", + folder_path=tmp_path, + max_workers=1, + use_cache=False, + disable_tqdm=True, + ) + + assert client.upload_blob.call_count == 4 + sent_sizes = [len(call.kwargs["operations"]) for call in client.create_commit.call_args_list] + assert sent_sizes[-2:] == [2, 2] + assert all(size == 4 for size in sent_sizes[:-2]) + + def test_upload_file_still_fails_fast_on_a_permanent_error() -> None: manager, client = _make_manager() client.create_commit.side_effect = InvalidParameter("path is not allowed") @@ -864,6 +1086,53 @@ def test_upload_file_still_fails_fast_on_a_permanent_error() -> None: assert client.create_commit.call_count == 1 +def test_folder_capacity_limits_warn_but_do_not_block_upload(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setattr(upload_module, "UPLOAD_MAX_FILE_COUNT", 1) + monkeypatch.setattr(upload_module, "UPLOAD_MAX_FILES_PER_DIRECTORY", 1) + monkeypatch.setattr(upload_module, "UPLOAD_MAX_FILE_SIZE_BYTES", 1) + monkeypatch.setattr(upload_module, "UPLOAD_NORMAL_FILES_TOTAL_SIZE_BYTES", 1) + manager, client = _make_manager() + warnings = MagicMock() + monkeypatch.setattr(upload_module.logger, "warning", warnings) + (tmp_path / "a.txt").write_bytes(b"aa") + (tmp_path / "b.txt").write_bytes(b"bb") + + manager.upload_folder( + repo_id="owner/repo", + repo_type="dataset", + folder_path=tmp_path, + max_workers=1, + use_cache=False, + disable_tqdm=True, + ) + + client.create_commit.assert_called_once() + messages = " ".join(call.args[0] for call in warnings.call_args_list) + assert "advisory limit" in messages + assert "continuing" in messages + + +def test_upload_file_capacity_limit_warns_but_does_not_block(monkeypatch) -> None: + monkeypatch.setattr(upload_module, "UPLOAD_MAX_FILE_SIZE_BYTES", 1) + monkeypatch.setattr(upload_module, "UPLOAD_NORMAL_FILES_TOTAL_SIZE_BYTES", 1) + manager, client = _make_manager() + warnings = MagicMock() + monkeypatch.setattr(upload_module.logger, "warning", warnings) + + manager.upload_file( + repo_id="owner/repo", + repo_type="model", + path_or_fileobj=b"hello", + path_in_repo="notes.txt", + disable_tqdm=True, + ) + + client.create_commit.assert_called_once() + messages = " ".join(call.args[0] for call in warnings.call_args_list) + assert "single-file limit" in messages + assert "normal (non-LFS)" in messages + + def test_prepare_upload_folder_reports_sizes_for_reuse(tmp_path: Path) -> None: # Planning needs these sizes immediately afterwards; re-stating the tree # costs one syscall per file and yields nothing new. diff --git a/tests/test_upload_config.py b/tests/test_upload_config.py index acaf9ab..05df365 100644 --- a/tests/test_upload_config.py +++ b/tests/test_upload_config.py @@ -331,13 +331,13 @@ def test_canonical_lfs_threshold_wins_over_deprecated_aliases() -> None: ( {"MODELSCOPE_UPLOAD_LFS_FORCE_THRESHOLD": "32ZB"}, "UPLOAD_LFS_FORCE_THRESHOLD_BYTES", - 1024 * 1024, + 64 * 1024, "unknown size unit", ), ( {"MODELSCOPE_UPLOAD_LFS_FORCE_THRESHOLD": "-1"}, "UPLOAD_LFS_FORCE_THRESHOLD_BYTES", - 1024 * 1024, + 64 * 1024, "must not be negative", ), ( @@ -394,7 +394,7 @@ def test_new_upload_knobs_are_registered_under_upload() -> None: "MODELSCOPE_UPLOAD_INLINE_METADATA_PATHS", ): assert registry[name]["category"] == "Upload" - assert registry["MODELSCOPE_UPLOAD_LFS_FORCE_THRESHOLD"]["default"] == "1MiB" + assert registry["MODELSCOPE_UPLOAD_LFS_FORCE_THRESHOLD"]["default"] == "64KiB" assert "MODELSCOPE_UPLOAD_LFS_FORCE_THRESHOLD_MB" not in registry