0.7.1: read a directory's documents at once

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Changelog
 
+## 0.7.1 (2026-10-23)
+
+Faster opens over http. No placement change, no unit table change, no API change.
+
+- The documents a call needs from one directory are read at once instead of in turn. An
+  http open of a zarr v2 level made its eight requests one after another (five probe keys,
+  the level's `.zattrs`, the parent's `zarr.json` and `.zattrs`); now they go in three
+  rounds. At 10 ms an answer, `wa.open` took 98 ms and takes 52 (cellmap-flow's
+  measurement). A server that never answers still costs one key's attempts, not five.
+
 ## 0.7.0 (2026-10-23)
 
 Opens time out. No placement change and no unit table change.

docs/spec.md

@@ -490,7 +490,9 @@ this document and that one differ, fix one of them.
 
 5. **Stores are read through tensorstore's kvstore only**, never zarr-python: `zarr.json`,
    `.zarray`, `.zattrs`, `.zgroup`, `attributes.json`, `info`. Format detection probes keys
-   in this order: `zarr.json`, `.zarray`, `.zgroup`, `info`, `attributes.json`; when none is
+   in this order: `zarr.json`, `.zarray`, `.zgroup`, `info`, `attributes.json`, the reads
+   begun all at once and the order deciding which document wins (0.7.1: one round trip
+   where an http open made five in a row, cellmap-flow's measurement); when none is
    found, the parent is probed for `info` and the last path segment is taken as a
    precomputed scale key (mesh-n-bone's case), then `s0` is probed so a group with no
    document of its own (an N5 directory holding `s0, s1, ...`) is still a group. The

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "worldarray"
-version = "0.7.0"
+version = "0.7.1"
 description = "Read zarr, N5 and Neuroglancer precomputed volumes in world coordinates through tensorstore: correct voxel placement, pyramid levels, padded region reads, resampling. numpy + tensorstore only."
 readme = "README.md"
 license = { file = "LICENSE" }

src/worldarray/__init__.py

@@ -28,7 +28,7 @@
     parse_chunk_key,
 )
 
-__version__ = "0.7.0"
+__version__ = "0.7.1"
 
 __all__ = [
     "ArrayMeta",

src/worldarray/store.py

@@ -254,18 +254,48 @@ def _metadata_kvstore(spec_json: str, gcs_url: str | None) -> ts.KvStore:
     return _wait(lambda: ts.KvStore.open(metadata_spec(spec)), spec, _where(spec), "open")
 
 
-def _read_key(spec: dict, key: str) -> bytes | None:
-    kv = _metadata_kvstore(
+def _metadata_kv(spec: dict) -> ts.KvStore:
+    return _metadata_kvstore(
         json.dumps(spec, sort_keys=True), os.environ.get("TENSORSTORE_GCS_HTTP_URL")
     )
-    result = _wait(lambda: kv.read(key), spec, _where(spec, key), "read")
-    return None if result.state == "missing" else bytes(result.value)
+
+
+def _read_keys(spec: dict, keys: Sequence[str]) -> dict[str, bytes | None | BaseException]:
+    """``keys`` under one way in, their reads begun at once: one round trip, not one per key
+    (an http open probes five). Each value is the document, None when it is missing, or the
+    error its read raised; after an error other than a refusal the rest are cancelled and
+    left out, so a server that never answers costs one key's attempts, not five."""
+    kv = _metadata_kv(spec)
+    started = {key: kv.read(key) for key in keys}
+    out: dict[str, bytes | None | BaseException] = {}
+    for key in keys:
+        first = [started[key]]
+
+        def start(key=key, first=first):
+            return first.pop() if first else kv.read(key)
+
+        try:
+            result = _wait(start, spec, _where(spec, key), "read")
+        except Exception as e:  # noqa: BLE001  the caller decides what a failed key means
+            out[key] = e
+            if not is_refused(e):
+                for future in started.values():
+                    future.cancel()
+                return out
+            continue
+        out[key] = None if result.state == "missing" else bytes(result.value)
+    return out
+
+
+def _read_key(spec: dict, key: str) -> bytes | None:
+    value = _read_keys(spec, [key])[key]
+    if isinstance(value, BaseException):
+        raise value
+    return value
 
 
 def _write_key(spec: dict, key: str, data: bytes) -> None:
-    kv = _metadata_kvstore(
-        json.dumps(spec, sort_keys=True), os.environ.get("TENSORSTORE_GCS_HTTP_URL")
-    )
+    kv = _metadata_kv(spec)
     _wait(lambda: kv.write(key, data), spec, _where(spec, key), "write")
 
 
@@ -292,20 +322,21 @@ def write_json(self, key: str, doc: dict) -> None:
         self.write(key, (json.dumps(doc, indent=2) + "\n").encode())
 
     def read_many(self, keys: Sequence[str]) -> dict[str, bytes | None]:
-        """All of ``keys`` under one way in. A bucket with listing off answers a refusal for a
-        missing key, so a refusal on one key counts as missing when another key was read
-        under the same way; only when every key is refused is the next way tried."""
+        """All of ``keys`` under one way in, read at once (``_read_keys``). A bucket with
+        listing off answers a refusal for a missing key, so a refusal on one key counts as
+        missing when another key was read under the same way; only when every key is refused
+        is the next way tried."""
 
         def attempt(spec: dict) -> dict[str, bytes | None]:
             found: dict[str, bytes | None] = {}
             refusals: dict[str, BaseException] = {}
-            for key in keys:
-                try:
-                    found[key] = _read_key(spec, key)
-                except Exception as e:
-                    if not is_refused(e):
-                        raise
-                    refusals[key] = e
+            for key, value in _read_keys(spec, keys).items():
+                if isinstance(value, BaseException):
+                    if not is_refused(value):
+                        raise value
+                    refusals[key] = value
+                else:
+                    found[key] = value
             if refusals and len(refusals) == len(keys):
                 raise Refused(next(iter(refusals.values())).args[0] if refusals else "refused")
             for key in refusals:

tests/test_access.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
+import functools
 import gc
+import http.server
 import logging
 import os
 import socket
@@ -56,6 +58,21 @@ def attempt(spec):
         store.with_access(specs, attempt)
 
 
+def _per_key(read_key):
+    """A ``store._read_keys`` stand-in built from a one-key fake."""
+
+    def read_keys(spec, keys):
+        out = {}
+        for key in keys:
+            try:
+                out[key] = read_key(spec, key)
+            except Exception as e:  # noqa: BLE001
+                out[key] = e
+        return out
+
+    return read_keys
+
+
 def test_a_refused_probe_key_counts_as_missing_when_another_key_was_read(monkeypatch):
     present = {"zarr.json": b'{"node_type": "array"}'}
 
@@ -66,7 +83,7 @@ def fake_read_key(spec, key):
             return present[key]
         raise RuntimeError("PERMISSION_DENIED: listing is off, so a missing key is a 403")
 
-    monkeypatch.setattr(store, "_read_key", fake_read_key)
+    monkeypatch.setattr(store, "_read_keys", _per_key(fake_read_key))
     loc = store.Location("s3://bucket/vol.zarr/s0")
     found = loc.read_many(store.PROBE_KEYS)
     assert found["zarr.json"] == present["zarr.json"]
@@ -79,7 +96,7 @@ def fake_read_key(spec, key):
             raise RuntimeError("PERMISSION_DENIED")
         return b"{}" if key == ".zarray" else None
 
-    monkeypatch.setattr(store, "_read_key", fake_read_key)
+    monkeypatch.setattr(store, "_read_keys", _per_key(fake_read_key))
     found = store.Location("s3://bucket/vol.zarr/s0").read_many(store.PROBE_KEYS)
     assert found[".zarray"] == b"{}" and found["zarr.json"] is None
 
@@ -358,3 +375,27 @@ def test_a_server_that_never_answers_times_out_the_open(silent_server):
         with pytest.raises(TimeoutError, match="127.0.0.1.*2 attempts"):
             open_it(f"{silent_server}/vol.zarr/s0")
         assert time.monotonic() - start < 10
+
+
+def test_an_http_open_probes_its_keys_at_once(tmp_path):
+    """Five probe keys, one round trip: each answer here takes 0.3 s."""
+    write_zarr2(tmp_path / "vol.zarr" / "s0", pattern((4, 4, 4)))
+
+    class Slow(http.server.SimpleHTTPRequestHandler):
+        def do_GET(self):
+            time.sleep(0.3)
+            super().do_GET()
+
+        def log_message(self, *args):
+            pass
+
+    handler = functools.partial(Slow, directory=str(tmp_path))
+    server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), handler)
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+    try:
+        start = time.monotonic()
+        node = wa.detect(f"http://127.0.0.1:{server.server_address[1]}/vol.zarr/s0")
+        assert node.format == "zarr2" and node.kind == "array"
+        assert time.monotonic() - start < 1.0  # in turn: 1.5 s
+    finally:
+        server.shutdown()