#!/usr/bin/env python3 """Controlled Zarr reads: absence, payload integrity, identity and metadata age. All mutations target newly created synthetic arrays inside TemporaryDirectory. No existing store, production data, network store or storage hardware is touched. """ import argparse import hashlib import importlib.metadata import json from pathlib import Path import platform import tempfile import numpy as np import zarr from zarr.codecs import BytesCodec, Crc32cCodec FILL = -999 VALUES = list(range(10, 18)) def require(condition, message): if not condition: raise RuntimeError(message) def inventory(path): return { p.relative_to(path).as_posix(): hashlib.sha256(p.read_bytes()).hexdigest() for p in sorted(path.rglob("*")) if p.is_file() } def exact_store_bytes(path): return {p.relative_to(path).as_posix(): p.read_bytes() for p in path.rglob("*") if p.is_file()} def compare_inventory(expected, actual): return { "missing": sorted(expected.keys() - actual.keys()), "unexpected": sorted(actual.keys() - expected.keys()), "changed": sorted(k for k in expected.keys() & actual.keys() if expected[k] != actual[k]), } def make_array(path, version, values=None): kwargs = {"compressors": None} if version == 2 else { "serializer": BytesCodec(endian="little"), "compressors": [Crc32cCodec()], } arr = zarr.create_array( store=path, zarr_format=version, shape=(8,), chunks=(4,), dtype="int32", fill_value=FILL, config={"write_empty_chunks": False}, **kwargs ) if values is not None: arr[:] = np.asarray(values, dtype="int32") return arr def read_array(path, selection=slice(None)): try: arr = zarr.open_array(store=path, mode="r") return {"values": arr[selection].tolist(), "error_type": None} except Exception as exc: return {"values": None, "error_type": type(exc).__name__} def reference_crc32c(data): crc = 0xFFFFFFFF for byte in data: crc ^= byte for _ in range(8): crc = (crc >> 1) ^ (0x82F63B78 if crc & 1 else 0) return crc ^ 0xFFFFFFFF def verify_raw_crc(path): payload = path.read_bytes() require(len(payload) == 20, "Expected four int32 values plus a four-byte CRC32C") return reference_crc32c(payload[:-4]) == int.from_bytes(payload[-4:], "little") def one_repetition(base, repetition): rows = [] def row(case, version, result, **detail): rows.append({"repetition": repetition, "case": case, "zarr_format": version, **result, **detail}) for version in (2, 3): sparse = base / f"v{version}-sparse" lost = base / f"v{version}-lost" a = make_array(sparse, version) a[:4] = VALUES[:4] make_array(lost, version, VALUES) key = "1" if version == 2 else "c/1" expected_inventory = inventory(lost) baseline = read_array(lost) require(baseline == {"values": VALUES, "error_type": None}, "Intact control failed") row("intact", version, baseline) (lost / key).unlink() absent_read = read_array(lost) sparse_read = read_array(sparse) expected = {"values": VALUES[:4] + [FILL] * 4, "error_type": None} require(absent_read == expected == sparse_read, "Missing chunk did not match sparse control") equal = exact_store_bytes(lost) == exact_store_bytes(sparse) require(equal, "Lost-chunk store differs from never-written store") row("deleted_chunk", version, absent_read, inventory_delta=compare_inventory(expected_inventory, inventory(lost))) row("never_written_chunk", version, sparse_read, byte_identical_to_deleted_store=equal) a[4:] = FILL require(not (sparse / key).exists(), "Explicit fill chunk should be elided") require(exact_store_bytes(sparse) == exact_store_bytes(lost), "Explicit fill changed the sparse store") row("explicit_fill_elided", version, read_array(sparse), byte_identical_to_never_written_store=True) damaged = base / "v3-damaged" make_array(damaged, 3, VALUES) expected_inventory = inventory(damaged) require(verify_raw_crc(damaged / "c/0") and verify_raw_crc(damaged / "c/1"), "Independent CRC control failed") payload = bytearray((damaged / "c/1").read_bytes()) payload[0] ^= 1 (damaged / "c/1").write_bytes(payload) require(not verify_raw_crc(damaged / "c/1"), "Mutation did not break the checksum") selected = read_array(damaged, slice(0, 4)) require(selected == {"values": VALUES[:4], "error_type": None}, "Unselected chunk was read") row("corruption_outside_selection", 3, selected, selection="0:4") full = read_array(damaged) require(full["values"] is None and full["error_type"] == "ValueError", f"Expected CRC failure, received {full}") row("corrupt_chunk_full_read", 3, full, inventory_delta=compare_inventory(expected_inventory, inventory(damaged))) (damaged / "c/1").unlink() no_damaged_chunk = read_array(damaged) require(no_damaged_chunk == {"values": VALUES[:4] + [FILL] * 4, "error_type": None}, "Deletion after CRC failure did not return fill") row("delete_corrupt_chunk", 3, no_damaged_chunk) (damaged / "c/0").unlink() empty = read_array(damaged) require(empty == {"values": [FILL] * 8, "error_type": None}, "All-missing result differs") row("all_chunks_deleted", 3, empty) swapped = base / "v3-swapped" make_array(swapped, 3, VALUES) expected_inventory = inventory(swapped) first, second = (swapped / "c/0").read_bytes(), (swapped / "c/1").read_bytes() (swapped / "c/0").write_bytes(second) (swapped / "c/1").write_bytes(first) require(verify_raw_crc(swapped / "c/0") and verify_raw_crc(swapped / "c/1"), "Moving a whole payload should preserve its own CRC") moved = read_array(swapped) require(moved == {"values": VALUES[4:] + VALUES[:4], "error_type": None}, "Whole-chunk exchange control failed") row("whole_chunks_swapped", 3, moved, inventory_delta=compare_inventory(expected_inventory, inventory(swapped))) # V2 metadata consolidation is deliberately separated from the V3 codec controls. group_path = base / "v2-consolidation" writer = zarr.open_group(store=group_path, mode="w", zarr_format=2) arr = writer.create_array("signal", shape=(8,), chunks=(4,), dtype="int32", fill_value=FILL, compressors=None) arr[:] = VALUES zarr.consolidate_metadata(group_path) old_handle = zarr.open_group(store=group_path, mode="r", use_consolidated=True) before = inventory(group_path) arr.resize((12,)) arr[8:] = [18, 19, 20, 21] arr[0] = 999 writer.create_array("calibration", data=np.asarray([7, 8], dtype="int32"), compressors=None) require(inventory(group_path)[".zmetadata"] == before[".zmetadata"], "Writer unexpectedly refreshed the consolidated metadata") def group_read(group): signal = group["signal"] return {"values": signal[:].tolist(), "error_type": None, "shape": list(signal.shape), "array_names": sorted(group.array_keys())} expected_old = {"values": [999] + VALUES[1:], "error_type": None, "shape": [8], "array_names": ["signal"]} expected_new = {"values": [999] + VALUES[1:] + [18, 19, 20, 21], "error_type": None, "shape": [12], "array_names": ["calibration", "signal"]} stale = group_read(zarr.open_group(store=group_path, mode="r", use_consolidated=True)) fresh = group_read(zarr.open_group(store=group_path, mode="r", use_consolidated=False)) default = group_read(zarr.open_group(store=group_path, mode="r")) require(stale == expected_old == default, "Stale/default metadata controls differ") require(fresh == expected_new, "Unconsolidated metadata control differs") row("stale_consolidated_new_handle", 2, stale) row("default_prefers_stale_consolidated", 2, default) row("fresh_unconsolidated", 2, fresh) zarr.consolidate_metadata(group_path) reconsolidated = group_read(zarr.open_group(store=group_path, mode="r", use_consolidated=True)) retained = group_read(old_handle) require(reconsolidated == expected_new, "Reconsolidation did not refresh a new reader") require(retained == expected_old, "Previously opened handle unexpectedly refreshed") row("reconsolidated_new_handle", 2, reconsolidated) row("retained_handle_after_reconsolidation", 2, retained) return rows def run(repetitions=3): require(repetitions >= 1, "Repetitions must be positive") rows = [] with tempfile.TemporaryDirectory(prefix="hesela-zarr-") as directory: for i in range(1, repetitions + 1): root = Path(directory) / str(i) root.mkdir() rows.extend(one_repetition(root, i)) packages = ["zarr", "numpy", "numcodecs", "crc32c", "donfig", "PyYAML", "packaging", "typing_extensions"] return { "schema_version": "1.0.0", "study": "Zarr missing chunks, CRC32C scope and consolidated metadata", "evidence_type": "controlled software observations on synthetic inputs", "environment": {"python": platform.python_version(), "system": platform.system(), "machine": platform.machine(), "packages": {name: importlib.metadata.version(name) for name in packages}}, "input": {"shape": [8], "chunks": [4], "dtype": "int32", "fill_value": FILL, "values": VALUES, "v2_compression": None, "v3_codecs": ["bytes(little)", "crc32c"]}, "repetitions": repetitions, "case_count": len(rows) // repetitions, "observations": rows, "limitations": [ "Local directory stores only; no cloud consistency, crash or power-loss experiment.", "Mutations are deliberate fault injections, not observed field failure frequencies.", "Three repetitions check reproducibility in this environment, not statistical confidence.", "Metadata-consolidation observations use format 2; no V3 consolidation or sharding claims.", "A trusted pre-mutation inventory is assumed by the manifest comparison controls.", "One pinned Python implementation; no claim about all Zarr versions or readers.", ], } def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--output", type=Path, required=True) parser.add_argument("--repetitions", type=int, default=3) args = parser.parse_args() result = run(args.repetitions) args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8") print(f"PASS: {result['case_count']} cases x {result['repetitions']} repetitions") if __name__ == "__main__": main()