fix(sharding): reject a stored shard too short to hold its index on every path

changes/4475.bugfix.md

@@ -0,0 +1 @@
+A stored shard too short to hold its shard index, including a zero-length value, now raises a `ValueError` saying the shard is truncated or corrupt, on every read and partial-write path under either codec pipeline. Previously the asynchronous full-shard read and partial write treated a zero-length shard as missing (reading the fill value, or overwriting it), while every other path failed with a checksum mismatch or a decode error. zarrs and tensorstore also reject such a shard.

src/zarr/codecs/sharding.py

@@ -355,7 +355,7 @@ def create_empty(
 
     def __getitem__(self, chunk_coords: tuple[int, ...]) -> Buffer:
         chunk_byte_slice = self.index.get_chunk_slice(chunk_coords)
-        if chunk_byte_slice:
+        if chunk_byte_slice is not None:
             return self.buf[chunk_byte_slice[0] : chunk_byte_slice[1]]
         raise KeyError
 
@@ -667,6 +667,7 @@ def _decode_shard_index_sync(
         self, index_bytes: Buffer, chunks_per_shard: tuple[int, ...]
     ) -> _ShardIndex:
         """Decode shard index synchronously using ChunkTransform."""
+        self._check_shard_index_size(index_bytes, chunks_per_shard)
         index_transform = self._get_index_chunk_transform(chunks_per_shard)
         index_spec = self._get_index_chunk_spec(chunks_per_shard)
         index_array = index_transform.decode_chunk(index_bytes, index_spec)
@@ -1403,7 +1404,8 @@ async def _encode_partial_single(
                 prototype=chunk_spec.prototype,
                 chunks_per_shard=chunks_per_shard,
             )
-            shard_reader = shard_reader or _ShardReader.create_empty(chunks_per_shard)
+            if shard_reader is None:
+                shard_reader = _ShardReader.create_empty(chunks_per_shard)
             # Use vectorized lookup for better performance. The lexicographic
             # coordinate array and keys are cached, so neither is rebuilt on
             # every write.
@@ -1547,6 +1549,7 @@ async def _decode_shard_index(
         # the synchronous read paths cannot use but this async path still can.
         if self._index_codecs_sync_capable():
             return self._decode_shard_index_sync(index_bytes, chunks_per_shard)
+        self._check_shard_index_size(index_bytes, chunks_per_shard)
         index_array = next(
             iter(
                 await get_pipeline_class()
@@ -1581,6 +1584,23 @@ async def _encode_shard_index(self, index: _ShardIndex) -> Buffer:
             raise RuntimeError("Encoding the shard index produced no bytes.")
         return index_bytes
 
+    def _check_shard_index_size(
+        self, index_bytes: Buffer, chunks_per_shard: tuple[int, ...]
+    ) -> None:
+        """Raise if `index_bytes` is shorter than the encoded shard index.
+
+        A stored shard always holds its index, so a shorter read means the
+        stored value is truncated (a zero-length value included). Every read
+        and write path decodes the index through here, so they all reject it
+        alike. zarrs and tensorstore also reject such a shard.
+        """
+        expected = self._shard_index_size(chunks_per_shard)
+        if len(index_bytes) < expected:
+            raise ValueError(
+                f"The stored shard is too short to hold its index: read {len(index_bytes)} "
+                f"bytes of index, expected {expected}. The shard is truncated or corrupt."
+            )
+
     def _shard_index_size(self, chunks_per_shard: tuple[int, ...]) -> int:
         return (
             get_pipeline_class()
@@ -1669,7 +1689,7 @@ async def _load_full_shard_maybe(
 
         return (
             await _ShardReader.from_bytes(shard_bytes, self, chunks_per_shard)
-            if shard_bytes
+            if shard_bytes is not None
             else None
         )
 

tests/test_codecs/test_sharding.py

@@ -2,7 +2,7 @@
 import math
 import pickle
 import warnings
-from typing import Any, cast, get_args
+from typing import TYPE_CHECKING, Any, cast, get_args
 from unittest.mock import AsyncMock
 
 import numpy as np
@@ -40,6 +40,9 @@
 from ..conftest import ArrayRequest
 from .test_codecs import _AsyncArrayProxy, order_from_dim
 
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
 
 def _reads_are_sync(store_mock: AsyncMock) -> bool:
     """True when the partial-shard read for this store+pipeline goes through the
@@ -1395,3 +1398,41 @@ def test_sharding_orthogonal_set_multiple_array_dims(
         expected[ix] = value.reshape(expected[ix].shape)
         assert np.array_equal(a[:], expected)
         assert np.array_equal(a.oindex[selection], value)
+
+
+@pytest.mark.parametrize(
+    "pipeline_path",
+    [
+        "zarr.core.codec_pipeline.FusedCodecPipeline",
+        "zarr.core.codec_pipeline.BatchedCodecPipeline",
+    ],
+)
+@pytest.mark.parametrize("store", ["memory", "local"], indirect=True)
+@pytest.mark.parametrize("index_location", ["start", "end"])
+@pytest.mark.parametrize("stored", [b"", b"abc"])
+@pytest.mark.parametrize("op", ["full read", "partial read", "partial write"])
+def test_sharding_truncated_shard_raises(
+    pipeline_path: str, store: Store, index_location: IndexLocation, stored: bytes, op: str
+) -> None:
+    """A stored shard shorter than its index (zero-length included) is
+    rejected by every read and partial-write path, under either pipeline,
+    rather than read as missing by some paths and failing the checksum in
+    others."""
+    with zarr.config.set({"codec_pipeline.path": pipeline_path}):
+        arr = zarr.create_array(
+            store,
+            shape=(8,),
+            chunks=(2,),
+            shards={"shape": (4,), "index_location": index_location},
+            dtype="i4",
+            compressors=None,
+            fill_value=-1,
+        )
+        zarr.core.sync.sync(store.set("c/0", default_buffer_prototype().buffer.from_bytes(stored)))
+        ops: dict[str, Callable[[], object]] = {
+            "full read": lambda: arr[0:4],
+            "partial read": lambda: arr[0:2],
+            "partial write": lambda: arr.__setitem__(slice(0, 2), 7),
+        }
+        with pytest.raises(ValueError, match="too short to hold its index"):
+            ops[op]()