From 2c6d405e157dcbd906f0868261615c7ae54ca075 Mon Sep 17 00:00:00 2001 From: Jacob Nesbitt Date: Mon, 31 Aug 2026 13:40:10 -0400 Subject: [PATCH 01/21] Extract blob multipart upload into a reusable helper Move the S3 multipart upload logic out of `LocalFileAsset.iter_upload()` and into a `multipart_upload()` generator, which yields the same progress `dict`\s and returns the deserialized `validate` response. The caller selects the endpoint family via `upload_root` and supplies the matching `init_fields` identifying what is being uploaded, so that Zarr chunks can reuse the same machinery as asset blobs. No behavior change for asset blob uploads. --- dandi/files/bases.py | 236 ++++++++++++++++++++++++------------------- 1 file changed, 134 insertions(+), 102 deletions(-) diff --git a/dandi/files/bases.py b/dandi/files/bases.py index db502dcc3..b983a29db 100644 --- a/dandi/files/bases.py +++ b/dandi/files/bases.py @@ -2,7 +2,7 @@ from abc import ABC, abstractmethod from collections import deque -from collections.abc import Iterator +from collections.abc import Generator, Iterator from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass from datetime import datetime @@ -364,37 +364,17 @@ def iter_upload( ``"done"`` and an ``"asset"`` key containing the resulting `RemoteAsset`. """ - # Avoid heavy import by importing within function: - from dandi.support.digests import get_dandietag - asset_path = metadata.setdefault("path", self.path) set_asset_schema_key(metadata) client = dandiset.client - yield {"status": "calculating etag"} - etagger = get_dandietag(self.filepath) - filetag = etagger.as_str() - lgr.debug("Calculated dandi-etag of %s for %s", filetag, self.filepath) - digest = metadata.get("digest", {}) - if "dandi:dandi-etag" in digest: - if digest["dandi:dandi-etag"] != filetag: - raise RuntimeError( - f"{self.filepath}: File etag changed; was originally" - f" {digest['dandi:dandi-etag']} but is now {filetag}" - ) - yield {"status": "initiating upload"} - lgr.debug("%s: Beginning upload", asset_path) - total_size = pre_upload_size_check(self.filepath) try: - resp = client.post( - "/uploads/initialize/", - json={ - "contentSize": total_size, - "digest": { - "algorithm": "dandi:dandi-etag", - "value": filetag, - }, - "dandiset": dandiset.identifier, - }, + resp = yield from multipart_upload( + client=client, + filepath=self.filepath, + asset_path=asset_path, + init_fields={"dandiset": dandiset.identifier}, + expected_etag=metadata.get("digest", {}).get("dandi:dandi-etag"), + jobs=jobs, ) except requests.HTTPError as e: if e.response is not None and e.response.status_code == 409: @@ -403,80 +383,7 @@ def iter_upload( else: raise else: - try: - upload_id = resp["upload_id"] - parts = resp["parts"] - if len(parts) != etagger.part_qty: - raise RuntimeError( - f"Server and client disagree on number of parts for upload;" - f" server says {len(parts)}, client says {etagger.part_qty}" - ) - parts_out = [] - bytes_uploaded = 0 - lgr.debug("Uploading %s in %d parts", self.filepath, len(parts)) - with RESTFullAPIClient("http://nil.nil") as storage: - with self.filepath.open("rb") as fp: - with ThreadPoolExecutor(max_workers=jobs or 5) as executor: - lock = Lock() - futures = [ - executor.submit( - _upload_blob_part, - storage_session=storage, - fp=fp, - lock=lock, - etagger=etagger, - asset_path=asset_path, - part=part, - ) - for part in parts - ] - for fut in as_completed(futures): - out_part = fut.result() - bytes_uploaded += out_part["size"] - yield { - "status": "uploading", - "progress": 100 * bytes_uploaded / total_size, - "current": bytes_uploaded, - } - parts_out.append(out_part) - lgr.debug("%s: Completing upload", asset_path) - resp = client.post( - f"/uploads/{upload_id}/complete/", - json={"parts": parts_out}, - ) - lgr.debug( - "%s: Announcing completion to %s", - asset_path, - resp["complete_url"], - ) - r = storage.post( - resp["complete_url"], data=resp["body"], json_resp=False - ) - lgr.debug( - "%s: Upload completed. Response content: %s", - asset_path, - r.content, - ) - rxml = fromstring(r.text) - m = re.match(r"\{.+?\}", rxml.tag) - ns = m.group(0) if m else "" - final_etag = rxml.findtext(f"{ns}ETag") - if final_etag is not None: - final_etag = final_etag.strip('"') - if final_etag != filetag: - raise RuntimeError( - "Server and client disagree on final ETag of" - f" uploaded file; server says {final_etag}," - f" client says {filetag}" - ) - # else: Error? Warning? - resp = client.post(f"/uploads/{upload_id}/validate/") - blob_id = resp["blob_id"] - except Exception: - post_upload_size_check(self.filepath, total_size, True) - raise - else: - post_upload_size_check(self.filepath, total_size, False) + blob_id = resp["blob_id"] lgr.debug("%s: Assigning asset blob to dandiset & version", asset_path) yield {"status": "producing asset"} if replacing is not None: @@ -733,6 +640,131 @@ def _upload_blob_part( } +def multipart_upload( + client: RESTFullAPIClient, + filepath: Path, + asset_path: str, + init_fields: dict[str, Any], + expected_etag: str | None = None, + jobs: int | None = None, + upload_root: str = "/uploads", +) -> Generator[dict, None, dict]: + """ + Upload ``filepath`` to the archive via the S3 multipart upload API, + yielding progress `dict`\\s and returning the deserialized response of the + ``validate`` endpoint. + + Asset blobs and Zarr chunks have separate multipart upload endpoints, so + the caller selects one via ``upload_root`` — ``"/uploads"`` for an asset + blob or ``"/zarr/uploads"`` for a Zarr chunk — and supplies the matching + ``init_fields`` identifying what is being uploaded to the ``initialize`` + endpoint: ``{"dandiset": ...}`` for an asset blob or + ``{"zarr_id": ..., "chunk_key": ...}`` for a Zarr chunk. + + If ``expected_etag`` is non-`None` and does not match the etag computed for + ``filepath``, `RuntimeError` is raised. An HTTP 409 from ``initialize`` + (i.e., the blob already exists) propagates to the caller. + + :meta private: + """ + # Avoid heavy import by importing within function: + from dandi.support.digests import get_dandietag + + yield {"status": "calculating etag"} + etagger = get_dandietag(filepath) + filetag = etagger.as_str() + lgr.debug("Calculated dandi-etag of %s for %s", filetag, filepath) + if expected_etag is not None and expected_etag != filetag: + raise RuntimeError( + f"{filepath}: File etag changed; was originally" + f" {expected_etag} but is now {filetag}" + ) + yield {"status": "initiating upload"} + lgr.debug("%s: Beginning upload", asset_path) + total_size = pre_upload_size_check(filepath) + resp = client.post( + f"{upload_root}/initialize/", + json={ + "contentSize": total_size, + "digest": {"algorithm": "dandi:dandi-etag", "value": filetag}, + **init_fields, + }, + ) + try: + upload_id = resp["upload_id"] + parts = resp["parts"] + if len(parts) != etagger.part_qty: + raise RuntimeError( + f"Server and client disagree on number of parts for upload;" + f" server says {len(parts)}, client says {etagger.part_qty}" + ) + parts_out = [] + bytes_uploaded = 0 + lgr.debug("Uploading %s in %d parts", filepath, len(parts)) + with RESTFullAPIClient("http://nil.nil") as storage: + with filepath.open("rb") as fp: + with ThreadPoolExecutor(max_workers=jobs or 5) as executor: + lock = Lock() + futures = [ + executor.submit( + _upload_blob_part, + storage_session=storage, + fp=fp, + lock=lock, + etagger=etagger, + asset_path=asset_path, + part=part, + ) + for part in parts + ] + for fut in as_completed(futures): + out_part = fut.result() + bytes_uploaded += out_part["size"] + yield { + "status": "uploading", + "progress": 100 * bytes_uploaded / total_size, + "current": bytes_uploaded, + } + parts_out.append(out_part) + lgr.debug("%s: Completing upload", asset_path) + resp = client.post( + f"{upload_root}/{upload_id}/complete/", + json={"parts": parts_out}, + ) + lgr.debug( + "%s: Announcing completion to %s", + asset_path, + resp["complete_url"], + ) + r = storage.post(resp["complete_url"], data=resp["body"], json_resp=False) + lgr.debug( + "%s: Upload completed. Response content: %s", + asset_path, + r.content, + ) + rxml = fromstring(r.text) + m = re.match(r"\{.+?\}", rxml.tag) + ns = m.group(0) if m else "" + final_etag = rxml.findtext(f"{ns}ETag") + if final_etag is not None: + final_etag = final_etag.strip('"') + if final_etag != filetag: + raise RuntimeError( + "Server and client disagree on final ETag of" + f" uploaded file; server says {final_etag}," + f" client says {filetag}" + ) + # else: Error? Warning? + validated = client.post(f"{upload_root}/{upload_id}/validate/") + except Exception: + post_upload_size_check(filepath, total_size, True) + raise + else: + post_upload_size_check(filepath, total_size, False) + assert isinstance(validated, dict) + return validated + + def _check_required_fields( d: dict, required: list[str], file_path: str ) -> list[ValidationResult]: From 496fe18c099dc8bad2b2bb6da382ff9b339562f9 Mon Sep 17 00:00:00 2001 From: Jacob Nesbitt Date: Mon, 31 Aug 2026 13:40:29 -0400 Subject: [PATCH 02/21] Add per-upload-scheme Zarr entry digest helpers An entry uploaded to S3 via a single-part PUT is stored under its plain MD5 ETag, while one uploaded via S3 multipart upload is stored under a multipart ETag; since a Zarr's checksum is an aggregate over its entries' S3 ETags, the two schemes give a Zarr different digests for identical content. Add `dandietag_nocache()` and `get_zarr_multipart_checksum()` as the multipart counterparts of `md5file_nocache()` and `get_zarr_checksum()`, factoring the shared tree walk into `_zarr_checksum()`, so callers name the upload scheme they mean rather than passing a flag. Also add `zarr_has_oversized_entry()` for detecting Zarrs that cannot be uploaded single-part at all. --- dandi/support/digests.py | 110 +++++++++++++++++++++++----- dandi/support/tests/test_digests.py | 40 +++++++++- 2 files changed, 128 insertions(+), 22 deletions(-) diff --git a/dandi/support/digests.py b/dandi/support/digests.py index 7a69a1629..2ac076c80 100644 --- a/dandi/support/digests.py +++ b/dandi/support/digests.py @@ -19,6 +19,7 @@ from dataclasses import dataclass, field import hashlib import logging +import os import os.path from pathlib import Path @@ -28,6 +29,7 @@ from zarr_checksum.tree import ZarrChecksumTree from .threaded_walk import threaded_walk +from ..consts import S3_MAX_SINGLE_PART_UPLOAD from ..utils import Hasher, exclude_from_zarr lgr = logging.getLogger("dandi.support.digests") @@ -98,43 +100,111 @@ def get_dandietag(filepath: str | Path) -> DandiETag: return DandiETag.from_file(filepath) -def get_zarr_checksum(path: Path, known: dict[str, str] | None = None) -> str: +def zarr_has_oversized_entry(path: Path) -> bool: + """ + Return whether the Zarr at ``path`` contains any entry larger than + `S3_MAX_SINGLE_PART_UPLOAD`. Such a Zarr must be uploaded via S3 multipart + upload, since S3 rejects single-part PUTs above that size. """ - Compute the Zarr checksum for a file or directory tree. + for dirpath, dirnames, filenames in os.walk(path): + dp = Path(dirpath) + dirnames[:] = [d for d in dirnames if not exclude_from_zarr(dp / d)] + for fn in filenames: + fp = dp / fn + if exclude_from_zarr(fp): + continue + if os.path.getsize(fp) > S3_MAX_SINGLE_PART_UPLOAD: + return True + return False - If the digests for any files in the Zarr are already known, they can be - passed in the ``known`` argument, which must be a `dict` mapping - slash-separated paths relative to the root of the Zarr to hex digests. + +def md5file_nocache(filepath: str | Path) -> str: """ - if path.is_file(): - s = get_digest(path, "md5") - assert isinstance(s, str) - return s - if known is None: - known = {} + Compute the plain MD5 digest of a file, bypassing the fscacher cache (which + has been shown to slow things down for the large numbers of files typically + present in Zarrs). + + This is the digest of an entry of a **single-part** Zarr, which S3 stores + under its plain MD5 ETag. For the multipart counterpart, see + `dandietag_nocache`. + """ + return Digester(["md5"])(filepath)["md5"] + + +def dandietag_nocache(filepath: str | Path) -> str: + """ + Compute the S3 multipart ETag (a.k.a. DANDI etag) of a file, bypassing the + fscacher cache (cf. `md5file_nocache`). + + This is the digest of an entry of a **multipart** Zarr, which S3 stores + under its multipart ETag; multipart is the scheme `dandi upload` uses for + new Zarrs. For the single-part counterpart, see `md5file_nocache`. + """ + s = DandiETag.from_file(filepath).as_str() + assert isinstance(s, str) + return s + + +def _zarr_checksum( + path: Path, known: dict[str, str], digest_file: Callable[[Path], str] +) -> str: + """ + Compute a Zarr checksum for the directory tree ``path``, digesting each + entry not already present in ``known`` with ``digest_file``. The two public + entry points — `get_zarr_checksum` (single-part) and + `get_zarr_multipart_checksum` (multipart) — differ only in that per-entry + digest function. - def digest_file(f: Path) -> tuple[Path, str, int]: - assert known is not None + :meta private: + """ + + def digest_entry(f: Path) -> tuple[Path, str, int]: relpath = f.relative_to(path).as_posix() try: dgst = known[relpath] except KeyError: - dgst = md5file_nocache(f) + dgst = digest_file(f) return (f, dgst, os.path.getsize(f)) zcc = ZarrChecksumTree() - for p, digest, size in threaded_walk(path, digest_file, exclude=exclude_from_zarr): + for p, digest, size in threaded_walk(path, digest_entry, exclude=exclude_from_zarr): zcc.add_leaf(p.relative_to(path), size, digest) return str(zcc.process()) -def md5file_nocache(filepath: str | Path) -> str: +def get_zarr_checksum(path: Path, known: dict[str, str] | None = None) -> str: """ - Compute the MD5 digest of a file without caching with fscacher, which has - been shown to slow things down for the large numbers of files typically - present in Zarrs + Compute the **single-part** Zarr checksum for a file or directory tree: + every entry is digested with its plain MD5, as S3 stores it for a + single-part upload. This is the checksum of a single-part Zarr; for the + multipart counterpart, see `get_zarr_multipart_checksum`. + + If the digests for any files in the Zarr are already known, they can be + passed in the ``known`` argument, which must be a `dict` mapping + slash-separated paths relative to the root of the Zarr to hex digests. """ - return Digester(["md5"])(filepath)["md5"] + if path.is_file(): + s = get_digest(path, "md5") + assert isinstance(s, str) + return s + return _zarr_checksum(path, known or {}, md5file_nocache) + + +def get_zarr_multipart_checksum(path: Path, known: dict[str, str] | None = None) -> str: + """ + Compute the **multipart** Zarr checksum for a file or directory tree: every + entry is digested with its S3 multipart ETag, as S3 stores it for a + multipart upload (the scheme `dandi upload` uses for new Zarrs). This is + the checksum of a multipart Zarr; for the single-part counterpart, see + `get_zarr_checksum`. + + If the digests for any files in the Zarr are already known, they can be + passed in the ``known`` argument, which must be a `dict` mapping + slash-separated paths relative to the root of the Zarr to hex digests. + """ + if path.is_file(): + return dandietag_nocache(path) + return _zarr_checksum(path, known or {}, dandietag_nocache) def checksum_zarr_dir( diff --git a/dandi/support/tests/test_digests.py b/dandi/support/tests/test_digests.py index af37214ea..603fc68bb 100644 --- a/dandi/support/tests/test_digests.py +++ b/dandi/support/tests/test_digests.py @@ -15,7 +15,15 @@ from pytest_mock import MockerFixture from .. import digests -from ..digests import Digester, checksum_zarr_dir, get_zarr_checksum +from ..digests import ( + Digester, + checksum_zarr_dir, + dandietag_nocache, + get_dandietag, + get_zarr_checksum, + get_zarr_multipart_checksum, + md5file_nocache, +) def test_digester(tmp_path): @@ -78,11 +86,17 @@ def test_get_zarr_checksum(mocker: MockerFixture, tmp_path: Path) -> None: assert ( get_zarr_checksum(tmp_path / "file1.txt") == "d0aa42f003e36c1ecaf9aa8f20b6f1ad" ) + # get_zarr_checksum is the single-part codepath: entries digested with MD5. assert get_zarr_checksum(tmp_path) == "25627e0fc7c609d10100d020f7782a25-8--197" assert get_zarr_checksum(sub1) == "64af93ad7f8d471c00044d1ddbd4c0ba-4--97" - assert get_zarr_checksum(empty) == "481a2f77ab786a0f45aafd5db0971caa-0--0" + # get_zarr_multipart_checksum is the distinct multipart codepath: entries + # are digested differently, so it yields a different checksum for the same + # content. (An empty Zarr's checksum does not depend on the scheme.) + assert get_zarr_multipart_checksum(tmp_path) != get_zarr_checksum(tmp_path) + assert get_zarr_multipart_checksum(empty) == get_zarr_checksum(empty) + spy = mocker.spy(digests, "md5file_nocache") assert ( get_zarr_checksum( @@ -157,3 +171,25 @@ def test_checksum_zarr_dir( checksum: str, ) -> None: assert checksum_zarr_dir(files=files, directories=directories) == checksum + + +@pytest.mark.ai_generated +def test_md5file_nocache_single_part(tmp_path: Path) -> None: + """An entry of a single-part Zarr is digested with its plain MD5.""" + f = tmp_path / "sample.txt" + f.write_bytes(b"123") + assert md5file_nocache(f) == "202cb962ac59075b964b07152d234b70" + + +@pytest.mark.ai_generated +def test_dandietag_nocache_multipart(tmp_path: Path) -> None: + """ + An entry of a multipart Zarr is digested with its multipart ETag rather than + its MD5, as that is what the server ends up storing for it. This is the + distinct multipart codepath, separate from `md5file_nocache`. + """ + f = tmp_path / "sample.txt" + f.write_bytes(b"123") + digest = dandietag_nocache(f) + assert digest != md5file_nocache(f) + assert digest == get_dandietag(f).as_str() From 37b3e28b2cec8ce57a44ef9dc0fa6ad22499fd6c Mon Sep 17 00:00:00 2001 From: Jacob Nesbitt Date: Mon, 31 Aug 2026 13:40:53 -0400 Subject: [PATCH 03/21] Upload new Zarrs via S3 multipart upload `dandi upload` now creates Zarrs with `upload_type: multipart` and uploads every one of their entries via S3 multipart upload, lifting the previous limit that made a Zarr entry larger than `S3_MAX_SINGLE_PART_UPLOAD` (5 GiB) unuploadable. A Zarr's upload scheme is fixed by the archive when the Zarr is created and applies to all of its entries, since its checksum is an aggregate over per-entry S3 ETags. The scheme the server actually recorded is therefore what drives the client: entries are digested with `dandietag_nocache()` for a multipart Zarr and `md5file_nocache()` for a single-part one, both when deciding what to re-upload and when computing the checksum to report. A pre-existing Zarr keeps whatever scheme it was created with, and an archive predating `upload_type` always creates single-part Zarrs; uploading oversized content to such a Zarr is rejected up front with an explanatory error rather than failing partway through. Entries of a batch are uploaded concurrently, each driving its own multipart upload in a worker thread while progress is reported from the main thread. --- dandi/consts.py | 16 ++- dandi/files/zarr.py | 244 +++++++++++++++++++++++++++++++------- dandi/support/digests.py | 2 +- dandi/tests/test_files.py | 71 ++++++++--- 4 files changed, 269 insertions(+), 64 deletions(-) diff --git a/dandi/consts.py b/dandi/consts.py index baca5465c..ecc82b6e1 100644 --- a/dandi/consts.py +++ b/dandi/consts.py @@ -203,10 +203,22 @@ def urls(self) -> Iterator[str]: ZARR_MIME_TYPE = "application/x-zarr" #: Maximum file size for a single S3 PUT upload (5 GiB). -#: S3 rejects single-part PUTs larger than this; such files would need -#: multipart upload which is not yet supported for zarr chunks. +#: S3 rejects single-part PUTs larger than this. A Zarr that contains any +#: entry above this size must therefore be uploaded via S3 multipart upload; +#: the archive records this per Zarr with an immutable ``multipart`` flag, set +#: when the Zarr is created. All entries of a multipart Zarr are uploaded via +#: multipart upload (and digested with their S3 multipart ETag), while all +#: entries of a single-part Zarr are uploaded via single-part PUT (and digested +#: with plain MD5); the two schemes cannot be mixed within one Zarr, since its +#: checksum is an aggregate over per-entry S3 ETags. S3_MAX_SINGLE_PART_UPLOAD = 5 * 1024**3 +#: Values of a Zarr's ``upload_type`` field in the archive API, recording +#: whether its entries are uploaded via single-part PUT or S3 multipart upload. +#: The scheme is fixed when the Zarr is created (see `S3_MAX_SINGLE_PART_UPLOAD`). +ZARR_UPLOAD_TYPE_SINGLEPART = "singlepart" +ZARR_UPLOAD_TYPE_MULTIPART = "multipart" + #: Maximum number of Zarr directory entries to upload at once ZARR_UPLOAD_BATCH_SIZE = 255 diff --git a/dandi/files/zarr.py b/dandi/files/zarr.py index 25394769e..60177ea27 100644 --- a/dandi/files/zarr.py +++ b/dandi/files/zarr.py @@ -21,7 +21,7 @@ if TYPE_CHECKING: from ..upload import ZarrMode -from dandischema.models import BareAsset, DigestType +from dandischema.models import BareAsset from pydantic import BaseModel, ConfigDict, ValidationError import requests from zarr_checksum.tree import ZarrChecksumTree @@ -34,6 +34,7 @@ ZARR_DELETE_BATCH_SIZE, ZARR_MIME_TYPE, ZARR_UPLOAD_BATCH_SIZE, + ZARR_UPLOAD_TYPE_MULTIPART, ) from dandi.dandiapi import ( RemoteAsset, @@ -55,7 +56,7 @@ pre_upload_size_check, ) -from .bases import LocalDirectoryAsset +from .bases import LocalDirectoryAsset, multipart_upload from ..validate._types import ( ORIGIN_VALIDATION_DANDI_ZARR, MissingFileContent, @@ -349,19 +350,18 @@ def iterdir(self) -> Iterator[LocalZarrEntry]: def get_digest(self) -> Digest: """ - Calculate the DANDI etag digest for the entry. If the entry is a - directory, the algorithm will be the DANDI Zarr checksum algorithm; if - it is a file, it will be MD5. + Calculate the digest of the entry as it would be stored in the archive. + If the entry is a directory, the algorithm is the DANDI Zarr checksum + algorithm; if it is a file, it is the S3 multipart ETag (DANDI etag), + matching how `dandi upload` stores an entry of a (multipart) Zarr. """ # Avoid heavy import by importing within function: - from dandi.support.digests import get_digest, get_zarr_checksum + from dandi.support.digests import get_digest, get_zarr_multipart_checksum if self.is_dir(): - return Digest.dandi_zarr(get_zarr_checksum(self.filepath)) + return Digest.dandi_zarr(get_zarr_multipart_checksum(self.filepath)) else: - return Digest( - algorithm=DigestType.md5, value=get_digest(self.filepath, "md5") - ) + return Digest.dandi_etag(get_digest(self.filepath, "dandi-etag")) @property def size(self) -> int: @@ -424,11 +424,12 @@ def filetree(self) -> LocalZarrEntry: def stat(self) -> ZarrStat: """Return various details about the Zarr asset""" + # Avoid heavy import by importing within function: + from dandi.support.digests import checksum_zarr_dir, dandietag_nocache + # Digest entries as they would be uploaded: `dandi upload` creates new + # Zarrs with multipart upload, so every entry gets its multipart ETag. def dirstat(dirpath: LocalZarrEntry) -> ZarrStat: - # Avoid heavy import by importing within function: - from dandi.support.digests import checksum_zarr_dir, md5file_nocache - size = 0 dir_info = {} file_info = {} @@ -441,7 +442,7 @@ def dirstat(dirpath: LocalZarrEntry) -> ZarrStat: files.extend(st.files) else: size += p.size - file_info[p.name] = (md5file_nocache(p.filepath), p.size) + file_info[p.name] = (dandietag_nocache(p.filepath), p.size) files.append(p) return ZarrStat( size=size, @@ -454,9 +455,11 @@ def dirstat(dirpath: LocalZarrEntry) -> ZarrStat: def get_digest(self) -> Digest: """Calculate a dandi-zarr-checksum digest for the asset""" # Avoid heavy import by importing within function: - from dandi.support.digests import get_zarr_checksum + from dandi.support.digests import get_zarr_multipart_checksum - return Digest.dandi_zarr(get_zarr_checksum(self.filepath)) + # `dandi upload` creates new Zarrs with multipart upload, so report the + # checksum the asset will have once uploaded. + return Digest.dandi_zarr(get_zarr_multipart_checksum(self.filepath)) def get_metadata( self, @@ -594,11 +597,29 @@ def iter_upload( lgr.debug("%s: Producing asset", asset_path) yield {"status": "producing asset"} - def mkzarr() -> str: + # Avoid heavy import by importing within function: + from dandi.support.digests import zarr_has_oversized_entry + + # New Zarrs are created with multipart upload enabled. The archive + # records the scheme per Zarr in an immutable ``upload_type`` field set + # at creation time; all of a Zarr's entries must use the same scheme, + # since its checksum is an aggregate over per-entry S3 ETags. A + # pre-existing Zarr keeps whatever scheme it was created with. An entry + # too large for a single-part S3 PUT *requires* multipart upload, so + # such content cannot be uploaded to a single-part Zarr (a pre-existing + # one, or any Zarr on an archive predating the field, which always + # creates single-part Zarrs). + needs_multipart = zarr_has_oversized_entry(self.filepath) + + def mkzarr() -> tuple[str, bool]: try: r = client.post( "/zarr/", - json={"name": asset_path, "dandiset": dandiset.identifier}, + json={ + "name": asset_path, + "dandiset": dandiset.identifier, + "upload_type": ZARR_UPLOAD_TYPE_MULTIPART, + }, ) except requests.HTTPError as e: if e.response is not None and ( @@ -618,12 +639,17 @@ def mkzarr() -> str: }, ) zarr_id = old_zarr["zarr_id"] + zarr_multipart = _zarr_is_multipart(old_zarr) else: raise else: zarr_id = r["zarr_id"] + # Honour the scheme the server actually recorded: an archive + # that predates the ``upload_type`` field ignores it and always + # creates a single-part Zarr. + zarr_multipart = _zarr_is_multipart(r) assert isinstance(zarr_id, str) - return zarr_id + return zarr_id, zarr_multipart if replacing is not None: lgr.debug("%s: Replacing pre-existing asset", asset_path) @@ -632,22 +658,33 @@ def mkzarr() -> str: "%s: Pre-existing asset is a Zarr; reusing & updating", asset_path ) zarr_id = replacing.zarr + # The reused Zarr's upload scheme was fixed when it was created. + multipart = _zarr_is_multipart(client.get(f"/zarr/{zarr_id}/")) else: lgr.debug( "%s: Pre-existing asset is not a Zarr; minting new Zarr", asset_path ) - zarr_id = mkzarr() + zarr_id, multipart = mkzarr() r = client.put( replacing.api_path, json={"metadata": metadata, "zarr_id": zarr_id}, ) else: lgr.debug("%s: Minting new Zarr", asset_path) - zarr_id = mkzarr() + zarr_id, multipart = mkzarr() r = client.post( f"{dandiset.version_api_path}assets/", json={"metadata": metadata, "zarr_id": zarr_id}, ) + + if needs_multipart and not multipart: + raise UploadError( + f"{asset_path}: this Zarr contains an entry larger than" + f" {S3_MAX_SINGLE_PART_UPLOAD / 1024**3:.0f} GiB and so requires" + f" multipart upload, but the target Zarr does not support it." + f" The archive may not support multipart Zarr upload, or the" + f" Zarr being replaced was created as single-part." + ) a = RemoteAsset.from_data(dandiset, r) assert isinstance(a, RemoteZarrAsset) mismatched = True @@ -659,7 +696,7 @@ def mkzarr() -> str: str(e): e for e in a.iterfiles() } total_size = 0 - to_upload = EntryUploadTracker() + to_upload = EntryUploadTracker(multipart=multipart) if old_zarr_entries: to_delete: list[RemoteZarrEntry] = [] digesting: list[Future[tuple[LocalZarrEntry, str, bool]]] = [] @@ -713,6 +750,7 @@ def mkzarr() -> str: asset_path, local_entry, remote_entry.digest.value, + multipart, ) ) for dgstfut in as_completed(digesting): @@ -783,16 +821,63 @@ def mkzarr() -> str: for i, items in enumerate( chunked(upload_items, ZARR_UPLOAD_BATCH_SIZE), start=1 ): - # Items to upload in this batch (may be retried e.g. due to - # 403 errors because of timed-out upload URLs) - items_to_upload = list(items) - max_retries = 5 - retry_count = 0 - current_jobs = jobs or 5 + batch = list(items) # Add all items to checksum tree (only done once) - for it in items_to_upload: + for it in batch: zcc.add_leaf(Path(it.entry_path), it.size, it.digest) + if multipart: + # Every entry of a multipart Zarr is uploaded via S3 + # multipart upload. Upload the batch's entries + # concurrently, driving each entry's multipart upload to + # completion in a worker thread (a generator cannot yield + # from within a worker), and report progress from this, + # the main thread, as entries finish. Each entry still + # parallelizes its own parts across ``jobs`` threads, + # which matters for the occasional very large entry. + def upload_one(it: UploadItem) -> int: + lgr.debug( + "%s: Uploading Zarr entry %s (%.2f GiB) via" + " multipart upload", + asset_path, + it.entry_path, + it.size / 1024**3, + ) + for _status in _upload_zarr_entry_multipart( + client=client, zarr_id=zarr_id, item=it, jobs=jobs + ): + pass + return it.size + + with ThreadPoolExecutor(max_workers=jobs or 5) as executor: + entry_futures = [ + executor.submit(upload_one, it) for it in batch + ] + try: + for entry_fut in as_completed(entry_futures): + bytes_uploaded += entry_fut.result() + changed = True + yield { + "status": "uploading", + "progress": 100 + * bytes_uploaded + / to_upload.total_size, + "current": bytes_uploaded, + } + except BaseException: + for f in entry_futures: + f.cancel() + raise + lgr.debug("%s: Completing upload of batch #%d", asset_path, i) + continue + + # Single-part Zarr: upload the batch of entries via + # single-part PUTs. Items may be retried, e.g. due to 403 + # errors because of timed-out upload URLs. + items_to_upload = list(batch) + max_retries = 5 + retry_count = 0 + current_jobs = jobs or 5 while items_to_upload and retry_count <= max_retries: # Prepare upload requests for current items uploading = [it.upload_request() for it in items_to_upload] @@ -969,6 +1054,17 @@ def mkzarr() -> str: yield {"status": "done", "asset": a} +def _zarr_is_multipart(zarr: dict[str, Any]) -> bool: + """ + Return whether the archive's serialization of a Zarr indicates multipart + upload. An archive predating the ``upload_type`` field omits it, which + means single-part upload. + + :meta private: + """ + return zarr.get("upload_type") == ZARR_UPLOAD_TYPE_MULTIPART + + def _handle_failed_items_and_raise( executor: ThreadPoolExecutor, failed_items: list, futures: list ) -> None: @@ -996,6 +1092,51 @@ def _handle_failed_items_and_raise( raise failed_items[0][1] +def _upload_zarr_entry_multipart( + client: RESTFullAPIClient, + zarr_id: str, + item: UploadItem, + jobs: int | None = None, +) -> Generator[dict, None, None]: + """ + Upload an entry of a multipart Zarr via S3 multipart upload, yielding the + status `dict`\\s of the underlying multipart upload. + + :meta private: + """ + try: + resp = yield from multipart_upload( + client=client, + filepath=item.filepath, + asset_path=item.entry_path, + init_fields={"zarr_id": zarr_id, "chunk_key": item.entry_path}, + expected_etag=item.digest, + jobs=jobs, + upload_root="/zarr/uploads", + ) + except requests.HTTPError as e: + if e.response is not None and e.response.status_code in (400, 404): + # A 404 means the archive lacks the Zarr multipart upload endpoint + # entirely; a 400 means it rejected this multipart upload (e.g. the + # Zarr is not marked for multipart upload). Either way a multipart + # upload of a Zarr entry cannot succeed against this archive. + raise UploadError( + f"{item.entry_path}: server rejected the multipart upload of" + f" this Zarr entry; the archive may not support multipart upload" + f" of Zarr chunks. Server response: {e.response.text}" + ) from e + raise + # The server reports back the key it stored the entry under. If that isn't + # the key we asked for, the entry landed elsewhere in the Zarr, which would + # otherwise surface only as an unexplained Zarr checksum mismatch. + chunk_key = resp.get("chunk_key") + if chunk_key is not None and chunk_key != item.entry_path: + raise UploadError( + f"{item.entry_path}: server stored this Zarr entry under the" + f" unexpected key {chunk_key!r}" + ) + + def _upload_zarr_file( storage_session: RESTFullAPIClient, dandiset: RemoteDandiset, @@ -1098,6 +1239,9 @@ class EntryUploadTracker: :meta private: """ + #: Whether the Zarr is uploaded via S3 multipart upload, which selects how + #: its entries are digested: `dandietag_nocache` if so, else `md5file_nocache`. + multipart: bool = False total_size: int = 0 digested_entries: list[UploadItem] = field(default_factory=list) fresh_entries: list[LocalZarrEntry] = field(default_factory=list) @@ -1109,12 +1253,16 @@ def register(self, e: LocalZarrEntry, digest: str | None = None) -> None: self.fresh_entries.append(e) self.total_size += e.size - @staticmethod - def _mkitem(e: LocalZarrEntry) -> UploadItem: + def _mkitem(self, e: LocalZarrEntry) -> UploadItem: # Avoid heavy import by importing within function: - from dandi.support.digests import md5file_nocache + from dandi.support.digests import dandietag_nocache, md5file_nocache - digest = md5file_nocache(e.filepath) + # Dispatch to the digest matching the Zarr's upload scheme. + digest = ( + dandietag_nocache(e.filepath) + if self.multipart + else md5file_nocache(e.filepath) + ) return UploadItem.from_entry(e, digest) def get_items(self, jobs: int = 5) -> Generator[UploadItem, None, None]: @@ -1166,13 +1314,6 @@ def from_entry(cls, e: LocalZarrEntry, digest: str) -> UploadItem: else: content_type = None size = pre_upload_size_check(e.filepath) - if size > S3_MAX_SINGLE_PART_UPLOAD: - raise ValueError( - f"Zarr chunk {e.filepath} is {size / 1024**3:.2f} GiB," - f" exceeding the S3 single-part upload limit of" - f" {S3_MAX_SINGLE_PART_UPLOAD / 1024**3:.0f} GiB." - f" Multipart upload for zarr chunks is not yet supported." - ) return cls( entry_path=str(e), filepath=e.filepath, @@ -1183,6 +1324,15 @@ def from_entry(cls, e: LocalZarrEntry, digest: str) -> UploadItem: @property def base64_digest(self) -> str: + # An entry of a multipart Zarr is digested with its S3 multipart ETag + # (``-``), which is not a plain MD5 and so has no base64 MD5 + # representation. Such entries are uploaded via multipart upload and do + # not go through the single-part path that needs this header. + if "-" in self.digest: + raise ValueError( + f"{self.entry_path}: digest {self.digest!r} is a multipart" + f" ETag, which has no base64 MD5 representation" + ) return b64encode(bytes.fromhex(self.digest)).decode("us-ascii") def upload_request(self) -> dict[str, str | None]: @@ -1190,12 +1340,20 @@ def upload_request(self) -> dict[str, str | None]: def _cmp_digests( - asset_path: str, local_entry: LocalZarrEntry, remote_digest: str + asset_path: str, + local_entry: LocalZarrEntry, + remote_digest: str, + multipart: bool = False, ) -> tuple[LocalZarrEntry, str, bool]: # Avoid heavy import by importing within function: - from dandi.support.digests import md5file_nocache + from dandi.support.digests import dandietag_nocache, md5file_nocache - local_digest = md5file_nocache(local_entry.filepath) + # Dispatch to the digest matching the Zarr's upload scheme. + local_digest = ( + dandietag_nocache(local_entry.filepath) + if multipart + else md5file_nocache(local_entry.filepath) + ) if local_digest != remote_digest: lgr.debug( "%s: Path %s in Zarr differs from local file; re-uploading", diff --git a/dandi/support/digests.py b/dandi/support/digests.py index 2ac076c80..89adbb282 100644 --- a/dandi/support/digests.py +++ b/dandi/support/digests.py @@ -90,7 +90,7 @@ def get_digest(filepath: str | Path, digest: str = "sha256") -> str: assert isinstance(s, str) return s elif digest == "zarr-checksum": - return get_zarr_checksum(Path(filepath)) + return get_zarr_multipart_checksum(Path(filepath)) else: return Digester([digest])(filepath)[digest] diff --git a/dandi/tests/test_files.py b/dandi/tests/test_files.py index bc73aece2..0fae4c9e7 100644 --- a/dandi/tests/test_files.py +++ b/dandi/tests/test_files.py @@ -31,6 +31,8 @@ dandi_file, find_dandi_files, ) +from ..files.zarr import UploadItem +from ..support.digests import dandietag_nocache, md5file_nocache lgr = get_logger() @@ -472,30 +474,30 @@ def test_upload_zarr(new_dandiset, tmp_path): _ZARR_PROPERTIES_EXPECTED = { "2": { "total_size": 1516, - "total_digest": "4313ab36412db2981c3ed391b38604d6-5--1516", + "total_digest": "8411ffbee3d86259ddbbf9d0d9c754bb-5--1516", "entries": [ - (".zgroup", 24, "e20297935e73dd0154104d4ea53040ab"), - ("arr_0", 746, "51c74ec257069ce3a555bdddeb50230a-2--746"), - ("arr_0/.zarray", 315, "9e30a0a1a465e24220d4132fdd544634"), - ("arr_0/0", 431, "ed4e934a474f1d2096846c6248f18c00"), - ("arr_1", 746, "7b99a0ad9bd8bb3331657e54755b1a31-2--746"), - ("arr_1/.zarray", 315, "9e30a0a1a465e24220d4132fdd544634"), - ("arr_1/0", 431, "fba4dee03a51bde314e9713b00284a93"), + (".zgroup", 24, "c42a84c3473618a4013cbb106ada1c14-1"), + ("arr_0", 746, "02d454d2efbae1c8359399af1006bec4-2--746"), + ("arr_0/.zarray", 315, "980c7afd2e491fd1448dec190f96ed66-1"), + ("arr_0/0", 431, "41a358317ae6e63108b0df7a6d210334-1"), + ("arr_1", 746, "f0c0530e7ad17b8aab0a7136ce784d93-2--746"), + ("arr_1/.zarray", 315, "980c7afd2e491fd1448dec190f96ed66-1"), + ("arr_1/0", 431, "5f534328920f7ee3ac417fbf4ac1826e-1"), ], }, "3": { "total_size": 3935, - "total_digest": "00157f091c9a6295e89eb3c4c2efaeff-5--3935", + "total_digest": "5dda89edeb0a78f05e5a990a55059223-5--3935", "entries": [ - ("arr_0", 2192, "ae16256ae750e4303674ccf1e23fa3c6-2--2192"), - ("arr_0/c", 1573, "93912a45f2107a08090f7b283297d662-1--1573"), - ("arr_0/c/0", 1573, "6c237f8d2d4a41bc1e26e31518dafd9e"), - ("arr_0/zarr.json", 619, "850fae056c97aa9c76df0a52411f4086"), - ("arr_1", 1677, "debc9ca4b2184a6ef1a3d6fcf7d79fd9-2--1677"), - ("arr_1/c", 1058, "2642f5d2df2cddf469313abd9910b371-1--1058"), - ("arr_1/c/0", 1058, "084d662af7251a807649fb48edc36e95"), - ("arr_1/zarr.json", 619, "850fae056c97aa9c76df0a52411f4086"), - ("zarr.json", 66, "457126c0639af2eba0140851c39c1aad"), + ("arr_0", 2192, "0db93f2603c99fd46e40e88e2fc55513-2--2192"), + ("arr_0/c", 1573, "5174f547f2b6124aa6c10b4085eba421-1--1573"), + ("arr_0/c/0", 1573, "0f9a883ef281510b0a1c003dc7da5ea1-1"), + ("arr_0/zarr.json", 619, "8dec6b2c24e4625f746482abbd0d2028-1"), + ("arr_1", 1677, "0be9308adf874a99bca107ac507ec7ec-2--1677"), + ("arr_1/c", 1058, "8ca53b9c7ba6f1b52b1962f9de3a3a50-1--1058"), + ("arr_1/c/0", 1058, "e8f546d8c4808292e5c7d2d1b066ff24-1"), + ("arr_1/zarr.json", 619, "8dec6b2c24e4625f746482abbd0d2028-1"), + ("zarr.json", 66, "ad2687bd67b5a2a11bba074b0a9bdf22-1"), ], }, } @@ -561,6 +563,39 @@ def test_upload_zarr_entry_content_type(new_dandiset, tmp_path): assert r.headers["Content-Type"] == "application/json" +@pytest.mark.ai_generated +def test_zarr_upload_item_single_part(tmp_path: Path) -> None: + """ + An entry of a single-part Zarr carries a plain MD5 digest, which has a + base64 representation used by the single-part path as the Content-MD5 + header. + """ + zarr_path = tmp_path / "example.zarr" + zarr.save(zarr_path, np.arange(1000), np.arange(1000, 0, -1)) + zf = dandi_file(zarr_path) + assert isinstance(zf, ZarrAsset) + entry = next(e for e in zf.iterfiles() if e.is_file()) + item = UploadItem.from_entry(entry, md5file_nocache(entry.filepath)) + assert item.base64_digest + + +@pytest.mark.ai_generated +def test_zarr_upload_item_multipart(tmp_path: Path) -> None: + """ + Every entry of a multipart Zarr is uploaded via multipart upload and carries + a multipart ETag rather than an MD5 digest, regardless of the entry's size. + """ + zarr_path = tmp_path / "example.zarr" + zarr.save(zarr_path, np.arange(1000), np.arange(1000, 0, -1)) + zf = dandi_file(zarr_path) + assert isinstance(zf, ZarrAsset) + entry = next(e for e in zf.iterfiles() if e.is_file()) + item = UploadItem.from_entry(entry, dandietag_nocache(entry.filepath)) + # A multipart ETag is not a hex digest, so it has no base64 MD5 form. + with pytest.raises(ValueError, match="multipart ETag"): + item.base64_digest + + def test_validate_deep_zarr(tmp_path: Path) -> None: zarr_path = tmp_path / "foo.zarr" zarr.save(zarr_path, np.arange(1000), np.arange(1000, 0, -1)) From 12dc50f9d4aaf78678628bf8f4717074c8174202 Mon Sep 17 00:00:00 2001 From: Jacob Nesbitt Date: Thu, 17 Sep 2026 14:02:49 -0400 Subject: [PATCH 04/21] Verify multipart-uploaded Zarrs on download An entry of a Zarr uploaded via S3 multipart upload is stored under a multipart ETag (-) rather than a plain MD5, so downloads of such a Zarr failed verification: both the per-entry digest check and the whole-Zarr checksum assumed MD5 unconditionally. Add `is_multipart_etag()` and use it in `RemoteZarrEntry.from_server_data` to report the entry's digest as `dandi_etag` when the server's ETag is a multipart one. `_download_zarr` then accepts either algorithm per entry, passes the matching digest name down to `_download_file`, and computes the final checksum with `get_zarr_multipart_checksum` when the Zarr's entries are multipart. Co-Authored-By: Claude Opus 5 --- dandi/dandiapi.py | 14 +++++++++++++- dandi/download.py | 27 ++++++++++++++++++++++----- dandi/support/digests.py | 16 ++++++++++++++++ 3 files changed, 51 insertions(+), 6 deletions(-) diff --git a/dandi/dandiapi.py b/dandi/dandiapi.py index ccf17239b..fe7dd18c6 100644 --- a/dandi/dandiapi.py +++ b/dandi/dandiapi.py @@ -2239,12 +2239,24 @@ def from_server_data( cls, asset: BaseRemoteZarrAsset, data: ZarrEntryServerData ) -> RemoteZarrEntry: """:meta private:""" + # Avoid heavy import by importing within function: + from dandi.support.digests import is_multipart_etag + + # An entry's digest is the ETag S3 stores it under, which is a plain MD5 + # for a single-part upload and a multipart ETag (a.k.a. the DANDI etag) + # for a multipart one; the algorithm has to follow suit, or the entry + # cannot be verified on download. + algorithm = ( + models.DigestType.dandi_etag + if is_multipart_etag(data.etag) + else models.DigestType.md5 + ) return cls( client=asset.client, zarr_id=asset.zarr, parts=tuple(data.key.split("/")), modified=data.last_modified, - digest=Digest(algorithm=models.DigestType.md5, value=data.etag), + digest=Digest(algorithm=algorithm, value=data.etag), size=data.size, ) diff --git a/dandi/download.py b/dandi/download.py index a87d5380e..46e494e6f 100644 --- a/dandi/download.py +++ b/dandi/download.py @@ -1043,7 +1043,7 @@ def _download_zarr( zarr_entry_filter: Callable[[str], bool] | None = None, ) -> Iterator[dict]: # Avoid heavy import by importing within function: - from .support.digests import get_zarr_checksum + from .support.digests import get_zarr_checksum, get_zarr_multipart_checksum # we will collect them all while starting the download # with the first page of entries received from the server. @@ -1052,7 +1052,10 @@ def _download_zarr( pc = ProgressCombiner(zarr_size=asset.size) def digest_callback(path: str, algoname: str, d: str) -> None: - if algoname == "md5": + # A Zarr's entries are all digested the same way -- with plain MD5 for a + # single-part Zarr and with the multipart ETag for a multipart one -- + # and either is what the Zarr's checksum is computed from. + if algoname in ("md5", "dandi-etag"): digests[path] = d def downloads_gen(): @@ -1062,7 +1065,12 @@ def downloads_gen(): continue entries.append(entry) etag = entry.digest - assert etag.algorithm is DigestType.md5 + # An entry of a multipart Zarr is stored under an S3 multipart ETag + # rather than a plain MD5 (see `RemoteZarrEntry.from_server_data`). + assert etag.algorithm in (DigestType.md5, DigestType.dandi_etag) + etag_algo = ( + "dandi-etag" if etag.algorithm is DigestType.dandi_etag else "md5" + ) yield pairing( entry_path, _download_file( @@ -1072,7 +1080,7 @@ def downloads_gen(): size=entry.size, mtime=entry.modified, existing=existing, - digests={"md5": etag.value}, + digests={etag_algo: etag.value}, lock=lock, digest_callback=partial(digest_callback, entry_path), ), @@ -1152,7 +1160,16 @@ def downloads_gen(): if "skipped" not in final_out["message"]: zarr_checksum = asset.get_digest().value - local_checksum = get_zarr_checksum(zarr_basepath, known=digests) + # A multipart Zarr's checksum aggregates its entries' multipart + # ETags, so it has to be recomputed the same way; every entry of a + # Zarr uses the one scheme the Zarr was created with. + multipart = any( + e.digest.algorithm is DigestType.dandi_etag for e in entries + ) + checksummer = ( + get_zarr_multipart_checksum if multipart else get_zarr_checksum + ) + local_checksum = checksummer(zarr_basepath, known=digests) if zarr_checksum != local_checksum: msg = f"Zarr checksum: downloaded {local_checksum} != {zarr_checksum}" yield {"checksum": "differs", "status": "error", "message": msg} diff --git a/dandi/support/digests.py b/dandi/support/digests.py index 89adbb282..221d39d81 100644 --- a/dandi/support/digests.py +++ b/dandi/support/digests.py @@ -22,6 +22,7 @@ import os import os.path from pathlib import Path +import re from dandischema.digests.dandietag import DandiETag from fscacher import PersistentCache @@ -100,6 +101,21 @@ def get_dandietag(filepath: str | Path) -> DandiETag: return DandiETag.from_file(filepath) +#: Pattern of an S3 multipart ETag: an MD5 hex digest, a hyphen, and the number +#: of parts (e.g. ``d41d8cd98f00b204e9800998ecf8427e-3``). +_MULTIPART_ETAG_RE = re.compile(r"[0-9a-f]{32}-[1-9][0-9]*\Z") + + +def is_multipart_etag(digest: str) -> bool: + """ + Return whether ``digest`` is an S3 multipart ETag (``-``) rather + than a plain MD5 digest. An entry of a multipart Zarr is stored under such + an ETag (see `dandietag_nocache`), so this distinguishes the digests of a + multipart Zarr's entries from a single-part Zarr's. + """ + return bool(_MULTIPART_ETAG_RE.match(digest)) + + def zarr_has_oversized_entry(path: Path) -> bool: """ Return whether the Zarr at ``path`` contains any entry larger than From 52947773556ef9a4ed7542bdd1101aa12cd9a50d Mon Sep 17 00:00:00 2001 From: Jacob Nesbitt Date: Thu, 17 Sep 2026 14:03:01 -0400 Subject: [PATCH 05/21] Share the S3 upload retry conditions across upload paths `_retry_zarr_file` recognized two transient S3 failures (a 501 from a chunked-encoding fallback and a 400 read/write timeout) that a plain retry of the same request resolves, but only the single-part Zarr upload used it. Move it to `dandi/files/bases.py` as `_retry_s3_upload` and use it for blob part uploads as well, so multipart uploads recover from the same conditions. Co-Authored-By: Claude Opus 5 --- dandi/files/bases.py | 25 +++++++++++++++++++++++++ dandi/files/zarr.py | 20 ++------------------ 2 files changed, 27 insertions(+), 18 deletions(-) diff --git a/dandi/files/bases.py b/dandi/files/bases.py index b983a29db..e39cff965 100644 --- a/dandi/files/bases.py +++ b/dandi/files/bases.py @@ -583,6 +583,30 @@ def size(self) -> int: return sum(p.size for p in self.iterfiles()) +def _retry_s3_upload(r: requests.Response) -> bool: + """ + Return whether a failed S3 upload request should be retried as-is. These + are transient conditions that a fresh attempt at the same presigned URL can + resolve; an expired URL (403), by contrast, has to be re-signed and so is + handled by the caller. + + :meta private: + """ + return ( + # Some sort of filesystem hiccup can cause requests to be unable to get + # the filesize, leading to it falling back to "chunked" transfer + # encoding, which S3 doesn't support. + r.status_code == 501 + and "header you provided implies functionality that is not implemented" + in r.text + ) or ( + # Network issue or rate limiting can cause a timeout, which results in a + # 400. Case: https://github.com/dandi/dandi-cli/issues/1662 + r.status_code == 400 + and "was not read from or written to within the timeout period" in r.text + ) + + def _upload_blob_part( storage_session: RESTFullAPIClient, fp: IO[bytes], @@ -618,6 +642,7 @@ def _upload_blob_part( data=chunk, json_resp=False, retry_statuses=[500], + retry_if=_retry_s3_upload, ) server_etag = r.headers["ETag"].strip('"') lgr.debug( diff --git a/dandi/files/zarr.py b/dandi/files/zarr.py index 60177ea27..a227a6499 100644 --- a/dandi/files/zarr.py +++ b/dandi/files/zarr.py @@ -56,7 +56,7 @@ pre_upload_size_check, ) -from .bases import LocalDirectoryAsset, multipart_upload +from .bases import LocalDirectoryAsset, _retry_s3_upload, multipart_upload from ..validate._types import ( ORIGIN_VALIDATION_DANDI_ZARR, MissingFileContent, @@ -1177,7 +1177,7 @@ def _upload_zarr_file( upload_url, data=fp, json_resp=False, - retry_if=_retry_zarr_file, + retry_if=_retry_s3_upload, headers=headers, timeout=(60, 7200), ) @@ -1215,22 +1215,6 @@ def _upload_zarr_file( return UploadResult(item=item, status=UploadStatus.SUCCESS, size=item.size) -def _retry_zarr_file(r: requests.Response) -> bool: - return ( - # Some sort of filesystem hiccup can cause requests to be unable to get the - # filesize, leading to it falling back to "chunked" transfer encoding, - # which S3 doesn't support. - r.status_code == 501 - and "header you provided implies functionality that is not implemented" - in r.text - ) or ( - # Network issue or rate limiting can cause a timeout, which results in a 400. - # Case: https://github.com/dandi/dandi-cli/issues/1662 - r.status_code == 400 - and "was not read from or written to within the timeout period" in r.text - ) - - @dataclass class EntryUploadTracker: """ From b824e2d1869e880f7a1e2f4b15a7410e339dc50c Mon Sep 17 00:00:00 2001 From: Jacob Nesbitt Date: Thu, 17 Sep 2026 14:03:05 -0400 Subject: [PATCH 06/21] Raise an HTTP error when a part upload returns no ETag A part upload that comes back without an ETag header cannot be completed, since the ETag is how the completion request identifies the part, but it surfaced as a bare `KeyError` that carried no information about the response. Raise a `requests.HTTPError` holding the response instead, so callers can inspect it and decide whether the condition is retryable. Co-Authored-By: Claude Opus 5 --- dandi/files/bases.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/dandi/files/bases.py b/dandi/files/bases.py index e39cff965..baca238fe 100644 --- a/dandi/files/bases.py +++ b/dandi/files/bases.py @@ -644,6 +644,16 @@ def _upload_blob_part( retry_statuses=[500], retry_if=_retry_s3_upload, ) + if "ETag" not in r.headers: + # A part upload that reports no ETag cannot be completed, as the ETag is + # what the completion request identifies the part by. Raise it as an + # HTTP error carrying the response, so that the caller can decide from + # the response whether the condition is worth retrying. + raise requests.HTTPError( + f"{asset_path}: upload of part {part['part_number']} returned no" + f" ETag (status {r.status_code})", + response=r, + ) server_etag = r.headers["ETag"].strip('"') lgr.debug( "%s: Part upload finished ETag=%s Content-Length=%s", From afcfc928272c8d2caebca12fbcf81850eefc2cab Mon Sep 17 00:00:00 2001 From: Jacob Nesbitt Date: Thu, 17 Sep 2026 14:03:12 -0400 Subject: [PATCH 07/21] Send a Zarr entry's Content-Type when initializing a multipart upload S3 fixes an object's Content-Type when the multipart upload is created, which happens server-side, so it cannot be set on the part uploads the way the single-part path sets it as a header on its own PUT. Pass the entry's content type to the `/zarr/uploads/initialize` endpoint instead. Co-Authored-By: Claude Opus 5 --- dandi/files/zarr.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/dandi/files/zarr.py b/dandi/files/zarr.py index a227a6499..e6faa25f9 100644 --- a/dandi/files/zarr.py +++ b/dandi/files/zarr.py @@ -1104,12 +1104,19 @@ def _upload_zarr_entry_multipart( :meta private: """ + init_fields: dict[str, Any] = {"zarr_id": zarr_id, "chunk_key": item.entry_path} + if item.content_type is not None: + # S3 fixes an object's Content-Type when the multipart upload is + # created, which happens server-side, so -- unlike the single-part + # path, which sets it as a header on its own PUT -- it has to be sent + # to the initialize endpoint. + init_fields["content_type"] = item.content_type try: resp = yield from multipart_upload( client=client, filepath=item.filepath, asset_path=item.entry_path, - init_fields={"zarr_id": zarr_id, "chunk_key": item.entry_path}, + init_fields=init_fields, expected_etag=item.digest, jobs=jobs, upload_root="/zarr/uploads", From 20e534d09bb1ee3260f223c09153a5716fad72d0 Mon Sep 17 00:00:00 2001 From: Jacob Nesbitt Date: Thu, 17 Sep 2026 14:03:13 -0400 Subject: [PATCH 08/21] Only treat a failed initialize as a lack of multipart support `_upload_zarr_entry_multipart` fell back to the single-part upload path on any 400 or 404, but only the `initialize` request says anything about whether the archive supports this upload at all. A 400 or 404 from a part upload or from the completion request is an ordinary upload failure, which the caller may want to retry; silently restarting the entry as a single-part upload hides it. Restrict the fallback to errors whose response URL is the initialize endpoint. Co-Authored-By: Claude Opus 5 --- dandi/files/zarr.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/dandi/files/zarr.py b/dandi/files/zarr.py index e6faa25f9..725ae2c6e 100644 --- a/dandi/files/zarr.py +++ b/dandi/files/zarr.py @@ -1122,7 +1122,14 @@ def _upload_zarr_entry_multipart( upload_root="/zarr/uploads", ) except requests.HTTPError as e: - if e.response is not None and e.response.status_code in (400, 404): + # Only the ``initialize`` request tells us whether the archive supports + # this upload at all; a failure of a part upload or of completion is an + # ordinary upload error, which the caller may retry. + if ( + e.response is not None + and e.response.status_code in (400, 404) + and "/zarr/uploads/initialize" in str(getattr(e.response, "url", "")) + ): # A 404 means the archive lacks the Zarr multipart upload endpoint # entirely; a 400 means it rejected this multipart upload (e.g. the # Zarr is not marked for multipart upload). Either way a multipart From 3b2e6759a22cbc3dcd5109254b0108cb1bcc3fe2 Mon Sep 17 00:00:00 2001 From: Jacob Nesbitt Date: Thu, 17 Sep 2026 14:03:19 -0400 Subject: [PATCH 09/21] Retry Zarr entries whose multipart uploads hit retryable errors The multipart entry batch uploaded each entry once and let any failure abort the whole upload, so a batch of large entries could be lost to a presigned part URL expiring (403) or to one of the transient S3 conditions `_retry_s3_upload` recognizes. Give the batch the same structure the single-part path already has: each entry reports an `UploadResult`, entries with retryable errors are re-uploaded from a fresh multipart upload (their part URLs cannot be re-signed in place), and each round halves the worker count and backs off exponentially with jitter, up to five attempts. Non-retryable failures still abort immediately, and entries still failing after the last attempt raise an `UploadError`. Co-Authored-By: Claude Opus 5 --- dandi/files/zarr.py | 156 +++++++++++++++++++++++++++++++++++++------- 1 file changed, 131 insertions(+), 25 deletions(-) diff --git a/dandi/files/zarr.py b/dandi/files/zarr.py index 725ae2c6e..c42654fa9 100644 --- a/dandi/files/zarr.py +++ b/dandi/files/zarr.py @@ -835,7 +835,16 @@ def mkzarr() -> tuple[str, bool]: # the main thread, as entries finish. Each entry still # parallelizes its own parts across ``jobs`` threads, # which matters for the occasional very large entry. - def upload_one(it: UploadItem) -> int: + # + # As in the single-part path, entries whose presigned + # part URLs have timed out (403) are retried with freshly + # initialized uploads at reduced parallelism. + items_to_upload = list(batch) + max_retries = 5 + retry_count = 0 + current_jobs = jobs or 5 + + def upload_one(it: UploadItem) -> UploadResult: lgr.debug( "%s: Uploading Zarr entry %s (%.2f GiB) via" " multipart upload", @@ -843,31 +852,128 @@ def upload_one(it: UploadItem) -> int: it.entry_path, it.size / 1024**3, ) - for _status in _upload_zarr_entry_multipart( - client=client, zarr_id=zarr_id, item=it, jobs=jobs - ): - pass - return it.size - - with ThreadPoolExecutor(max_workers=jobs or 5) as executor: - entry_futures = [ - executor.submit(upload_one, it) for it in batch - ] try: - for entry_fut in as_completed(entry_futures): - bytes_uploaded += entry_fut.result() - changed = True - yield { - "status": "uploading", - "progress": 100 - * bytes_uploaded - / to_upload.total_size, - "current": bytes_uploaded, - } - except BaseException: - for f in entry_futures: - f.cancel() - raise + for _status in _upload_zarr_entry_multipart( + client=client, zarr_id=zarr_id, item=it, jobs=jobs + ): + pass + except requests.HTTPError as e: + # A 403 means the presigned part URLs timed out; + # the other conditions are transient S3 hiccups + # (see `_retry_s3_upload`). Either way the entry + # is retried from a fresh multipart upload, since + # its part URLs cannot be re-signed in place. + if e.response is not None and ( + e.response.status_code == 403 + or _retry_s3_upload(e.response) + ): + lgr.debug( + "Got %d error uploading Zarr entry %s" + " (%d bytes), will retry with a new" + " multipart upload: %s", + e.response.status_code, + it.filepath, + it.size, + str(e), + ) + return UploadResult( + status=UploadStatus.RETRY_NEEDED, item=it + ) + return UploadResult( + status=UploadStatus.FAILED, item=it, error=e + ) + except Exception as e: + return UploadResult( + status=UploadStatus.FAILED, item=it, error=e + ) + return UploadResult( + status=UploadStatus.SUCCESS, item=it, size=it.size + ) + + while items_to_upload and retry_count <= max_retries: + if retry_count == 0: + lgr.debug( + "%s: Uploading Zarr entry batch #%d (%s) via" + " multipart upload", + asset_path, + i, + pluralize(len(items_to_upload), "file"), + ) + else: + lgr.debug( + "%s: Retrying %s from batch #%d (attempt %d/%d)", + asset_path, + pluralize(len(items_to_upload), "file"), + i, + retry_count, + max_retries, + ) + with ThreadPoolExecutor( + max_workers=current_jobs + ) as executor: + entry_futures = [ + executor.submit(upload_one, it) + for it in items_to_upload + ] + retry_items = [] + failed_items = [] + try: + for entry_fut in as_completed(entry_futures): + result = entry_fut.result() + if result.status == UploadStatus.SUCCESS: + bytes_uploaded += result.size + changed = True + yield { + "status": "uploading", + "progress": 100 + * bytes_uploaded + / to_upload.total_size, + "current": bytes_uploaded, + } + elif result.status == UploadStatus.RETRY_NEEDED: + retry_items.append(result.item) + else: + assert result.status == UploadStatus.FAILED + failed_items.append( + (result.item, result.error) + ) + except BaseException: + for f in entry_futures: + f.cancel() + raise + + if failed_items: + _handle_failed_items_and_raise( + executor, failed_items, entry_futures + ) + + if items_to_upload := retry_items: + retry_count += 1 + current_jobs = max(1, math.ceil(current_jobs / 2)) + if retry_count <= max_retries: + lgr.info( + "%s: %s got retryable errors," + " requesting new URLs" + " (attempt %d/%d, workers: %d)", + asset_path, + pluralize(len(items_to_upload), "file"), + retry_count, + max_retries, + current_jobs, + ) + # Exponential backoff with jitter before retry + sleep( + min(2**retry_count * 5, 120) + + random.uniform(0, 5) + ) + + if items_to_upload: + nfiles_str = pluralize(len(items_to_upload), "file") + raise UploadError( + f"{asset_path}: failed to upload {nfiles_str} " + f"after {max_retries} retries due to repeated" + f" retryable upload errors" + ) lgr.debug("%s: Completing upload of batch #%d", asset_path, i) continue From d7ad2fb50f36d6b9658da8787eff591cc6cb38c1 Mon Sep 17 00:00:00 2001 From: Jacob Nesbitt Date: Thu, 17 Sep 2026 14:03:23 -0400 Subject: [PATCH 10/21] Count only PUTs as upload attempts in the Zarr retry test A multipart upload also POSTs to the object's URL to complete it, which `test_zarr_upload_400_timeout_retry` counted as another attempt at uploading the bytes and so mis-tracked its per-URL attempt numbers. Co-Authored-By: Claude Opus 5 --- dandi/tests/test_upload.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/dandi/tests/test_upload.py b/dandi/tests/test_upload.py index 8d1c372d7..d3c4973e1 100644 --- a/dandi/tests/test_upload.py +++ b/dandi/tests/test_upload.py @@ -590,9 +590,12 @@ def test_zarr_upload_400_timeout_retry( original_request = RESTFullAPIClient.request def mock_request(self, method, path, **kwargs): - # Track attempts for each request + # Track upload attempts for each request. Only PUTs are counted: a + # multipart upload also POSTs to the object's URL to complete it, which + # is not another attempt at uploading the bytes. urlpath = urlparse(path).path if path.startswith("http") else path - request_attempts[urlpath] += 1 + if method == "PUT": + request_attempts[urlpath] += 1 # Simulate 400 timeout on first attempt for files containing "arr_0" if method == "PUT" and "arr_0" in path and request_attempts[urlpath] == 1: From 966f5cf70a468358912054e6f83ea092eb4cf89b Mon Sep 17 00:00:00 2001 From: Jacob Nesbitt Date: Thu, 17 Sep 2026 14:18:20 -0400 Subject: [PATCH 11/21] Fix comment --- dandi/consts.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/dandi/consts.py b/dandi/consts.py index ecc82b6e1..d42c08c5c 100644 --- a/dandi/consts.py +++ b/dandi/consts.py @@ -205,12 +205,12 @@ def urls(self) -> Iterator[str]: #: Maximum file size for a single S3 PUT upload (5 GiB). #: S3 rejects single-part PUTs larger than this. A Zarr that contains any #: entry above this size must therefore be uploaded via S3 multipart upload; -#: the archive records this per Zarr with an immutable ``multipart`` flag, set -#: when the Zarr is created. All entries of a multipart Zarr are uploaded via -#: multipart upload (and digested with their S3 multipart ETag), while all -#: entries of a single-part Zarr are uploaded via single-part PUT (and digested -#: with plain MD5); the two schemes cannot be mixed within one Zarr, since its -#: checksum is an aggregate over per-entry S3 ETags. +#: the archive records this per Zarr in an immutable ``upload_type`` field, set +#: when the Zarr is created. All entries of a ``multipart`` Zarr are uploaded +#: via multipart upload (and digested with their S3 multipart ETag), while all +#: entries of a ``singlepart`` Zarr are uploaded via single-part PUT (and +#: digested with plain MD5); the two schemes cannot be mixed within one Zarr, +#: since its checksum is an aggregate over per-entry S3 ETags. S3_MAX_SINGLE_PART_UPLOAD = 5 * 1024**3 #: Values of a Zarr's ``upload_type`` field in the archive API, recording From d57b4c66fdc21bcc0278f08418fde19bc45bdf4a Mon Sep 17 00:00:00 2001 From: Jacob Nesbitt Date: Thu, 17 Sep 2026 14:35:18 -0400 Subject: [PATCH 12/21] Unify the Zarr batch upload retry loop The multipart and single-part batch uploads had grown into two copies of the same retry loop, one of them nested five levels deep inside `iter_upload`, which is already a long generator. Extract that loop as `_upload_zarr_batch`, a generator yielding the size of each entry as it finishes so the caller can report progress. The two schemes differ only in how an attempt's items are handed to the worker pool, which is now a `submit` callback -- `_submit_zarr_entries` for multipart uploads and `_submit_zarr_files` for presigned PUTs -- bound once per upload instead of branched per batch. The per-entry multipart closure becomes `_upload_zarr_entry` at module level. Behavior is unchanged apart from two incidental gains for the single-part path, which now also cancels outstanding futures when the loop is interrupted and marks the upload as changed per successful entry rather than per attempt. Co-Authored-By: Claude Opus 5 --- dandi/files/zarr.py | 508 ++++++++++++++++++++++---------------------- 1 file changed, 257 insertions(+), 251 deletions(-) diff --git a/dandi/files/zarr.py b/dandi/files/zarr.py index c42654fa9..1d5bf4847 100644 --- a/dandi/files/zarr.py +++ b/dandi/files/zarr.py @@ -2,12 +2,13 @@ from base64 import b64encode from collections import Counter -from collections.abc import Generator, Iterator +from collections.abc import Callable, Generator, Iterator from concurrent.futures import Future, ThreadPoolExecutor, as_completed from contextlib import closing from dataclasses import dataclass, field, replace from datetime import datetime from enum import Enum +from functools import partial import json import math import os @@ -818,6 +819,30 @@ def mkzarr() -> tuple[str, bool]: closing(to_upload.get_items()) as upload_items, ): bytes_uploaded = 0 + # A batch of entries is uploaded the same way under either + # scheme -- concurrently, with retries -- and the schemes differ + # only in how an attempt's items are handed to the worker pool. + submit_batch: _BatchSubmitter + if multipart: + submit_batch = partial( + _submit_zarr_entries, + client=client, + zarr_id=zarr_id, + jobs=jobs, + asset_path=asset_path, + ) + via = " via multipart upload" + retry_reason = "retryable errors" + else: + submit_batch = partial( + _submit_zarr_files, + client=client, + zarr_id=zarr_id, + storage=storage, + dandiset=dandiset, + ) + via = "" + retry_reason = "403 errors" for i, items in enumerate( chunked(upload_items, ZARR_UPLOAD_BATCH_SIZE), start=1 ): @@ -826,256 +851,22 @@ def mkzarr() -> tuple[str, bool]: for it in batch: zcc.add_leaf(Path(it.entry_path), it.size, it.digest) - if multipart: - # Every entry of a multipart Zarr is uploaded via S3 - # multipart upload. Upload the batch's entries - # concurrently, driving each entry's multipart upload to - # completion in a worker thread (a generator cannot yield - # from within a worker), and report progress from this, - # the main thread, as entries finish. Each entry still - # parallelizes its own parts across ``jobs`` threads, - # which matters for the occasional very large entry. - # - # As in the single-part path, entries whose presigned - # part URLs have timed out (403) are retried with freshly - # initialized uploads at reduced parallelism. - items_to_upload = list(batch) - max_retries = 5 - retry_count = 0 - current_jobs = jobs or 5 - - def upload_one(it: UploadItem) -> UploadResult: - lgr.debug( - "%s: Uploading Zarr entry %s (%.2f GiB) via" - " multipart upload", - asset_path, - it.entry_path, - it.size / 1024**3, - ) - try: - for _status in _upload_zarr_entry_multipart( - client=client, zarr_id=zarr_id, item=it, jobs=jobs - ): - pass - except requests.HTTPError as e: - # A 403 means the presigned part URLs timed out; - # the other conditions are transient S3 hiccups - # (see `_retry_s3_upload`). Either way the entry - # is retried from a fresh multipart upload, since - # its part URLs cannot be re-signed in place. - if e.response is not None and ( - e.response.status_code == 403 - or _retry_s3_upload(e.response) - ): - lgr.debug( - "Got %d error uploading Zarr entry %s" - " (%d bytes), will retry with a new" - " multipart upload: %s", - e.response.status_code, - it.filepath, - it.size, - str(e), - ) - return UploadResult( - status=UploadStatus.RETRY_NEEDED, item=it - ) - return UploadResult( - status=UploadStatus.FAILED, item=it, error=e - ) - except Exception as e: - return UploadResult( - status=UploadStatus.FAILED, item=it, error=e - ) - return UploadResult( - status=UploadStatus.SUCCESS, item=it, size=it.size - ) - - while items_to_upload and retry_count <= max_retries: - if retry_count == 0: - lgr.debug( - "%s: Uploading Zarr entry batch #%d (%s) via" - " multipart upload", - asset_path, - i, - pluralize(len(items_to_upload), "file"), - ) - else: - lgr.debug( - "%s: Retrying %s from batch #%d (attempt %d/%d)", - asset_path, - pluralize(len(items_to_upload), "file"), - i, - retry_count, - max_retries, - ) - with ThreadPoolExecutor( - max_workers=current_jobs - ) as executor: - entry_futures = [ - executor.submit(upload_one, it) - for it in items_to_upload - ] - retry_items = [] - failed_items = [] - try: - for entry_fut in as_completed(entry_futures): - result = entry_fut.result() - if result.status == UploadStatus.SUCCESS: - bytes_uploaded += result.size - changed = True - yield { - "status": "uploading", - "progress": 100 - * bytes_uploaded - / to_upload.total_size, - "current": bytes_uploaded, - } - elif result.status == UploadStatus.RETRY_NEEDED: - retry_items.append(result.item) - else: - assert result.status == UploadStatus.FAILED - failed_items.append( - (result.item, result.error) - ) - except BaseException: - for f in entry_futures: - f.cancel() - raise - - if failed_items: - _handle_failed_items_and_raise( - executor, failed_items, entry_futures - ) - - if items_to_upload := retry_items: - retry_count += 1 - current_jobs = max(1, math.ceil(current_jobs / 2)) - if retry_count <= max_retries: - lgr.info( - "%s: %s got retryable errors," - " requesting new URLs" - " (attempt %d/%d, workers: %d)", - asset_path, - pluralize(len(items_to_upload), "file"), - retry_count, - max_retries, - current_jobs, - ) - # Exponential backoff with jitter before retry - sleep( - min(2**retry_count * 5, 120) - + random.uniform(0, 5) - ) - - if items_to_upload: - nfiles_str = pluralize(len(items_to_upload), "file") - raise UploadError( - f"{asset_path}: failed to upload {nfiles_str} " - f"after {max_retries} retries due to repeated" - f" retryable upload errors" - ) - lgr.debug("%s: Completing upload of batch #%d", asset_path, i) - continue - - # Single-part Zarr: upload the batch of entries via - # single-part PUTs. Items may be retried, e.g. due to 403 - # errors because of timed-out upload URLs. - items_to_upload = list(batch) - max_retries = 5 - retry_count = 0 - current_jobs = jobs or 5 - while items_to_upload and retry_count <= max_retries: - # Prepare upload requests for current items - uploading = [it.upload_request() for it in items_to_upload] - - if retry_count == 0: - lgr.debug( - "%s: Uploading Zarr file batch #%d (%s)", - asset_path, - i, - pluralize(len(uploading), "file"), - ) - else: - lgr.debug( - "%s: Retrying %s from batch #%d (attempt %d/%d)", - asset_path, - pluralize(len(uploading), "file"), - i, - retry_count, - max_retries, - ) - - # Get signed URLs for items - r = client.post(f"/zarr/{zarr_id}/files/", json=uploading) - - # Upload files in parallel - with ThreadPoolExecutor(max_workers=current_jobs) as executor: - futures = [ - executor.submit( - _upload_zarr_file, - storage_session=storage, - dandiset=dandiset, - upload_url=signed_url, - item=it, - ) - for (signed_url, it) in zip(r, items_to_upload) - ] - - changed = True - retry_items = [] - failed_items = [] - - for fut in as_completed(futures): - result = fut.result() - - if result.status == UploadStatus.SUCCESS: - bytes_uploaded += result.size - yield { - "status": "uploading", - "progress": 100 - * bytes_uploaded - / to_upload.total_size, - "current": bytes_uploaded, - } - elif result.status == UploadStatus.RETRY_NEEDED: - retry_items.append(result.item) - else: - assert result.status == UploadStatus.FAILED - failed_items.append((result.item, result.error)) - - # Handle failed items (non-403 errors) - if failed_items: - _handle_failed_items_and_raise( - executor, failed_items, futures - ) - - # Prepare for next iteration with retry items - if items_to_upload := retry_items: - retry_count += 1 - current_jobs = max(1, math.ceil(current_jobs / 2)) - if retry_count <= max_retries: - lgr.info( - "%s: %s got 403 errors, requesting new URLs" - " (attempt %d/%d, workers: %d)", - asset_path, - pluralize(len(items_to_upload), "file"), - retry_count, - max_retries, - current_jobs, - ) - # Exponential backoff with jitter before retry - sleep( - min(2**retry_count * 5, 120) - + random.uniform(0, 5) - ) - - # Check if we exhausted retries - if items_to_upload: - nfiles_str = pluralize(len(items_to_upload), "file") - raise UploadError( - f"{asset_path}: failed to upload {nfiles_str} " - f"after {max_retries} retries due to repeated 403 errors" - ) + for size in _upload_zarr_batch( + asset_path=asset_path, + batch_number=i, + items=batch, + submit=submit_batch, + jobs=jobs, + via=via, + retry_reason=retry_reason, + ): + bytes_uploaded += size + changed = True + yield { + "status": "uploading", + "progress": 100 * bytes_uploaded / to_upload.total_size, + "current": bytes_uploaded, + } lgr.debug("%s: Completing upload of batch #%d", asset_path, i) lgr.debug("%s: All files uploaded", asset_path) if zarr_mode == "full": @@ -1160,6 +951,221 @@ def upload_one(it: UploadItem) -> UploadResult: yield {"status": "done", "asset": a} +#: Hands an attempt's worth of a batch's items to a worker pool, returning one +#: future per item +_BatchSubmitter = Callable[ + [ThreadPoolExecutor, list["UploadItem"]], list[Future[UploadResult]] +] + +#: Number of times a batch of Zarr entries is retried before giving up +_MAX_BATCH_RETRIES = 5 + + +def _upload_zarr_batch( + *, + asset_path: str, + batch_number: int, + items: list[UploadItem], + submit: _BatchSubmitter, + jobs: int | None, + retry_reason: str, + via: str = "", +) -> Iterator[int]: + """ + Upload one batch of Zarr entries, yielding the size of each entry as it + finishes so that the caller can report progress. + + ``submit`` hands the batch's items to a worker pool. It is called afresh + on every attempt, since a retry is generally needed because the presigned + URLs of the previous attempt timed out and so have to be re-requested. + Entries that failed for a retryable reason are re-submitted with the worker + count halved, after an exponential backoff, for up to + `_MAX_BATCH_RETRIES` attempts; any other failure aborts the batch at once. + + ``retry_reason`` names the retryable condition for log messages, and + ``via`` describes the upload scheme. + + :meta private: + """ + items_to_upload = list(items) + retry_count = 0 + current_jobs = jobs or 5 + while items_to_upload and retry_count <= _MAX_BATCH_RETRIES: + if retry_count == 0: + lgr.debug( + "%s: Uploading Zarr batch #%d (%s)%s", + asset_path, + batch_number, + pluralize(len(items_to_upload), "file"), + via, + ) + else: + lgr.debug( + "%s: Retrying %s from batch #%d (attempt %d/%d)", + asset_path, + pluralize(len(items_to_upload), "file"), + batch_number, + retry_count, + _MAX_BATCH_RETRIES, + ) + with ThreadPoolExecutor(max_workers=current_jobs) as executor: + futures = submit(executor, items_to_upload) + retry_items: list[UploadItem] = [] + failed_items: list[tuple[UploadItem, Exception | None]] = [] + try: + for fut in as_completed(futures): + result = fut.result() + if result.status == UploadStatus.SUCCESS: + yield result.size + elif result.status == UploadStatus.RETRY_NEEDED: + retry_items.append(result.item) + else: + assert result.status == UploadStatus.FAILED + failed_items.append((result.item, result.error)) + except BaseException: + for f in futures: + f.cancel() + raise + + if failed_items: + _handle_failed_items_and_raise(executor, failed_items, futures) + + if items_to_upload := retry_items: + retry_count += 1 + current_jobs = max(1, math.ceil(current_jobs / 2)) + if retry_count <= _MAX_BATCH_RETRIES: + lgr.info( + "%s: %s got %s, requesting new URLs" + " (attempt %d/%d, workers: %d)", + asset_path, + pluralize(len(items_to_upload), "file"), + retry_reason, + retry_count, + _MAX_BATCH_RETRIES, + current_jobs, + ) + # Exponential backoff with jitter before retry + sleep(min(2**retry_count * 5, 120) + random.uniform(0, 5)) + + if items_to_upload: + nfiles_str = pluralize(len(items_to_upload), "file") + raise UploadError( + f"{asset_path}: failed to upload {nfiles_str} after" + f" {_MAX_BATCH_RETRIES} retries due to repeated {retry_reason}" + ) + + +def _submit_zarr_files( + executor: ThreadPoolExecutor, + items: list[UploadItem], + *, + client: RESTFullAPIClient, + zarr_id: str, + storage: RESTFullAPIClient, + dandiset: RemoteDandiset, +) -> list[Future[UploadResult]]: + """ + Submit the entries of a single-part Zarr for upload, each via a presigned + PUT. The URLs are requested anew on every attempt, as a retry is usually + needed because they timed out. + + :meta private: + """ + signed_urls = client.post( + f"/zarr/{zarr_id}/files/", json=[it.upload_request() for it in items] + ) + return [ + executor.submit( + _upload_zarr_file, + storage_session=storage, + dandiset=dandiset, + upload_url=signed_url, + item=it, + ) + for (signed_url, it) in zip(signed_urls, items) + ] + + +def _submit_zarr_entries( + executor: ThreadPoolExecutor, + items: list[UploadItem], + *, + client: RESTFullAPIClient, + zarr_id: str, + jobs: int | None, + asset_path: str, +) -> list[Future[UploadResult]]: + """ + Submit the entries of a multipart Zarr for upload, each via its own S3 + multipart upload. Each entry's upload is driven to completion in a worker + thread, since a generator cannot yield from within one, while the entry + still parallelizes its own parts across ``jobs`` threads, which matters for + the occasional very large entry. + + :meta private: + """ + return [ + executor.submit( + _upload_zarr_entry, + client=client, + zarr_id=zarr_id, + item=it, + jobs=jobs, + asset_path=asset_path, + ) + for it in items + ] + + +def _upload_zarr_entry( + *, + client: RESTFullAPIClient, + zarr_id: str, + item: UploadItem, + jobs: int | None, + asset_path: str, +) -> UploadResult: + """ + Upload one entry of a multipart Zarr via S3 multipart upload, reporting the + outcome as an `UploadResult` rather than raising, so that the batch can + retry or abort as the error warrants. + + :meta private: + """ + lgr.debug( + "%s: Uploading Zarr entry %s (%.2f GiB) via multipart upload", + asset_path, + item.entry_path, + item.size / 1024**3, + ) + try: + for _status in _upload_zarr_entry_multipart( + client=client, zarr_id=zarr_id, item=item, jobs=jobs + ): + pass + except requests.HTTPError as e: + # A 403 means the presigned part URLs timed out; the other conditions + # are transient S3 hiccups (see `_retry_s3_upload`). Either way the + # entry is retried from a fresh multipart upload, since its part URLs + # cannot be re-signed in place. + if e.response is not None and ( + e.response.status_code == 403 or _retry_s3_upload(e.response) + ): + lgr.debug( + "Got %d error uploading Zarr entry %s (%d bytes), will retry" + " with a new multipart upload: %s", + e.response.status_code, + item.filepath, + item.size, + str(e), + ) + return UploadResult(status=UploadStatus.RETRY_NEEDED, item=item) + return UploadResult(status=UploadStatus.FAILED, item=item, error=e) + except Exception as e: + return UploadResult(status=UploadStatus.FAILED, item=item, error=e) + return UploadResult(status=UploadStatus.SUCCESS, item=item, size=item.size) + + def _zarr_is_multipart(zarr: dict[str, Any]) -> bool: """ Return whether the archive's serialization of a Zarr indicates multipart From 9122c9346924379a193beccf5d759ceafaf77b60 Mon Sep 17 00:00:00 2001 From: Jacob Nesbitt Date: Thu, 17 Sep 2026 14:56:23 -0400 Subject: [PATCH 13/21] Name the two Zarr checksums apart in get_digest A Zarr's checksum is an aggregate over its entries' S3 ETags, which differ by the scheme the Zarr was uploaded with, so "the zarr-checksum of this directory" is no longer a well-posed question. `get_digest` had been switched over to the multipart checksum wholesale, which silently changed what `dandi digest -d zarr-checksum` and `dandi ls` report for a Zarr and left no way to ask for the single-part checksum an existing archive Zarr has -- and broke three `dandi digest` tests. Restore ``"zarr-checksum"`` as the single-part checksum and add ``"zarr-checksum-multipart"`` beside it, as with the underlying `get_zarr_checksum` / `get_zarr_multipart_checksum` pair, so callers name the scheme they mean. Both are offered by `dandi digest`, and `dandi ls` asks for the multipart one, since that is what `dandi upload` produces for a new Zarr. Co-Authored-By: Claude Opus 5 --- dandi/cli/cmd_digest.py | 20 +++++++++++++-- dandi/cli/cmd_ls.py | 5 +++- dandi/cli/tests/test_digest.py | 46 ++++++++++++++++++++++++++++++++++ dandi/support/digests.py | 12 +++++++++ 4 files changed, 80 insertions(+), 3 deletions(-) diff --git a/dandi/cli/cmd_digest.py b/dandi/cli/cmd_digest.py index 639f1bda9..43572ee58 100644 --- a/dandi/cli/cmd_digest.py +++ b/dandi/cli/cmd_digest.py @@ -11,7 +11,15 @@ "--digest", "digest_alg", type=click.Choice( - ["dandi-etag", "md5", "sha1", "sha256", "sha512", "zarr-checksum"], + [ + "dandi-etag", + "md5", + "sha1", + "sha256", + "sha512", + "zarr-checksum", + "zarr-checksum-multipart", + ], case_sensitive=False, ), default="dandi-etag", @@ -21,7 +29,15 @@ @click.argument("paths", nargs=-1, type=click.Path(exists=True)) @map_to_click_exceptions def digest(paths: tuple[str, ...], digest_alg: str) -> None: - """Calculate file digests""" + """Calculate file digests + + A Zarr's checksum depends on the scheme its entries were uploaded with, so + the two schemes are named apart: use "zarr-checksum" for a Zarr uploaded + via single-part PUTs and "zarr-checksum-multipart" for one uploaded via S3 + multipart upload, which is the scheme `dandi upload` uses for new Zarrs. + + Example: dandi digest --digest zarr-checksum-multipart sample.zarr + """ # Avoid heavy import by importing within function: from ..support.digests import get_digest diff --git a/dandi/cli/cmd_ls.py b/dandi/cli/cmd_ls.py index 1969236ad..f016e3ded 100644 --- a/dandi/cli/cmd_ls.py +++ b/dandi/cli/cmd_ls.py @@ -371,7 +371,10 @@ def fn(): digest = "0" * 32 + "-0--0" else: lgr.info("Calculating digest for %s", path) - digest = get_digest(path, digest="zarr-checksum") + # `dandi upload` creates new Zarrs with multipart + # upload, so report the checksum this Zarr would + # have in the archive. + digest = get_digest(path, digest="zarr-checksum-multipart") rec = get_metadata(path, Digest.dandi_zarr(digest)) else: if use_fake_digest: diff --git a/dandi/cli/tests/test_digest.py b/dandi/cli/tests/test_digest.py index 9c8fcccbf..64d62ac86 100644 --- a/dandi/cli/tests/test_digest.py +++ b/dandi/cli/tests/test_digest.py @@ -106,3 +106,49 @@ def test_digest_zarr_with_excluded_dotfiles( r = runner.invoke(digest, ["--digest", "zarr-checksum", "sample.zarr"]) assert r.exit_code == 0 assert r.output == f"sample.zarr: {expected}\n" + + +@pytest.mark.ai_generated +def test_digest_zarr_multipart_differs_from_singlepart( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """ + The two Zarr checksums are distinct digests of the same content, since S3 + stores an entry under a plain MD5 ETag for a single-part upload and under a + multipart ETag for a multipart one. + """ + runner = CliRunner() + monkeypatch.chdir(tmp_path) + dt = np.dtype(" None: + """ + Applied to a single file, the multipart Zarr checksum is that entry's own + multipart ETag, as `dandi upload` would store it. + """ + runner = CliRunner() + monkeypatch.chdir(tmp_path) + Path("file.txt").write_bytes(b"123") + r = runner.invoke(digest, ["--digest", "zarr-checksum-multipart", "file.txt"]) + assert r.exit_code == 0 + assert r.output == "file.txt: d022646351048ac0ba397d12dfafa304-1\n" diff --git a/dandi/support/digests.py b/dandi/support/digests.py index 221d39d81..f50265dc1 100644 --- a/dandi/support/digests.py +++ b/dandi/support/digests.py @@ -86,11 +86,23 @@ def __call__(self, fpath: str | Path) -> dict[str, str]: @checksums.memoize_path def get_digest(filepath: str | Path, digest: str = "sha256") -> str: + """ + Compute the digest of ``filepath`` under the named algorithm. + + Besides the hashlib algorithms, ``digest`` may be ``"dandi-etag"`` or one + of the two Zarr checksums. A Zarr's checksum is an aggregate over its + entries' S3 ETags, which differ by the scheme the Zarr was uploaded with, + so the two schemes have to be named apart: ``"zarr-checksum"`` is the + checksum of a single-part Zarr and ``"zarr-checksum-multipart"`` that of a + multipart one, the scheme `dandi upload` uses for new Zarrs. + """ if digest == "dandi-etag": s = get_dandietag(filepath).as_str() assert isinstance(s, str) return s elif digest == "zarr-checksum": + return get_zarr_checksum(Path(filepath)) + elif digest == "zarr-checksum-multipart": return get_zarr_multipart_checksum(Path(filepath)) else: return Digester([digest])(filepath)[digest] From b75b7fba3a77836e909167197b25d22f70bf2b40 Mon Sep 17 00:00:00 2001 From: Jacob Nesbitt Date: Thu, 17 Sep 2026 16:21:04 -0400 Subject: [PATCH 14/21] Test uploading a Zarr entry above the single-part limit The 5 GiB threshold this branch exists to lift had no end-to-end coverage: every entry of a multipart Zarr takes the same path regardless of size, so the machinery was exercised, but nothing proved that an entry S3 will not accept as a single-part PUT actually uploads. Add two tests over a Zarr whose single chunk is 1 MiB past the limit. The first asserts the entry lands under a multipart ETag of more than one part -- which a single-part PUT could not produce -- and that the checksum the archive ingests matches the one computed locally for a multipart Zarr. The second mints the Zarr as single-part beforehand, standing in for one created before the archive supported multipart upload, and asserts the upload is refused up front. The chunk is written directly rather than through zarr, so that several GiB never has to be held in memory, and both tests are gated behind DANDI_TESTS_OVERSIZED_ZARR, as they cost several GiB of disk and need an archive that supports multipart Zarr upload. Co-Authored-By: Claude Opus 5 --- dandi/tests/test_upload.py | 118 ++++++++++++++++++++++++++++++++++++- 1 file changed, 117 insertions(+), 1 deletion(-) diff --git a/dandi/tests/test_upload.py b/dandi/tests/test_upload.py index d3c4973e1..90a1ce8ce 100644 --- a/dandi/tests/test_upload.py +++ b/dandi/tests/test_upload.py @@ -4,7 +4,7 @@ from datetime import datetime, timezone import os from pathlib import Path -from shutil import copyfile, rmtree +from shutil import copyfile, copytree, rmtree from typing import Any from unittest.mock import Mock from urllib.parse import urlparse @@ -22,7 +22,9 @@ from .test_helpers import assert_dirtrees_eq from ..consts import ( DOWNLOAD_SUFFIX, + S3_MAX_SINGLE_PART_UPLOAD, ZARR_MIME_TYPE, + ZARR_UPLOAD_TYPE_SINGLEPART, EmbargoStatus, SyncMode, dandiset_metadata_file, @@ -1093,3 +1095,117 @@ def test_upload_modified_zarr_reports_descriptive_message( assert "adding " in msg, msg assert "modifying " in msg, msg assert "deleting " in msg, msg + + +#: Size of a Zarr entry that S3 cannot accept as a single-part PUT, and which +#: therefore can only be uploaded via S3 multipart upload. +OVERSIZED_ENTRY_SIZE = S3_MAX_SINGLE_PART_UPLOAD + 1024**2 + +#: These tests write an oversized entry to disk and upload it, which costs +#: several GiB and several minutes, and they need an archive that supports +#: multipart Zarr upload, so they are opt-in. +oversized_zarr = pytest.mark.skipif( + not os.environ.get("DANDI_TESTS_OVERSIZED_ZARR"), + reason=( + "Set DANDI_TESTS_OVERSIZED_ZARR=1 to run tests that write a >5 GiB file" + " against a multipart-capable archive" + ), +) + + +@pytest.fixture(scope="module") +def oversized_zarr_path(tmp_path_factory: pytest.TempPathFactory) -> Path: + """ + A valid Zarr whose single chunk is just over `S3_MAX_SINGLE_PART_UPLOAD`. + + The array is created without data and its one chunk is then written + directly, so that the entry never has to be held in memory. The chunk is + filled with random bytes, both because the uncompressed array stores them + verbatim and so that the digests under test are of content that cannot be + trivially deduplicated. + """ + path = tmp_path_factory.mktemp("oversized") / "oversized.zarr" + root = zarr.open_group(str(path), mode="w") + root.create_dataset( + "arr", + shape=(OVERSIZED_ENTRY_SIZE,), + chunks=(OVERSIZED_ENTRY_SIZE,), + dtype="u1", + compressor=None, + ) + chunk = path / "arr" / "0" + with chunk.open("wb") as fp: + remaining = OVERSIZED_ENTRY_SIZE + while remaining > 0: + block = os.urandom(min(64 * 1024**2, remaining)) + fp.write(block) + remaining -= len(block) + assert chunk.stat().st_size == OVERSIZED_ENTRY_SIZE + return path + + +@pytest.mark.ai_generated +@oversized_zarr +def test_upload_zarr_oversized_entry( + new_dandiset: SampleDandiset, oversized_zarr_path: Path +) -> None: + """ + An entry too large for a single-part S3 PUT uploads via S3 multipart + upload, and the Zarr the archive ingests has the checksum computed locally + for a multipart Zarr. + """ + # Avoid heavy import by importing within function: + from ..support.digests import get_zarr_multipart_checksum, is_multipart_etag + + copytree(oversized_zarr_path, new_dandiset.dspath / "oversized.zarr") + new_dandiset.upload() + + (asset,) = new_dandiset.dandiset.get_assets() + assert isinstance(asset, RemoteZarrAsset) + assert asset.path == "oversized.zarr" + + # The oversized entry is stored under a multipart ETag, which is what makes + # it a multipart upload rather than a single-part PUT that happened to work. + entries = {str(e): e for e in asset.iterfiles()} + oversized = entries["arr/0"] + assert oversized.size == OVERSIZED_ENTRY_SIZE + assert is_multipart_etag(oversized.digest.value) + # S3 records the number of parts after the hyphen; more than one of them is + # what distinguishes a genuine multipart upload from a single-part PUT. + assert int(oversized.digest.value.split("-")[1]) > 1 + + # The archive's own checksum of what it ingested must match the checksum we + # compute locally for a multipart Zarr. + assert asset.get_digest().value == get_zarr_multipart_checksum( + new_dandiset.dspath / "oversized.zarr" + ) + + +@pytest.mark.ai_generated +@oversized_zarr +def test_upload_zarr_oversized_entry_to_singlepart_zarr( + new_dandiset: SampleDandiset, oversized_zarr_path: Path +) -> None: + """ + An oversized entry cannot be uploaded to a Zarr that was created as + single-part, since a Zarr's entries all have to use the scheme it was + created with. The upload is refused up front rather than failing partway + through. + """ + copytree(oversized_zarr_path, new_dandiset.dspath / "oversized.zarr") + + # Mint the Zarr ahead of the upload as a single-part one, so that `dandi + # upload` finds and reuses it instead of creating a multipart Zarr. This + # stands in for a Zarr created before the archive supported multipart + # upload. + new_dandiset.client.post( + "/zarr/", + json={ + "name": "oversized.zarr", + "dandiset": new_dandiset.dandiset_id, + "upload_type": ZARR_UPLOAD_TYPE_SINGLEPART, + }, + ) + + with pytest.raises(UploadError, match="requires multipart upload"): + new_dandiset.upload() From 962ad79ede57de20f3f374026f9529a34c37091a Mon Sep 17 00:00:00 2001 From: Jacob Nesbitt Date: Tue, 22 Sep 2026 13:21:33 -0400 Subject: [PATCH 15/21] Build a fresh ETagHashlike for each download attempt The dandi-etag hasher was instantiated once, outside the lambda, so every retry of a download reused the same `ETagHashlike` and kept feeding bytes into the partially-filled hasher of the previous attempt. The hashlib branch builds a fresh object per attempt; this one now does too. Previously only blob assets took this branch, but every entry of a multipart Zarr now does. Co-Authored-By: Claude Opus 5 --- dandi/download.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/dandi/download.py b/dandi/download.py index 46e494e6f..493ea199e 100644 --- a/dandi/download.py +++ b/dandi/download.py @@ -788,10 +788,13 @@ def _download_file( # TODO: reuse that sorting based on speed for algo, digest in digests.items(): if algo == "dandi-etag" and size is not None: - # Instantiate outside the lambda so that mypy is assured that - # `size` is not None: - hasher = ETagHashlike(size) - digester = lambda: hasher # noqa: E731 + # Bind `size` to a local so that mypy is assured it is not + # None. The lambda must construct a fresh `ETagHashlike` on + # each call, as the hashlib branch does: a download attempt + # that is retried starts the hashing over, and an + # `ETagHashlike` fed past `size` bytes raises `ValueError`. + etag_size = size + digester = lambda: ETagHashlike(etag_size) # noqa: E731 else: digester = getattr(hashlib, algo, None) if digester is not None: From c27d4804e29d8ed1c14e1be49762f464643e356c Mon Sep 17 00:00:00 2001 From: Jacob Nesbitt Date: Tue, 22 Sep 2026 13:22:00 -0400 Subject: [PATCH 16/21] Digest an empty Zarr entry the way S3 stores it `DandiETag` gives an empty file zero parts and so an etag of `-0`, but S3 rejects a multipart upload with no parts and stores an empty object under its plain MD5 under either upload scheme. A Zarr containing an empty entry therefore got a multipart checksum that nothing in the archive could ever reproduce -- from `dandi digest`, `dandi ls --metadata`, `ZarrAsset.get_digest()` and `.stat()` alike -- and an unchanged empty entry compared unequal on every re-upload. `dandietag_nocache` now falls back to the plain MD5 for an empty file. The upload path cannot be salvaged here: `contentSize` must be positive at the archive's initialize endpoint, so an empty entry cannot be uploaded to a multipart Zarr at all. Say that outright instead of letting it surface as an opaque 400 reported as "the archive may not support multipart upload". Co-Authored-By: Claude Opus 5 --- dandi/files/zarr.py | 9 +++++++++ dandi/support/digests.py | 11 ++++++++++- dandi/support/tests/test_digests.py | 30 +++++++++++++++++++++++++++++ 3 files changed, 49 insertions(+), 1 deletion(-) diff --git a/dandi/files/zarr.py b/dandi/files/zarr.py index 1d5bf4847..cc5a9239c 100644 --- a/dandi/files/zarr.py +++ b/dandi/files/zarr.py @@ -1216,6 +1216,15 @@ def _upload_zarr_entry_multipart( :meta private: """ + if item.size == 0: + # `contentSize` must be positive at the archive's initialize endpoint, + # and S3 rejects a multipart upload with no parts, so an empty entry + # cannot go through this path at all. Say so, rather than let it + # surface as an opaque 400. + raise UploadError( + f"{item.entry_path}: this Zarr entry is empty, and an empty entry" + f" cannot be uploaded to a multipart Zarr" + ) init_fields: dict[str, Any] = {"zarr_id": zarr_id, "chunk_key": item.entry_path} if item.content_type is not None: # S3 fixes an object's Content-Type when the multipart upload is diff --git a/dandi/support/digests.py b/dandi/support/digests.py index f50265dc1..1070dcc6c 100644 --- a/dandi/support/digests.py +++ b/dandi/support/digests.py @@ -114,7 +114,9 @@ def get_dandietag(filepath: str | Path) -> DandiETag: #: Pattern of an S3 multipart ETag: an MD5 hex digest, a hyphen, and the number -#: of parts (e.g. ``d41d8cd98f00b204e9800998ecf8427e-3``). +#: of parts (e.g. ``d41d8cd98f00b204e9800998ecf8427e-3``). The part count is +#: never zero: S3 rejects a multipart upload with no parts, so an empty object +#: is always stored under its plain MD5 (see `dandietag_nocache`). _MULTIPART_ETAG_RE = re.compile(r"[0-9a-f]{32}-[1-9][0-9]*\Z") @@ -167,7 +169,14 @@ def dandietag_nocache(filepath: str | Path) -> str: This is the digest of an entry of a **multipart** Zarr, which S3 stores under its multipart ETag; multipart is the scheme `dandi upload` uses for new Zarrs. For the single-part counterpart, see `md5file_nocache`. + + An empty file is the one exception: `DandiETag` gives it zero parts and so + an etag of ``-0``, but S3 rejects a multipart upload with no parts and + stores an empty object under its plain MD5 under either scheme. The plain + MD5 is therefore what the archive will have digested the entry as. """ + if os.path.getsize(filepath) == 0: + return md5file_nocache(filepath) s = DandiETag.from_file(filepath).as_str() assert isinstance(s, str) return s diff --git a/dandi/support/tests/test_digests.py b/dandi/support/tests/test_digests.py index 603fc68bb..57aa92d34 100644 --- a/dandi/support/tests/test_digests.py +++ b/dandi/support/tests/test_digests.py @@ -22,6 +22,7 @@ get_dandietag, get_zarr_checksum, get_zarr_multipart_checksum, + is_multipart_etag, md5file_nocache, ) @@ -193,3 +194,32 @@ def test_dandietag_nocache_multipart(tmp_path: Path) -> None: digest = dandietag_nocache(f) assert digest != md5file_nocache(f) assert digest == get_dandietag(f).as_str() + + +@pytest.mark.ai_generated +def test_dandietag_nocache_empty_file(tmp_path: Path) -> None: + """ + S3 rejects a multipart upload with no parts, so an empty object is stored + under its plain MD5 under either upload scheme. `DandiETag` would give it + ``-0``, which is not an ETag S3 ever produces, so `dandietag_nocache` + has to fall back to the plain MD5 or the digest would match nothing in the + archive. + """ + f = tmp_path / "empty.txt" + f.write_bytes(b"") + assert dandietag_nocache(f) == md5file_nocache(f) + assert dandietag_nocache(f) == "d41d8cd98f00b204e9800998ecf8427e" + assert not is_multipart_etag(dandietag_nocache(f)) + + +@pytest.mark.ai_generated +def test_zarr_checksums_agree_on_empty_entry(tmp_path: Path) -> None: + """ + An empty entry is stored under the same ETag under either scheme, so a Zarr + of nothing but empty entries has the same checksum either way. + """ + zarr_path = tmp_path / "empty.zarr" + (zarr_path / "sub").mkdir(parents=True) + (zarr_path / "a").write_bytes(b"") + (zarr_path / "sub" / "b").write_bytes(b"") + assert get_zarr_multipart_checksum(zarr_path) == get_zarr_checksum(zarr_path) From b406546b774109e266db358adfc6e9d1e4cdad7e Mon Sep 17 00:00:00 2001 From: Jacob Nesbitt Date: Tue, 22 Sep 2026 13:22:30 -0400 Subject: [PATCH 17/21] Check for an oversized Zarr entry before minting its asset Three problems with `zarr_has_oversized_entry`: The walk ran on every Zarr upload, although only a single-part Zarr can be rejected by it -- the uncommon case. It is now deferred until the scheme is known to be single-part. It raised after the asset had already been POSTed (or PUT), leaving behind an asset for a Zarr that cannot be uploaded. The check now runs first. It was a third hand-rolled `os.walk` of a Zarr, and `os.walk` does not descend symlinked directories while `iterfiles()` does, so an oversized entry behind a symlink slipped past the guard and failed later, inside `iter_upload`. Replaced by a `ZarrAsset` method over `iterfiles()`. Co-Authored-By: Claude Opus 5 --- dandi/files/zarr.py | 46 +++++++++++++++++++++++++++++----------- dandi/support/digests.py | 19 ----------------- 2 files changed, 34 insertions(+), 31 deletions(-) diff --git a/dandi/files/zarr.py b/dandi/files/zarr.py index cc5a9239c..77ed47a79 100644 --- a/dandi/files/zarr.py +++ b/dandi/files/zarr.py @@ -560,6 +560,16 @@ def _is_too_deep(self) -> bool: return True return False + def _has_oversized_entry(self) -> bool: + """ + Whether any entry is larger than `S3_MAX_SINGLE_PART_UPLOAD`. Such a + Zarr can only be uploaded via S3 multipart upload, since S3 rejects + single-part PUTs above that size. + + :meta private: + """ + return any(e.size > S3_MAX_SINGLE_PART_UPLOAD for e in self.iterfiles()) + def iter_upload( self, dandiset: RemoteDandiset, @@ -598,9 +608,6 @@ def iter_upload( lgr.debug("%s: Producing asset", asset_path) yield {"status": "producing asset"} - # Avoid heavy import by importing within function: - from dandi.support.digests import zarr_has_oversized_entry - # New Zarrs are created with multipart upload enabled. The archive # records the scheme per Zarr in an immutable ``upload_type`` field set # at creation time; all of a Zarr's entries must use the same scheme, @@ -610,7 +617,6 @@ def iter_upload( # such content cannot be uploaded to a single-part Zarr (a pre-existing # one, or any Zarr on an archive predating the field, which always # creates single-part Zarrs). - needs_multipart = zarr_has_oversized_entry(self.filepath) def mkzarr() -> tuple[str, bool]: try: @@ -666,6 +672,7 @@ def mkzarr() -> tuple[str, bool]: "%s: Pre-existing asset is not a Zarr; minting new Zarr", asset_path ) zarr_id, multipart = mkzarr() + _check_single_part_ok(self, asset_path, multipart) r = client.put( replacing.api_path, json={"metadata": metadata, "zarr_id": zarr_id}, @@ -673,19 +680,12 @@ def mkzarr() -> tuple[str, bool]: else: lgr.debug("%s: Minting new Zarr", asset_path) zarr_id, multipart = mkzarr() + _check_single_part_ok(self, asset_path, multipart) r = client.post( f"{dandiset.version_api_path}assets/", json={"metadata": metadata, "zarr_id": zarr_id}, ) - if needs_multipart and not multipart: - raise UploadError( - f"{asset_path}: this Zarr contains an entry larger than" - f" {S3_MAX_SINGLE_PART_UPLOAD / 1024**3:.0f} GiB and so requires" - f" multipart upload, but the target Zarr does not support it." - f" The archive may not support multipart Zarr upload, or the" - f" Zarr being replaced was created as single-part." - ) a = RemoteAsset.from_data(dandiset, r) assert isinstance(a, RemoteZarrAsset) mismatched = True @@ -951,6 +951,28 @@ def mkzarr() -> tuple[str, bool]: yield {"status": "done", "asset": a} +def _check_single_part_ok(zarr: ZarrAsset, asset_path: str, multipart: bool) -> None: + """ + Raise `UploadError` if ``zarr`` can only be uploaded via multipart upload + but the target Zarr is single-part. Called before the asset is minted or + updated, so that a Zarr that cannot be uploaded leaves no asset behind. + + The tree is walked only when the scheme is single-part, which is the + uncommon case; nothing about a multipart Zarr turns on the answer. + + :meta private: + """ + if multipart or not zarr._has_oversized_entry(): + return + raise UploadError( + f"{asset_path}: this Zarr contains an entry larger than" + f" {S3_MAX_SINGLE_PART_UPLOAD / 1024**3:.0f} GiB and so requires" + f" multipart upload, but the target Zarr does not support it." + f" The archive may not support multipart Zarr upload, or the" + f" Zarr being replaced was created as single-part." + ) + + #: Hands an attempt's worth of a batch's items to a worker pool, returning one #: future per item _BatchSubmitter = Callable[ diff --git a/dandi/support/digests.py b/dandi/support/digests.py index 1070dcc6c..c5c3f7547 100644 --- a/dandi/support/digests.py +++ b/dandi/support/digests.py @@ -30,7 +30,6 @@ from zarr_checksum.tree import ZarrChecksumTree from .threaded_walk import threaded_walk -from ..consts import S3_MAX_SINGLE_PART_UPLOAD from ..utils import Hasher, exclude_from_zarr lgr = logging.getLogger("dandi.support.digests") @@ -130,24 +129,6 @@ def is_multipart_etag(digest: str) -> bool: return bool(_MULTIPART_ETAG_RE.match(digest)) -def zarr_has_oversized_entry(path: Path) -> bool: - """ - Return whether the Zarr at ``path`` contains any entry larger than - `S3_MAX_SINGLE_PART_UPLOAD`. Such a Zarr must be uploaded via S3 multipart - upload, since S3 rejects single-part PUTs above that size. - """ - for dirpath, dirnames, filenames in os.walk(path): - dp = Path(dirpath) - dirnames[:] = [d for d in dirnames if not exclude_from_zarr(dp / d)] - for fn in filenames: - fp = dp / fn - if exclude_from_zarr(fp): - continue - if os.path.getsize(fp) > S3_MAX_SINGLE_PART_UPLOAD: - return True - return False - - def md5file_nocache(filepath: str | Path) -> str: """ Compute the plain MD5 digest of a file, bypassing the fscacher cache (which From 075a8bfcc363dd1caf3cd3eec6638bff178aa468 Mon Sep 17 00:00:00 2001 From: Jacob Nesbitt Date: Tue, 22 Sep 2026 13:23:31 -0400 Subject: [PATCH 18/21] Reuse a Zarr entry's DandiETag instead of hashing it twice `EntryUploadTracker._mkitem` and `_cmp_digests` computed an entry's etag with the uncached `dandietag_nocache`, and then `multipart_upload` computed it again via the fscacher-memoized `get_dandietag`. Since the first pass never populates that cache, the second was a guaranteed full re-read of the file, plus an fscacher lookup and write per entry -- every byte of every multipart Zarr hashed twice. `UploadItem` now carries the `DandiETag` its digest came from, and `multipart_upload` takes it as `etagger` and skips the recomputation. The new `_digest_entry` also collapses the two copies of the `dandietag_nocache`-or-`md5file_nocache` dispatch. `base64_digest` gets `is_multipart_etag` in place of testing for a hyphen. Co-Authored-By: Claude Opus 5 --- dandi/files/bases.py | 15 ++++--- dandi/files/zarr.py | 87 ++++++++++++++++++++++++++------------- dandi/tests/test_files.py | 27 +++++++++++- 3 files changed, 94 insertions(+), 35 deletions(-) diff --git a/dandi/files/bases.py b/dandi/files/bases.py index baca238fe..d2bacb75e 100644 --- a/dandi/files/bases.py +++ b/dandi/files/bases.py @@ -681,6 +681,7 @@ def multipart_upload( asset_path: str, init_fields: dict[str, Any], expected_etag: str | None = None, + etagger: DandiETag | None = None, jobs: int | None = None, upload_root: str = "/uploads", ) -> Generator[dict, None, dict]: @@ -696,9 +697,12 @@ def multipart_upload( endpoint: ``{"dandiset": ...}`` for an asset blob or ``{"zarr_id": ..., "chunk_key": ...}`` for a Zarr chunk. - If ``expected_etag`` is non-`None` and does not match the etag computed for - ``filepath``, `RuntimeError` is raised. An HTTP 409 from ``initialize`` - (i.e., the blob already exists) propagates to the caller. + A caller that has already computed ``filepath``'s `DandiETag` — as the Zarr + upload path does, in order to compare against the remote — should pass it + as ``etagger`` so that the file is not hashed a second time here. + Otherwise, if ``expected_etag`` is non-`None` and does not match the etag + computed for ``filepath``, `RuntimeError` is raised. An HTTP 409 from + ``initialize`` (i.e., the blob already exists) propagates to the caller. :meta private: """ @@ -706,9 +710,10 @@ def multipart_upload( from dandi.support.digests import get_dandietag yield {"status": "calculating etag"} - etagger = get_dandietag(filepath) + if etagger is None: + etagger = get_dandietag(filepath) + lgr.debug("Calculated dandi-etag of %s for %s", etagger.as_str(), filepath) filetag = etagger.as_str() - lgr.debug("Calculated dandi-etag of %s for %s", filetag, filepath) if expected_etag is not None and expected_etag != filetag: raise RuntimeError( f"{filepath}: File etag changed; was originally" diff --git a/dandi/files/zarr.py b/dandi/files/zarr.py index 77ed47a79..c3d3af9f6 100644 --- a/dandi/files/zarr.py +++ b/dandi/files/zarr.py @@ -22,6 +22,7 @@ if TYPE_CHECKING: from ..upload import ZarrMode +from dandischema.digests.dandietag import DandiETag from dandischema.models import BareAsset from pydantic import BaseModel, ConfigDict, ValidationError import requests @@ -700,7 +701,9 @@ def mkzarr() -> tuple[str, bool]: to_upload = EntryUploadTracker(multipart=multipart) if old_zarr_entries: to_delete: list[RemoteZarrEntry] = [] - digesting: list[Future[tuple[LocalZarrEntry, str, bool]]] = [] + digesting: list[ + Future[tuple[LocalZarrEntry, str, DandiETag | None, bool]] + ] = [] yield {"status": "comparing against remote Zarr"} with ThreadPoolExecutor(max_workers=jobs or 5) as executor: for local_entry in self.iterfiles(): @@ -762,9 +765,9 @@ def mkzarr() -> tuple[str, bool]: d.cancel() raise else: - local_entry, local_digest, differs = item + local_entry, local_digest, etagger, differs = item if differs: - to_upload.register(local_entry, local_digest) + to_upload.register(local_entry, local_digest, etagger) else: zcc.add_leaf( Path(str(local_entry)), @@ -1261,6 +1264,7 @@ def _upload_zarr_entry_multipart( asset_path=item.entry_path, init_fields=init_fields, expected_etag=item.digest, + etagger=item.etagger, jobs=jobs, upload_root="/zarr/uploads", ) @@ -1387,24 +1391,21 @@ class EntryUploadTracker: digested_entries: list[UploadItem] = field(default_factory=list) fresh_entries: list[LocalZarrEntry] = field(default_factory=list) - def register(self, e: LocalZarrEntry, digest: str | None = None) -> None: + def register( + self, + e: LocalZarrEntry, + digest: str | None = None, + etagger: DandiETag | None = None, + ) -> None: if digest is not None: - self.digested_entries.append(UploadItem.from_entry(e, digest)) + self.digested_entries.append(UploadItem.from_entry(e, digest, etagger)) else: self.fresh_entries.append(e) self.total_size += e.size def _mkitem(self, e: LocalZarrEntry) -> UploadItem: - # Avoid heavy import by importing within function: - from dandi.support.digests import dandietag_nocache, md5file_nocache - - # Dispatch to the digest matching the Zarr's upload scheme. - digest = ( - dandietag_nocache(e.filepath) - if self.multipart - else md5file_nocache(e.filepath) - ) - return UploadItem.from_entry(e, digest) + digest, etagger = _digest_entry(e.filepath, self.multipart) + return UploadItem.from_entry(e, digest, etagger) def get_items(self, jobs: int = 5) -> Generator[UploadItem, None, None]: # Note: In order for the ThreadPoolExecutor to be closed if an error @@ -1438,9 +1439,15 @@ class UploadItem: digest: str size: int content_type: str | None + #: The `DandiETag` that ``digest`` was computed from, when the entry is + #: digested for a multipart upload, so that the upload need not hash the + #: file again + etagger: DandiETag | None = None @classmethod - def from_entry(cls, e: LocalZarrEntry, digest: str) -> UploadItem: + def from_entry( + cls, e: LocalZarrEntry, digest: str, etagger: DandiETag | None = None + ) -> UploadItem: # JSON metadata files. ``.zarray`` / ``.zattrs`` / ``.zgroup`` / # ``.zmetadata`` are the V2 names; ``zarr.json`` is the V3 name (a # single file per group/array containing all metadata). @@ -1461,15 +1468,19 @@ def from_entry(cls, e: LocalZarrEntry, digest: str) -> UploadItem: digest=digest, size=size, content_type=content_type, + etagger=etagger, ) @property def base64_digest(self) -> str: + # Avoid heavy import by importing within function: + from dandi.support.digests import is_multipart_etag + # An entry of a multipart Zarr is digested with its S3 multipart ETag # (``-``), which is not a plain MD5 and so has no base64 MD5 # representation. Such entries are uploaded via multipart upload and do # not go through the single-part path that needs this header. - if "-" in self.digest: + if is_multipart_etag(self.digest): raise ValueError( f"{self.entry_path}: digest {self.digest!r} is a multipart" f" ETag, which has no base64 MD5 representation" @@ -1480,31 +1491,49 @@ def upload_request(self) -> dict[str, str | None]: return {"path": self.entry_path, "base64md5": self.base64_digest} +def _digest_entry(filepath: Path, multipart: bool) -> tuple[str, DandiETag | None]: + """ + Digest a Zarr entry the way the archive will have stored it: with its S3 + multipart ETag for an entry of a multipart Zarr, or with its plain MD5 for + a single-part one. + + For a multipart entry the `DandiETag` that produced the digest is returned + alongside it, so that uploading the entry need not hash the file a second + time. There is none for an empty entry, which S3 stores under its plain + MD5 under either scheme (see `dandietag_nocache`). + + :meta private: + """ + # Avoid heavy import by importing within function: + from dandi.support.digests import dandietag_nocache, md5file_nocache + + if not multipart: + return (md5file_nocache(filepath), None) + if os.path.getsize(filepath) == 0: + return (dandietag_nocache(filepath), None) + etagger = DandiETag.from_file(filepath) + digest = etagger.as_str() + assert isinstance(digest, str) + return (digest, etagger) + + def _cmp_digests( asset_path: str, local_entry: LocalZarrEntry, remote_digest: str, multipart: bool = False, -) -> tuple[LocalZarrEntry, str, bool]: - # Avoid heavy import by importing within function: - from dandi.support.digests import dandietag_nocache, md5file_nocache - - # Dispatch to the digest matching the Zarr's upload scheme. - local_digest = ( - dandietag_nocache(local_entry.filepath) - if multipart - else md5file_nocache(local_entry.filepath) - ) +) -> tuple[LocalZarrEntry, str, DandiETag | None, bool]: + local_digest, etagger = _digest_entry(local_entry.filepath, multipart) if local_digest != remote_digest: lgr.debug( "%s: Path %s in Zarr differs from local file; re-uploading", asset_path, local_entry, ) - return (local_entry, local_digest, True) + return (local_entry, local_digest, etagger, True) else: lgr.debug("%s: File %s already on server; skipping", asset_path, local_entry) - return (local_entry, local_digest, False) + return (local_entry, local_digest, etagger, False) def _rmfiles( diff --git a/dandi/tests/test_files.py b/dandi/tests/test_files.py index 0fae4c9e7..cfed8fd67 100644 --- a/dandi/tests/test_files.py +++ b/dandi/tests/test_files.py @@ -31,7 +31,7 @@ dandi_file, find_dandi_files, ) -from ..files.zarr import UploadItem +from ..files.zarr import EntryUploadTracker, UploadItem from ..support.digests import dandietag_nocache, md5file_nocache lgr = get_logger() @@ -596,6 +596,31 @@ def test_zarr_upload_item_multipart(tmp_path: Path) -> None: item.base64_digest +@pytest.mark.ai_generated +def test_zarr_upload_item_carries_etagger(tmp_path: Path) -> None: + """ + An entry digested for a multipart upload keeps the `DandiETag` it was + digested with, so that uploading it does not read & hash the file a second + time. There is none to keep for a single-part entry, nor for an empty one, + which S3 stores under its plain MD5 under either scheme. + """ + zarr_path = tmp_path / "example.zarr" + zarr.save(zarr_path, np.arange(1000), np.arange(1000, 0, -1)) + (zarr_path / "empty").write_bytes(b"") + zf = dandi_file(zarr_path) + assert isinstance(zf, ZarrAsset) + entries = {str(e): e for e in zf.iterfiles()} + nonempty = next(e for e in entries.values() if e.size > 0) + + tracker = EntryUploadTracker(multipart=True) + item = tracker._mkitem(nonempty) + assert item.etagger is not None + assert item.etagger.as_str() == item.digest + + assert tracker._mkitem(entries["empty"]).etagger is None + assert EntryUploadTracker(multipart=False)._mkitem(nonempty).etagger is None + + def test_validate_deep_zarr(tmp_path: Path) -> None: zarr_path = tmp_path / "foo.zarr" zarr.save(zarr_path, np.arange(1000), np.arange(1000, 0, -1)) From 3b67a4b26992a64c3bf1ad505b6ac98636a537f7 Mon Sep 17 00:00:00 2001 From: Jacob Nesbitt Date: Tue, 22 Sep 2026 13:23:45 -0400 Subject: [PATCH 19/21] Only treat a 409 from initialize as a pre-existing blob Extracting `multipart_upload` widened the 409 handler in `LocalFileAsset.iter_upload`: on master it wrapped only the call to `/uploads/initialize/`, but it now wraps the whole upload. Only that endpoint answers 409 with a `Location` header naming the existing blob. `/uploads/{id}/validate/` also answers 409 -- "An identical blob has already been uploaded", when two clients race to upload the same one -- with no such header, as does any part PUT or the completion request. Such a response reached `e.response.headers["Location"]` and raised `KeyError: 'Location'`, hiding the real error. `multipart_upload` now raises `BlobExistsError` from the initialize call alone; a 409 from anywhere else propagates as the ordinary `HTTPError` it was before this branch. Co-Authored-By: Claude Opus 5 --- dandi/exceptions.py | 15 +++++++++++ dandi/files/bases.py | 43 ++++++++++++++++++----------- dandi/tests/test_files.py | 57 +++++++++++++++++++++++++++++++++++++-- 3 files changed, 97 insertions(+), 18 deletions(-) diff --git a/dandi/exceptions.py b/dandi/exceptions.py index f323d2a42..096d74ff8 100644 --- a/dandi/exceptions.py +++ b/dandi/exceptions.py @@ -93,6 +93,21 @@ class UploadError(Exception): pass +class BlobExistsError(UploadError): + """ + Raised when the archive reports, via an HTTP 409 from the ``initialize`` + endpoint of a multipart upload, that the blob being uploaded is already + present. Only that endpoint identifies the existing blob; a 409 from any + other point of an upload is an ordinary error and propagates as such. + """ + + def __init__(self, blob_id: str) -> None: + super().__init__(f"Blob already exists on server with ID {blob_id}") + #: The ID of the pre-existing blob, from the response's ``Location`` + #: header + self.blob_id = blob_id + + class UploadValidationError(UploadError): """An upload could not proceed because an asset failed validation.""" diff --git a/dandi/files/bases.py b/dandi/files/bases.py index d2bacb75e..61f51b345 100644 --- a/dandi/files/bases.py +++ b/dandi/files/bases.py @@ -31,6 +31,7 @@ RESTFullAPIClient, set_asset_schema_key, ) +from dandi.exceptions import BlobExistsError from dandi.metadata.core import get_default_metadata from dandi.misctypes import DUMMY_DANDI_ETAG, Digest, LocalReadableFile, P from dandi.utils import post_upload_size_check, pre_upload_size_check, yaml_load @@ -376,12 +377,9 @@ def iter_upload( expected_etag=metadata.get("digest", {}).get("dandi:dandi-etag"), jobs=jobs, ) - except requests.HTTPError as e: - if e.response is not None and e.response.status_code == 409: - lgr.debug("%s: Blob already exists on server", asset_path) - blob_id = e.response.headers["Location"] - else: - raise + except BlobExistsError as e: + lgr.debug("%s: Blob already exists on server", asset_path) + blob_id = e.blob_id else: blob_id = resp["blob_id"] lgr.debug("%s: Assigning asset blob to dandiset & version", asset_path) @@ -701,8 +699,12 @@ def multipart_upload( upload path does, in order to compare against the remote — should pass it as ``etagger`` so that the file is not hashed a second time here. Otherwise, if ``expected_etag`` is non-`None` and does not match the etag - computed for ``filepath``, `RuntimeError` is raised. An HTTP 409 from - ``initialize`` (i.e., the blob already exists) propagates to the caller. + computed for ``filepath``, `RuntimeError` is raised. + + An HTTP 409 from ``initialize`` — the blob is already present — is raised + as `BlobExistsError`. A 409 from any other point of the upload (e.g. from + ``validate``, when two clients race to upload the same blob) identifies no + blob and so propagates as an ordinary `requests.HTTPError`. :meta private: """ @@ -722,14 +724,23 @@ def multipart_upload( yield {"status": "initiating upload"} lgr.debug("%s: Beginning upload", asset_path) total_size = pre_upload_size_check(filepath) - resp = client.post( - f"{upload_root}/initialize/", - json={ - "contentSize": total_size, - "digest": {"algorithm": "dandi:dandi-etag", "value": filetag}, - **init_fields, - }, - ) + try: + resp = client.post( + f"{upload_root}/initialize/", + json={ + "contentSize": total_size, + "digest": {"algorithm": "dandi:dandi-etag", "value": filetag}, + **init_fields, + }, + ) + except requests.HTTPError as e: + if ( + e.response is not None + and e.response.status_code == 409 + and (blob_id := e.response.headers.get("Location")) is not None + ): + raise BlobExistsError(blob_id) from e + raise try: upload_id = resp["upload_id"] parts = resp["parts"] diff --git a/dandi/tests/test_files.py b/dandi/tests/test_files.py index cfed8fd67..5b9b8d435 100644 --- a/dandi/tests/test_files.py +++ b/dandi/tests/test_files.py @@ -4,11 +4,12 @@ import os from pathlib import Path import subprocess -from unittest.mock import ANY +from unittest.mock import ANY, Mock from dandischema.models import get_schema_version import numpy as np import pytest +import requests import zarr from .fixtures import SampleDandiset @@ -16,7 +17,7 @@ from .. import get_logger from ..consts import ZARR_MIME_TYPE, dandiset_metadata_file from ..dandiapi import AssetType, RemoteZarrAsset -from ..exceptions import UnknownAssetError +from ..exceptions import BlobExistsError, UnknownAssetError from ..files import ( BIDSDatasetDescriptionAsset, DandisetMetadataFile, @@ -31,6 +32,7 @@ dandi_file, find_dandi_files, ) +from ..files.bases import multipart_upload from ..files.zarr import EntryUploadTracker, UploadItem from ..support.digests import dandietag_nocache, md5file_nocache @@ -621,6 +623,57 @@ def test_zarr_upload_item_carries_etagger(tmp_path: Path) -> None: assert EntryUploadTracker(multipart=False)._mkitem(nonempty).etagger is None +def _http_error( + status: int, headers: dict[str, str] | None = None +) -> requests.HTTPError: + r = requests.Response() + r.status_code = status + r.headers.update(headers or {}) + return requests.HTTPError(f"{status} error", response=r) + + +@pytest.mark.ai_generated +def test_multipart_upload_blob_exists(tmp_path: Path) -> None: + """ + A 409 from ``initialize`` means the blob is already present, and the + response's ``Location`` header identifies it. + """ + f = tmp_path / "blob.dat" + f.write_bytes(b"data") + client = Mock() + client.post.side_effect = _http_error(409, {"Location": "some-blob-id"}) + with pytest.raises(BlobExistsError) as excinfo: + for _ in multipart_upload( + client=client, + filepath=f, + asset_path="blob.dat", + init_fields={"dandiset": "000001"}, + ): + pass + assert excinfo.value.blob_id == "some-blob-id" + + +@pytest.mark.ai_generated +def test_multipart_upload_conflict_without_location(tmp_path: Path) -> None: + """ + Only ``initialize`` reports a pre-existing blob. A 409 that identifies no + blob -- as ``validate`` returns when two clients race to upload the same + one -- is an ordinary error and must not be mistaken for one. + """ + f = tmp_path / "blob.dat" + f.write_bytes(b"data") + client = Mock() + client.post.side_effect = _http_error(409) + with pytest.raises(requests.HTTPError): + for _ in multipart_upload( + client=client, + filepath=f, + asset_path="blob.dat", + init_fields={"dandiset": "000001"}, + ): + pass + + def test_validate_deep_zarr(tmp_path: Path) -> None: zarr_path = tmp_path / "foo.zarr" zarr.save(zarr_path, np.arange(1000), np.arange(1000, 0, -1)) From d1f845fd213f8e783c03d3bf5b7af58f3bf62647 Mon Sep 17 00:00:00 2001 From: Jacob Nesbitt Date: Tue, 22 Sep 2026 13:24:00 -0400 Subject: [PATCH 20/21] Document the new digest choice and the oversized-Zarr test switch `dandi digest` gained a user-visible `zarr-checksum-multipart` choice that the command's page did not list, and `DANDI_TESTS_OVERSIZED_ZARR` was not listed alongside the other `DANDI_TESTS_*` variables. Co-Authored-By: Claude Opus 5 --- DEVELOPMENT.md | 5 +++++ docs/source/cmdline/digest.rst | 8 +++++++- 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index 9a2e4f3b9..ac029ec35 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -337,6 +337,11 @@ New markers must be registered in `pytest_configure()` in - `DANDI_TESTS_NO_VCR` — When set, the use of vcrpy to playback captured HTTP requests during testing will be disabled +- `DANDI_TESTS_OVERSIZED_ZARR` -- When set to a non-empty value, enables the + tests that upload a Zarr entry larger than `S3_MAX_SINGLE_PART_UPLOAD` + (5 GiB). These write several GiB to disk, take several minutes, and require + an archive that supports multipart Zarr upload, so they are opt-in. + - `DANDI_TESTS_INSTANCE_NAME` -- Sets the instance name for the dandi-archive instance used for testing. Defaults to `"DANDI"`. Useful for testing dandi-cli against a dandi-archive instance with a particular vendor information. diff --git a/docs/source/cmdline/digest.rst b/docs/source/cmdline/digest.rst index 793de616b..eb060f6a6 100644 --- a/docs/source/cmdline/digest.rst +++ b/docs/source/cmdline/digest.rst @@ -10,6 +10,12 @@ Calculate file digests Options ------- -.. option:: -d, --digest [dandi-etag|md5|sha1|sha256|sha512|zarr-checksum] +.. option:: -d, --digest [dandi-etag|md5|sha1|sha256|sha512|zarr-checksum|zarr-checksum-multipart] Digest algorithm to use [default: ``dandi-etag``] + + A Zarr's checksum is an aggregate over its entries' S3 ETags, which differ + by the scheme the Zarr was uploaded with, so the two schemes are named + apart: ``zarr-checksum`` is the checksum of a Zarr uploaded via single-part + PUTs, and ``zarr-checksum-multipart`` that of one uploaded via S3 multipart + upload, the scheme :program:`dandi upload` uses for new Zarrs. From 40134ad082c3b4db75f4eabf70c1af61e7b65e18 Mon Sep 17 00:00:00 2001 From: Jacob Nesbitt Date: Tue, 22 Sep 2026 16:40:44 -0400 Subject: [PATCH 21/21] [WIP] Abort a Zarr entry's multipart upload when giving up on it The archive refuses to finalize a Zarr while any of its uploads are still outstanding, since such an upload's chunk can land after ingestion has already listed the Zarr's contents. But a retryable failure -- a 403 on expired part URLs, or one of the transient S3 conditions `_retry_s3_upload` covers -- had the client abandon its upload and *initialize* a fresh one, leaving the old record behind. A single 403 during an otherwise successful upload was therefore enough to have finalize rejected, and re-running it kept failing until garbage collection expired the record a week later. `multipart_upload` now releases an upload that fails after initialize, which also stops orphaning the S3 multipart upload behind it. This is best-effort: an archive without the endpoint leaves the record for garbage collection, as before, and a failure to abort never displaces the error that caused it. Should a record be left behind anyway -- by an older client, or by this one being killed mid-upload -- finalize's refusal is now reported as something the user can act on rather than a bare HTTPError. Co-Authored-By: Claude Opus 5 --- dandi/files/bases.py | 36 +++++++++++++++++++++++++++++++ dandi/files/zarr.py | 22 +++++++++++++++++-- dandi/tests/test_files.py | 45 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 101 insertions(+), 2 deletions(-) diff --git a/dandi/files/bases.py b/dandi/files/bases.py index 61f51b345..5ae27d16e 100644 --- a/dandi/files/bases.py +++ b/dandi/files/bases.py @@ -673,6 +673,36 @@ def _upload_blob_part( } +def _abort_upload( + client: RESTFullAPIClient, upload_root: str, upload_id: str, asset_path: str +) -> None: + """ + Release a multipart upload that will not be completed, so that the archive + does not keep an upload record (and an in-progress S3 multipart upload) + around until it is garbage-collected a week later. Until such a record is + gone it counts as an active upload, which blocks finalizing the Zarr it + belongs to and unembargoing its Dandiset. + + Best-effort: a failure here must not displace whatever went wrong with the + upload itself, and an archive with no such endpoint simply leaves the + record for garbage collection, as before. + + :meta private: + """ + try: + client.delete(f"{upload_root}/{upload_id}/") + except Exception as e: + lgr.debug( + "%s: Failed to abort upload %s: %s: %s", + asset_path, + upload_id, + type(e).__name__, + str(e), + ) + else: + lgr.debug("%s: Aborted upload %s", asset_path, upload_id) + + def multipart_upload( client: RESTFullAPIClient, filepath: Path, @@ -701,6 +731,9 @@ def multipart_upload( Otherwise, if ``expected_etag`` is non-`None` and does not match the etag computed for ``filepath``, `RuntimeError` is raised. + An upload that fails after ``initialize`` is aborted, so that the archive + does not count it as active until garbage collection (see `_abort_upload`). + An HTTP 409 from ``initialize`` — the blob is already present — is raised as `BlobExistsError`. A 409 from any other point of the upload (e.g. from ``validate``, when two clients race to upload the same blob) identifies no @@ -741,6 +774,7 @@ def multipart_upload( ): raise BlobExistsError(blob_id) from e raise + upload_id: str | None = None try: upload_id = resp["upload_id"] parts = resp["parts"] @@ -809,6 +843,8 @@ def multipart_upload( validated = client.post(f"{upload_root}/{upload_id}/validate/") except Exception: post_upload_size_check(filepath, total_size, True) + if upload_id is not None: + _abort_upload(client, upload_root, upload_id, asset_path) raise else: post_upload_size_check(filepath, total_size, False) diff --git a/dandi/files/zarr.py b/dandi/files/zarr.py index c3d3af9f6..e54d7ad26 100644 --- a/dandi/files/zarr.py +++ b/dandi/files/zarr.py @@ -900,7 +900,24 @@ def mkzarr() -> tuple[str, bool]: "%s: Waiting for server to calculate Zarr checksum", asset_path ) yield {"status": "server calculating checksum"} - client.post(f"/zarr/{zarr_id}/finalize/") + try: + client.post(f"/zarr/{zarr_id}/finalize/") + except requests.HTTPError as e: + # The archive refuses to ingest a Zarr with uploads still + # outstanding, as their chunks could land after it has + # listed the Zarr's contents. An upload this client gave + # up on is aborted (see `_abort_upload`), so reaching this + # means some upload was left behind -- by an older client, + # or by this one being killed mid-upload -- and only + # garbage collection will clear it. + if e.response is not None and ("active uploads" in e.response.text): + raise UploadError( + f"{asset_path}: the archive will not finalize this" + f" Zarr while it has uploads outstanding; they are" + f" cleared by garbage collection once they expire." + f" Server response: {e.response.text}" + ) from e + raise while True: sleep(2) r = client.get(f"/zarr/{zarr_id}/") @@ -1172,7 +1189,8 @@ def _upload_zarr_entry( # A 403 means the presigned part URLs timed out; the other conditions # are transient S3 hiccups (see `_retry_s3_upload`). Either way the # entry is retried from a fresh multipart upload, since its part URLs - # cannot be re-signed in place. + # cannot be re-signed in place; `multipart_upload` has already aborted + # the one that failed, which the Zarr cannot be finalized until. if e.response is not None and ( e.response.status_code == 403 or _retry_s3_upload(e.response) ): diff --git a/dandi/tests/test_files.py b/dandi/tests/test_files.py index 5b9b8d435..f60709500 100644 --- a/dandi/tests/test_files.py +++ b/dandi/tests/test_files.py @@ -674,6 +674,51 @@ def test_multipart_upload_conflict_without_location(tmp_path: Path) -> None: pass +@pytest.mark.ai_generated +def test_multipart_upload_aborts_on_failure(tmp_path: Path) -> None: + """ + An upload that fails after initialize is released, so that the archive does + not go on counting it as active -- which would block finalizing the Zarr it + belongs to, and unembargoing its Dandiset -- until garbage collection. + """ + f = tmp_path / "blob.dat" + f.write_bytes(b"data") + client = Mock() + # Initialize succeeds, but the part list disagrees with what we computed. + client.post.return_value = {"upload_id": "upload-1", "parts": []} + with pytest.raises(RuntimeError, match="number of parts"): + for _ in multipart_upload( + client=client, + filepath=f, + asset_path="blob.dat", + init_fields={"dandiset": "000001"}, + upload_root="/zarr/uploads", + ): + pass + client.delete.assert_called_once_with("/zarr/uploads/upload-1/") + + +@pytest.mark.ai_generated +def test_multipart_upload_abort_failure_is_not_masked(tmp_path: Path) -> None: + """ + Aborting is best-effort: against an archive with no such endpoint it must + not displace the error that actually stopped the upload. + """ + f = tmp_path / "blob.dat" + f.write_bytes(b"data") + client = Mock() + client.post.return_value = {"upload_id": "upload-1", "parts": []} + client.delete.side_effect = _http_error(404) + with pytest.raises(RuntimeError, match="number of parts"): + for _ in multipart_upload( + client=client, + filepath=f, + asset_path="blob.dat", + init_fields={"dandiset": "000001"}, + ): + pass + + def test_validate_deep_zarr(tmp_path: Path) -> None: zarr_path = tmp_path / "foo.zarr" zarr.save(zarr_path, np.arange(1000), np.arange(1000, 0, -1))