fix bug in windows

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.

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:<module>` (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`)
 

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:

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()

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)

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)