@@ -1 +1,5 @@
-Add protocols for stores that support byte-range-writes. This is necessary to support in-place writes of sharded arrays.
+Add the `SupportsSetRange` protocol for stores that support writing to a byte range within an existing value, implemented by `LocalStore` and `MemoryStore`. This is necessary to support in-place writes of sharded arrays (e.g. writing a single subchunk without rewriting the entire shard).
+
+Byte-range writes are exposed as an opt-in protocol rather than a method on the `Store` ABC. Only a few stores can perform them natively, and most cannot. A universal method with a read-modify-write fallback (as in the Rust `zarrs` crate) would let every store participate, but for the motivating use case that fallback would silently rewrite an entire shard, defeating the purpose. The opt-in protocol keeps the cost model honest and keeps `set_range` out of the signatures of stores that will never support it; any fallback strategy is left to the caller (the sharding codec). Stores satisfy the protocol structurally, so `GpuMemoryStore` (which has no use case for in-place GPU byte-range writes) disclaims it and is correctly reported as unsupported by `isinstance`.
+
+It is entirely the caller's responsibility to ensure consistency: concurrent writes to overlapping ranges are order-dependent, `set_range` racing against `set`/`delete` is a race, and writes are not guaranteed to be atomic with respect to a process crash. A write that does not fit within the existing value raises `ValueError` consistently across implementations.
@@ -788,16 +788,55 @@ async def delete(self) -> None: ...
     async def set_if_not_exists(self, default: Buffer) -> None: ...
 
 
+# Design note: byte-range writes are exposed as an opt-in protocol rather than a
+# method on the `Store` ABC. Only a few stores can do them natively (`LocalStore`,
+# `MemoryStore`); most (cloud, zip, read-only) cannot. A universal `Store.set_range`
+# with a read-modify-write fallback (as in the Rust `zarrs` crate's
+# `WritableStorageTraits::set_partial` + `supports_set_partial`) would let every
+# store participate, but for our motivating use case — writing one subchunk without
+# rewriting the whole shard — that fallback is a footgun: it would silently rewrite
+# an entire (possibly multi-GB) shard, defeating the purpose while appearing to
+# succeed. An opt-in protocol instead keeps the cost model honest (a store either
+# supports cheap ranged writes or doesn't advertise the capability at all) and keeps
+# `set_range` out of the signatures of stores that will never support it.
+#
+# Stores satisfy this protocol *structurally* (by defining the methods), not by
+# nominal inheritance, so a subclass can disclaim it by setting the methods to `None`
+# (see `GpuMemoryStore`). Any read-modify-write fallback strategy belongs in the
+# caller (the sharding codec), which already has to decide between in-place and
+# buffer-and-rewrite — mirroring the zarrs layering (storage writes bytes, codec owns
+# strategy) without making every store carry the method. If broad-backend partial
+# encoding is wanted later, adding a `supports_set_range()` capability flag plus a
+# codec-level fallback is an additive change that does not require retrofitting stores.
 @runtime_checkable
 class SupportsSetRange(Protocol):
     """Protocol for stores that support writing to a byte range within an existing value.
 
-    Overwrites ``len(value)`` bytes starting at byte offset ``start`` within the
-    existing stored value for ``key``. The key must already exist and the write
-    must fit within the existing value (i.e., ``start + len(value) <= len(existing)``).
-
-    Behavior when the write extends past the end of the existing value is
-    implementation-specific and should not be relied upon.
+    Overwrites `len(value)` bytes starting at byte offset `start` within the
+    existing stored value for `key`. The key must already exist and the write
+    must fit within the existing value (i.e., `start + len(value) <= len(existing)`);
+    a write that does not fit raises `ValueError`.
+
+    Concurrency and atomicity
+    -------------------------
+    **It is entirely the caller's responsibility to ensure consistency.** Any
+    coordination needed to keep stored values consistent must be arranged by the
+    caller. In particular:
+
+    - Concurrent `set_range` calls that write to **disjoint** byte ranges of the
+      same key are safe.
+    - Concurrent `set_range` calls that write to **overlapping** ranges of the same
+      key have order-dependent, unspecified results. The caller must serialize them.
+    - A `set_range` racing against a `set` or `delete` on the same key is a race
+      condition, just as concurrent `set` calls are. The caller must serialize these.
+    - Writes are **not** guaranteed to be atomic with respect to a process crash:
+      a crash mid-write may leave the value partially updated. The caller is
+      responsible for any durability or recovery guarantees it requires.
+
+    What an implementation does to honor (or fall short of) this contract — locking,
+    atomic replacement, and so on — is documented on the implementing store, not here.
+    The intended consumer (the sharding codec writing inner chunks of deterministic
+    compressed size) coordinates writes so that they target disjoint ranges.
     """
 
     async def set_range(self, key: str, value: Buffer, start: int) -> None: ...
 
@@ -18,11 +18,11 @@
     RangeByteRequest,
     Store,
     SuffixByteRequest,
-    SupportsSetRange,
 )
 from zarr.core.buffer import Buffer
 from zarr.core.buffer.core import default_buffer_prototype
 from zarr.core.common import AccessModeLiteral, concurrent_map
+from zarr.storage._utils import _check_set_range_bounds
 
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator, Iterable, Iterator
@@ -96,6 +96,12 @@ def _put_range(path: Path, value: Buffer, start: int) -> None:
     """Write bytes at a specific offset within an existing file."""
     view = value.as_buffer_like()
     with path.open("r+b") as f:
+        # Validate bounds before writing: a bare seek+write would silently extend the
+        # file (zero-filling any gap), but the SupportsSetRange contract requires the
+        # write to fit within the existing value, so we fail consistently with
+        # MemoryStore instead.
+        existing_length = f.seek(0, os.SEEK_END)
+        _check_set_range_bounds(existing_length, start, len(value))
         f.seek(start)
         f.write(view)
 
@@ -108,7 +114,7 @@ def _put(path: Path, value: Buffer, exclusive: bool = False) -> int:
         return f.write(view)
 
 
-class LocalStore(Store, SupportsSetRange):
+class LocalStore(Store):
     """
     Store for the local file system.
 
 
@@ -7,11 +7,12 @@
 from logging import getLogger
 from typing import TYPE_CHECKING, Any, Self
 
-from zarr.abc.store import ByteRequest, Store, SupportsSetRange
+from zarr.abc.store import ByteRequest, Store
 from zarr.core.buffer import Buffer, gpu
 from zarr.core.buffer.core import default_buffer_prototype
 from zarr.core.common import concurrent_map
 from zarr.storage._utils import (
+    _check_set_range_bounds,
     _join_paths,
     _normalize_byte_range_index,
     normalize_path,
@@ -27,7 +28,7 @@
 logger = getLogger(__name__)
 
 
-class MemoryStore(Store, SupportsSetRange):
+class MemoryStore(Store):
     """
     Store for local memory.
 
@@ -202,6 +203,7 @@ async def delete(self, key: str) -> None:
     def _set_range_impl(self, key: str, value: Buffer, start: int) -> None:
         buf = self._store_dict[key]
         target = buf.as_numpy_array()
+        _check_set_range_bounds(len(target), start, len(value))
         if not target.flags.writeable:
             target = target.copy()
             self._store_dict[key] = buf.__class__(target)
@@ -566,6 +568,19 @@ async def set(self, key: str, value: Buffer, byte_range: tuple[int, int] | None
         gpu_value = value if isinstance(value, gpu.Buffer) else gpu.Buffer.from_buffer(value)
         await super().set(key, gpu_value, byte_range=byte_range)
 
+    # ``GpuMemoryStore`` deliberately does not support byte-range writes, so it must
+    # not satisfy ``SupportsSetRange``. The inherited ``MemoryStore`` implementation
+    # mutates a *host* copy of the GPU buffer (via ``as_numpy_array``) and would
+    # silently lose the write, and there is no use case for in-place byte-range writes
+    # into GPU memory (the intended ``set_range`` consumer targets cpu/local storage).
+    # Disclaiming the inherited methods by setting them to ``None`` makes
+    # ``isinstance(gpu_store, SupportsSetRange)`` return ``False`` (``runtime_checkable``
+    # treats a ``None`` attribute as "method absent"). This works because
+    # ``MemoryStore`` satisfies the protocol structurally rather than by nominal
+    # inheritance; mirrors the ``__hash__ = None`` idiom.
+    set_range = None  # type: ignore[assignment]
+    set_range_sync = None  # type: ignore[assignment]
+
 
 # -----------------------------------------------------------------------------
 # ManagedMemoryStore and its registry
@@ -860,6 +875,14 @@ async def set_if_not_exists(self, key: str, value: Buffer) -> None:
         # docstring inherited
         return await super().set_if_not_exists(_join_paths([self.path, key]), value)
 
+    async def set_range(self, key: str, value: Buffer, start: int) -> None:
+        # docstring inherited
+        return await super().set_range(_join_paths([self.path, key]), value, start)
+
+    def set_range_sync(self, key: str, value: Buffer, start: int) -> None:
+        # docstring inherited
+        return super().set_range_sync(_join_paths([self.path, key]), value, start)
+
     async def delete(self, key: str) -> None:
         # docstring inherited
         return await super().delete(_join_paths([self.path, key]))
 
@@ -160,6 +160,23 @@ def _normalize_byte_range_index(data: Buffer, byte_range: ByteRequest | None) ->
     return (start, stop)
 
 
+def _check_set_range_bounds(existing_length: int, start: int, value_length: int) -> None:
+    """
+    Validate that a ``set_range`` write fits within an existing value.
+
+    Stores implementing ``SupportsSetRange`` use this so the out-of-bounds case fails
+    the same way everywhere (a clear ``ValueError``) rather than silently extending the
+    value (as a file seek+write would) or raising an opaque numpy shape error.
+    """
+    if start < 0:
+        raise ValueError(f"set_range start must be non-negative, got {start}.")
+    if start + value_length > existing_length:
+        raise ValueError(
+            f"set_range write of {value_length} bytes at offset {start} does not fit "
+            f"within the existing value of length {existing_length}."
+        )
+
+
 def _join_paths(paths: Iterable[str]) -> str:
     """
     Filter out instances of '' and join the remaining strings with '/'.
 
@@ -213,6 +213,23 @@ def test_set_range_sync(
         assert result is not None
         assert result.to_bytes() == expected
 
+    @pytest.mark.parametrize(
+        ("start", "patch"),
+        [(9, b"XX"), (10, b"X"), (0, b"ZZZZZZZZZZZ")],
+        ids=["overhang", "past-end", "too-long"],
+    )
+    async def test_set_range_out_of_bounds(
+        self, store: LocalStore, start: int, patch: bytes
+    ) -> None:
+        """A write that does not fit within the existing value raises, not extends."""
+        await store.set("test/key", cpu.Buffer.from_bytes(b"AAAAAAAAAA"))
+        with pytest.raises(ValueError, match="does not fit within the existing value"):
+            await store.set_range("test/key", cpu.Buffer.from_bytes(patch), start=start)
+        # The file is left unchanged (not zero-filled / extended).
+        result = await store.get("test/key", prototype=cpu.buffer_prototype)
+        assert result is not None
+        assert result.to_bytes() == b"AAAAAAAAAA"
+
     async def test_set_range_not_open(self, store_not_open: LocalStore) -> None:
         """set_range auto-opens a closed store."""
         assert not store_not_open._is_open
 
@@ -178,6 +178,20 @@ def test_set_range_sync(
         assert result is not None
         assert result.to_bytes() == expected
 
+    @pytest.mark.parametrize(
+        ("start", "patch"),
+        [(9, b"XX"), (10, b"X"), (0, b"ZZZZZZZZZZZ")],
+        ids=["overhang", "past-end", "too-long"],
+    )
+    async def test_set_range_out_of_bounds(
+        self, store: MemoryStore, start: int, patch: bytes
+    ) -> None:
+        """A write that does not fit within the existing value raises consistently."""
+        store._store_dict["test/key"] = cpu.Buffer.from_bytes(b"AAAAAAAAAA")
+        with pytest.raises(ValueError, match="does not fit within the existing value"):
+            await store.set_range("test/key", cpu.Buffer.from_bytes(patch), start=start)
+        assert store._store_dict["test/key"].to_bytes() == b"AAAAAAAAAA"
+
     async def test_set_range_not_open(self, store_not_open: MemoryStore) -> None:
         """set_range auto-opens a closed store."""
         assert not store_not_open._is_open
@@ -296,6 +310,18 @@ def test_from_dict(self) -> None:
         for v in result._store_dict.values():
             assert type(v) is gpu.Buffer
 
+    def test_set_range_not_supported(self, store: GpuMemoryStore) -> None:
+        """GpuMemoryStore deliberately does not satisfy SupportsSetRange.
+
+        Capability detection via isinstance must report False so a consumer (e.g. the
+        sharding codec) does not select it and crash. The methods are disclaimed
+        (set to None), so isinstance returns False rather than a false positive.
+        """
+        # mypy statically knows GpuMemoryStore cannot satisfy the protocol (the methods
+        # are None), which is exactly what we want — but it then flags this runtime
+        # assertion as unreachable. Keep the runtime check as a regression guard.
+        assert not isinstance(store, SupportsSetRange)  # type: ignore[unreachable]
+
 
 class TestManagedMemoryStore(StoreTests[ManagedMemoryStore, cpu.Buffer]):
     store_cls = ManagedMemoryStore
@@ -565,6 +591,26 @@ async def test_path_prefix_operations(self) -> None:
         assert result2 is not None
         assert result2.to_bytes() == b"value"
 
+    def test_supports_set_range(self, store: ManagedMemoryStore) -> None:
+        assert isinstance(store, SupportsSetRange)
+
+    async def test_set_range_applies_path_prefix(self) -> None:
+        """set_range must prepend the store's path prefix, matching set/get.
+
+        Regression: an unprefixed inherited set_range would target the wrong key.
+        """
+        store = ManagedMemoryStore(name="set-range-path-test", path="subdir")
+        await store.set("k", self.buffer_cls.from_bytes(b"AAAAAAAAAA"))
+        # set() writes to the prefixed backing key.
+        assert "subdir/k" in store._store_dict
+        await store.set_range("k", self.buffer_cls.from_bytes(b"XX"), start=2)
+        store.set_range_sync("k", self.buffer_cls.from_bytes(b"YY"), start=6)
+        # Both writes landed on the same prefixed value that set/get use.
+        observed = await store.get("k")
+        assert observed is not None
+        assert observed.to_bytes() == b"AAXXAAYYAA"
+        assert store._store_dict["subdir/k"].to_bytes() == b"AAXXAAYYAA"
+
     async def test_path_list_operations(self) -> None:
         """Test that list operations filter by path prefix."""
         store = ManagedMemoryStore(name="list-test")