fix(core.utils): narrow PermissionError swallowing to sharing violations

Roborev review #1735 flagged two issues with the unlink-retry commit:

1. **HIGH**: schema-mismatch wipe in __init__ suppressed every
   PermissionError then advanced the schema marker anyway. If a
   Windows-locked entry survived the wipe, future opens would skip
   the wipe and lookups could return bytes from the incompatible
   old format.

2. **MEDIUM**: _enforce_size_cap and _prune_if_stat_unchanged
   suppressed every PermissionError, hiding real configuration
   failures (POSIX ACL issues, Windows non-sharing winerrors).

Fix: add ``_is_windows_sharing_violation(exc)`` so best-effort callers
can filter the exhausted-Windows-sharing case while letting other
PermissionError instances propagate. Apply at:

* ``_prune_if_stat_unchanged``: catch + filter; non-sharing PE re-raises.
* Schema-mismatch wipe: track ``wipe_complete``; only advance the
  schema marker if every old entry was successfully removed.
* ``_enforce_size_cap``: catch + filter.

Tests cover the new propagation paths (POSIX PermissionError surfaces,
not silently swallowed) and the schema-marker preservation when an
old entry is locked.

Author: cpcloud

cuda_core/cuda/core/utils/_program_cache.py

@@ -1194,6 +1194,23 @@ def _touch_atime(path: Path, st_before: os.stat_result) -> None:
         os.utime(path, ns=(new_atime_ns, st_before.st_mtime_ns))
 
 
+def _is_windows_sharing_violation(exc: BaseException) -> bool:
+    """Return True if ``exc`` is a Windows sharing/lock violation that
+    :func:`_unlink_with_sharing_retry` would have retried.
+
+    Used by best-effort callers to filter out the exhausted-retry case
+    while letting other ``PermissionError`` instances (POSIX ACL
+    issues, Windows non-sharing winerrors) propagate -- those are real
+    configuration problems, not transient contention.
+    """
+    if not _IS_WINDOWS:
+        return False
+    if not isinstance(exc, PermissionError):
+        return False
+    winerror = getattr(exc, "winerror", None)
+    return winerror in _SHARING_VIOLATION_WINERRORS or exc.errno == errno.EACCES
+
+
 def _unlink_with_sharing_retry(path: Path) -> None:
     """Unlink with Windows-specific retry on sharing/lock violations.
 
@@ -1208,7 +1225,9 @@ def _unlink_with_sharing_retry(path: Path) -> None:
     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.
+    ``PermissionError``) immediately. Best-effort callers should use
+    :func:`_is_windows_sharing_violation` to filter the exhausted-retry
+    case and re-raise any other ``PermissionError``.
     """
     last_exc: PermissionError | None = None
     for delay in _REPLACE_RETRY_DELAYS:
@@ -1254,8 +1273,16 @@ 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, PermissionError):
+    try:
         _unlink_with_sharing_retry(path)
+    except FileNotFoundError:
+        pass
+    except PermissionError as exc:
+        # Swallow only the exhausted-Windows-sharing case. POSIX ACL
+        # errors and Windows non-sharing winerrors are real configuration
+        # problems and must surface, not be silently lost during a prune.
+        if not _is_windows_sharing_violation(exc):
+            raise
 
 
 class FileStreamProgramCache(ProgramCacheResource):
@@ -1348,11 +1375,23 @@ def __init__(
                 # 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.
+                # be reading an old entry; in that case leave the OLD schema
+                # marker so the next open re-runs the wipe. Advancing the
+                # marker before every old entry is gone would cause future
+                # opens to skip the wipe and let lookups return bytes from
+                # the incompatible format.
+                wipe_complete = True
                 for entry in list(self._iter_entry_paths()):
-                    with contextlib.suppress(FileNotFoundError, PermissionError):
+                    try:
                         _unlink_with_sharing_retry(entry)
-                self._schema_path.write_text(expected)
+                    except FileNotFoundError:
+                        pass
+                    except PermissionError as exc:
+                        if not _is_windows_sharing_violation(exc):
+                            raise
+                        wipe_complete = False
+                if wipe_complete:
+                    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
         # other processes are preserved.
@@ -1559,9 +1598,14 @@ def _enforce_size_cap(self) -> None:
             # 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.
+            # the stat-guard above had triggered. Other PermissionErrors
+            # (POSIX ACL, Windows non-sharing winerrors) are real config
+            # problems -- surface them rather than silently exceed the cap.
             try:
                 _unlink_with_sharing_retry(path)
                 total -= size
-            except (FileNotFoundError, PermissionError):
+            except FileNotFoundError:
                 pass
+            except PermissionError as exc:
+                if not _is_windows_sharing_violation(exc):
+                    raise

cuda_core/tests/test_program_cache.py

@@ -814,6 +814,66 @@ def _always_locked(self, *args, **kwargs):
         cache[b"c"] = b"c" * 100
 
 
+def test_filestream_cache_eviction_propagates_non_sharing_permission_error(tmp_path, monkeypatch):
+    """Eviction must NOT silently swallow PermissionErrors that aren't
+    Windows sharing violations -- those (POSIX ACL, Windows non-sharing
+    winerrors) are real config issues and would otherwise let the cache
+    sit over its cap with no signal to the caller."""
+    from pathlib import Path
+
+    from cuda.core.utils import FileStreamProgramCache, _program_cache
+
+    monkeypatch.setattr(_program_cache, "_IS_WINDOWS", False)
+
+    with FileStreamProgramCache(tmp_path / "fc", max_size_bytes=200) as cache:
+        cache[b"a"] = b"a" * 100
+        cache[b"b"] = b"b" * 100
+
+        def _denied(self, *args, **kwargs):
+            raise PermissionError("acl denied")
+
+        monkeypatch.setattr(Path, "unlink", _denied)
+        with pytest.raises(PermissionError, match="acl denied"):
+            cache[b"c"] = b"c" * 100
+
+
+def test_filestream_cache_schema_mismatch_keeps_old_marker_when_locked(tmp_path, monkeypatch):
+    """If a schema-mismatch wipe fails to remove an entry due to a Windows
+    sharing violation, the old schema marker must NOT be advanced --
+    otherwise a future open skips the wipe and a future lookup may return
+    bytes from the incompatible old format."""
+    from pathlib import Path
+
+    from cuda.core.utils import FileStreamProgramCache, _program_cache
+
+    root = tmp_path / "fc"
+    # Seed a cache with one entry under the current schema.
+    with FileStreamProgramCache(root) as cache:
+        cache[b"k"] = b"v"
+
+    # Pretend the on-disk schema is from an old version.
+    schema_path = root / "SCHEMA_VERSION"
+    schema_path.write_text("0.0")
+    expected_marker = schema_path.read_text()
+    assert expected_marker == "0.0"
+
+    # Make every unlink raise a Windows sharing violation. Re-opening should
+    # attempt the wipe, fail to remove the locked entry, and LEAVE the old
+    # schema marker so a later open re-runs the wipe.
+    monkeypatch.setattr(_program_cache, "_IS_WINDOWS", True)
+
+    def _always_locked(self, *args, **kwargs):
+        exc = PermissionError("sharing violation")
+        exc.winerror = 32
+        raise exc
+
+    monkeypatch.setattr(Path, "unlink", _always_locked)
+
+    # The wipe is best-effort but must not silently advance the marker.
+    FileStreamProgramCache(root)
+    assert schema_path.read_text() == "0.0", "schema marker advanced despite locked entries surviving the wipe"
+
+
 @pytest.mark.parametrize(
     "option_kw",
     [