fix: gate fused sync fast paths on full store sync capability; wrappers forward it

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.

changes/4206.bugfix.md

@@ -0,0 +1 @@
+Fixed `FusedCodecPipeline`'s gating of its synchronous fast paths: stores exposing only part of the sync surface (e.g. `set_sync` without `get_sync`) now fall back cleanly to the async path instead of failing mid-write, and `WrapperStore` now forwards `get_sync`/`set_sync`/`delete_sync` to the wrapped store so wrapped sync-capable stores keep the fast path. The capability decision uses a private, interim convention (`zarr.abc.store._store_supports_sync_io`) rather than new public API, pending a formal sync/async store architecture. Also fixed `LatencyStore`: synchronous reads and writes now pay the configured latency, `get_ranges`/`get_partial_values` no longer bypass latency injection, and derived stores (e.g. from `with_read_only`) keep a stochastic `(loc, scale)` latency configuration instead of freezing a single sample.

changes/4219.bugfix.md

@@ -0,0 +1,3 @@
+`DefaultChunkKeyEncoding.decode_chunk_key` now validates that a chunk key
+starts with the configured `c<separator>` prefix and raises `ValueError` for
+malformed keys, instead of silently decoding them incorrectly.

pyproject.toml

@@ -94,18 +94,18 @@ homepage = "https://github.com/zarr-developers/zarr-python"
 # pins deliberately, e.g. via dependabot or `uv lock --upgrade`.
 [dependency-groups]
 test = [
-    "coverage==7.14.3",
+    "coverage==7.15.2",
     "pytest==9.1.1",
     "pytest-asyncio==1.4.0",
     "pytest-cov==7.1.0",
     "pytest-accept==0.3.0",
     "numpydoc==1.10.0",
-    "hypothesis==6.155.7",
+    "hypothesis==6.160.0",
     "pytest-xdist==3.8.0",
     "pytest-benchmark==5.2.3",
     "pytest-codspeed==5.0.3",
-    "tomlkit==0.15.0",
-    "uv==0.11.26",
+    "tomlkit==0.15.1",
+    "uv==0.11.31",
 ]
 remote-tests = [
     {include-group = "test"},
@@ -121,15 +121,15 @@ release = [
 ]
 docs = [
     # Doc building
-    "mkdocs-material[imaging]==9.7.6",
+    "mkdocs-material[imaging]==9.7.7",
     "mkdocs==1.6.1",
-    "mkdocstrings==1.0.4",
+    "mkdocstrings==1.0.6",
     "mkdocstrings-python==2.0.5",
     "mike==2.2.0",
     "mkdocs-redirects==1.2.3",
-    "markdown-exec[ansi]==1.12.1",
+    "markdown-exec[ansi]==1.12.3",
     "griffe-inherited-docstrings==1.1.3",
-    "ruff==0.15.20",
+    "ruff==0.15.22",
     # Changelog generation
     {include-group = "release"},
     # Optional dependencies to run examples
@@ -143,7 +143,7 @@ dev = [
     {include-group = "remote-tests"},
     {include-group = "docs"},
     "universal-pathlib",
-    "mypy==2.1.0",
+    "mypy==2.3.0",
 ]
 
 [tool.coverage.report]

src/zarr/abc/store.py

@@ -662,6 +662,15 @@ def delete_sync(self) -> None: ...
 
 @runtime_checkable
 class SupportsGetSync(Protocol):
+    """Store protocol for synchronous reads (`get_sync`).
+
+    The store sync surface is all-or-nothing: a store implementing any of the
+    `*_sync` methods must implement all of them (`SupportsSyncStore`), because
+    consumers mix sync reads, writes, and deletes within one operation.
+    Capability-gated callers consult `_store_supports_sync_io` rather than the
+    individual protocols.
+    """
+
     def get_sync(
         self,
         key: str,
@@ -673,16 +682,52 @@ def get_sync(
 
 @runtime_checkable
 class SupportsSetSync(Protocol):
+    """Store protocol for synchronous writes (`set_sync`).
+
+    See `SupportsGetSync` for the all-or-nothing contract on the store sync
+    surface.
+    """
+
     def set_sync(self, key: str, value: Buffer) -> None: ...
 
 
 @runtime_checkable
 class SupportsDeleteSync(Protocol):
+    """Store protocol for synchronous deletes (`delete_sync`).
+
+    See `SupportsGetSync` for the all-or-nothing contract on the store sync
+    surface.
+    """
+
     def delete_sync(self, key: str) -> None: ...
 
 
 @runtime_checkable
-class SupportsSyncStore(SupportsGetSync, SupportsSetSync, SupportsDeleteSync, Protocol): ...
+class SupportsSyncStore(SupportsGetSync, SupportsSetSync, SupportsDeleteSync, Protocol):
+    """The full store sync surface: `get_sync`, `set_sync`, and `delete_sync`."""
+
+
+def _store_supports_sync_io(store: object) -> bool:
+    """Whether `store` can serve the full synchronous IO surface right now.
+
+    Structural membership in `SupportsSyncStore` is necessary but not always
+    sufficient: a store can present the `*_sync` methods while its ability to
+    run them depends on runtime state the type system cannot see. Wrapper
+    stores are the canonical case — `WrapperStore` delegates the sync methods
+    to the store it wraps, so they only work when the wrapped store is itself
+    sync-capable. Such stores opt out dynamically via a `_supports_sync_io`
+    attribute/property (absent means capable).
+
+    This is an interim, private convention pending a formal sync/async store
+    architecture — the store-side twin of the codec-side `_sync_capable`
+    convention consulted by `zarr.abc.codec._codec_supports_sync`.
+
+    Synchronous IO is all-or-nothing: consumers such as the fused codec
+    pipeline mix synchronous reads, writes, and deletes within one batch
+    (e.g. a partial-chunk write reads existing bytes and an all-fill chunk is
+    deleted), so a partial sync surface never satisfies this predicate.
+    """
+    return isinstance(store, SupportsSyncStore) and getattr(store, "_supports_sync_io", True)
 
 
 async def set_or_delete(byte_setter: ByteSetter, value: Buffer | None) -> None:

src/zarr/codecs/sharding.py

@@ -23,7 +23,7 @@
     RangeByteRequest,
     Store,
     SuffixByteRequest,
-    SupportsGetSync,
+    _store_supports_sync_io,
 )
 from zarr.codecs._deprecated_enum import _coerce_enum_input, _DeprecatedStrEnumMeta
 from zarr.codecs.bytes import BytesCodec
@@ -1721,7 +1721,7 @@ def _load_partial_shard_maybe_sync(
 
         shard_dict: ShardMutableMapping = {}
         store = byte_getter.store if hasattr(byte_getter, "store") else None
-        if isinstance(store, Store) and isinstance(store, SupportsGetSync):
+        if isinstance(store, Store) and _store_supports_sync_io(store):
             # External store: coalesce via get_ranges_sync (mirrors get_ranges).
             byte_ranges = [byte_range for _, byte_range in chunk_coord_byte_ranges]
             try:

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_key_encodings.py

@@ -79,7 +79,11 @@ def __post_init__(self) -> None:
     def decode_chunk_key(self, chunk_key: str) -> tuple[int, ...]:
         if chunk_key == "c":
             return ()
-        return tuple(map(int, chunk_key[1:].split(self.separator)))
+        # Strip the "c<sep>" prefix (e.g. "c/" or "c.") before splitting.
+        prefix = "c" + self.separator
+        if chunk_key.startswith(prefix):
+            return tuple(map(int, chunk_key[len(prefix) :].split(self.separator)))
+        raise ValueError(f"Invalid chunk key for default encoding: {chunk_key!r}")
 
     def encode_chunk_key(self, chunk_coords: tuple[int, ...]) -> str:
         return self.separator.join(map(str, ("c",) + chunk_coords))

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)