fixup! feat(core.utils): clear() preserves young temp files (concurrent-writer race)

Author: cpcloud

cuda_core/cuda/core/utils/_program_cache.py

@@ -1233,15 +1233,10 @@ def clear(self) -> None:
         for path in list(self._iter_entry_paths()):
             with contextlib.suppress(FileNotFoundError):
                 path.unlink()
-        # The user explicitly asked to wipe this cache, so also drop every
-        # temp file we can see (whether stale or in flight from this process).
-        # Other processes' in-flight writes will still complete to ``entries``
-        # via ``os.replace``, but their staging files are intentionally gone.
-        if self._tmp.exists():
-            for tmp in list(self._tmp.iterdir()):
-                if tmp.is_file():
-                    with contextlib.suppress(FileNotFoundError):
-                        tmp.unlink()
+        # Sweep ONLY stale temp files. Deleting a young temp would race with
+        # another process between ``mkstemp`` and ``os.replace`` and turn its
+        # write into ``FileNotFoundError`` instead of a successful commit.
+        self._sweep_stale_tmp_files()
         # Remove empty subdirs (best-effort; concurrent writers may re-create).
         if self._entries.exists():
             for sub in sorted(self._entries.iterdir(), reverse=True):

cuda_core/tests/test_program_cache.py

@@ -1411,21 +1411,59 @@ def test_filestream_cache_sweeps_stale_tmp_files_on_open(tmp_path):
         assert not ancient.exists(), "ancient temp file should have been swept"
 
 
-def test_filestream_cache_clear_drops_all_tmp_files(tmp_path):
-    """clear() is an explicit user wipe, so it removes every temp file too
-    -- including young ones (other processes' in-flight writes will still
-    complete to entries/, just without their staging file)."""
-    from cuda.core.utils import FileStreamProgramCache
+def test_filestream_cache_clear_preserves_young_tmp_files(tmp_path):
+    """clear() must not delete young temp files: another process could be
+    mid-write between ``mkstemp`` and ``os.replace``, and unlinking under
+    it turns the writer's harmless rename into ``FileNotFoundError``.
+    Stale temps (older than the threshold) are still swept."""
+    import os as _os
+
+    from cuda.core.utils import FileStreamProgramCache, _program_cache
 
     root = tmp_path / "fc"
     with FileStreamProgramCache(root) as cache:
         cache[b"k"] = _fake_object_code(b"v")
     young_tmp = root / "tmp" / "entry-young"
     young_tmp.write_bytes(b"in-flight")
+    ancient_tmp = root / "tmp" / "entry-ancient"
+    ancient_tmp.write_bytes(b"crashed")
+    ancient_mtime = time.time() - _program_cache._TMP_STALE_AGE_SECONDS - 60
+    _os.utime(ancient_tmp, (ancient_mtime, ancient_mtime))
 
     with FileStreamProgramCache(root) as cache:
         cache.clear()
-    assert not young_tmp.exists()
+    # Committed entry is gone, ancient orphan is gone, young temp survives.
+    assert list((root / "entries").rglob("*.*")) == []  # no committed entry files
+    assert young_tmp.exists()
+    assert not ancient_tmp.exists()
+
+
+def test_filestream_cache_clear_does_not_break_concurrent_writer(tmp_path):
+    """Simulate a writer that has already produced a temp file but has not
+    yet executed ``os.replace``; a concurrent ``clear()`` from another
+    cache instance must NOT unlink that temp, so the writer's
+    ``os.replace`` still succeeds."""
+    import os as _os
+
+    from cuda.core.utils import FileStreamProgramCache
+
+    root = tmp_path / "fc"
+    with FileStreamProgramCache(root) as cache:
+        cache[b"seed"] = _fake_object_code(b"seed")
+
+    # Stage a temp file that mimics an in-flight write.
+    inflight_tmp = root / "tmp" / "entry-inflight"
+    inflight_tmp.write_bytes(b"\x80\x05fake-pickle")  # contents do not matter
+
+    # Concurrent clear() from another cache handle.
+    with FileStreamProgramCache(root) as other:
+        other.clear()
+
+    # The writer can now finish: rename the staged file into entries/.
+    target = root / "entries" / "ab" / "cdef"
+    target.parent.mkdir(parents=True, exist_ok=True)
+    _os.replace(inflight_tmp, target)
+    assert target.exists()
 
 
 def test_filestream_cache_size_cap_counts_tmp_files(tmp_path):