fix: byte-range path skips empty chunks and deletes empty shards

The byte-range write path in ShardingCodec._encode_partial_sync was
encoding every affected inner chunk regardless of whether it was empty,
and never deleting fully-empty shards.  This made test_delete_empty_shards
fail under the SyncCodecPipeline (the bench branch's default).

Now matches the full-rewrite path's behavior:
  * When skip_empty (write_empty_chunks=False) is set and a merged chunk
    equals the fill value, mark the chunk absent in the index instead of
    writing data.
  * After updating all affected entries, if the index is fully empty,
    delete the shard rather than writing an empty index.

Also copy the shard index before mutating it; 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` test.
The asyncio.run call leaked event-loop self-pipe sockets that pytest's
unraisableexception plugin would later attribute to unrelated tests,
producing spurious "flaky" failures across the suite.

Author: d-v-b

src/zarr/codecs/sharding.py

@@ -622,7 +622,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))}
 
@@ -648,6 +651,11 @@ def _byte_offset(coords: tuple[int, ...]) -> int:
                         ).copy()
                         chunk_array[chunk_sel] = chunk_value
 
+                    if skip_empty and chunk_array.all_equal(fill_value):
+                        # Mark absent in the index; leave the data region as-is.
+                        index.set_chunk_slice(chunk_coords, None)
+                        continue
+
                     encoded = inner_transform.encode_chunk(chunk_array, chunk_spec)
                     if encoded is not None:
                         store.set_range_sync(key, encoded, byte_offset)
@@ -656,6 +664,11 @@ def _byte_offset(coords: tuple[int, ...]) -> int:
                             slice(byte_offset, byte_offset + chunk_byte_length),
                         )
 
+                # If every inner chunk is now absent, delete the entire shard.
+                if index.is_all_empty():
+                    byte_setter.delete_sync()
+                    return
+
                 index_bytes = self._encode_shard_index_sync(index)
                 if self.index_location == ShardingCodecIndexLocation.start:
                     store.set_range_sync(key, index_bytes, 0)

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: