use protocols for new sync behavior

Author: d-v-b

src/zarr/abc/codec.py

@@ -2,7 +2,7 @@
 
 from abc import abstractmethod
 from collections.abc import Mapping
-from typing import TYPE_CHECKING, Generic, TypeGuard, TypeVar
+from typing import TYPE_CHECKING, Generic, Protocol, TypeGuard, TypeVar, runtime_checkable
 
 from typing_extensions import ReadOnly, TypedDict
 
@@ -32,6 +32,7 @@
     "CodecInput",
     "CodecOutput",
     "CodecPipeline",
+    "SupportsSyncCodec",
 ]
 
 CodecInput = TypeVar("CodecInput", bound=NDBuffer | Buffer)
@@ -59,6 +60,19 @@ def _check_codecjson_v2(data: object) -> TypeGuard[CodecJSON_V2[str]]:
 """The widest type of JSON-like input that could specify a codec."""
 
 
+@runtime_checkable
+class SupportsSyncCodec(Protocol):
+    """Protocol for codecs that support synchronous encode/decode."""
+
+    def _decode_sync(
+        self, chunk_data: NDBuffer | Buffer, chunk_spec: ArraySpec
+    ) -> NDBuffer | Buffer: ...
+
+    def _encode_sync(
+        self, chunk_data: NDBuffer | Buffer, chunk_spec: ArraySpec
+    ) -> NDBuffer | Buffer | None: ...
+
+
 class BaseCodec(Metadata, Generic[CodecInput, CodecOutput]):
     """Generic base class for codecs.
 
@@ -137,21 +151,6 @@ def validate(
             The array chunk grid
         """
 
-    def _decode_sync(self, chunk_data: CodecOutput, chunk_spec: ArraySpec) -> CodecInput:
-        """Synchronously decode a single chunk. Override in subclasses to enable
-        sync codec pipeline support."""
-        raise NotImplementedError  # pragma: no cover
-
-    def _encode_sync(self, chunk_data: CodecInput, chunk_spec: ArraySpec) -> CodecOutput | None:
-        """Synchronously encode a single chunk. Override in subclasses to enable
-        sync codec pipeline support."""
-        raise NotImplementedError  # pragma: no cover
-
-    @property
-    def supports_sync(self) -> bool:
-        """Whether this codec has synchronous encode/decode implementations."""
-        return type(self)._decode_sync is not BaseCodec._decode_sync
-
     async def _decode_single(self, chunk_data: CodecOutput, chunk_spec: ArraySpec) -> CodecInput:
         raise NotImplementedError  # pragma: no cover
 
@@ -491,7 +490,7 @@ def supports_sync_io(self) -> bool:
         """Whether this pipeline can run read/write entirely on the calling thread.
 
         True when:
-        - All codecs support synchronous encode/decode (_decode_sync/_encode_sync)
+        - All codecs implement ``SupportsSyncCodec``
         - The pipeline's read_sync/write_sync methods are implemented
 
         Checked by ``Array._can_use_sync_path()`` to decide whether to bypass

src/zarr/abc/store.py

@@ -16,7 +16,14 @@
 
     from zarr.core.buffer import Buffer, BufferPrototype
 
-__all__ = ["ByteGetter", "ByteSetter", "Store", "set_or_delete"]
+__all__ = [
+    "ByteGetter",
+    "ByteSetter",
+    "Store",
+    "SyncByteGetter",
+    "SyncByteSetter",
+    "set_or_delete",
+]
 
 
 @dataclass
@@ -524,61 +531,6 @@ def supports_partial_writes(self) -> Literal[False]:
         """
         return False
 
-    # -----------------------------------------------------------------------
-    # Synchronous IO interface (opt-in)
-    #
-    # These methods enable the codec pipeline to bypass the event loop
-    # entirely for store IO. The default implementations raise
-    # NotImplementedError; stores that wrap fundamentally synchronous
-    # operations (MemoryStore, LocalStore) override them with direct
-    # implementations. Remote/cloud stores (FsspecStore) leave them as-is
-    # and remain async-only.
-    # -----------------------------------------------------------------------
-
-    @property
-    def supports_sync(self) -> bool:
-        """Whether this store has native synchronous get/set/delete methods.
-
-        When True, the codec pipeline's ``read_sync`` / ``write_sync`` will
-        call ``get_sync`` / ``set_sync`` / ``delete_sync`` directly on the
-        calling thread, avoiding the event loop overhead of the async
-        equivalents.
-
-        Subclasses that override the sync methods below should also override
-        this property to return True.
-        """
-        return False
-
-    def get_sync(
-        self,
-        key: str,
-        prototype: BufferPrototype,
-        byte_range: ByteRequest | None = None,
-    ) -> Buffer | None:
-        """Synchronous version of ``get()``.
-
-        Called by the codec pipeline's ``read_sync`` to fetch chunk bytes without
-        going through the event loop. Only called when ``supports_sync`` is
-        True, so the default ``NotImplementedError`` is never hit in practice.
-        """
-        raise NotImplementedError
-
-    def set_sync(self, key: str, value: Buffer) -> None:
-        """Synchronous version of ``set()``.
-
-        Called by the codec pipeline's ``write_sync`` to persist encoded chunk
-        bytes without going through the event loop.
-        """
-        raise NotImplementedError
-
-    def delete_sync(self, key: str) -> None:
-        """Synchronous version of ``delete()``.
-
-        Called by the codec pipeline's ``write_sync`` when a chunk should be
-        removed (e.g. an empty chunk with ``write_empty_chunks=False``).
-        """
-        raise NotImplementedError
-
     @property
     @abstractmethod
     def supports_listing(self) -> bool:
@@ -755,6 +707,24 @@ async def delete(self) -> None: ...
     async def set_if_not_exists(self, default: Buffer) -> None: ...
 
 
+@runtime_checkable
+class SyncByteGetter(Protocol):
+    """Protocol for stores that support synchronous byte reads."""
+
+    def get_sync(
+        self, prototype: BufferPrototype, byte_range: ByteRequest | None = None
+    ) -> Buffer | None: ...
+
+
+@runtime_checkable
+class SyncByteSetter(SyncByteGetter, Protocol):
+    """Protocol for stores that support synchronous byte reads, writes, and deletes."""
+
+    def set_sync(self, value: Buffer) -> None: ...
+
+    def delete_sync(self) -> None: ...
+
+
 async def set_or_delete(byte_setter: ByteSetter, value: Buffer | None) -> None:
     """Set or delete a value in a byte setter
 

src/zarr/codecs/sharding.py

@@ -16,6 +16,7 @@
     ArrayBytesCodecPartialEncodeMixin,
     Codec,
     CodecPipeline,
+    SupportsSyncCodec,
 )
 from zarr.abc.store import (
     ByteGetter,
@@ -968,15 +969,15 @@ def _decode_shard_index_sync(
             spec = bb.resolve_metadata(spec)
 
         # Decode: reverse bb, then ab, then reverse aa
-        chunk_bytes: Buffer = index_bytes
+        bb_out: Any = index_bytes
         for bb_codec, s in reversed(bb_with_spec):
-            chunk_bytes = bb_codec._decode_sync(chunk_bytes, s)
-        chunk_array: NDBuffer = ab_codec._decode_sync(chunk_bytes, ab_spec)
+            bb_out = cast("SupportsSyncCodec", bb_codec)._decode_sync(bb_out, s)
+        ab_out: Any = cast("SupportsSyncCodec", ab_codec)._decode_sync(bb_out, ab_spec)
         for aa_codec, s in reversed(aa_with_spec):
-            chunk_array = aa_codec._decode_sync(chunk_array, s)
+            ab_out = cast("SupportsSyncCodec", aa_codec)._decode_sync(ab_out, s)
 
-        assert chunk_array is not None
-        return _ShardIndex(chunk_array.as_numpy_array())
+        assert ab_out is not None
+        return _ShardIndex(ab_out.as_numpy_array())
 
     def _encode_shard_index_sync(self, index: _ShardIndex) -> Buffer:
         """Encode shard index synchronously by running index codecs inline."""
@@ -986,25 +987,25 @@ def _encode_shard_index_sync(self, index: _ShardIndex) -> Buffer:
 
         aa_codecs, ab_codec, bb_codecs = codecs_from_list(list(self.index_codecs))
 
-        aa_out: NDBuffer | None = get_ndbuffer_class().from_numpy_array(index.offsets_and_lengths)
+        aa_out: Any = get_ndbuffer_class().from_numpy_array(index.offsets_and_lengths)
 
         # Encode: aa forward, then ab, then bb forward
         spec = index_chunk_spec
         for aa_codec in aa_codecs:
             assert aa_out is not None
-            aa_out = aa_codec._encode_sync(aa_out, spec)
+            aa_out = cast("SupportsSyncCodec", aa_codec)._encode_sync(aa_out, spec)
             spec = aa_codec.resolve_metadata(spec)
         assert aa_out is not None
-        chunk_bytes = ab_codec._encode_sync(aa_out, spec)
+        bb_out: Any = cast("SupportsSyncCodec", ab_codec)._encode_sync(aa_out, spec)
         spec = ab_codec.resolve_metadata(spec)
         for bb_codec in bb_codecs:
-            assert chunk_bytes is not None
-            chunk_bytes = bb_codec._encode_sync(chunk_bytes, spec)
+            assert bb_out is not None
+            bb_out = cast("SupportsSyncCodec", bb_codec)._encode_sync(bb_out, spec)
             spec = bb_codec.resolve_metadata(spec)
 
-        assert chunk_bytes is not None
-        assert isinstance(chunk_bytes, Buffer)
-        return chunk_bytes
+        assert bb_out is not None
+        assert isinstance(bb_out, Buffer)
+        return bb_out
 
     async def _decode_shard_index(
         self, index_bytes: Buffer, chunks_per_shard: tuple[int, ...]

src/zarr/core/array.py

@@ -1978,12 +1978,12 @@ def _can_use_sync_path(self) -> bool:
 
         Two conditions must hold:
 
-        1. The codec pipeline supports fully synchronous IO (all codecs in
-           the chain have _decode_sync/_encode_sync). This is True for
+        1. The codec pipeline supports fully synchronous IO (all codecs
+           implement ``SupportsSyncCodec``). This is True for
            BatchedCodecPipeline when all codecs support sync.
 
-        2. The store supports synchronous operations (MemoryStore, LocalStore).
-           Remote stores like FsspecStore remain async-only.
+        2. The store supports synchronous operations (has ``get_sync``).
+           MemoryStore and LocalStore provide this; remote stores do not.
 
         When both hold, the selection methods below call
         _get_selection_sync / _set_selection_sync directly, running the
@@ -1992,10 +1992,8 @@ def _can_use_sync_path(self) -> bool:
         is used automatically.
         """
         pipeline = self.async_array.codec_pipeline
-        store_path = self.async_array.store_path
-        return getattr(pipeline, "supports_sync_io", False) and getattr(
-            store_path, "supports_sync", False
-        )
+        store = self.async_array.store_path.store
+        return getattr(pipeline, "supports_sync_io", False) and hasattr(store, "get_sync")
 
     @classmethod
     @deprecated("Use zarr.create_array instead.", category=ZarrDeprecationWarning)