fix(store): retry the LocalStore rename when the destination is busy

changes/3522.bugfix.md

@@ -0,0 +1,9 @@
+`LocalStore` now retries the rename that publishes a written file when Windows
+reports the destination as transiently busy (`ERROR_ACCESS_DENIED` or
+`ERROR_SHARING_VIOLATION`). Replacing a name that was itself replaced moments
+earlier intermittently fails this way with no second process and no open handle
+involved, which aborted otherwise ordinary writes; the retry is bounded, is a
+no-op off Windows, and deliberately does not cover the `exclusive` path, whose
+`FileExistsError` reports an existing node rather than a busy one. Zarr v2 had
+the equivalent retry from #698 and it was not carried over when atomic writes
+arrived in #3412.

src/zarr/storage/_local.py

@@ -6,6 +6,7 @@
 import os
 import shutil
 import sys
+import time
 import uuid
 from pathlib import Path
 from typing import TYPE_CHECKING, BinaryIO, Literal, Self
@@ -22,7 +23,7 @@
 from zarr.core.common import AccessModeLiteral, concurrent_map
 
 if TYPE_CHECKING:
-    from collections.abc import AsyncIterator, Iterable, Iterator
+    from collections.abc import AsyncIterator, Callable, Iterable, Iterator
 
     from zarr.core.buffer import BufferPrototype
 
@@ -58,6 +59,53 @@ def _safe_move(src: Path, dst: Path) -> None:
         os.unlink(src)
 
 
+# Windows error codes meaning the destination could not be superseded *right
+# now*, as opposed to a permission problem that will not clear. Replacing a name
+# that was itself replaced moments earlier intermittently fails this way, with no
+# second process and no open handle involved, and a retry clears it in well under
+# a millisecond. Zarr v2 hit the same thing and fixed it in #698.
+#
+# Nothing else is retried. In particular the FileExistsError that the exclusive
+# path relies on to report an existing node is ERROR_ALREADY_EXISTS (183), so it
+# is excluded here by construction and still propagates on the first attempt.
+_TRANSIENT_WINERRORS = frozenset(
+    {
+        5,  # ERROR_ACCESS_DENIED
+        32,  # ERROR_SHARING_VIOLATION
+    }
+)
+
+# Delay before each successive attempt; the leading 0.0 is the original attempt.
+# Measured on Windows 11 over 20,000 replaces onto an existing destination: 720
+# failed with no retry and none with, 475 of those clearing on the second attempt
+# and the worst on the fourth. The tail is headroom for a busier machine, and the
+# whole sequence sums to under a second so a genuine failure still surfaces
+# promptly.
+_RETRY_DELAYS = (0.0, 0.001, 0.005, 0.02, 0.05, 0.2)
+
+
+def _move_with_retry(tmp_path: Path, path: Path, move: Callable[[Path, Path], object]) -> None:
+    """Run ``move(tmp_path, path)``, retrying while the destination is busy.
+
+    This is a single attempt on every platform but Windows, without needing to
+    test for one: only ``winerror`` values are ever retried, and off Windows an
+    ``OSError`` does not carry one.
+    """
+    last_error: OSError
+    for delay in _RETRY_DELAYS:
+        if delay:
+            time.sleep(delay)
+        try:
+            move(tmp_path, path)
+        except OSError as e:
+            if getattr(e, "winerror", None) not in _TRANSIENT_WINERRORS:
+                raise
+            last_error = e
+        else:
+            return
+    raise last_error
+
+
 @contextlib.contextmanager
 def _atomic_write(
     path: Path,
@@ -69,9 +117,9 @@ def _atomic_write(
         with tmp_path.open(mode) as f:
             yield f
         if exclusive:
-            _safe_move(tmp_path, path)
+            _move_with_retry(tmp_path, path, _safe_move)
         else:
-            tmp_path.replace(path)
+            _move_with_retry(tmp_path, path, Path.replace)
     except Exception:
         tmp_path.unlink(missing_ok=True)
         raise

tests/test_store/test_local.py

@@ -2,6 +2,7 @@
 
 import pathlib
 import re
+from typing import TYPE_CHECKING
 
 import numpy as np
 import pytest
@@ -10,10 +11,13 @@
 from zarr import create_array
 from zarr.core.buffer import Buffer, cpu
 from zarr.storage import LocalStore
-from zarr.storage._local import _atomic_write
+from zarr.storage._local import _RETRY_DELAYS, _atomic_write, _move_with_retry
 from zarr.testing.store import StoreTests
 from zarr.testing.utils import assert_bytes_equal
 
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
 
 class TestLocalStore(StoreTests[LocalStore, cpu.Buffer]):
     store_cls = LocalStore
@@ -164,3 +168,84 @@ def test_atomic_write_exclusive_preexisting(tmp_path: pathlib.Path) -> None:
             f.write(b"abc")
     assert path.read_bytes() == b"xyz"
     assert list(path.parent.iterdir()) == [path]  # no temp files
+
+
+def _oserror(winerror: int) -> OSError:
+    """An OSError shaped like the one a failed MoveFileEx produces."""
+    error = OSError(13, "Access is denied")
+    error.winerror = winerror  # type: ignore[attr-defined]
+    return error
+
+
+def _flaky_move(failures: int, error: OSError) -> Callable[[pathlib.Path, pathlib.Path], None]:
+    """A move that raises ``error`` the first ``failures`` times it is called."""
+    attempts = 0
+
+    def move(src: pathlib.Path, dst: pathlib.Path) -> None:
+        nonlocal attempts
+        attempts += 1
+        if attempts <= failures:
+            raise error
+        src.replace(dst)
+
+    move.attempts = lambda: attempts  # type: ignore[attr-defined]
+    return move
+
+
+@pytest.mark.parametrize("winerror", [5, 32])
+@pytest.mark.parametrize("failures", [1, 2, 3])
+def test_move_with_retry_recovers(tmp_path: pathlib.Path, winerror: int, failures: int) -> None:
+    """A destination that is briefly busy is retried, not reported."""
+    src = tmp_path / "src"
+    dst = tmp_path / "dst"
+    src.write_bytes(b"abc")
+    move = _flaky_move(failures, _oserror(winerror))
+
+    _move_with_retry(src, dst, move)
+
+    assert dst.read_bytes() == b"abc"
+    assert move.attempts() == failures + 1  # type: ignore[attr-defined]
+
+
+def test_move_with_retry_gives_up(tmp_path: pathlib.Path) -> None:
+    """A destination that never frees still raises, after a bounded wait."""
+    src = tmp_path / "src"
+    dst = tmp_path / "dst"
+    src.write_bytes(b"abc")
+    move = _flaky_move(len(_RETRY_DELAYS), _oserror(5))
+
+    with pytest.raises(OSError, match="Access is denied"):
+        _move_with_retry(src, dst, move)
+
+    assert move.attempts() == len(_RETRY_DELAYS)  # type: ignore[attr-defined]
+    assert not dst.exists()
+
+
+@pytest.mark.parametrize("winerror", [2, 3, 183, None])
+def test_move_with_retry_does_not_retry_other_errors(
+    tmp_path: pathlib.Path, winerror: int | None
+) -> None:
+    """Only a busy destination is transient; everything else fails at once.
+
+    183 is the case that matters: ``ERROR_ALREADY_EXISTS`` is how the
+    ``exclusive`` path reports that a node is already there, and retrying it
+    would overwrite what ``_safe_move`` refused to touch.
+    """
+    src = tmp_path / "src"
+    dst = tmp_path / "dst"
+    src.write_bytes(b"abc")
+    error = OSError(17, "boom")
+    if winerror is not None:
+        error.winerror = winerror  # type: ignore[attr-defined]
+    move = _flaky_move(1, error)
+
+    with pytest.raises(OSError, match="boom"):
+        _move_with_retry(src, dst, move)
+
+    assert move.attempts() == 1  # type: ignore[attr-defined]
+
+
+def test_retry_delays_are_bounded() -> None:
+    """A stuck destination must not turn a fast failure into a long hang."""
+    assert sum(_RETRY_DELAYS) < 1.0
+    assert _RETRY_DELAYS[0] == 0.0  # the original attempt is not delayed