From d03c005f50c3e64a7489b15cf89312b7a74ec43e Mon Sep 17 00:00:00 2001 From: 7487 <1042653432@qq.com> Date: Wed, 2 Sep 2026 10:04:36 +0800 Subject: [PATCH 1/2] fix: write file cache entries atomically to avoid read races FileCache.put opened the cache entry with "wb", truncating it in place. A concurrent reader (e.g. another pytest-xdist worker sharing the cache directory) could open the file between the truncate and the completed write and fail to unpickle it ("Ran out of input"), causing flaky Windows CI runs such as test_multi_repo_macro_references. Write the gzip/pickle payload to a temp file in the cache directory and os.replace it onto the entry path so readers only ever see complete entries. The replace is best-effort: on Windows it can fail if a reader still has the target open, in which case the existing entry is left intact. Fixes #6010 Co-Authored-By: Claude Fable 5 Signed-off-by: 7487 <1042653432@qq.com> --- sqlmesh/utils/cache.py | 23 +++++++++++++++++++++-- tests/utils/test_cache.py | 16 ++++++++++++++++ 2 files changed, 37 insertions(+), 2 deletions(-) diff --git a/sqlmesh/utils/cache.py b/sqlmesh/utils/cache.py index e1ff59a4a7..a6bf6c6a32 100644 --- a/sqlmesh/utils/cache.py +++ b/sqlmesh/utils/cache.py @@ -2,8 +2,10 @@ import gzip import logging +import os import pickle import shutil +import tempfile import typing as t from pathlib import Path @@ -125,8 +127,25 @@ def put(self, name: str, entry_id: str = "", *, value: T) -> None: if not self._path.is_dir(): raise SQLMeshError(f"Cache path '{self._path}' is not a directory.") - with gzip.open(self._cache_entry_path(name, entry_id), "wb", compresslevel=1) as fd: - pickle.dump(value, fd) + # Write to a temporary file and then atomically move it into place. Writing the + # entry in place ("wb" truncates the target file first) means that a concurrent + # reader (e.g. another pytest-xdist worker) could observe a partially written + # entry and fail to unpickle it. + tmp_fd, tmp_name = tempfile.mkstemp(dir=self._path, prefix=f"{self._cache_version}__tmp") + try: + with os.fdopen(tmp_fd, "wb") as raw_fd: + with gzip.open(raw_fd, "wb", compresslevel=1) as fd: + pickle.dump(value, fd) + os.replace(tmp_name, self._cache_entry_path(name, entry_id)) + except OSError as ex: + # Storing an entry is best-effort; e.g. on Windows os.replace fails if a + # concurrent reader still has the target file open. + logger.warning("Failed to store a cache entry '%s': %s", name, ex) + finally: + try: + os.unlink(tmp_name) + except OSError: + pass def exists(self, name: str, entry_id: str = "") -> bool: """Returns true if the cache entry with the given name and ID exists, false otherwise. diff --git a/tests/utils/test_cache.py b/tests/utils/test_cache.py index e6e041e30a..aa11b6c846 100644 --- a/tests/utils/test_cache.py +++ b/tests/utils/test_cache.py @@ -42,6 +42,22 @@ def test_file_cache(tmp_path: Path, mocker: MockerFixture): assert "客户数据" in cache._cache_entry_path("客户数据").name +def test_file_cache_put_is_atomic(tmp_path: Path, mocker: MockerFixture) -> None: + cache: FileCache[_TestEntry] = FileCache(tmp_path) + + old_entry = _TestEntry(value="old") + cache.put("test_name", value=old_entry) + + # Simulate os.replace failing, e.g. on Windows when a concurrent reader still has the + # target file open. The existing entry must never be truncated / partially overwritten. + mocker.patch("sqlmesh.utils.cache.os.replace", side_effect=PermissionError("file in use")) + cache.put("test_name", value=_TestEntry(value="new")) + + assert cache.get("test_name") == old_entry + # The temporary file should have been cleaned up. + assert len(list(tmp_path.glob("*"))) == 1 + + def test_optimized_query_cache(tmp_path: Path, mocker: MockerFixture): model = SqlModel( name="test_model", From 21588c5064a0dbea79e62f14c271aacdcd3700c9 Mon Sep 17 00:00:00 2001 From: 7487 <1042653432@qq.com> Date: Fri, 4 Sep 2026 18:20:37 +0800 Subject: [PATCH 2/2] refactor: narrow put error handling to os.replace per review Co-Authored-By: Claude Fable 5 Signed-off-by: 7487 <1042653432@qq.com> --- sqlmesh/utils/cache.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/sqlmesh/utils/cache.py b/sqlmesh/utils/cache.py index a6bf6c6a32..d7df2ef0d2 100644 --- a/sqlmesh/utils/cache.py +++ b/sqlmesh/utils/cache.py @@ -136,11 +136,11 @@ def put(self, name: str, entry_id: str = "", *, value: T) -> None: with os.fdopen(tmp_fd, "wb") as raw_fd: with gzip.open(raw_fd, "wb", compresslevel=1) as fd: pickle.dump(value, fd) - os.replace(tmp_name, self._cache_entry_path(name, entry_id)) - except OSError as ex: - # Storing an entry is best-effort; e.g. on Windows os.replace fails if a - # concurrent reader still has the target file open. - logger.warning("Failed to store a cache entry '%s': %s", name, ex) + try: + os.replace(tmp_name, self._cache_entry_path(name, entry_id)) + except OSError as ex: + # Windows os.replace fails if a concurrent reader still has the target file open. + logger.warning("Failed to store a cache entry '%s': %s", name, ex) finally: try: os.unlink(tmp_name)