fix: minor correctness and hygiene fixes from the sync-pipeline audit

changes/4205.bugfix.md

@@ -0,0 +1,9 @@
+Fixed several small correctness issues from the codec-pipeline performance work: construction-time
+codec warnings (e.g. sharding's "disables partial reads" warning) no longer fire twice per array
+open — including for `FusedCodecPipeline`, which previously re-warned via its own codec-chain
+reconstruction and, on the async fallback path, on every decode/encode call; `concurrent_iter` now
+schedules its tasks eagerly, matching its documented contract; an invalid
+`codec_pipeline.max_workers` config/environment value now warns and falls back to the default
+instead of raising mid-read; and `FusedCodecPipeline`'s async fallback helpers now cancel
+already-spawned fetch/decode/write tasks instead of abandoning them in the background when one
+fails.

src/zarr/core/array.py

@@ -228,6 +228,12 @@ def create_codec_pipeline(metadata: ArrayMetadata, *, store: Store | None = None
             pass
 
     if isinstance(metadata, ArrayV3Metadata):
+        # The pipeline built here is a throwaway: `evolve_from_array_spec` below
+        # reconstructs codecs against the evolved spec. `from_codecs` is the
+        # chain's first construction, so its advisory warnings (e.g. sharding's
+        # "disables partial reads" warning) fire here; `evolve_from_array_spec`
+        # re-splits the same already-warned-about chain via
+        # `codecs_from_list_unchecked`, so it does not re-emit them.
         pipeline = get_pipeline_class().from_codecs(metadata.codecs)
         from zarr.core.metadata.v3 import RegularChunkGridMetadata
 

src/zarr/core/chunk_utils.py

@@ -238,7 +238,7 @@ class ChunkTransform:
     )
 
     def __post_init__(self) -> None:
-        from zarr.core.codec_pipeline import codecs_from_list
+        from zarr.core.codec_pipeline import codecs_from_list_unchecked
 
         # _codec_supports_sync, not a bare isinstance check: a codec can satisfy
         # the SupportsSyncCodec protocol structurally yet be unable to run
@@ -253,7 +253,11 @@ def __post_init__(self) -> None:
                 f"All codecs must implement SupportsSyncCodec. The following do not: {names}"
             )
 
-        aa, ab, bb = codecs_from_list(list(self.codecs))
+        # `ChunkTransform` is built from a codec chain that already went
+        # through `codecs_from_list` when the owning pipeline was constructed
+        # (see `FusedCodecPipeline.evolve_from_array_spec`), so re-splitting it
+        # here must not re-emit that chain's advisory warnings.
+        aa, ab, bb = codecs_from_list_unchecked(list(self.codecs))
         # SupportsSyncCodec was verified above; the cast is purely for mypy.
         self._aa_codecs = cast("tuple[SupportsSyncCodec[NDBuffer, NDBuffer], ...]", tuple(aa))
         self._ab_codec = cast("SupportsSyncCodec[NDBuffer, Buffer]", ab)

src/zarr/core/codec_pipeline.py

@@ -4,7 +4,7 @@
 import threading
 from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass, field
-from itertools import batched, pairwise
+from itertools import batched, chain, pairwise
 from typing import TYPE_CHECKING, Any, cast
 from warnings import warn
 
@@ -54,10 +54,23 @@ def _resolve_max_workers() -> int:
     """Helper for getting the maximum number of workers available to the `FusedCodecPipeline`"""
     import os as _os
 
+    default = _os.cpu_count() or 1
     cfg = config.get("codec_pipeline.max_workers", default=None)
     if cfg is None:
-        return _os.cpu_count() or 1
-    return max(1, int(cfg))
+        return default
+    try:
+        return max(1, int(cfg))
+    except (TypeError, ValueError):
+        # This value arrives via the config/env layer (e.g.
+        # `ZARR_CODEC_PIPELINE__MAX_WORKERS`), so tolerate bad input here
+        # instead of raising mid-read.
+        warn(
+            f"Ignoring invalid `codec_pipeline.max_workers` config value {cfg!r}; "
+            f"falling back to {default}.",
+            category=ZarrUserWarning,
+            stacklevel=2,
+        )
+        return default
 
 
 def _get_pool(max_workers: int) -> ThreadPoolExecutor:
@@ -169,6 +182,23 @@ def pipeline_supports_partial_encode(
     return isinstance(array_bytes_codec, ArrayBytesCodecPartialEncodeMixin)
 
 
+async def _cancel_and_drain(futures: Iterable[asyncio.Future[Any]]) -> None:
+    """Cancel every not-yet-done future/task and await its outcome.
+
+    Used to clean up work spawned by a drain loop (`asyncio.as_completed` +
+    `await`) when the loop exits early via exception. Without this, tasks
+    already spawned keep running unattended after the caller has moved on,
+    and an eventual failure surfaces as an unraisable "exception was never
+    retrieved" warning instead of being observed here.
+    """
+    pending = [f for f in futures if not f.done()]
+    if len(pending) == 0:
+        return
+    for f in pending:
+        f.cancel()
+    await asyncio.gather(*pending, return_exceptions=True)
+
+
 async def _fetch_and_decode_as_completed(
     batch: Sequence[tuple[ByteGetter | None, ArraySpec]],
     transform: ChunkTransform,
@@ -201,20 +231,29 @@ def _decode(buffer: Buffer | None, chunk_spec: ArraySpec) -> NDBuffer | None:
         _fetch,
         config.get("async.concurrency"),
     )
-    for fetch_coro in asyncio.as_completed(fetch_tasks):
-        idx, buffer = await fetch_coro
-        chunk_spec = batch[idx][1]
-        # Bridge both paths to asyncio.Future so the final collection loop
-        # can `await` uniformly without blocking the event loop. For the
-        # pool path that means `wrap_future` (not `pool.submit(...).result()`,
-        # which would block the loop thread for the duration of every decode
-        # — freezing any unrelated coroutines sharing this loop).
-        if pool is None:
-            decode_futures[idx].set_result(_decode(buffer, chunk_spec))
-        else:
-            decode_futures[idx] = asyncio.wrap_future(pool.submit(_decode, buffer, chunk_spec))
+    try:
+        for fetch_coro in asyncio.as_completed(fetch_tasks):
+            idx, buffer = await fetch_coro
+            chunk_spec = batch[idx][1]
+            # Bridge both paths to asyncio.Future so the final collection loop
+            # can `await` uniformly without blocking the event loop. For the
+            # pool path that means `wrap_future` (not `pool.submit(...).result()`,
+            # which would block the loop thread for the duration of every decode
+            # — freezing any unrelated coroutines sharing this loop).
+            if pool is None:
+                decode_futures[idx].set_result(_decode(buffer, chunk_spec))
+            else:
+                decode_futures[idx] = asyncio.wrap_future(pool.submit(_decode, buffer, chunk_spec))
 
-    return await asyncio.gather(*decode_futures)
+        return await asyncio.gather(*decode_futures)
+    finally:
+        # On the happy path every future here is already done, so this is a
+        # no-op; on failure it stops abandoned fetches/decodes from
+        # continuing to run unattended after this function has raised. A
+        # single call over both iterables (not two sequential calls) so that
+        # outer-task cancellation during the first drain can't skip the
+        # second, leaving its futures/tasks unobserved.
+        await _cancel_and_drain(chain(fetch_tasks, decode_futures))
 
 
 async def _encode_and_write_as_completed(
@@ -263,10 +302,20 @@ async def _write(idx: int, chunk_bytes: Buffer | None) -> None:
     # Kick off each chunk's write the instant its encode lands, so writes of
     # already-compressed chunks proceed while the rest are still encoding.
     write_tasks: list[asyncio.Task[None]] = []
-    for encode_coro in asyncio.as_completed(encode_futures):
-        idx, chunk_bytes = await encode_coro
-        write_tasks.append(asyncio.ensure_future(_write(idx, chunk_bytes)))
-    await asyncio.gather(*write_tasks)
+    try:
+        for encode_coro in asyncio.as_completed(encode_futures):
+            idx, chunk_bytes = await encode_coro
+            write_tasks.append(asyncio.ensure_future(_write(idx, chunk_bytes)))
+        await asyncio.gather(*write_tasks)
+    finally:
+        # On the happy path every future here is already done, so this is a
+        # no-op; on failure (an encode or a write raising) it stops
+        # already-spawned writes from continuing in the background after
+        # this function has raised. A single call over both iterables (not
+        # two sequential calls) so that outer-task cancellation during the
+        # first drain can't skip the second, leaving its futures/tasks
+        # unobserved.
+        await _cancel_and_drain(chain(write_tasks, encode_futures))
 
 
 async def _async_read_fallback(
@@ -468,7 +517,11 @@ class AsyncChunkTransform:
     _bb_codecs: tuple[BytesBytesCodec, ...] = field(init=False, repr=False, compare=False)
 
     def __post_init__(self) -> None:
-        aa, ab, bb = codecs_from_list(list(self.codecs))
+        # `AsyncChunkTransform` is (re)constructed per decode/encode call from a
+        # codec chain that already went through `codecs_from_list` when the
+        # pipeline itself was built, so re-splitting it here must not re-emit
+        # that chain's advisory warnings on every call.
+        aa, ab, bb = codecs_from_list_unchecked(list(self.codecs))
         self._aa_codecs = aa
         self._ab_codec = ab
         self._bb_codecs = bb
@@ -532,7 +585,19 @@ class BatchedCodecPipeline(CodecPipeline):
     batch_size: int
 
     def evolve_from_array_spec(self, array_spec: ArraySpec) -> Self:
-        return type(self).from_codecs(evolve_codecs(self, array_spec))
+        # Re-splits an already-`codecs_from_list`-validated (and warned-about)
+        # chain against the evolved spec, so this uses the quiet variant rather
+        # than routing through `from_codecs` (which would re-warn).
+        evolved_codecs = evolve_codecs(self, array_spec)
+        array_array_codecs, array_bytes_codec, bytes_bytes_codecs = codecs_from_list_unchecked(
+            evolved_codecs
+        )
+        return type(self)(
+            array_array_codecs=array_array_codecs,
+            array_bytes_codec=array_bytes_codec,
+            bytes_bytes_codecs=bytes_bytes_codecs,
+            batch_size=self.batch_size,
+        )
 
     @classmethod
     def from_codecs(cls, codecs: Iterable[Codec], *, batch_size: int | None = None) -> Self:
@@ -794,21 +859,44 @@ async def write(
 def codecs_from_list(
     codecs: Iterable[Codec],
 ) -> tuple[tuple[ArrayArrayCodec, ...], ArrayBytesCodec, tuple[BytesBytesCodec, ...]]:
+    """Split `codecs` into `(array_array, array_bytes, bytes_bytes)`, validating order.
+
+    Emits user-facing advisory warnings about the codec chain (e.g. sharding's
+    "disables partial reads" warning). Use this for the FIRST construction of a
+    codec chain from user-supplied codecs. Use `codecs_from_list_unchecked` when
+    re-splitting a chain that was already validated and warned about by a prior
+    `codecs_from_list` call (e.g. `evolve_from_array_spec` re-splitting the same
+    codecs against an evolved spec) — re-warning there would fire the same
+    advisory once per reconstruction instead of once per user-facing chain.
+    """
     from zarr.codecs.sharding import ShardingCodec
 
     codecs = tuple(codecs)  # materialize to avoid generator consumption issues
 
-    array_array: tuple[ArrayArrayCodec, ...] = ()
-    array_bytes_maybe: ArrayBytesCodec | None = None
-    bytes_bytes: tuple[BytesBytesCodec, ...] = ()
-
     if any(isinstance(codec, ShardingCodec) for codec in codecs) and len(codecs) > 1:
         warn(
             "Combining a `sharding_indexed` codec disables partial reads and "
             "writes, which may lead to inefficient performance.",
             category=ZarrUserWarning,
             stacklevel=3,
         )
+    return codecs_from_list_unchecked(codecs)
+
+
+def codecs_from_list_unchecked(
+    codecs: Iterable[Codec],
+) -> tuple[tuple[ArrayArrayCodec, ...], ArrayBytesCodec, tuple[BytesBytesCodec, ...]]:
+    """Split `codecs` into `(array_array, array_bytes, bytes_bytes)`, validating order.
+
+    Same structural validation as `codecs_from_list` (raises on bad codec
+    ordering or a missing/duplicate array-bytes codec) but does NOT emit
+    user-facing advisory warnings. See `codecs_from_list` for when to use each.
+    """
+    codecs = tuple(codecs)  # materialize to avoid generator consumption issues
+
+    array_array: tuple[ArrayArrayCodec, ...] = ()
+    array_bytes_maybe: ArrayBytesCodec | None = None
+    bytes_bytes: tuple[BytesBytesCodec, ...] = ()
 
     for prev_codec, cur_codec in pairwise((None, *codecs)):
         if isinstance(cur_codec, ArrayArrayCodec):
@@ -911,8 +999,11 @@ def from_codecs(cls, codecs: Iterable[Codec], *, batch_size: int | None = None)
         )
 
     def evolve_from_array_spec(self, array_spec: ArraySpec) -> Self:
+        # Re-splits an already-`codecs_from_list`-validated (and warned-about)
+        # chain against the evolved spec, so this uses the quiet variant to
+        # avoid re-emitting the same advisory warning on every array open.
         evolved_codecs = evolve_codecs(self.codecs, array_spec)
-        aa, ab, bb = codecs_from_list(evolved_codecs)
+        aa, ab, bb = codecs_from_list_unchecked(evolved_codecs)
 
         try:
             sync_transform: ChunkTransform | None = ChunkTransform(codecs=evolved_codecs)

src/zarr/core/common.py

@@ -93,34 +93,34 @@ def concurrent_iter[T: tuple[Any, ...], V](
     items: Iterable[T],
     func: Callable[..., Awaitable[V]],
     limit: int | None = None,
-) -> Iterator[asyncio.Task[V]]:
+) -> list[asyncio.Task[V]]:
     """Launch `func(*item)` for each item concurrently, returning the tasks.
 
     When `limit` is set, no more than `limit` calls are in flight at once.
     Tasks are returned in input order; callers that want completion order
     should wrap the result in `asyncio.as_completed`.
 
-    Note on `ensure_future`: when the result is passed to `asyncio.gather` or
-    `asyncio.as_completed`, those already wrap awaitables into tasks, so the
-    `ensure_future` here is redundant. It matters for callers that iterate and
-    await tasks one at a time — without eager scheduling, each coroutine would
-    only start when individually awaited, serializing the work and defeating
-    the semaphore. It also makes the return type honest (real `Task`s support
-    `.cancel()`, `.done()`, callbacks) rather than bare coroutines.
+    Every task is scheduled (via `ensure_future`) before this function
+    returns, not on first iteration of the result. That matters for callers
+    that await the returned tasks one at a time — without eager scheduling,
+    each coroutine would only start when individually awaited, serializing
+    the work and defeating the semaphore. It also makes the return type
+    honest (real `Task`s support `.cancel()`, `.done()`, callbacks) rather
+    than bare coroutines.
 
     See https://docs.python.org/3/library/asyncio-task.html#coroutines:
     "Note that simply calling a coroutine will not schedule it to be executed:"
     """
     if limit is None:
-        return (asyncio.ensure_future(func(*item)) for item in items)
+        return [asyncio.ensure_future(func(*item)) for item in items]
 
     sem = asyncio.Semaphore(limit)
 
     async def run(item: T) -> V:
         async with sem:
             return await func(*item)
 
-    return (asyncio.ensure_future(run(item)) for item in items)
+    return [asyncio.ensure_future(run(item)) for item in items]
 
 
 async def concurrent_map[T: tuple[Any, ...], V](

tests/benchmarks/test_e2e.py

@@ -63,7 +63,8 @@ def _data(shape: tuple[int]) -> np.ndarray:
     noise_level = 1
     pattern = (np.sin(np.linspace(0, 2 * np.pi, period)) * 50 + 128).round().astype(np.uint8)
     data = np.tile(pattern, int(np.ceil(n / period)))[:n].astype(np.int16)
-    data += np.random.randint(-noise_level, noise_level + 1, size=n, dtype=np.int16)
+    rng = np.random.default_rng(0)
+    data += rng.integers(-noise_level, noise_level + 1, size=n, dtype=np.int16)
     return np.clip(data, 0, 255).astype(np.uint8)
 
 
@@ -189,7 +190,7 @@ def test_read_array(
     get_data: Callable[[tuple[int]], np.ndarray | int],
 ) -> None:
     """
-    Test the time required to fill an array with a single value
+    Test the time required to read the entirety of an array
     """
     arr = create_array(
         bench_store,

tests/test_codecs/test_codecs.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import warnings
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any
 
@@ -22,7 +23,7 @@
 from zarr.core.metadata.v3 import ArrayV3Metadata
 from zarr.dtype import UInt8
 from zarr.errors import ZarrUserWarning
-from zarr.storage import StorePath
+from zarr.storage import MemoryStore, StorePath
 
 if TYPE_CHECKING:
     from zarr.abc.codec import Codec
@@ -375,6 +376,49 @@ def test_invalid_metadata_create_array() -> None:
         )
 
 
+@pytest.mark.parametrize(
+    "pipeline_path",
+    [
+        "zarr.core.codec_pipeline.BatchedCodecPipeline",
+        "zarr.core.codec_pipeline.FusedCodecPipeline",
+    ],
+)
+def test_sharding_warning_fires_once_per_open(pipeline_path: str) -> None:
+    """Construction-time codec warnings (e.g. sharding's partial-reads warning)
+    must fire exactly once per array open, not once per internal codec-chain
+    reconstruction.
+
+    `create_codec_pipeline` builds a throwaway pipeline via `from_codecs` (which
+    warns) and then calls `evolve_from_array_spec` on it, which re-splits the
+    (already-warned-about) codec chain against the evolved spec. That re-split
+    goes through `codecs_from_list_unchecked` rather than `codecs_from_list`, so
+    it does not re-emit the warning. `FusedCodecPipeline` additionally builds a
+    `ChunkTransform` (and, on the async fallback path, an `AsyncChunkTransform`
+    per call) from the same evolved codec chain, which must use the same quiet
+    variant.
+    """
+    with config.set({"codec_pipeline.path": pipeline_path}):
+        store = MemoryStore()
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore")
+            zarr.create_array(
+                store,
+                shape=(16, 16),
+                chunks=(16, 16),
+                dtype=np.dtype("uint8"),
+                fill_value=0,
+                serializer=ShardingCodec(chunk_shape=(8, 8)),
+                compressors=[GzipCodec()],
+            )
+
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter("always")
+            zarr.open_array(store, mode="r")
+
+        matches = [w for w in caught if "disables partial reads" in str(w.message)]
+        assert len(matches) == 1
+
+
 @pytest.mark.parametrize("store", ["local", "memory"], indirect=["store"])
 async def test_resize(store: Store) -> None:
     data = np.zeros((16, 18), dtype="uint16")