diff --git a/py_hamt/store_httpx.py b/py_hamt/store_httpx.py index f5ea6a6..dec0d01 100644 --- a/py_hamt/store_httpx.py +++ b/py_hamt/store_httpx.py @@ -713,7 +713,19 @@ def __init__( self.gateway_base_urls = [_normalize_gateway_base_url(gateway_base_url)] pin_string: str = "true" if pin_on_add else "false" - self.rpc_url: str = f"{rpc_base_url}/api/v0/add?hash={self.hasher}&chunker={self.chunker}&pin={pin_string}" + # cid-version=1 is required, not cosmetic. Kubo returns a CIDv0 for any + # add that does not ask for v1, and a CIDv0 is dag-pb by definition -- + # so with sha2-256 (a CIDv0-representable hasher) the daemon would wrap + # the payload in a UnixFS dag-pb node and hand back Qm... regardless of + # the codec this store asked for. That breaks two things: + # * save() cannot relabel the CID's codec, because the stored block is + # the protobuf wrapper rather than the bytes passed in, so the digest + # would no longer match the block. + # * _cid_is_verifiable() declines to check dag-pb responses, so + # verify_content would silently pass every block through unverified. + # Requesting v1 makes Kubo store the raw bytes under the codec asked + # for, which is what blake3 (not CIDv0-representable) already got. + self.rpc_url: str = f"{rpc_base_url}/api/v0/add?hash={self.hasher}&chunker={self.chunker}&pin={pin_string}&cid-version=1" """@private""" self.gateway_base_url: str = self.gateway_base_urls[0] """@private""" @@ -1085,6 +1097,23 @@ async def save(self, data: bytes, codec: ContentAddressedStore.CodecInput) -> CI cid: CID = CID.decode(cid_str) if cid.codec.code != self.DAG_PB_MARKER: cid = cid.set(codec=codec) + elif self.verify_content: + # Kubo splits payloads larger than ``chunker`` into a UnixFS + # dag-pb tree, so the root block is the protobuf node rather + # than the bytes handed in. The requested codec cannot be + # applied (the digest would stop matching the block), and + # _cid_is_verifiable() skips dag-pb, so verify_content + # silently does nothing for this object. Warn rather than + # fail: the data still round-trips correctly, and the + # threshold depends on the caller's chunker setting. + warnings.warn( + f"Saved {len(data)} bytes exceeded the '{self.chunker}' " + f"chunker, so Kubo returned a dag-pb CID ({cid}). " + "Content verification is not possible for this object; " + "raise the chunker size to keep payloads in one block.", + RuntimeWarning, + stacklevel=2, + ) return cid except httpx.RequestError: diff --git a/tests/conftest.py b/tests/conftest.py index 67a63fd..da2f96f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -22,6 +22,13 @@ def pytest_addoption(parser): default=False, help="run tests that require a Kubo daemon", ) + parser.addoption( + "--benchmark-report", + action="store_true", + default=False, + help="print timing and request-count tables from the benchmark suite " + "(combine with -s); benchmarks assert on request counts either way", + ) def pytest_configure(config): diff --git a/tests/test_benchmark_stores.py b/tests/test_benchmark_stores.py index b2b9534..dd1865f 100644 --- a/tests/test_benchmark_stores.py +++ b/tests/test_benchmark_stores.py @@ -1,310 +1,611 @@ -# import time - -# import numpy as np -# import pandas as pd -# import pytest -# import xarray as xr -# from dag_cbor.ipld import IPLDKind - -# # Import both store implementations -# from py_hamt import HAMT, KuboCAS, FlatZarrStore, ShardedZarrStore -# from py_hamt.zarr_hamt_store import ZarrHAMTStore - - -# @pytest.fixture(scope="module") -# def random_zarr_dataset(): -# """Creates a random xarray Dataset for benchmarking.""" -# # Using a slightly larger dataset for a more meaningful benchmark -# times = pd.date_range("2024-01-01", periods=100) -# lats = np.linspace(-90, 90, 18) -# lons = np.linspace(-180, 180, 36) - -# temp = np.random.randn(len(times), len(lats), len(lons)) -# precip = np.random.gamma(2, 0.5, size=(len(times), len(lats), len(lons))) - -# ds = xr.Dataset( -# { -# "temp": (["time", "lat", "lon"], temp), -# }, -# coords={"time": times, "lat": lats, "lon": lons}, -# ) - -# # Define chunking for the store -# ds = ds.chunk({"time": 20, "lat": 18, "lon": 36}) -# yield ds - -# @pytest.fixture(scope="module") -# def random_shard_dataset(): -# """Creates a random xarray Dataset for benchmarking.""" -# # Using a slightly larger dataset for a more meaningful benchmark -# times = pd.date_range("2024-01-01", periods=100) -# lats = np.linspace(-90, 90, 18) -# lons = np.linspace(-180, 180, 36) - -# temp = np.random.randn(len(times), len(lats), len(lons)) -# precip = np.random.gamma(4, 2.5, size=(len(times), len(lats), len(lons))) - -# ds = xr.Dataset( -# { -# "precip": (["time", "lat", "lon"], precip), -# }, -# coords={"time": times, "lat": lats, "lon": lons}, -# ) - -# # Define chunking for the store -# ds = ds.chunk({"time": 20, "lat": 18, "lon": 36}) -# yield ds - - -# # # ### -# # # BENCHMARK FOR THE NEW FlatZarrStore -# # # ### -# # @pytest.mark.asyncio(loop_scope="session") -# # async def test_benchmark_flat_store( -# # create_ipfs: tuple[str, str], -# # random_zarr_dataset: xr.Dataset, -# # ): -# # """Benchmarks write and read performance for the new FlatZarrStore.""" -# # print("\n\n" + "=" * 80) -# # print("šŸš€ STARTING BENCHMARK for FlatZarrStore") -# # print("=" * 80) - -# # rpc_base_url, gateway_base_url = create_ipfs -# # # rpc_base_url = f"https://ipfs-gateway.dclimate.net" -# # # gateway_base_url = f"https://ipfs-gateway.dclimate.net" -# # # headers = { -# # # "X-API-Key": "", -# # # } -# # headers = {} -# # test_ds = random_zarr_dataset - -# # async with KuboCAS( -# # rpc_base_url=rpc_base_url, gateway_base_url=gateway_base_url, headers=headers -# # ) as kubo_cas: -# # # --- Write --- -# # # The full shape after appending -# # appended_shape = list(test_ds.dims.values()) -# # time_axis_index = list(test_ds.dims).index("time") -# # appended_shape[time_axis_index] *= 2 -# # final_array_shape = tuple(appended_shape) - -# # final_chunk_shape = [] -# # for dim_name in test_ds.dims: # Preserves dimension order -# # if dim_name in test_ds.chunks: -# # # test_ds.chunks[dim_name] is a tuple e.g. (20,) -# # final_chunk_shape.append(test_ds.chunks[dim_name][0]) -# # else: -# # # Fallback if a dimension isn't explicitly chunked (should use its full size) -# # final_chunk_shape.append(test_ds.dims[dim_name]) -# # final_chunk_shape = tuple(final_chunk_shape) - -# # store_write = await FlatZarrStore.open( -# # cas=kubo_cas, -# # read_only=False, -# # array_shape=final_array_shape, -# # chunk_shape=final_chunk_shape, -# # ) - -# # start_write = time.perf_counter() -# # # Perform an initial write and an append -# # test_ds.to_zarr(store=store_write, mode="w") -# # test_ds.to_zarr(store=store_write, mode="a", append_dim="time") -# # root_cid = await store_write.flush() # Flush to get the final CID -# # end_write = time.perf_counter() - -# # print(f"\n--- [FlatZarr] Write Stats ---") -# # print(f"Total time to write and append: {end_write - start_write:.2f} seconds") -# # print(f"Final Root CID: {root_cid}") - -# # # --- Read --- -# # store_read = await FlatZarrStore.open( -# # cas=kubo_cas, read_only=True, root_cid=root_cid -# # ) - -# # start_read = time.perf_counter() -# # ipfs_ds = xr.open_zarr(store=store_read) -# # # Force a read of some data to ensure it's loaded -# # _ = ipfs_ds.temp.isel(time=0).values -# # end_read = time.perf_counter() - -# # print(f"\n--- [FlatZarr] Read Stats ---") -# # print(f"Total time to open and read: {end_read - start_read:.2f} seconds") - -# # # --- Verification --- -# # full_test_ds = xr.concat([test_ds, test_ds], dim="time") -# # xr.testing.assert_identical(full_test_ds, ipfs_ds) -# # print("\nāœ… [FlatZarr] Data verification successful.") -# # print("=" * 80) - -# @pytest.mark.asyncio(loop_scope="session") -# async def test_benchmark_sharded_store( # Renamed function -# create_ipfs: tuple[str, str], -# random_shard_dataset: xr.Dataset, -# ): -# """Benchmarks write and read performance for the new ShardedZarrStore.""" # Updated docstring -# print("\n\n" + "=" * 80) -# print("šŸš€ STARTING BENCHMARK for ShardedZarrStore") # Updated print -# print("=" * 80) - -# rpc_base_url, gateway_base_url = create_ipfs - -# rpc_base_url = f"https://ipfs-gateway.dclimate.net" -# gateway_base_url = f"https://ipfs-gateway.dclimate.net" -# headers = { -# "X-API-Key": "", -# } -# # headers = {} -# test_ds = random_shard_dataset - -# # Define chunks_per_shard for the ShardedZarrStore -# chunks_per_shard_config = 50 # Configuration for sharding - -# async with KuboCAS( -# rpc_base_url=rpc_base_url, gateway_base_url=gateway_base_url, headers=headers -# ) as kubo_cas: -# # --- Write --- -# # The full shape after appending -# appended_shape = list(test_ds.dims.values()) -# time_axis_index = list(test_ds.dims).index("time") -# appended_shape[time_axis_index] *= 2 # Simulating appending along time dimension -# final_array_shape = tuple(appended_shape) - -# # Determine chunk shape from the dataset's encoding or dimensions -# final_chunk_shape_list = [] -# for dim_name in test_ds.dims: # Preserves dimension order from the dataset -# if dim_name in test_ds.chunks: -# # test_ds.chunks is a dict like {'time': (20,), 'y': (20,), 'x': (20,)} -# final_chunk_shape_list.append(test_ds.chunks[dim_name][0]) -# else: -# # Fallback if a dimension isn't explicitly chunked (should use its full size) -# final_chunk_shape_list.append(test_ds.dims[dim_name]) -# final_chunk_shape = tuple(final_chunk_shape_list) - -# # Use ShardedZarrStore and provide chunks_per_shard -# store_write = await ShardedZarrStore.open( -# cas=kubo_cas, -# read_only=False, -# array_shape=final_array_shape, -# chunk_shape=final_chunk_shape, -# chunks_per_shard=chunks_per_shard_config # Added new parameter -# ) - -# start_write = time.perf_counter() -# # Perform an initial write and an append -# test_ds.to_zarr(store=store_write, mode="w") -# test_ds.to_zarr(store=store_write, mode="a", append_dim="time") -# root_cid = await store_write.flush() # Flush to get the final CID -# end_write = time.perf_counter() - -# print(f"\n--- [ShardedZarr] Write Stats (chunks_per_shard={chunks_per_shard_config}) ---") # Updated print -# print(f"Total time to write and append: {end_write - start_write:.2f} seconds") -# print(f"Final Root CID: {root_cid}") - -# print(f"\n--- [ShardedZarr] STARTING READ ---") # Updated print -# # --- Read --- -# # When opening for read, chunks_per_shard is read from the store's metadata -# store_read = await ShardedZarrStore.open( # Use ShardedZarrStore -# cas=kubo_cas, read_only=True, root_cid=root_cid -# ) - -# start_read = time.perf_counter() -# ipfs_ds = xr.open_zarr(store=store_read) -# # Force a read of some data to ensure it's loaded (e.g., first time slice of 'temp' variable) -# if "precip" in ipfs_ds.variables and "time" in ipfs_ds.coords: -# # _ = ipfs_ds.temp.isel(time=0).values -# data_fetched = ipfs_ds.precip.isel(time=slice(0, 1)).values - -# # Calculate the size of the fetched data -# data_size = data_fetched.nbytes if data_fetched is not None else 0 -# print(f"Fetched data size: {data_size / (1024 * 1024):.4f} MB") -# elif len(ipfs_ds.data_vars) > 0 : # Fallback: try to read from the first data variable -# first_var_name = list(ipfs_ds.data_vars.keys())[0] -# # Construct a minimal selection based on available dimensions -# selection = {dim: 0 for dim in ipfs_ds[first_var_name].dims} -# if selection: -# _ = ipfs_ds[first_var_name].isel(**selection).values -# else: # If no dimensions, try loading the whole variable (e.g. scalar) -# _ = ipfs_ds[first_var_name].values -# end_read = time.perf_counter() - -# print(f"\n--- [ShardedZarr] Read Stats ---") # Updated print -# print(f"Total time to open and read some data: {end_read - start_read:.2f} seconds") - -# # --- Verification --- -# # Create the expected full dataset after append operation -# full_test_ds = xr.concat([test_ds, test_ds], dim="time") -# xr.testing.assert_identical(full_test_ds, ipfs_ds) -# print("\nāœ… [ShardedZarr] Data verification successful.") # Updated print -# print("=" * 80) - -# # ### -# # BENCHMARK FOR THE ORIGINAL ZarrHAMTStore -# # ### -# @pytest.mark.asyncio(loop_scope="session") -# async def test_benchmark_hamt_store( -# create_ipfs: tuple[str, str], -# random_zarr_dataset: xr.Dataset, -# ): -# """Benchmarks write and read performance for the ZarrHAMTStore.""" -# print("\n\n" + "=" * 80) -# print("šŸš€ STARTING BENCHMARK for ZarrHAMTStore") -# print("=" * 80) - -# rpc_base_url, gateway_base_url = create_ipfs - -# # rpc_base_url = f"https://ipfs-gateway.dclimate.net" -# # gateway_base_url = f"https://ipfs-gateway.dclimate.net" -# # headers = { -# # "X-API-Key": "", -# # } -# headers = {} -# test_ds = random_zarr_dataset - -# async with KuboCAS( -# rpc_base_url=rpc_base_url, gateway_base_url=gateway_base_url, headers=headers -# ) as kubo_cas: -# # --- Write --- -# print("Building HAMT store...") -# hamt = await HAMT.build(cas=kubo_cas, values_are_bytes=True) -# print("HAMT store built successfully.") -# zhs = ZarrHAMTStore(hamt) -# print("ZarrHAMTStore created successfully.") - -# start_write = time.perf_counter() -# # Perform an initial write and an append to simulate a common workflow -# test_ds.to_zarr(store=zhs, mode="w") -# print("Initial write completed, now appending...") -# test_ds.to_zarr(store=zhs, mode="a", append_dim="time") -# await hamt.make_read_only() # Flush and freeze to get the final CID -# end_write = time.perf_counter() - -# cid: IPLDKind = hamt.root_node_id -# print(f"\n--- [HAMT] Write Stats ---") -# print(f"Total time to write and append: {end_write - start_write:.2f} seconds") -# print(f"Final Root CID: {cid}") - -# # --- Read --- -# hamt_ro = await HAMT.build( -# cas=kubo_cas, root_node_id=cid, values_are_bytes=True, read_only=True -# ) -# zhs_ro = ZarrHAMTStore(hamt_ro, read_only=True) - -# start_read = time.perf_counter() -# ipfs_ds = xr.open_zarr(store=zhs_ro) -# # Force a read of some data to ensure it's loaded -# data_fetched = ipfs_ds.temp.isel(time=slice(0, 1)).values - -# # Calculate the size of the fetched data -# data_size = data_fetched.nbytes if data_fetched is not None else 0 -# print(f"Fetched data size: {data_size / (1024 * 1024):.4f} MB") -# end_read = time.perf_counter() - -# print(f"\n--- [HAMT] Read Stats ---") -# print(f"Total time to open and read: {end_read - start_read:.2f} seconds") - - -# # --- Verification --- -# full_test_ds = xr.concat([test_ds, test_ds], dim="time") -# xr.testing.assert_identical(full_test_ds, ipfs_ds) -# print("\nāœ… [HAMT] Data verification successful.") -# print("=" * 80) +"""Performance benchmarks for the Zarr store backends (issue #58). + +This file replaces a commented-out scratch script. That script could not run: +it imported ``FlatZarrStore`` (no longer exported), pointed at the production +``ipfs-gateway.dclimate.net`` with a blank ``X-API-Key``, used +``Dataset.dims.values()`` (removed in modern xarray), and ``print``ed timings +without asserting anything -- so it could not fail, and therefore protected +nothing. + +What is asserted here is **request counts**, not wall-clock time. Counts are +deterministic, so they hold up in CI. Timings are still measured and reported +via ``--benchmark-report`` for humans, but nothing fails on them. + +A caveat these benchmarks exist to make visible: ``ShardedZarrStore`` defaults to +``shard_read_mode="sparse"``, which fetches one entry per chunk to keep *point* +reads cheap (PR #87: ~9x on 30 years of ERA5 dailies). On a whole-array scan that +is the wrong mode by design and costs roughly 2x the ``"full"`` path. Every scan +benchmark below therefore sets the mode explicitly, and +``test_sparse_mode_trades_scan_cost_for_point_read_latency`` pins the trade-off +so a scan measured on the default is not misread as a regression. + +Note also that these are network-round-trip benchmarks. PR #88's headline numbers +measure different quantities -- bulk writes against ``InMemoryCAS``, warm-cache +get latency, and per-request RTT depth -- so they are not comparable to the +totals reported here. + +Run: + pytest tests/test_benchmark_stores.py --ipfs + pytest tests/test_benchmark_stores.py --ipfs --benchmark-report -s +""" + +import math +import time +from dataclasses import dataclass, field +from typing import Any, Optional + +import httpx +import numpy as np +import pandas as pd +import pytest +import xarray as xr + +from py_hamt import HAMT, KuboCAS, ShardedZarrStore, ZarrHAMTStore +from py_hamt import instrumentation as instr + +pytestmark = pytest.mark.ipfs + + +# -------------------------------------------------------------------------- +# Harness +# -------------------------------------------------------------------------- + + +@dataclass +class BenchmarkResult: + """One measured phase: how long it took and what it cost in requests.""" + + label: str + seconds: float + counters: dict[str, float] = field(default_factory=dict) + unique_cids: int = 0 + duplicate_requests: int = 0 + + @property + def cas_loads(self) -> float: + return self.counters.get("cas_load.total", 0.0) + + @property + def retries(self) -> float: + return self.counters.get("cas_load.retries", 0.0) + + @property + def shard_loads(self) -> float: + """Shard cache accesses, hits and misses together.""" + return self.counters.get("sharded_store.shard_cache.total", 0.0) + + @property + def shard_fetches(self) -> float: + """Shard cache misses -- i.e. shards actually fetched over the network.""" + return self.counters.get("sharded_store.shard_cache.miss", 0.0) + + @property + def work_units(self) -> float: + """Total storage operations, whichever backend performed them. + + ``ShardedZarrStore`` writes buffer into shards and flush, so they emit + ``sharded_store.*`` rather than ``cas_load.*``. Comparing raw + ``cas_load`` totals across the two backends would therefore read a + sharded write as "no work done". + """ + return self.cas_loads + self.shard_loads + + def report(self) -> str: + return ( + f" {self.label:<34} {self.seconds:7.2f}s " + f"cas_loads={self.cas_loads:<7.0f} " + f"unique={self.unique_cids:<7} " + f"dupes={self.duplicate_requests:<5} " + f"retries={self.retries:.0f}" + ) + + +class _Phase: + """Context manager timing a block and snapshotting instrumentation.""" + + def __init__(self, label: str) -> None: + self.label = label + self.result: Optional[BenchmarkResult] = None + + def __enter__(self) -> "_Phase": + instr.reset(self.label) + self._start = time.perf_counter() + return self + + def __exit__(self, *exc: Any) -> None: + elapsed = time.perf_counter() - self._start + snap = instr.snapshot() + self.result = BenchmarkResult( + label=self.label, + seconds=elapsed, + counters=dict(snap.get("counters", {})), + unique_cids=int(snap.get("cas_load_unique_cids", 0)), + duplicate_requests=int(snap.get("cas_load_duplicate_requests", 0)), + ) + + +@pytest.fixture(autouse=True) +def _trace_enabled(monkeypatch: pytest.MonkeyPatch) -> None: + """Instrumentation is opt-in via env var; benchmarks always need it.""" + monkeypatch.setenv("PY_HAMT_TRACE", "1") + + +@pytest.fixture +def report(request: pytest.FixtureRequest) -> Any: + """Collect phase results and print them when --benchmark-report is passed.""" + collected: list[BenchmarkResult] = [] + + def add(result: Optional[BenchmarkResult]) -> BenchmarkResult: + assert result is not None, "phase did not complete" + collected.append(result) + return result + + yield add + + if request.config.getoption("--benchmark-report") and collected: + print(f"\n{'=' * 78}\n{request.node.name}\n{'=' * 78}") + for result in collected: + print(result.report()) + + +# -------------------------------------------------------------------------- +# Datasets +# -------------------------------------------------------------------------- + + +def _make_dataset( + variable: str, *, periods: int = 100, time_chunk: int = 20 +) -> xr.Dataset: + """A small, deterministically-seeded gridded dataset. + + Seeded so chunk bytes -- and therefore CIDs and request counts -- are stable + across runs. An unseeded dataset would make the count assertions flaky. + """ + rng = np.random.default_rng(0) + times = pd.date_range("2024-01-01", periods=periods) + lats = np.linspace(-90, 90, 18) + lons = np.linspace(-180, 180, 36) + + data = rng.standard_normal((len(times), len(lats), len(lons))) + ds = xr.Dataset( + {variable: (["time", "lat", "lon"], data)}, + coords={"time": times, "lat": lats, "lon": lons}, + ) + return ds.chunk({"time": time_chunk, "lat": 18, "lon": 36}) + + +@pytest.fixture(scope="module") +def hamt_dataset() -> xr.Dataset: + return _make_dataset("temp") + + +@pytest.fixture(scope="module") +def shard_dataset() -> xr.Dataset: + return _make_dataset("precip") + + +def _chunk_shape(ds: xr.Dataset) -> tuple[int, ...]: + """Chunk shape in dimension order, via .sizes (``.dims`` is deprecated).""" + return tuple( + ds.chunks[dim][0] if dim in ds.chunks else ds.sizes[dim] for dim in ds.sizes + ) + + +def _appended_shape(ds: xr.Dataset) -> tuple[int, ...]: + """Array shape after one append along ``time``.""" + return tuple(size * 2 if dim == "time" else size for dim, size in ds.sizes.items()) + + +@dataclass +class Tally: + """Request counts shared across per-loop transport instances.""" + + adds: int = 0 + gets: int = 0 + in_flight: int = 0 + peak_in_flight: int = 0 + + +class CountingTransport(httpx.AsyncHTTPTransport): + """Wraps the real transport to count requests and observe parallelism. + + Two things the ``instrumentation`` module cannot report: + + * **Saves.** It records ``cas_load.*`` but has no save-side counters, so + ``/api/v0/add`` POSTs are invisible to it. Asserting on load counts would + say nothing about write amplification. + * **Concurrency.** Passing ``concurrency=N`` proves nothing unless something + observes how many requests are actually in flight at once. + """ + + def __init__(self, tally: Optional["Tally"] = None) -> None: + super().__init__() + # Counts live in a shared Tally so several transport instances -- one + # per event loop, since httpx clients are not loop-portable -- can + # aggregate into a single set of numbers. + self.tally = tally if tally is not None else Tally() + + @property + def adds(self) -> int: + return self.tally.adds + + @property + def gets(self) -> int: + return self.tally.gets + + @property + def peak_in_flight(self) -> int: + return self.tally.peak_in_flight + + async def handle_async_request(self, request: httpx.Request) -> httpx.Response: + tally = self.tally + if request.method == "POST" and "/api/v0/add" in request.url.path: + tally.adds += 1 + else: + tally.gets += 1 + + tally.in_flight += 1 + tally.peak_in_flight = max(tally.peak_in_flight, tally.in_flight) + try: + return await super().handle_async_request(request) + finally: + tally.in_flight -= 1 + + +# -------------------------------------------------------------------------- +# Benchmarks +# -------------------------------------------------------------------------- + + +async def test_benchmark_hamt_store( + create_ipfs: tuple[str, str], hamt_dataset: xr.Dataset, report: Any +) -> None: + """Write-then-append-then-read through ZarrHAMTStore.""" + rpc, gateway = create_ipfs + ds = hamt_dataset + n_chunks = math.prod(len(chunks) for chunks in ds.chunks.values()) + tally = Tally() + + async with KuboCAS( + rpc_base_url=rpc, + gateway_base_url=gateway, + # client_factory, not client: a user-supplied client is not reused + # across event loops, and the replacement would drop this transport + # (and with it the counts). A fresh transport per call keeps httpx's + # loop-bound state per-loop while the shared Tally aggregates counts. + client_factory=lambda: httpx.AsyncClient(transport=CountingTransport(tally)), + ) as cas: + with _Phase("ZarrHAMTStore write+append") as phase: + hamt = await HAMT.build(cas=cas, values_are_bytes=True) + store = ZarrHAMTStore(hamt) + ds.to_zarr(store=store, mode="w") + ds.to_zarr(store=store, mode="a", append_dim="time") + await hamt.make_read_only() + write = report(phase.result) + + root = hamt.root_node_id + + with _Phase("ZarrHAMTStore read") as phase: + read_hamt = await HAMT.build( + cas=cas, root_node_id=root, values_are_bytes=True, read_only=True + ) + actual = xr.open_zarr(store=ZarrHAMTStore(read_hamt, read_only=True)) + actual.load() + read = report(phase.result) + + xr.testing.assert_identical(xr.concat([ds, ds], dim="time"), actual) + + # Write amplification, bounded rather than merely non-zero. A two-pass write + # of this fixture currently costs 32 POSTs for 5 chunks -- chunk bodies plus + # metadata and interior HAMT nodes, which dominate at this size. The bound + # has ~50% headroom for tree-shape churn while still failing long before a + # per-chunk or per-node blowup. + max_adds = 48 + assert 0 < tally.adds <= max_adds, ( + f"{tally.adds} /api/v0/add POSTs for a two-pass write of " + f"{n_chunks} chunks (bound {max_adds}); writes are amplifying" + ) + + assert read.cas_loads > 0, "read did no content-store work" + # A clean local daemon should never need to retry. + assert write.retries == 0 + assert read.retries == 0 + + +async def test_benchmark_sharded_store( + create_ipfs: tuple[str, str], shard_dataset: xr.Dataset, report: Any +) -> None: + """Write-then-append-then-read through ShardedZarrStore.""" + rpc, gateway = create_ipfs + ds = shard_dataset + + async with KuboCAS(rpc_base_url=rpc, gateway_base_url=gateway) as cas: + with _Phase("ShardedZarrStore write+append") as phase: + store = await ShardedZarrStore.open( + cas=cas, + read_only=False, + array_shape=_appended_shape(ds), + chunk_shape=_chunk_shape(ds), + chunks_per_shard=50, + ) + ds.to_zarr(store=store, mode="w") + ds.to_zarr(store=store, mode="a", append_dim="time") + root_cid = await store.flush() + write = report(phase.result) + + with _Phase("ShardedZarrStore read") as phase: + # Explicit: this phase is a whole-array scan, which is what "full" + # is for. The default is "sparse", tuned for point reads instead. + read_store = await ShardedZarrStore.open( + cas=cas, read_only=True, root_cid=root_cid, shard_read_mode="full" + ) + actual = xr.open_zarr(store=read_store) + actual.load() + read = report(phase.result) + + xr.testing.assert_identical(xr.concat([ds, ds], dim="time"), actual) + + assert write.work_units > 0, "write did no storage work" + assert read.work_units > 0, "read did no storage work" + assert read.retries == 0 + + +async def test_sharded_full_mode_read_batches_chunks_into_shard_fetches( + create_ipfs: tuple[str, str], report: Any +) -> None: + """Sharding's reason for existing, asserted rather than assumed. + + In ``shard_read_mode="full"`` one shard fetch serves every chunk in that + shard, so a whole-array scan costs roughly ``num_shards`` loads instead of + one per chunk. Without this in a test, a regression reintroducing per-chunk + round trips on the full path would pass CI silently. + + Note the mode is explicit here. The store's default is ``"sparse"``, which + is deliberately the opposite trade-off -- see + ``test_sparse_mode_trades_scan_cost_for_point_read_latency``. + + ``chunks_per_shard`` is set low enough to force several shards. With a + single shard the assertion below would hold even if every chunk triggered + its own fetch, so the multi-shard setup is what gives it teeth. + """ + rpc, gateway = create_ipfs + # 100 steps chunked by 5 along a single chunk of lat/lon => 20 chunks. + ds = _make_dataset("temp", periods=100, time_chunk=5) + n_chunks = math.prod(len(chunks) for chunks in ds.chunks.values()) + chunks_per_shard = 5 + expected_shards = math.ceil(n_chunks / chunks_per_shard) + assert expected_shards > 1, "setup must span multiple shards to be meaningful" + + async with KuboCAS(rpc_base_url=rpc, gateway_base_url=gateway) as cas: + hamt = await HAMT.build(cas=cas, values_are_bytes=True) + ds.to_zarr(store=ZarrHAMTStore(hamt), mode="w") + await hamt.make_read_only() + + with _Phase("HAMT read") as phase: + read_hamt = await HAMT.build( + cas=cas, + root_node_id=hamt.root_node_id, + values_are_bytes=True, + read_only=True, + ) + hamt_result = xr.open_zarr(store=ZarrHAMTStore(read_hamt, read_only=True)) + hamt_result.load() + hamt_read = report(phase.result) + + sharded = await ShardedZarrStore.open( + cas=cas, + read_only=False, + array_shape=tuple(ds.sizes.values()), + chunk_shape=_chunk_shape(ds), + chunks_per_shard=chunks_per_shard, + ) + ds.to_zarr(store=sharded, mode="w") + root_cid = await sharded.flush() + + with _Phase("Sharded read (full)") as phase: + read_store = await ShardedZarrStore.open( + cas=cas, + read_only=True, + root_cid=root_cid, + shard_read_mode="full", + ) + sharded_result = xr.open_zarr(store=read_store) + sharded_result.load() + sharded_read = report(phase.result) + + # Both must return the same data, or the comparison is meaningless. + xr.testing.assert_identical(hamt_result, sharded_result) + + # The batching claim, pinned to an exact count rather than a loose bound: + # each shard is fetched once and serves every chunk it covers. Asserted on + # misses (actual network fetches) rather than shard_cache.total, which also + # counts cache hits and so would not distinguish batching from per-chunk + # refetching that happened to be cached. + assert sharded_read.shard_fetches == expected_shards, ( + f"{sharded_read.shard_fetches:.0f} shard fetches for {n_chunks} chunks " + f"across {expected_shards} shards; full mode should fetch each shard " + "exactly once and batch every chunk reference it covers" + ) + # Batching should also mean no block is pulled twice. + assert sharded_read.duplicate_requests == 0, ( + f"full-mode scan refetched {sharded_read.duplicate_requests} blocks; " + "one shard fetch should serve every chunk it covers" + ) + assert sharded_read.cas_loads <= hamt_read.cas_loads, ( + f"sharded read cost {sharded_read.cas_loads:.0f} CAS loads vs HAMT's " + f"{hamt_read.cas_loads:.0f}; sharding is supposed to reduce round trips" + ) + + +async def test_sparse_mode_trades_scan_cost_for_point_read_latency( + create_ipfs: tuple[str, str], report: Any +) -> None: + """Characterize the default read mode, so its cost is not mistaken for a bug. + + ``shard_read_mode`` defaults to ``"sparse"`` (PR #87): a read-only cache miss + with no byte range fetches just the requested entry rather than decoding the + whole shard. That is a large win for point reads over long time ranges -- the + AEGIS workload the mode was built for -- and a deliberate loss on a full + scan, where every entry is wanted anyway and per-chunk fetches add up. + + Asserted here so the trade-off is visible and intentional: a full scan in + sparse mode legitimately costs *more* than in full mode. Anyone benchmarking + a whole-array read should set ``shard_read_mode="full"``. + """ + rpc, gateway = create_ipfs + ds = _make_dataset("temp", periods=100, time_chunk=5) + + async with KuboCAS(rpc_base_url=rpc, gateway_base_url=gateway) as cas: + store = await ShardedZarrStore.open( + cas=cas, + read_only=False, + array_shape=tuple(ds.sizes.values()), + chunk_shape=_chunk_shape(ds), + chunks_per_shard=50, + ) + ds.to_zarr(store=store, mode="w") + root_cid = await store.flush() + + results: dict[str, BenchmarkResult] = {} + for mode in ("sparse", "full"): + with _Phase(f"full scan @ shard_read_mode={mode}") as phase: + read_store = await ShardedZarrStore.open( + cas=cas, read_only=True, root_cid=root_cid, shard_read_mode=mode + ) + scanned = xr.open_zarr(store=read_store) + scanned.load() + results[mode] = report(phase.result) + xr.testing.assert_identical(ds, scanned) + + assert results["sparse"].cas_loads > results["full"].cas_loads, ( + "expected sparse mode to cost more on a full scan; if this now holds " + "the other way, sparse decoding has changed and the default may want " + "revisiting" + ) + + +async def test_sharded_store_defaults_to_sparse_read_mode( + create_ipfs: tuple[str, str], +) -> None: + """Pin the default, since it determines which trade-off users get.""" + rpc, gateway = create_ipfs + + async with KuboCAS(rpc_base_url=rpc, gateway_base_url=gateway) as cas: + store = await ShardedZarrStore.open( + cas=cas, + read_only=False, + array_shape=(10, 10), + chunk_shape=(5, 5), + chunks_per_shard=4, + ) + + assert store.shard_read_mode == "sparse" + + +async def test_read_cache_prevents_duplicate_cid_fetches( + create_ipfs: tuple[str, str], report: Any +) -> None: + """Re-reading the same data must be served from cache, not refetched. + + Guards the read-cache behaviour that ``test_h5_read_cache_growth`` bounds the + memory of: that test proves the cache does not grow without limit, this one + proves it actually saves requests. + """ + rpc, gateway = create_ipfs + ds = _make_dataset("temp", periods=40) + + async with KuboCAS(rpc_base_url=rpc, gateway_base_url=gateway) as cas: + hamt = await HAMT.build(cas=cas, values_are_bytes=True) + ds.to_zarr(store=ZarrHAMTStore(hamt), mode="w") + await hamt.make_read_only() + + read_hamt = await HAMT.build( + cas=cas, + root_node_id=hamt.root_node_id, + values_are_bytes=True, + read_only=True, + ) + store = ZarrHAMTStore(read_hamt, read_only=True) + + with _Phase("first read (cold)") as phase: + xr.open_zarr(store=store).load() + cold = report(phase.result) + + with _Phase("second read (warm)") as phase: + xr.open_zarr(store=store).load() + warm = report(phase.result) + + assert cold.cas_loads > 0 + assert warm.cas_loads < cold.cas_loads, ( + f"warm read cost {warm.cas_loads:.0f} loads vs cold {cold.cas_loads:.0f}; " + "the read cache is not being used" + ) + + +@pytest.mark.parametrize("concurrency", [1, 8, 32]) +async def test_benchmark_concurrency_levels( + create_ipfs: tuple[str, str], concurrency: int, report: Any +) -> None: + """Read throughput across concurrency settings (also covers issue #59). + + Correctness must hold at every level, the work done must not depend on how + many requests are allowed in flight, and -- measured via + ``CountingTransport`` -- the configured ceiling must actually be respected. + Without that last check the parametrization proves nothing: a ``KuboCAS`` + that ignored the argument and serialized everything would still satisfy data + identity and a zero retry count at every level. + + Wall-clock is reported but not asserted on. + """ + rpc, gateway = create_ipfs + ds = _make_dataset("temp", periods=40) + + async with KuboCAS(rpc_base_url=rpc, gateway_base_url=gateway) as cas: + hamt = await HAMT.build(cas=cas, values_are_bytes=True) + ds.to_zarr(store=ZarrHAMTStore(hamt), mode="w") + await hamt.make_read_only() + root = hamt.root_node_id + + tally = Tally() + async with KuboCAS( + rpc_base_url=rpc, + gateway_base_url=gateway, + concurrency=concurrency, + client_factory=lambda: httpx.AsyncClient(transport=CountingTransport(tally)), + ) as cas: + with _Phase(f"read @ concurrency={concurrency}") as phase: + read_hamt = await HAMT.build( + cas=cas, root_node_id=root, values_are_bytes=True, read_only=True + ) + actual = xr.open_zarr(store=ZarrHAMTStore(read_hamt, read_only=True)) + actual.load() + result = report(phase.result) + + xr.testing.assert_identical(ds, actual) + assert result.unique_cids > 0 + assert result.retries == 0, ( + f"concurrency={concurrency} forced {result.retries:.0f} retries" + ) + + # The ceiling is honoured: never more requests in flight than configured. + assert tally.peak_in_flight <= concurrency, ( + f"peak {tally.peak_in_flight} concurrent requests exceeded the " + f"configured concurrency={concurrency}" + ) + if concurrency == 1: + # The floor case proves the semaphore is enforced at all: a peak of 1 + # is only reachable if requests are genuinely serialized. + assert tally.peak_in_flight == 1, ( + "concurrency=1 should serialize requests, but peak in-flight was " + f"{tally.peak_in_flight}" + ) + else: + # Above the floor, prove requests actually overlap -- otherwise a + # KuboCAS that silently serialized everything would pass the ceiling + # check above. Not asserted equal to `concurrency`: this fixture only + # exposes ~4 independent fetches, so the limit is not the binding + # constraint at 8 or 32. + assert tally.peak_in_flight > 1, ( + f"concurrency={concurrency} never ran two requests at once " + "(peak in-flight was 1); reads appear to be serialized" + ) diff --git a/tests/test_k16_sha2_256_cid_version.py b/tests/test_k16_sha2_256_cid_version.py new file mode 100644 index 0000000..ceb62eb --- /dev/null +++ b/tests/test_k16_sha2_256_cid_version.py @@ -0,0 +1,224 @@ +"""Regression tests for issue #43. + +Kubo returns a CIDv0 for any ``/api/v0/add`` that does not explicitly request +version 1, and a CIDv0 is ``dag-pb`` by definition. With a hasher that CIDv0 can +represent -- ``sha2-256`` -- the daemon therefore wrapped the payload in a UnixFS +``dag-pb`` node and returned ``Qm...`` no matter which codec ``save()`` asked +for. Two consequences, both covered here: + +* ``save()`` deliberately declines to relabel a ``dag-pb`` CID's codec, because + the block Kubo stored is the protobuf wrapper rather than the bytes passed in. + A relabelled CID would not hash to its own block. +* ``_cid_is_verifiable()`` declines to check ``dag-pb`` responses, so a + ``verify_content=True`` store built on ``sha2-256`` silently returned every + block unverified. + +The fix pins ``cid-version=1`` on the add URL. ``blake3`` -- the default, and not +CIDv0-representable -- was never affected, so it is asserted alongside as a +control. +""" + +import os +import warnings + +import httpx +import pytest +from multiformats import CID, multihash + +from py_hamt import HAMT, ContentAddressedStore, KuboCAS +from py_hamt.store_httpx import _cid_is_verifiable + + +def test_add_url_requests_cid_version_1() -> None: + """The add URL must pin cid-version=1 for every hasher.""" + cas = KuboCAS(hasher="sha2-256") + + assert "cid-version=1" in cas.rpc_url + + +@pytest.mark.parametrize("hasher", ["sha2-256", "blake3"]) +def test_add_url_pins_cid_version_for_any_hasher(hasher: str) -> None: + cas = KuboCAS(hasher=hasher) + + assert "cid-version=1" in cas.rpc_url + assert f"hash={hasher}" in cas.rpc_url + + +@pytest.mark.ipfs +@pytest.mark.parametrize( + ("hasher", "codec"), + [ + ("sha2-256", "raw"), + ("sha2-256", "dag-cbor"), + ("blake3", "raw"), + ("blake3", "dag-cbor"), + ], +) +async def test_save_returns_cidv1_with_requested_codec( + create_ipfs: tuple[str, str], + hasher: str, + codec: ContentAddressedStore.CodecInput, +) -> None: + """sha2-256 must yield a CIDv1 carrying the codec save() was asked for.""" + rpc, gw = create_ipfs + payload = b"py-hamt issue 43: " + hasher.encode() + b"/" + codec.encode() + + async with KuboCAS(hasher=hasher, rpc_base_url=rpc, gateway_base_url=gw) as cas: + cid = await cas.save(payload, codec=codec) + + assert cid.version == 1, f"{hasher} still returns a CIDv0" + assert cid.codec.name == codec + assert cid.codec.code != KuboCAS.DAG_PB_MARKER + + +@pytest.mark.ipfs +@pytest.mark.parametrize("hasher", ["sha2-256", "blake3"]) +async def test_saved_cid_digest_matches_stored_block( + create_ipfs: tuple[str, str], hasher: str +) -> None: + """The returned CID must hash the bytes passed in, not a dag-pb wrapper. + + This is what a CIDv0 broke: Kubo stored ``0a24 0802 121e 181e`` + (the UnixFS protobuf) rather than the payload, so the digest committed to by + the CID was the wrapper's. + """ + rpc, gw = create_ipfs + payload = b"issue 43 digest check for " + hasher.encode() + + async with KuboCAS(hasher=hasher, rpc_base_url=rpc, gateway_base_url=gw) as cas: + cid = await cas.save(payload, codec="raw") + + raw_digest = bytes(cid.raw_digest) + computed = multihash.digest(payload, cid.hashfun.name, size=len(raw_digest)) + + assert bytes(multihash.unwrap(computed)) == raw_digest + + +@pytest.mark.ipfs +@pytest.mark.parametrize("hasher", ["sha2-256", "blake3"]) +async def test_kubo_stores_payload_verbatim_under_returned_cid( + create_ipfs: tuple[str, str], hasher: str +) -> None: + """Kubo's own block store must hold the payload, unwrapped, at that CID. + + Asserted against the daemon rather than the client so a future regression in + the add parameters is caught even if py-hamt's own bookkeeping stays + self-consistent. + """ + rpc, gw = create_ipfs + payload = b"issue 43 block-store check for " + hasher.encode() + + async with KuboCAS(hasher=hasher, rpc_base_url=rpc, gateway_base_url=gw) as cas: + cid = await cas.save(payload, codec="raw") + + async with httpx.AsyncClient() as client: + response = await client.post(f"{rpc}/api/v0/block/get?arg={cid}") + response.raise_for_status() + + assert response.content == payload + + +@pytest.mark.ipfs +@pytest.mark.parametrize("hasher", ["sha2-256", "blake3"]) +async def test_saved_cid_is_verifiable( + create_ipfs: tuple[str, str], hasher: str +) -> None: + """verify_content must actually verify, not fall through on dag-pb.""" + rpc, gw = create_ipfs + payload = b"issue 43 verifiability for " + hasher.encode() + + async with KuboCAS( + hasher=hasher, rpc_base_url=rpc, gateway_base_url=gw, verify_content=True + ) as cas: + cid = await cas.save(payload, codec="raw") + + assert _cid_is_verifiable(cid, None, None), ( + f"{hasher} blocks are unverifiable, so verify_content is a no-op" + ) + assert await cas.load(cid) == payload + + +@pytest.mark.ipfs +async def test_hamt_roundtrips_under_sha2_256(create_ipfs: tuple[str, str]) -> None: + """A whole HAMT must build and read back on sha2-256, verification on.""" + rpc, gw = create_ipfs + + async with KuboCAS( + hasher="sha2-256", rpc_base_url=rpc, gateway_base_url=gw, verify_content=True + ) as cas: + hamt = await HAMT.build(cas=cas) + for index in range(64): + await hamt.set(f"key-{index}", index) + await hamt.make_read_only() + root = hamt.root_node_id + + assert isinstance(root, CID) + assert root.version == 1 + assert root.codec.code != KuboCAS.DAG_PB_MARKER + + read = await HAMT.build(cas=cas, root_node_id=root, read_only=True) + assert await read.len() == 64 + for index in range(64): + assert await read.get(f"key-{index}") == index + + +@pytest.mark.ipfs +@pytest.mark.parametrize("hasher", ["sha2-256", "blake3"]) +async def test_payload_over_chunker_returns_dag_pb_and_warns( + create_ipfs: tuple[str, str], hasher: str +) -> None: + """A payload larger than ``chunker`` still yields an unverifiable dag-pb CID. + + ``cid-version=1`` only makes *leaf* blocks raw. Once the payload exceeds the + chunker size Kubo builds a UnixFS dag-pb tree, so the root block is the + protobuf node rather than the bytes handed in. The requested codec cannot be + applied and ``_cid_is_verifiable()`` skips dag-pb, so ``verify_content`` + cannot check the object. + + This is a pre-existing limitation, not a regression from the cid-version + fix, and it affects both hashers equally -- ``blake3`` is not special here. + The data still round-trips; only verification is unavailable. ``save()`` + now warns instead of failing silently. + """ + rpc, gw = create_ipfs + # Comfortably over the small chunker configured below. + payload = os.urandom(300_000) + + async with KuboCAS( + hasher=hasher, + rpc_base_url=rpc, + gateway_base_url=gw, + chunker="size-65536", + verify_content=True, + ) as cas: + with pytest.warns(RuntimeWarning, match="dag-pb"): + cid = await cas.save(payload, codec="raw") + + assert cid.codec.code == KuboCAS.DAG_PB_MARKER + assert not _cid_is_verifiable(cid, None, None) + # Data integrity is unaffected -- only verifiability is lost. + assert await cas.load(cid) == payload + + +@pytest.mark.ipfs +async def test_payload_under_chunker_stays_raw_and_verifiable( + create_ipfs: tuple[str, str], +) -> None: + """The companion case: within the chunker, CIDs stay raw and checkable.""" + rpc, gw = create_ipfs + payload = os.urandom(10_000) + + async with KuboCAS( + hasher="sha2-256", + rpc_base_url=rpc, + gateway_base_url=gw, + chunker="size-65536", + verify_content=True, + ) as cas: + with warnings.catch_warnings(): + warnings.simplefilter("error", RuntimeWarning) + cid = await cas.save(payload, codec="raw") + + assert cid.codec.name == "raw" + assert _cid_is_verifiable(cid, None, None) + assert await cas.load(cid) == payload diff --git a/uv.lock b/uv.lock index c6479d8..12c2bf1 100644 --- a/uv.lock +++ b/uv.lock @@ -1656,7 +1656,7 @@ wheels = [ [[package]] name = "py-hamt" -version = "3.4.1" +version = "3.5.0" source = { editable = "." } dependencies = [ { name = "dag-cbor" },