diff --git a/CHANGELOG.md b/CHANGELOG.md index a1ff3a37..57e50bc0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -60,6 +60,7 @@ plate.get_images(max_workers=8) # was get_images_ ### Fixes +- Windows: concurrent access to a store no longer fails with `PermissionError: [WinError 5]`/`[WinError 32]`. Windows refuses to replace or remove a file while another handle to it is open, so a concurrent *reader* of `zarr.json` could break a writer's atomic rename — including between parallel `atomic_add_image` workers, which ngio's lock cannot prevent since opening a group reads metadata before any lock exists. Store operations now absorb these transient conflicts with a short bounded retry, always on and independent of `io_retry`; the original error is raised once the bound is reached. No behaviour change on Linux or macOS. - `import ngio` no longer raises `AttributeError` when an s3fs older than 2026.2.0 is installed. - `concatenate_image_tables` built a wrong index: unnamed, and duplicated under `mode="lazy"`. - `Roi.union`/`intersection` dropped ROI name `""` and label `0`; `Roi.from_values` now validates its inputs. diff --git a/docs/getting_started/7_configuration.md b/docs/getting_started/7_configuration.md index adb24442..58787351 100644 --- a/docs/getting_started/7_configuration.md +++ b/docs/getting_started/7_configuration.md @@ -61,6 +61,7 @@ ngio's own errors (`NgioError` subclasses, e.g. validation errors) are never ret - **Zarr IO snapshots the policy at open time.** Every group ngio opens is backed by a store that copies the current `io_retry` at construction. The snapshot travels with the store — including into pickled dask task graphs, so workers retry with the policy that was active on the driver. Changing `get_config().io_retry` afterwards does not affect already-open containers. - **Non-zarr IO reads the policy at call time.** The table backends and store probes check the current global config on every call, so runtime changes apply immediately there. - Retries are logged as warnings, including the error, attempt count, and sleep time. ngio names its loggers `ngio:` (here `ngio:ngio.utils._retry`) — note the colon, which means they are not children of a `ngio` logger in Python's dot-separated hierarchy, so attach handlers to the full name. +- **Windows file-sharing conflicts are always retried**, whatever `io_retry` says. Windows refuses to replace or remove a file while another handle to it is open, so an unrelated concurrent *reader* can break a writer's atomic rename (`WinError 5`, `32` or `33`). The conflict clears in milliseconds, so ngio absorbs it with a short bounded retry (up to ~0.5s, logged at debug level) before the error ever reaches `io_retry`. This is a platform quirk rather than a policy, so it is not configurable; after the bound the original error is raised. Nothing changes on Linux or macOS. If `retry_on` also matches `PermissionError`, the two layers compose multiplicatively. ## s3fs retry markers (`s3fs`) diff --git a/src/ngio/utils/_retry.py b/src/ngio/utils/_retry.py index e998c189..a43809a8 100644 --- a/src/ngio/utils/_retry.py +++ b/src/ngio/utils/_retry.py @@ -2,16 +2,20 @@ The retry policy is defined by `ngio.config.RetryConfig`. With the default policy (`max_retries=0`) every helper here degrades to a plain call. + +Windows file-sharing conflicts are handled separately, by an always-on retry +that does not depend on the user policy. See `aretry_sharing_violation`. """ import asyncio import functools import logging +import sys import time from collections.abc import Awaitable, Callable from typing import ParamSpec, TypeVar -from ngio.config import RetryConfig, get_config +from ngio.config import ExponentialBackoff, RetryConfig, get_config from ngio.utils._errors import NgioError logger = logging.getLogger(f"ngio:{__name__}") @@ -22,6 +26,9 @@ # Kept as a module attribute so tests can monkeypatch it. _sleep = time.sleep +# Kept as a module attribute so tests can simulate Windows on any platform. +_IS_WINDOWS = sys.platform == "win32" + _NEVER_RETRY: tuple[type[BaseException], ...] = ( NgioError, KeyboardInterrupt, @@ -30,6 +37,13 @@ GeneratorExit, ) +# ERROR_ACCESS_DENIED, ERROR_SHARING_VIOLATION, ERROR_LOCK_VIOLATION. +_SHARING_VIOLATION_WINERRORS = frozenset({5, 32, 33}) +_SHARING_VIOLATION_ATTEMPTS = 8 +_SHARING_VIOLATION_BACKOFF = ExponentialBackoff( + delay_s=0.005, max_delay_s=0.2, jitter=True +) + def is_retryable(exc: BaseException, policy: RetryConfig) -> bool: """Return whether the error should be retried under the given policy. @@ -45,6 +59,55 @@ def is_retryable(exc: BaseException, policy: RetryConfig) -> bool: return any(marker in error_repr for marker in policy.retry_on) +def is_sharing_violation(exc: BaseException) -> bool: + """Return whether the error is a transient Windows file-sharing conflict. + + Matches on the Win32 error code rather than on `PermissionError`, so a + remote store's 403 (also raised as `PermissionError`) is never mistaken + for one. Ngio's own errors are never treated as retryable. + """ + if isinstance(exc, _NEVER_RETRY) or not isinstance(exc, OSError): + return False + return getattr(exc, "winerror", None) in _SHARING_VIOLATION_WINERRORS + + +async def aretry_sharing_violation( + fn: Callable[[], Awaitable[T]], *, op_name: str = "" +) -> T: + """Await `fn()`, retrying transient Windows file-sharing conflicts. + + Windows refuses `os.replace` and `shutil.rmtree` on a path while another + handle to it is open without `FILE_SHARE_DELETE`, which CPython never + passes. An unrelated concurrent *reader* therefore breaks a writer's + atomic rename. The condition clears in milliseconds, so this retry is + always on, unlike `io_retry` which is a user policy and defaults to off. + + Bounded: after `_SHARING_VIOLATION_ATTEMPTS` the original error is raised. + + Note: + Never use blocking sleeps here: this runs on zarr's single IO event + loop, where `asyncio.sleep` suspends only the retrying coroutine. + """ + for attempt in range(1, _SHARING_VIOLATION_ATTEMPTS): + try: + return await fn() + except OSError as e: + if not is_sharing_violation(e): + raise + delay = _SHARING_VIOLATION_BACKOFF.compute_delay(attempt) + logger.debug( + "Windows file-sharing conflict on %s (winerror %s), retrying in " + "%.3fs (attempt %d/%d)", + op_name or "IO operation", + getattr(e, "winerror", None), + delay, + attempt, + _SHARING_VIOLATION_ATTEMPTS, + ) + await asyncio.sleep(delay) + return await fn() + + def _log_retry( op_name: str, attempt: int, policy: RetryConfig, delay: float, exc: Exception ) -> None: diff --git a/src/ngio/utils/_store.py b/src/ngio/utils/_store.py index 54b38f72..6ac8d143 100644 --- a/src/ngio/utils/_store.py +++ b/src/ngio/utils/_store.py @@ -9,6 +9,7 @@ from __future__ import annotations import asyncio +import functools import json import logging import warnings @@ -23,8 +24,9 @@ from zarr.storage._common import make_store from ngio.config import RetryConfig, get_config +from ngio.utils import _retry from ngio.utils._errors import NgioValueError -from ngio.utils._retry import aretry_call +from ngio.utils._retry import aretry_call, aretry_sharing_violation from ngio.utils._warnings import NgioUserWarning if TYPE_CHECKING: @@ -237,6 +239,14 @@ async def _collect() -> list[str]: # ------------------------------------------------------------------ async def _io(self, op_name: str, fn: Callable[[], Awaitable[T]]) -> T: + # Windows refuses an atomic rename or a directory removal while any + # other handle to the target is open, so a concurrent reader breaks an + # unrelated writer. Always retried, independently of `io_retry`: it is a + # platform quirk with a millisecond lifetime, not a user IO policy. The + # module attribute is read at call time so it stays monkeypatchable and + # never travels into a pickled store. + if _retry._IS_WINDOWS: + fn = functools.partial(aretry_sharing_violation, fn, op_name=op_name) if self._retry.max_retries == 0: return await fn() return await aretry_call(fn, self._retry, op_name=op_name) @@ -246,14 +256,14 @@ def _retried_list( ) -> AsyncIterator[str]: # A partially-consumed generator cannot be safely retried, so the # listing is materialized inside the retried call and re-yielded. - if self._retry.max_retries == 0: + if self._retry.max_retries == 0 and not _retry._IS_WINDOWS: return factory() async def gen() -> AsyncGenerator[str, None]: async def collect() -> list[str]: return [key async for key in factory()] - for key in await aretry_call(collect, self._retry, op_name=op_name): + for key in await self._io(op_name, collect): yield key return gen() diff --git a/tests/unit/utils/test_retry.py b/tests/unit/utils/test_retry.py index c80126dc..cee279ca 100644 --- a/tests/unit/utils/test_retry.py +++ b/tests/unit/utils/test_retry.py @@ -12,8 +12,14 @@ NgioConfig, RetryConfig, ) -from ngio.utils import NgioValueError, retry_io -from ngio.utils._retry import aretry_call, is_retryable, retry_call +from ngio.utils import NgioFileExistsError, NgioValueError, retry_io +from ngio.utils._retry import ( + _SHARING_VIOLATION_WINERRORS, + aretry_call, + is_retryable, + is_sharing_violation, + retry_call, +) _NO_BACKOFF = ConstantBackoff(delay_s=0.0, jitter=False) @@ -61,6 +67,40 @@ def test_control_flow_exceptions_never_retried(self, exc): assert not is_retryable(exc, policy) +def _win_error(winerror: int) -> PermissionError: + """Build the error Windows raises for a file-sharing conflict. + + `winerror` is assignable on any platform, so these tests run everywhere. + """ + exc = PermissionError(13, "Access is denied") + exc.winerror = winerror + return exc + + +class TestIsSharingViolation: + @pytest.mark.parametrize("winerror", sorted(_SHARING_VIOLATION_WINERRORS)) + def test_matches_win32_codes(self, winerror): + assert is_sharing_violation(_win_error(winerror)) + + def test_other_win32_code_not_matched(self): + assert not is_sharing_violation(_win_error(2)) + + def test_permission_error_without_winerror_not_matched(self): + # The shape botocore/s3fs raise for an S3 403. + assert not is_sharing_violation(PermissionError("Access Denied")) + + def test_ngio_errors_never_matched(self): + # NgioFileExistsError is an OSError subclass, so it reaches the check. + exc = NgioFileExistsError("already there") + exc.winerror = 5 + assert not is_sharing_violation(exc) + + def test_non_os_error_not_matched(self): + exc = ValueError("boom") + exc.winerror = 5 + assert not is_sharing_violation(exc) + + class TestBackoffStrategies: def test_constant(self): backoff = ConstantBackoff(delay_s=0.5, jitter=False) diff --git a/tests/unit/utils/test_store.py b/tests/unit/utils/test_store.py index dd262acd..d4e3465e 100644 --- a/tests/unit/utils/test_store.py +++ b/tests/unit/utils/test_store.py @@ -21,8 +21,14 @@ from zarr.abc.store import ByteRequest from zarr.core.buffer import BufferPrototype +import ngio.utils._retry as retry_mod from ngio.config import ConstantBackoff, RetryConfig -from ngio.utils import NgioStore, NgioUserWarning, NgioValueError +from ngio.utils import ( + NgioFileExistsError, + NgioStore, + NgioUserWarning, + NgioValueError, +) _RETRY = RetryConfig( max_retries=3, @@ -246,6 +252,123 @@ async def collect(): assert flaky.attempts["get"] == 3 # 1 failure + 2 successes +class SharingViolationStore(MemoryStore): + """A MemoryStore raising a Windows sharing violation n times per method.""" + + def __init__(self, fail_times: int = 1, winerror: int = 5, exc=None, **kwargs): + super().__init__(**kwargs) + self.fail_times = fail_times + self.winerror = winerror + self.exc = exc + self.attempts: Counter[str] = Counter() + + def _flake(self, method: str) -> None: + self.attempts[method] += 1 + if self.attempts[method] > self.fail_times: + return + if self.exc is not None: + raise self.exc + exc = PermissionError(13, "Access is denied") + exc.winerror = self.winerror + raise exc + + async def set( + self, key: str, value: Buffer, byte_range: tuple[int, int] | None = None + ) -> None: + self._flake("set") + return await super().set(key, value, byte_range) + + async def exists(self, key: str) -> bool: + self._flake("exists") + return await super().exists(key) + + def list_dir(self, prefix: str) -> AsyncIterator[str]: + self._flake("list_dir") + return super().list_dir(prefix) + + +class TestSharingViolationRetry: + """Windows sharing violations are absorbed independently of `io_retry`.""" + + @pytest.fixture(autouse=True) + def _simulate_windows(self, monkeypatch): + monkeypatch.setattr(retry_mod, "_IS_WINDOWS", True) + monkeypatch.setattr( + retry_mod, + "_SHARING_VIOLATION_BACKOFF", + ConstantBackoff(delay_s=0.0, jitter=False), + ) + + @pytest.mark.parametrize("winerror", [5, 32, 33]) + def test_recovers_with_retries_disabled(self, winerror): + flaky = SharingViolationStore(fail_times=1, winerror=winerror) + store = NgioStore(flaky, retry=RetryConfig()) + assert sync(store.exists("nope")) is False + assert flaky.attempts["exists"] == 2 + + def test_exhaustion_raises_the_original_error(self): + flaky = SharingViolationStore(fail_times=100) + store = NgioStore(flaky, retry=RetryConfig()) + with pytest.raises(PermissionError) as excinfo: + sync(store.exists("nope")) + assert excinfo.value.winerror == 5 + assert flaky.attempts["exists"] == retry_mod._SHARING_VIOLATION_ATTEMPTS + + def test_not_retried_off_windows(self, monkeypatch): + monkeypatch.setattr(retry_mod, "_IS_WINDOWS", False) + flaky = SharingViolationStore(fail_times=1) + store = NgioStore(flaky, retry=RetryConfig()) + with pytest.raises(PermissionError): + sync(store.exists("nope")) + assert flaky.attempts["exists"] == 1 + + @pytest.mark.parametrize( + "exc", + [ + PermissionError("Access Denied"), # the s3fs 403 shape + NgioFileExistsError("already there"), + OSError("plain"), + ], + ) + def test_other_errors_propagate_immediately(self, exc): + flaky = SharingViolationStore(fail_times=1, exc=exc) + store = NgioStore(flaky, retry=RetryConfig()) + with pytest.raises(type(exc)): + sync(store.exists("nope")) + assert flaky.attempts["exists"] == 1 + + def test_writes_recover(self): + flaky = SharingViolationStore(fail_times=1) + store = NgioStore(flaky, retry=RetryConfig()) + group = zarr.open_group(store=store, mode="a") + group.attrs["marker"] = 42 + assert flaky.attempts["set"] > flaky.fail_times + + def test_listing_recovers(self): + flaky = SharingViolationStore(fail_times=1) + store = NgioStore(flaky, retry=RetryConfig()) + + async def collect(): + return [k async for k in store.list_dir("")] + + assert sync(collect()) == [] + assert flaky.attempts["list_dir"] == 2 + + def test_composes_multiplicatively_with_io_retry(self): + flaky = SharingViolationStore(fail_times=1000) + retry = RetryConfig( + max_retries=3, + retry_on=["PermissionError"], + backoff=ConstantBackoff(delay_s=0.0, jitter=False), + ) + store = NgioStore(flaky, retry=retry) + with pytest.raises(PermissionError): + sync(store.exists("nope")) + assert flaky.attempts["exists"] == retry_mod._SHARING_VIOLATION_ATTEMPTS * ( + retry.max_retries + 1 + ) + + class TestStoreBehavior: def test_pickle_roundtrip(self, tmp_path): store = NgioStore.from_any(tmp_path / "data.zarr", mode="a", retry=_RETRY)