diff --git a/CHANGELOG.md b/CHANGELOG.md index e894870..8840bb7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,18 @@ # Changelog +## 0.7.0 (2026-10-23) + +Opens time out. No placement change and no unit table change. + +- Every metadata document read or written and every store opened is timed out and + retried under `Retry`, as data reads are (Behaviour 6): by default 5 s times the attempt, + ten attempts. A server that accepts a connection and never answers held `wa.open` and + `wa.metadata` forever before; it now raises `TimeoutError` naming the URL and the + attempts (cellmap-flow, 2026-10-23). The policy is the call's `retries`, which + `wa.metadata`, `wa.detect` and `wa.open_tensorstore` now take as `wa.open`, + `wa.open_pyramid` and `wa.create` already did. Listings and the open that creates an + array are not timed out. + ## 0.6.0 (2026-10-23) One new argument, at chunkmirage's request. No placement change and no unit table change; diff --git a/README.md b/README.md index a012444..cd8d218 100644 --- a/README.md +++ b/README.md @@ -146,9 +146,10 @@ wa.open(path, *, level=None, voxel_size=None, level_mode="floor", on_mismatch="e metadata_conflicts="error", base_voxel_size=None, base_units=None, anchor=None, transform=None, strict_units=True) -> WorldArray wa.open_pyramid(path, ...) -> Pyramid # .names .levels .by_name() .select() .placements() .metas(): no opens needed -wa.metadata(path, level=None, units=None, open=True) -> ArrayMeta # open=False: from documents alone -wa.detect(path, missing_ok=False) -> Node # format, array or group, raw attrs -wa.open_tensorstore(path, level=None, context=None) +wa.metadata(path, level=None, units=None, open=True, retries=None) -> ArrayMeta # open=False: from documents alone +wa.detect(path, missing_ok=False, retries=None) -> Node # format, array or group, raw attrs +wa.open_tensorstore(path, level=None, context=None, retries=None) + # retries= times out and retries document reads and the open, as it does data reads wa.shared_context(cache_bytes=0, threads=None) ; wa.capped_context(cache_bytes=0, concurrency=1, network=True) ; wa.Retry(...) wa.thread_budget() # the shared context's default thread limit: OMP_NUM_THREADS, the job's slots, or CPUs; # threads=N limits it to N, threads=0 leaves tensorstore's own pool sizes diff --git a/docs/spec.md b/docs/spec.md index 072ecc8..c87ece8 100644 --- a/docs/spec.md +++ b/docs/spec.md @@ -177,10 +177,10 @@ pyr.voxel_sizes() ; pyr.translations() # where it lists the levels (OME-NGFF # choosing or placing a level opens nothing, whether the pyramid was opened by its # group path or by one of its level paths; else by opening each level -wa.metadata(path, level=None, units=None, open=True) -> ArrayMeta # open=False: from the documents alone -wa.detect(path, missing_ok=False) -> Node | None # format, kind (array|group), path, docs, scale_key; .attrs (raw user +wa.metadata(path, level=None, units=None, open=True, retries=None) -> ArrayMeta # open=False: from the documents alone +wa.detect(path, missing_ok=False, retries=None) -> Node | None # format, kind (array|group), path, docs, scale_key; .attrs (raw user # attributes), .multiscales (OME 0.5's `ome` unwrapped) -wa.open_tensorstore(path, level=None, context=None) -> tensorstore.TensorStore # reads no world metadata, so an +wa.open_tensorstore(path, level=None, context=None, retries=None) -> tensorstore.TensorStore # reads no world metadata, so an # array with odd placement attributes (a coordinate array) still opens wa.corner_to_centre(corner, voxel_size, axes=None) ; wa.default_axes(ndim) ; wa.index_range(begin, end, voxel_size, translation) wa.units.TABLE # the unit tables, public: {"length": {...metres per unit}, "time": {...seconds per unit}} @@ -521,7 +521,8 @@ this document and that one differ, fix one of them. one probe key counts as "missing" when another probe key was read under the same way, and only when every probe key is refused is the next way tried (see Open question 4). Metadata reads retry for about five seconds in total (`max_retries 5, initial_delay - 0.2s, max_delay 2s`); data reads keep tensorstore's defaults plus `Retry`. + 0.2s, max_delay 2s`); data reads keep tensorstore's defaults plus `Retry`, and each + document read and store open is also timed out under `Retry` (0.7, Behaviour 6). 6. **Writing is reading inverted, in one format.** The package writes OME-NGFF zarr only: zarr v2 with OME 0.4 (`.zgroup`, and `.zattrs` holding `multiscales`, each entry with @@ -649,9 +650,21 @@ this document and that one differ, fix one of them. still land after its cancel, with the same bytes the next attempt writes. Each retry is logged as a warning naming the error's first line, or its type when the message is empty (tensorstore's timeout is `TimeoutError()`); when every attempt times out the - `TimeoutError` names the operation, the path and the attempts. Metadata documents are not - retried this way (Convention 5). The policy is a per-handle attribute and a per-read and - per-write argument, never only constructor state. + `TimeoutError` names the operation, the path and the attempts. The policy is a + per-handle attribute and a per-read and per-write argument, never only constructor + state. + + Documents and opens (0.7): every metadata document read or written and every store + opened is timed out and retried the same way, as an operation of no bytes (by default + 5 s times the attempt, ten attempts), with tensorstore's own request retries + (Convention 5) running inside each attempt. A server that accepts a connection and + never answers used to hold `wa.open` and `wa.metadata` forever: cellmap-flow's picker + kept a 5 s probe of its own in front of them (2026-10-23). The policy is the `retries` + of the public call doing the reading: `open`, `open_pyramid`, `create`, `metadata`, + `detect` and `open_tensorstore`, which all take it; a document read later outside such + a call (a zarr v2 `.zattrs` read on demand, a pyramid placing its levels) waits under + `Retry()`. Listings are not timed out, their size being unknown, and neither is the + open that creates an array, which a second attempt would find already made. The watchdog (0.5): tensorstore up to at least 0.1.85 can strand a queued task. A pool worker idle for 20 ms exits, and a task queued as it exits gets no thread while the pool @@ -1019,7 +1032,8 @@ kicked by the watchdog, a dead context no longer watched, and a kick landing on context's own pools; `thread_budget` from each variable, a bad value skipped, and the shared context's limits. 0.6: `shared_context(threads=)` at `None`, `N` and `0`, one context per pair, a negative count refused, and a pickled handle coming back on the same -shared context. Opt-in (`WORLDARRAY_STRESS=1`, about a minute, +shared context. 0.7: `metadata`, `open` and `detect` against a server +that accepts and never answers raising `TimeoutError` after the attempts. Opt-in (`WORLDARRAY_STRESS=1`, about a minute, `tests/test_lost_wakeup.py`): worker processes aim tasks at the moment a one-thread pool's worker exits idle, so a plain context strands some for good (13 of about 7,000 on tensorstore 0.1.78; 40 to 44 of 12,000 to 15,000 with 16 workers on 0.1.75 and 0.1.85) @@ -1160,13 +1174,20 @@ over funlib geometry, by this spec's own assignment. a real bucket with listing disabled. One network test before 0.1.0 may change the rule. 4. **The network test** against a non-integer-voxel OpenOrganelle volume (Tests) is not yet written; it needs the dataset chosen from the catalog. -5. **Legacy attributes on new outputs.** cellmap-analyze writes funlib `voxel_size` and - `offset` (holding the centre) on every output and nothing in the five repos reads them; - dacapo's checkout has nine `open_ds` users, and cellmap-data and fibsem-tools were not - checked. Does any live consumer of cellmap-analyze or cellmap-flow outputs read funlib - attributes? If so, should the package offer `funlib_attrs(voxel_size, translation, axes, - units)`, writing `offset` as the corner, funlib's meaning, for `create(attrs=)`, or do - the repos keep writing their own? +5. **Legacy attributes on new outputs.** Resolved 2026-10-23: nobody needs them, so there + is no `funlib_attrs`. cellmap-analyze stopped writing funlib `voxel_size` and `offset` + (which held the centre) in 0.6.0, whose outputs carry only the OME documents `create` + writes. cellmap-flow writes them through funlib's own `prepare_ds`, `offset` as the + corner beside an agreeing OME 0.4 translation, and is their only reader. dacapo reads + funlib attributes through `open_ds` (`ome_metadata=False` by default), but nothing feeds + it these outputs, and on a zarr v3 one funlib 0.7 silently reads voxel size 1 and + offset 0. cellmap-data and fibsem-tools read OME only, and both pin `zarr<3`. funlib + 0.6 and 0.7 also require integer offsets and voxel sizes, so a float helper here would + either refuse placements or write numbers funlib truncates; a repo that wants the keys + passes them through `create(attrs=)`. Found on the way, outside this package: + cellmap-data (main 356c3ac, `image.py`) takes an OME translation as voxel 0's corner, + half a voxel off on every OME-correct store, and funlib's `open_ome_ds` puts the + translation straight into its corner `offset`. 6. **The block iterator's shape** (Behaviour 17). Resolved 2026-10-22: both sessions confirmed it. mesh-n-bone asked that `box` may reach past the array and that `clip` clips read boxes only, and derives its per-level boxes from one grid itself; @@ -1197,7 +1218,7 @@ over funlib geometry, by this spec's own assignment. unlisted has its placement nowhere until it is listed, so a `create(listed=False)` handle, and its pickles, would carry the placement from its arguments alone. Decide whether that is worth having, or whether an unlisted level should describe itself on - the array (open question 5's funlib attributes, or an OME-shaped array attribute). + the array (an OME-shaped array attribute; funlib keys are out, open question 5). 12. **Per-open object-store credentials.** cellmap-flow's MinIO writers open with a per-session key, secret and endpoint while the same process reads public S3 anonymously. tensorstore 0.1.85's `aws_credentials` takes no inline key and secret, and diff --git a/pyproject.toml b/pyproject.toml index f624a9c..a374b66 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "worldarray" -version = "0.6.0" +version = "0.7.0" 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" } diff --git a/src/worldarray/__init__.py b/src/worldarray/__init__.py index 2fc7822..253dcf5 100644 --- a/src/worldarray/__init__.py +++ b/src/worldarray/__init__.py @@ -28,7 +28,7 @@ parse_chunk_key, ) -__version__ = "0.6.0" +__version__ = "0.7.0" __all__ = [ "ArrayMeta", diff --git a/src/worldarray/array.py b/src/worldarray/array.py index 35c71bc..1f4c772 100644 --- a/src/worldarray/array.py +++ b/src/worldarray/array.py @@ -44,6 +44,7 @@ normalize_path, open_array, precomputed_scale_index, + retried_documents, ) logger = logging.getLogger(__name__) @@ -68,6 +69,7 @@ class WorldArray: ``voxel_size`` picks the level; a group with neither raises, listing its levels). """ + @retried_documents def __init__( self, path: str, @@ -1002,6 +1004,7 @@ class Pyramid: own metadata where it lists the levels' scales (OME-NGFF, COSEM N5, precomputed), and opens the levels only when it does not.""" + @retried_documents def __init__( self, path: str, @@ -1201,6 +1204,7 @@ def __len__(self) -> int: # Functions +@retried_documents def create( path: str, shape: Sequence[int] | None = None, @@ -1345,14 +1349,21 @@ def open_pyramid(path: str, **kwargs) -> Pyramid: return Pyramid(path, **kwargs) +@retried_documents def metadata( - path: str, *, level: str | int | None = None, units: Units = None, open: bool = True + path: str, + *, + level: str | int | None = None, + units: Units = None, + open: bool = True, + retries: Retry | None = None, ) -> ArrayMeta: """The ``ArrayMeta`` of an array (or of ``level`` under a group), in native units unless ``units`` says otherwise. ``open=False`` reads it from the documents alone, with no - tensorstore opened.""" + tensorstore opened. ``retries`` times out and retries each document read and the open + (Behaviour 6).""" if open: - vol = WorldArray(path, level=level, units=units) + vol = WorldArray(path, level=level, units=units, retries=retries) return vol.meta if units is None else vol.grid node = detect(path) if level == "first": @@ -1386,13 +1397,18 @@ def metadata( return meta.in_units(space=space, time=time) +@retried_documents def open_tensorstore( - path: str, *, level: str | int | None = None, context: ts.Context | None = None + path: str, + *, + level: str | int | None = None, + context: ts.Context | None = None, + retries: Retry | None = None, ) -> ts.TensorStore: """The raw tensorstore of an array or level: C order, origin 0, a precomputed single - channel dropped. The access fallbacks and shared context apply. No world metadata is - read, so an array whose placement attributes are odd (a coordinate array, a field) still - opens.""" + channel dropped. The access fallbacks and shared context apply, and ``retries`` to each + document read and the open (Behaviour 6). No world metadata is read, so an array whose + placement attributes are odd (a coordinate array, a field) still opens.""" context = context if context is not None else shared_context(0) node = detect(path) if level == "first": diff --git a/src/worldarray/store.py b/src/worldarray/store.py index b8aee74..60acce4 100644 --- a/src/worldarray/store.py +++ b/src/worldarray/store.py @@ -8,6 +8,8 @@ from __future__ import annotations +import contextlib +import contextvars import functools import json import logging @@ -21,6 +23,7 @@ import tensorstore as ts from .context import watch +from .retry import Retry, with_retries logger = logging.getLogger(__name__) T = TypeVar("T") @@ -33,6 +36,11 @@ _SCHEME = re.compile(r"^(s3|gs|http|https|file|memory)://") _REMOTE = ("s3://", "gs://", "http://", "https://") +# the Retry that documents and opens wait under: the policy of the public call they serve +_documents_retry: contextvars.ContextVar[Retry | None] = contextvars.ContextVar( + "worldarray_documents_retry", default=None +) + # --------------------------------------------------------------------------- # Paths @@ -189,6 +197,47 @@ def with_access(specs: Sequence[dict], attempt: Callable[[dict], T]) -> T: return result +@contextlib.contextmanager +def documents_retried(retry: Retry | None): + """Metadata documents read or written, and stores opened, inside this block are timed + out and retried per ``retry`` as reads of no bytes; ``None`` keeps the enclosing + policy, else ``Retry()`` (spec, Behaviour 6).""" + token = _documents_retry.set(retry) if retry is not None else None + try: + yield + finally: + if token is not None: + _documents_retry.reset(token) + + +def retried_documents(fn: Callable[..., T]) -> Callable[..., T]: + """``fn`` run under ``documents_retried(retries)``, ``retries`` being its keyword.""" + + @functools.wraps(fn) + def run(*args, **kwargs): + with documents_retried(kwargs.get("retries")): + return fn(*args, **kwargs) + + return run + + +def _where(spec: dict, key: str = "") -> str: + if spec["driver"] == "file": + return spec["path"] + key + if spec["driver"] == "http": + return spec["base_url"].rstrip("/") + "/" + spec.get("path", "").lstrip("/") + key + return f"{spec['driver']}://{spec.get('bucket', '')}/{spec.get('path', '')}{key}" + + +def _wait(start: Callable[[], Any], spec: dict, what: str, verb: str) -> Any: + """``start().result()`` for a document or an open: a server that accepts and never + answers would otherwise hold it forever (tensorstore's own request retries run inside + each attempt).""" + retry = _documents_retry.get() or Retry() + remote = spec["driver"] in ("http", "gcs", "s3") + return with_retries(start, 0, retry, remote=remote, what=what, verb=verb) + + def metadata_spec(spec: dict) -> dict: """``spec`` with its requests' retries capped, so a mistyped host fails in seconds.""" if spec["driver"] not in ("s3", "gcs", "http"): @@ -201,14 +250,15 @@ def _metadata_kvstore(spec_json: str, gcs_url: str | None) -> ts.KvStore: """One opened kvstore per spec, and per ``TENSORSTORE_GCS_HTTP_URL``: tensorstore's gcs driver reads that variable when the store is opened, so a cached store would keep talking to the host the variable named then.""" - return ts.KvStore.open(metadata_spec(json.loads(spec_json))).result() + spec = json.loads(spec_json) + 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( json.dumps(spec, sort_keys=True), os.environ.get("TENSORSTORE_GCS_HTTP_URL") ) - result = kv.read(key).result() + result = _wait(lambda: kv.read(key), spec, _where(spec, key), "read") return None if result.state == "missing" else bytes(result.value) @@ -216,7 +266,7 @@ 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.write(key, data).result() + _wait(lambda: kv.write(key, data), spec, _where(spec, key), "write") class Location: @@ -323,12 +373,14 @@ def _decode(docs: dict[str, bytes | None]) -> dict[str, Any]: return {k: json.loads(v) for k, v in docs.items() if v is not None} -def detect(path: str, *, missing_ok: bool = False) -> Node | None: +@retried_documents +def detect(path: str, *, missing_ok: bool = False, retries: Retry | None = None) -> Node | None: """Probe ``zarr.json``, ``.zarray``, ``.zgroup``, ``info``, ``attributes.json`` in that order; failing all, probe the parent for ``info`` and take the last path segment as a precomputed scale key, then probe ``s0`` for a group that has no document of its own. The ``.n5`` or ``.zarr`` suffix is never consulted. Nothing found raises - ``FileNotFoundError``, or returns None with ``missing_ok``.""" + ``FileNotFoundError``, or returns None with ``missing_ok``. ``retries`` times out and + retries each document read (Behaviour 6).""" path = normalize_path(path) docs = _decode(Location(path).read_many(PROBE_KEYS)) if "zarr.json" in docs: @@ -440,7 +492,12 @@ def attempt(kvstore: dict) -> ts.TensorStore: spec["assume_metadata"] = True if node.format == "precomputed" and scale_index is not None: spec["scale_index"] = scale_index - return ts.open(spec, read=True, write=write, context=watch(context)).result() + return _wait( + lambda: ts.open(spec, read=True, write=write, context=watch(context)), + kvstore, + node.path, + "open", + ) store = with_access(kvstore_specs(node.path), attempt) if node.format in ("n5", "precomputed"): diff --git a/tests/test_access.py b/tests/test_access.py index b9f2102..9d2a8ff 100644 --- a/tests/test_access.py +++ b/tests/test_access.py @@ -5,6 +5,8 @@ import gc import logging import os +import socket +import threading import time import weakref @@ -321,3 +323,38 @@ def test_a_shared_context_takes_the_callers_thread_count(monkeypatch): assert wa.context.from_recipe(("shared", 7)) is default with pytest.raises(ValueError): wa.shared_context(threads=-1) + + +@pytest.fixture +def silent_server(): + """An http address whose server accepts every connection and never answers.""" + srv = socket.socket() + srv.bind(("127.0.0.1", 0)) + srv.listen(16) + held = [] + + def accept(): + while True: + try: + held.append(srv.accept()[0]) + except OSError: + return + + threading.Thread(target=accept, daemon=True).start() + yield f"http://127.0.0.1:{srv.getsockname()[1]}" + srv.close() + for c in held: + c.close() + + +def test_a_server_that_never_answers_times_out_the_open(silent_server): + quick = wa.Retry(timeout=0.3, max_attempts=2, base_delay=0.0) + for open_it in ( + lambda url: wa.metadata(url, open=False, retries=quick), + lambda url: wa.open(url, retries=quick), + lambda url: wa.detect(url, retries=quick), + ): + start = time.monotonic() + with pytest.raises(TimeoutError, match="127.0.0.1.*2 attempts"): + open_it(f"{silent_server}/vol.zarr/s0") + assert time.monotonic() - start < 10 diff --git a/uv.lock b/uv.lock index 2c5e0df..48f0ffe 100644 --- a/uv.lock +++ b/uv.lock @@ -1247,7 +1247,7 @@ wheels = [ [[package]] name = "worldarray" -version = "0.6.0" +version = "0.7.0" source = { editable = "." } dependencies = [ { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" },