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

The fused write gate checked only SupportsSetSync, but write_sync also
needs get_sync (partial-chunk read-modify-write) and delete_sync
(all-fill chunk cleanup): a set-sync-only store passed the gate, wrote
some chunks, then died mid-batch with TypeError. And WrapperStore
forwarded no *_sync method, so every wrapped store (e.g. LatencyStore)
silently lost the sync fast path — latency benchmarks measured the async
fallback while claiming to measure the fused sync path.

Both gates now consult _store_supports_sync_io: structural membership in
SupportsSyncStore (the full get/set/delete sync surface) combined with a
per-instance _supports_sync_io opt-out (absent means capable). This is a
private, interim convention pending a formal sync/async store
architecture — the store-side twin of the codec-side _sync_capable
convention from #4179 — deliberately not new public API. WrapperStore
delegates the three sync methods and forwards the wrapped store's
capability, so wrapping a sync store keeps the fast path and wrapping an
async-only store falls back cleanly; LoggingStore logs the delegated
sync calls.

LatencyStore fixes: sync reads/writes now sleep the configured latency
on the worker thread; get_ranges/get_partial_values route through the
latency-injecting get instead of bypassing the wrapper; _with_store
passes the raw (loc, scale) latency config instead of a single sampled
float, so derived stores keep the distribution.

Assisted-by: ClaudeCode:claude-fable-5

Author: d-v-b

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.

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
@@ -1679,7 +1679,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/codec_pipeline.py

@@ -283,8 +283,8 @@ async def _async_read_fallback(
     then scatters each decoded chunk into `out` at its `out_selection`.
 
     Used by both `BatchedCodecPipeline.read_batch` (non-partial-decode
-    branch) and `FusedCodecPipeline.read` (when the store is not a
-    `SupportsGetSync` / sync transform is unavailable).
+    branch) and `FusedCodecPipeline.read` (when the store does not advertise
+    sync IO / sync transform is unavailable).
     """
 
     chunk_array_batch: list[NDBuffer | None]
@@ -345,8 +345,8 @@ async def _async_write_fallback(
     if encoding produced `None` or the chunk dropped).
 
     Used by both `BatchedCodecPipeline.write_batch` (non-partial-encode
-    branch) and `FusedCodecPipeline.write` (when the store is not a
-    `SupportsSetSync` / sync transform is unavailable).
+    branch) and `FusedCodecPipeline.write` (when the store does not advertise
+    sync IO / sync transform is unavailable).
     """
 
     if use_sync := (
@@ -1170,16 +1170,17 @@ async def read(
             return ()
 
         # Fast path: sync transform plus synchronous IO. For StorePath the gate
-        # is on the STORE's sync support (StorePath always has a get_sync
-        # method, but it only works when its store does); for other byte
-        # getters (e.g. the sharding codec's in-memory _ShardingByteGetter) the
-        # SyncByteGetter protocol is the gate.
-        from zarr.abc.store import SupportsGetSync, SyncByteGetter
+        # is the STORE's sync-IO capability (`_store_supports_sync_io`) (StorePath always has a
+        # get_sync method, but it only works when its store implements the full
+        # sync surface); for other byte getters (e.g. the sharding codec's
+        # in-memory _ShardingByteGetter) the SyncByteGetter protocol is the
+        # gate.
+        from zarr.abc.store import SyncByteGetter, _store_supports_sync_io
         from zarr.storage._common import StorePath
 
         first_bg = batch[0][0]
         if self.sync_transform is not None and (
-            (isinstance(first_bg, StorePath) and isinstance(first_bg.store, SupportsGetSync))
+            (isinstance(first_bg, StorePath) and _store_supports_sync_io(first_bg.store))
             or (not isinstance(first_bg, StorePath) and isinstance(first_bg, SyncByteGetter))
         ):
             # One thread hop for the WHOLE batch — not per chunk, so the fused
@@ -1233,14 +1234,17 @@ async def write(
             return
 
         # Fast path: sync transform plus synchronous IO. Mirrors `read`: gate
-        # StorePath on the store's sync support, other byte setters (e.g. the
-        # sharding codec's in-memory _ShardingByteSetter) on SyncByteSetter.
-        from zarr.abc.store import SupportsSetSync, SyncByteSetter
+        # StorePath on the store's sync-IO capability (`_store_supports_sync_io`) — write_sync
+        # needs the FULL sync surface (get_sync for partial-chunk
+        # read-modify-write, delete_sync for all-fill chunks), not just
+        # set_sync — and other byte setters (e.g. the sharding codec's
+        # in-memory _ShardingByteSetter) on SyncByteSetter.
+        from zarr.abc.store import SyncByteSetter, _store_supports_sync_io
         from zarr.storage._common import StorePath
 
         first_bs = batch[0][0]
         if self.sync_transform is not None and (
-            (isinstance(first_bs, StorePath) and isinstance(first_bs.store, SupportsSetSync))
+            (isinstance(first_bs, StorePath) and _store_supports_sync_io(first_bs.store))
             or (not isinstance(first_bs, StorePath) and isinstance(first_bs, SyncByteSetter))
         ):
             # One thread hop for the whole batch; see the matching comment in

src/zarr/storage/_logging.py

@@ -204,6 +204,27 @@ async def delete(self, key: str) -> None:
         with self.log(key):
             return await self._store.delete(key=key)
 
+    def get_sync(
+        self,
+        key: str,
+        *,
+        prototype: BufferPrototype | None = None,
+        byte_range: ByteRequest | None = None,
+    ) -> Buffer | None:
+        # docstring inherited
+        with self.log(key):
+            return super().get_sync(key, prototype=prototype, byte_range=byte_range)
+
+    def set_sync(self, key: str, value: Buffer) -> None:
+        # docstring inherited
+        with self.log(key):
+            return super().set_sync(key, value)
+
+    def delete_sync(self, key: str) -> None:
+        # docstring inherited
+        with self.log(key):
+            return super().delete_sync(key)
+
     async def list(self) -> AsyncGenerator[str, None]:
         # docstring inherited
         with self.log():

src/zarr/storage/_wrapper.py

@@ -11,7 +11,13 @@
     from zarr.abc.store import ByteRequest
     from zarr.core.buffer import BufferPrototype
 
-from zarr.abc.store import Store
+from zarr.abc.store import (
+    Store,
+    SupportsDeleteSync,
+    SupportsGetSync,
+    SupportsSetSync,
+    _store_supports_sync_io,
+)
 
 
 class WrapperStore[T_Store: Store](Store):
@@ -149,6 +155,40 @@ def supports_writes(self) -> bool:
     def supports_deletes(self) -> bool:
         return self._store.supports_deletes
 
+    @property
+    def _supports_sync_io(self) -> bool:
+        # The delegating `*_sync` methods below make every wrapper structurally
+        # satisfy `SupportsSyncStore`; whether they can actually run depends on
+        # the wrapped store, so forward its capability (see
+        # `zarr.abc.store._store_supports_sync_io`).
+        return _store_supports_sync_io(self._store)
+
+    def get_sync(
+        self,
+        key: str,
+        *,
+        prototype: BufferPrototype | None = None,
+        byte_range: ByteRequest | None = None,
+    ) -> Buffer | None:
+        """Forward `get_sync` to the wrapped store."""
+        if not isinstance(self._store, SupportsGetSync):
+            raise TypeError(f"Store {type(self._store).__name__} does not support synchronous get.")
+        return self._store.get_sync(key, prototype=prototype, byte_range=byte_range)  # type: ignore[unreachable]
+
+    def set_sync(self, key: str, value: Buffer) -> None:
+        """Forward `set_sync` to the wrapped store."""
+        if not isinstance(self._store, SupportsSetSync):
+            raise TypeError(f"Store {type(self._store).__name__} does not support synchronous set.")
+        self._store.set_sync(key, value)  # type: ignore[unreachable]
+
+    def delete_sync(self, key: str) -> None:
+        """Forward `delete_sync` to the wrapped store."""
+        if not isinstance(self._store, SupportsDeleteSync):
+            raise TypeError(
+                f"Store {type(self._store).__name__} does not support synchronous delete."
+            )
+        self._store.delete_sync(key)  # type: ignore[unreachable]
+
     async def delete(self, key: str) -> None:
         await self._store.delete(key)
 

src/zarr/testing/store.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import pickle
+import time
 from abc import abstractmethod
 from typing import TYPE_CHECKING, Self
 
@@ -10,6 +11,7 @@
 from zarr.storage import WrapperStore
 
 if TYPE_CHECKING:
+    from collections.abc import AsyncIterator, Iterable, Sequence
     from typing import Any
 
     from zarr.core.buffer.core import BufferPrototype
@@ -653,7 +655,10 @@ def set_latency(self) -> float:
         return max(0.0, np.random.normal(loc=self._set_latency[0], scale=self._set_latency[1]))
 
     def _with_store(self, store: Store) -> Self:
-        return type(self)(store, get_latency=self.get_latency, set_latency=self.set_latency)
+        # Pass the raw latency config, not the sampled `get_latency`/`set_latency`
+        # properties — sampling would freeze a `(loc, scale)` distribution into
+        # one fixed float on derived stores (e.g. via `with_read_only`).
+        return type(self)(store, get_latency=self._get_latency, set_latency=self._set_latency)
 
     async def set(self, key: str, value: Buffer) -> None:
         """
@@ -698,3 +703,76 @@ async def get(
         """
         await asyncio.sleep(self.get_latency)
         return await self._store.get(key, prototype=prototype, byte_range=byte_range)
+
+    def get_sync(
+        self,
+        key: str,
+        *,
+        prototype: BufferPrototype | None = None,
+        byte_range: ByteRequest | None = None,
+    ) -> Buffer | None:
+        """Add latency to `get_sync`.
+
+        Sleeps `self.get_latency` on the calling thread (the sync path runs on
+        worker threads, not the event loop) before delegating to the wrapped
+        store.
+        """
+        time.sleep(self.get_latency)
+        return super().get_sync(key, prototype=prototype, byte_range=byte_range)
+
+    def set_sync(self, key: str, value: Buffer) -> None:
+        """Add latency to `set_sync`.
+
+        Sleeps `self.set_latency` on the calling thread (the sync path runs on
+        worker threads, not the event loop) before delegating to the wrapped
+        store.
+        """
+        time.sleep(self.set_latency)
+        super().set_sync(key, value)
+
+    async def get_ranges(
+        self,
+        key: str,
+        byte_ranges: Sequence[ByteRequest | None],
+        *,
+        prototype: BufferPrototype,
+        max_concurrency: int | None = None,
+        max_gap_bytes: int | None = None,
+        max_coalesced_bytes: int | None = None,
+    ) -> AsyncIterator[Sequence[tuple[int, Buffer | None]]]:
+        """Byte-range reads built on `self.get`, so each fetch pays latency.
+
+        Routes through the coalescing `Store.get_ranges` default instead of the
+        `WrapperStore` delegation, which would bypass this wrapper's `get` and
+        therefore the synthetic latency. `None` for a coalescing kwarg means
+        "use the `Store` default".
+        """
+        kwargs: dict[str, int] = {}
+        if max_concurrency is not None:
+            kwargs["max_concurrency"] = max_concurrency
+        if max_gap_bytes is not None:
+            kwargs["max_gap_bytes"] = max_gap_bytes
+        if max_coalesced_bytes is not None:
+            kwargs["max_coalesced_bytes"] = max_coalesced_bytes
+        async for group in Store.get_ranges(self, key, byte_ranges, prototype=prototype, **kwargs):
+            yield group
+
+    async def get_partial_values(
+        self,
+        prototype: BufferPrototype,
+        key_ranges: Iterable[tuple[str, ByteRequest | None]],
+    ) -> list[Buffer | None]:
+        """Partial-value reads built on `self.get`, so each fetch pays latency.
+
+        Issues one `self.get` per `(key, byte_range)` pair instead of the
+        `WrapperStore` delegation, which would bypass this wrapper's `get` and
+        therefore the synthetic latency.
+        """
+        return list(
+            await asyncio.gather(
+                *(
+                    self.get(key, prototype=prototype, byte_range=byte_range)
+                    for key, byte_range in key_ranges
+                )
+            )
+        )

tests/test_codec_pipeline_suite.py

@@ -9,8 +9,8 @@
 Each test also runs over a *store axis* that exercises both code paths the
 synchronous pipelines branch on:
 
-* ``sync``  -> ``MemoryStore`` (supports ``get_sync``/``set_sync``: fast path)
-* ``async`` -> ``LatencyStore(MemoryStore())`` (NOT sync-capable: async fallback)
+* ``sync``  -> ``MemoryStore`` (full sync surface: fast path)
+* ``async`` -> ``_NoSyncIOStore(MemoryStore())`` (NOT sync-capable: async fallback)
 
 The async axis is deliberate: a regression that only affects the async fallback
 of the default pipeline (e.g. a codec-spec-evolution bug that surfaces only on
@@ -50,14 +50,22 @@
 STORE_KINDS = ["sync", "async"]
 
 
+class _NoSyncIOStore(LatencyStore):
+    """An in-memory store that advertises no sync IO capability, so a
+    synchronous pipeline must fall back to its async path. (A plain wrapper
+    won't do: `WrapperStore` forwards the wrapped store's sync capability.)"""
+
+    @property
+    def _supports_sync_io(self) -> bool:
+        return False
+
+
 def _make_store(kind: str) -> Store:
     if kind == "sync":
         # MemoryStore supports get_sync/set_sync -> synchronous fast path.
         return MemoryStore()
     if kind == "async":
-        # LatencyStore is NOT SupportsGetSync/SupportsSetSync, so a synchronous
-        # pipeline must fall back to its async path. Zero latency keeps it fast.
-        return LatencyStore(MemoryStore(), get_latency=0.0, set_latency=0.0)
+        return _NoSyncIOStore(MemoryStore(), get_latency=0.0, set_latency=0.0)
     raise AssertionError(kind)