fix(core.utils): retry path.unlink on Windows sharing violations

CI on win-64 hit ``PermissionError: [WinError 5] Access is denied`` in
``test_concurrent_eviction_does_not_delete_replaced_file``: an evictor
trying to unlink an entry that another process briefly held open for a
read crashed the writer that triggered eviction.

Python's ``open(p, "rb")`` does not pass ``FILE_SHARE_DELETE``, so a
reader from another process briefly blocks our unlink. Mirror the
existing ``_replace_with_sharing_retry`` and ``_stat_and_read_with_sharing_retry``
helpers with ``_unlink_with_sharing_retry``: same backoff budget, same
winerror set (5/32/33 + EACCES), same non-Windows passthrough.

Apply at every entry-file unlink site:
* ``_prune_if_stat_unchanged`` -- best-effort, catches both FileNotFoundError
  and the post-retry PermissionError.
* ``__init__`` schema-mismatch wipe -- same, with a comment that the
  next open will retry.
* ``__delitem__`` -- retries; FileNotFoundError still maps to KeyError;
  exhausted-budget PermissionError propagates so explicit deletes fail
  loudly.
* ``_enforce_size_cap`` -- retries; on exhausted budget the entry is
  skipped (a later eviction pass will retry) instead of crashing the
  writer that triggered eviction.

Tests cover the retry-then-succeed path, the non-Windows passthrough,
and the eviction skip-on-locked path.

Author: cpcloud

cuda_core/cuda/core/utils/_program_cache.py

@@ -1194,6 +1194,42 @@ def _touch_atime(path: Path, st_before: os.stat_result) -> None:
         os.utime(path, ns=(new_atime_ns, st_before.st_mtime_ns))
 
 
+def _unlink_with_sharing_retry(path: Path) -> None:
+    """Unlink with Windows-specific retry on sharing/lock violations.
+
+    On Windows, ``Path.unlink`` raises ``PermissionError`` (winerror 5,
+    32, or 33; sometimes bare ``EACCES``) when another process holds
+    the file open without ``FILE_SHARE_DELETE``. Python's default
+    ``open(p, "rb")`` does not pass that flag, so a reader from another
+    process briefly blocks our unlink while it reads. Retry with the
+    same backoff budget as :func:`_replace_with_sharing_retry` so
+    transient contention is not turned into a propagated error.
+
+    Raises ``FileNotFoundError`` if the file is absent; the last
+    ``PermissionError`` if the Windows retry budget is exhausted; and
+    propagates any non-sharing ``PermissionError`` (or any non-Windows
+    ``PermissionError``) immediately.
+    """
+    last_exc: PermissionError | None = None
+    for delay in _REPLACE_RETRY_DELAYS:
+        if delay:
+            time.sleep(delay)
+        try:
+            path.unlink()
+            return
+        except FileNotFoundError:
+            raise
+        except PermissionError as exc:
+            if not _IS_WINDOWS:
+                raise
+            winerror = getattr(exc, "winerror", None)
+            if winerror not in _SHARING_VIOLATION_WINERRORS and exc.errno != errno.EACCES:
+                raise
+            last_exc = exc
+    if last_exc is not None:
+        raise last_exc
+
+
 def _prune_if_stat_unchanged(path: Path, st_before: os.stat_result) -> None:
     """Unlink ``path`` iff its stat still matches ``st_before``.
 
@@ -1205,6 +1241,10 @@ def _prune_if_stat_unchanged(path: Path, st_before: os.stat_result) -> None:
     delete their work. The residual TOCTOU window between stat and
     unlink is narrow; worst case, a very-recently-written entry is
     removed and the next read recompiles.
+
+    Best-effort: a Windows sharing violation that survives the retry
+    budget leaves the file in place. The caller is in an eviction or
+    cleanup pass, so re-trying on the next pass is the right outcome.
     """
     try:
         st_now = path.stat()
@@ -1214,8 +1254,8 @@ def _prune_if_stat_unchanged(path: Path, st_before: os.stat_result) -> None:
     key_now = (st_now.st_ino, st_now.st_size, st_now.st_mtime_ns)
     if key_before != key_now:
         return
-    with contextlib.suppress(FileNotFoundError):
-        path.unlink()
+    with contextlib.suppress(FileNotFoundError, PermissionError):
+        _unlink_with_sharing_retry(path)
 
 
 class FileStreamProgramCache(ProgramCacheResource):
@@ -1307,9 +1347,11 @@ def __init__(
             if existing != expected:
                 # Schema mismatch: wipe incompatible entries. Losing cache
                 # contents is safe; returning bytes from an old format is not.
+                # Tolerate Windows sharing violations -- another process may
+                # be reading an old entry; the next open will retry the wipe.
                 for entry in list(self._iter_entry_paths()):
-                    with contextlib.suppress(FileNotFoundError):
-                        entry.unlink()
+                    with contextlib.suppress(FileNotFoundError, PermissionError):
+                        _unlink_with_sharing_retry(entry)
                 self._schema_path.write_text(expected)
         # Opportunistic startup sweep of orphaned temp files left by any
         # crashed writers. Age-based so concurrent in-flight writes from
@@ -1390,7 +1432,7 @@ def __setitem__(self, key: object, value: bytes | bytearray | memoryview | Objec
     def __delitem__(self, key: object) -> None:
         path = self._path_for_key(key)
         try:
-            path.unlink()
+            _unlink_with_sharing_retry(path)
         except FileNotFoundError:
             raise KeyError(key) from None
 
@@ -1514,6 +1556,12 @@ def _enforce_size_cap(self) -> None:
                 # below could declare us done while still over the limit.
                 total += stat_now.st_size - size
                 continue
-            with contextlib.suppress(FileNotFoundError):
-                path.unlink()
+            # Tolerate Windows sharing violations during eviction: another
+            # process may briefly hold the file open for a read. Skip this
+            # entry; a later eviction pass will retry. Same outcome as if
+            # the stat-guard above had triggered.
+            try:
+                _unlink_with_sharing_retry(path)
                 total -= size
+            except (FileNotFoundError, PermissionError):
+                pass

cuda_core/tests/test_program_cache.py

@@ -736,6 +736,84 @@ def _flaky_replace(src, dst):
         assert cache[b"k"] == b"v"
 
 
+def test_filestream_cache_unlink_retries_on_sharing_violation(tmp_path, monkeypatch):
+    """``Path.unlink`` retries on Windows sharing violations (winerror 5/32/33
+    or bare EACCES). Without this, an evictor unlinking an entry that another
+    process briefly holds open for a read crashes the writer that triggered
+    eviction -- which is what the win-64 multiprocess CI was hitting."""
+    from pathlib import Path
+
+    from cuda.core.utils import FileStreamProgramCache, _program_cache
+
+    monkeypatch.setattr(_program_cache, "_IS_WINDOWS", True)
+
+    real_unlink = Path.unlink
+    calls = {"n": 0}
+
+    def _flaky_unlink(self, *args, **kwargs):
+        calls["n"] += 1
+        if calls["n"] < 3:
+            exc = PermissionError("sharing violation")
+            exc.winerror = 32
+            raise exc
+        return real_unlink(self, *args, **kwargs)
+
+    with FileStreamProgramCache(tmp_path / "fc") as cache:
+        cache[b"k"] = _fake_object_code(b"v")
+        monkeypatch.setattr(Path, "unlink", _flaky_unlink)
+        del cache[b"k"]  # succeeds on third attempt
+        assert calls["n"] == 3
+        assert b"k" not in cache
+
+
+def test_filestream_cache_unlink_propagates_non_windows_permission_error(tmp_path, monkeypatch):
+    """On non-Windows, PermissionError on unlink is a real config error and
+    must surface, never silently retried."""
+    from pathlib import Path
+
+    from cuda.core.utils import FileStreamProgramCache, _program_cache
+
+    monkeypatch.setattr(_program_cache, "_IS_WINDOWS", False)
+
+    with FileStreamProgramCache(tmp_path / "fc") as cache:
+        cache[b"k"] = _fake_object_code(b"v")
+
+        def _denied(self, *args, **kwargs):
+            raise PermissionError("denied")
+
+        monkeypatch.setattr(Path, "unlink", _denied)
+        with pytest.raises(PermissionError, match="denied"):
+            del cache[b"k"]
+
+
+def test_filestream_cache_unlink_skips_locked_entries_in_eviction(tmp_path, monkeypatch):
+    """Eviction tolerates a Windows sharing violation: a locked entry is
+    skipped (a later eviction pass will retry), and the eviction does not
+    propagate the error to the writer that triggered it."""
+    from pathlib import Path
+
+    from cuda.core.utils import FileStreamProgramCache, _program_cache
+
+    monkeypatch.setattr(_program_cache, "_IS_WINDOWS", True)
+
+    # Tiny cap so every write triggers eviction.
+    with FileStreamProgramCache(tmp_path / "fc", max_size_bytes=200) as cache:
+        cache[b"a"] = b"a" * 100
+        cache[b"b"] = b"b" * 100  # both still fit
+
+        # Now make ALL unlink attempts fail with a sharing violation. The
+        # next write triggers eviction; eviction must swallow the error
+        # rather than crash the writer.
+        def _always_locked(self, *args, **kwargs):
+            exc = PermissionError("sharing violation")
+            exc.winerror = 32
+            raise exc
+
+        monkeypatch.setattr(Path, "unlink", _always_locked)
+        # Should not raise.
+        cache[b"c"] = b"c" * 100
+
+
 @pytest.mark.parametrize(
     "option_kw",
     [