fix: byte-range fast path requires write_empty_chunks=True

The byte-range path in ShardingCodec._encode_partial_sync writes
every affected inner chunk into its fixed-size data slot. This is
incompatible with write_empty_chunks=False (the default), which
expects empty chunks to be compacted out of the shard layout entirely.

Add `not skip_empty` to the gate. With the default config, partial
shard writes go through the full-rewrite path (which honors the
compact layout). With write_empty_chunks=True, the byte-range
optimization is used.

Also copy the shard index before mutating; the decoded view may be
backed by a read-only buffer (mmap-style reads from LocalStore).

Plus: convert tests/test_sync_pipeline.py::test_memory_store_set_range
from `asyncio.run(...)` inside a sync test to a plain `async def`.
The asyncio.run call leaked event-loop self-pipe sockets that pytest's
unraisableexception plugin would later attribute to unrelated tests.

Update test_partial_shard_write_uses_set_range to opt into
write_empty_chunks=True so the byte-range path is actually exercised.

Author: d-v-b

src/zarr/codecs/sharding.py

@@ -606,9 +606,14 @@ def _encode_partial_sync(
         is_scalar = len(value.shape) == 0
 
         # --- Byte-range fast path ---
+        # Only safe when we don't need to skip empty chunks: byte-range
+        # writes leave chunk presence unchanged (writes a fixed-size
+        # data slot for every affected chunk). Compacting empty chunks
+        # away requires rewriting the whole shard.
         store = byte_setter.store if hasattr(byte_setter, "store") else None
         if (
             not is_complete
+            and not skip_empty
             and self._inner_codecs_fixed_size
             and isinstance(store, SupportsSetRange)
         ):
@@ -622,7 +627,10 @@ def _encode_partial_sync(
             if existing is not None and len(existing) == total_shard_size:
                 key = byte_setter.path if hasattr(byte_setter, "path") else str(byte_setter)
                 shard_reader = self._shard_reader_from_bytes_sync(existing, chunks_per_shard)
-                index = shard_reader.index
+                # The decoded index may be a view of a read-only buffer (e.g.
+                # mmap-backed reads from LocalStore). Copy so set_chunk_slice
+                # below can mutate it.
+                index = _ShardIndex(shard_reader.index.offsets_and_lengths.copy())
 
                 rank_map = {c: r for r, c in enumerate(morton_order_iter(chunks_per_shard))}
 

tests/test_sync_pipeline.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import asyncio
 from typing import Any
 
 import numpy as np
@@ -440,23 +439,19 @@ def test_memory_store_supports_byte_range_setter() -> None:
     assert isinstance(store, SupportsSetRange)
 
 
-def test_memory_store_set_range() -> None:
+async def test_memory_store_set_range() -> None:
     """MemoryStore.set_range should overwrite bytes at the given offset."""
+    store = zarr.storage.MemoryStore()
+    await store._ensure_open()
+    buf = cpu.Buffer.from_bytes(b"AAAAAAAAAA")  # 10 bytes
+    await store.set("test/key", buf)
 
-    async def _test() -> None:
-        store = zarr.storage.MemoryStore()
-        await store._ensure_open()
-        buf = cpu.Buffer.from_bytes(b"AAAAAAAAAA")  # 10 bytes
-        await store.set("test/key", buf)
-
-        patch = cpu.Buffer.from_bytes(b"XX")
-        await store.set_range("test/key", patch, start=3)
-
-        result = await store.get("test/key", prototype=cpu.buffer_prototype)
-        assert result is not None
-        assert result.to_bytes() == b"AAAXXAAAAA"
+    patch = cpu.Buffer.from_bytes(b"XX")
+    await store.set_range("test/key", patch, start=3)
 
-    asyncio.run(_test())
+    result = await store.get("test/key", prototype=cpu.buffer_prototype)
+    assert result is not None
+    assert result.to_bytes() == b"AAAXXAAAAA"
 
 
 def test_sharding_codec_inner_codecs_fixed_size_no_compression() -> None:
@@ -528,6 +523,9 @@ def test_partial_shard_write_uses_set_range() -> None:
     from unittest.mock import patch
 
     store = zarr.storage.MemoryStore()
+    # write_empty_chunks=True keeps a fixed-size dense layout, which is
+    # required for the byte-range fast path (chunks never transition
+    # present <-> absent).
     arr = zarr.create_array(
         store=store,
         shape=(100,),
@@ -536,6 +534,7 @@ def test_partial_shard_write_uses_set_range() -> None:
         shards=(100,),
         compressors=None,
         fill_value=0.0,
+        config={"write_empty_chunks": True},
     )
     if not isinstance(arr._async_array.codec_pipeline, SyncCodecPipeline):
         pytest.skip("byte-range write optimization is specific to SyncCodecPipeline")