diff --git a/modelaudit/cache/cache_manager.py b/modelaudit/cache/cache_manager.py index 679f542a5..5039de064 100644 --- a/modelaudit/cache/cache_manager.py +++ b/modelaudit/cache/cache_manager.py @@ -87,6 +87,7 @@ def get_cached_result_with_stat( cached_result, file_identity = self.get_cached_result_with_identity( file_path, version_context=version_context, + file_stat=stat_result, include_private_metadata=include_private_metadata, ) try: @@ -100,17 +101,21 @@ def get_cached_result_with_identity( file_path: str, version_context: dict[str, Any] | None = None, *, + file_stat: os.stat_result | None = None, include_private_metadata: bool = False, ) -> tuple[dict[str, Any] | None, ScannedFileIdentity | None]: """Return a cache lookup and retain the monitored identity for a miss scan.""" if not self.enabled or not self.cache: return None, None - cached_result, file_identity = self.cache.get_cached_result_with_identity( - file_path, - version_context=version_context, - include_private_metadata=include_private_metadata, - ) + lookup_kwargs: dict[str, Any] = { + "version_context": version_context, + "include_private_metadata": include_private_metadata, + } + if file_stat is not None: + lookup_kwargs["file_stat"] = file_stat + + cached_result, file_identity = self.cache.get_cached_result_with_identity(file_path, **lookup_kwargs) if cached_result is not None and not cached_scan_result_dependencies_available(cached_result): logger.debug(f"Bypassing cached result with unavailable scanner dependencies: {Path(file_path).name}") return None, file_identity diff --git a/modelaudit/cache/scan_results_cache.py b/modelaudit/cache/scan_results_cache.py index 8204b4214..cdf230fe2 100644 --- a/modelaudit/cache/scan_results_cache.py +++ b/modelaudit/cache/scan_results_cache.py @@ -460,6 +460,7 @@ def get_cached_result_with_stat( cached_result, file_identity = self.get_cached_result_with_identity( file_path, version_context=version_context, + file_stat=file_stat, include_private_metadata=include_private_metadata, ) if file_identity is not None: @@ -471,6 +472,7 @@ def get_cached_result_with_identity( file_path: str, version_context: dict[str, Any] | None = None, *, + file_stat: os.stat_result | None = None, include_private_metadata: bool = False, ) -> tuple[dict[str, Any] | None, ScannedFileIdentity | None]: """Return a cache lookup plus the monitored identity reusable by a miss scan.""" @@ -480,7 +482,10 @@ def get_cached_result_with_identity( logger.debug("Bypassing scan-result cache lookup for symlinked path %s", file_path) return None, None - file_identity = self.capture_file_identity(file_path) + if file_stat is None: + file_identity = self.capture_file_identity(file_path) + else: + file_identity = self.capture_file_identity(file_path, file_stat=file_stat) file_stat, file_hash, _change_token, _ancestor_identity = file_identity cache_key, _content_hash = self._generate_cache_key_material( file_path, @@ -854,14 +859,20 @@ def store_result( temporary_cache_path.unlink(missing_ok=True) self.release_ancestor_identity(expected_ancestor_identity) - def capture_file_identity(self, file_path: str) -> ScannedFileIdentity: - """Capture a stable stat, content hash, and platform change token before scanning.""" + def capture_file_identity( + self, + file_path: str, + file_stat: os.stat_result | None = None, + ) -> ScannedFileIdentity: + """Capture a stable file identity, reusing a caller stat snapshot for the first probe when available.""" if self._path_has_symlink_component(file_path): raise ValueError(f"Symlinked paths are not cacheable: {file_path}") last_change_error: ValueError | None = None + stat_hint = file_stat for _capture_attempt in range(_MAX_IDENTITY_CAPTURE_ATTEMPTS): - preliminary_stat = os.stat(file_path) + preliminary_stat = stat_hint if stat_hint is not None else os.stat(file_path) + stat_hint = None probe = self._get_change_clock_probe(file_path, preliminary_stat.st_dev) preliminary_change_token = self._get_file_change_token(file_path, preliminary_stat) preliminary_ancestor_identity = self._capture_ancestor_identity(file_path) diff --git a/tests/cache/test_cache_correctness.py b/tests/cache/test_cache_correctness.py index f2025b577..5eb7b5d37 100644 --- a/tests/cache/test_cache_correctness.py +++ b/tests/cache/test_cache_correctness.py @@ -671,6 +671,79 @@ def capture_once(path: str) -> Any: assert capture_calls == 1 +def test_scan_results_cache_get_cached_result_with_stat_reuses_provided_stat( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + file_path = _make_cacheable_file(tmp_path, name="scan-results-stat.cache") + cache = ScanResultsCache(str(tmp_path / "scan-cache")) + version_context = build_cache_version_context({"timeout": 30}) + expected = {"checks": [], "issues": [], "metadata": {}, "scanner": "test", "success": True} + file_identity = cache.capture_file_identity(str(file_path)) + recorded_stats: list[os.stat_result | None] = [] + provided_stat = os.stat(file_path) + + def lookup_with_record( + path: str, + version_context: dict[str, Any] | None = None, + *, + file_stat: os.stat_result | None = None, + include_private_metadata: bool = False, + ) -> Any: + assert path == str(file_path) + assert version_context is not None + assert include_private_metadata is False + recorded_stats.append(file_stat) + return expected, file_identity + + monkeypatch.setattr(cache, "get_cached_result_with_identity", lookup_with_record) + + cached_result = cache.get_cached_result_with_stat(str(file_path), provided_stat, version_context=version_context) + + assert cached_result == expected + assert len(recorded_stats) == 1 + assert recorded_stats[0] is provided_stat + + +def test_cache_manager_get_cached_result_with_stat_reuses_provided_stat( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + file_path = _make_cacheable_file(tmp_path, name="cache-manager-stat.cache") + cache_manager = get_cache_manager(str(tmp_path / "cache"), enabled=True) + assert cache_manager.cache is not None + version_context = build_cache_version_context({"timeout": 30}) + expected = {"checks": [], "issues": [], "metadata": {}, "scanner": "test", "success": True} + file_identity = cache_manager.cache.capture_file_identity(str(file_path)) + recorded_stats: list[os.stat_result | None] = [] + provided_stat = os.stat(file_path) + + def lookup_with_record( + path: str, + version_context: dict[str, Any] | None = None, + *, + file_stat: os.stat_result | None = None, + include_private_metadata: bool = False, + ) -> Any: + assert path == str(file_path) + assert version_context is not None + assert include_private_metadata is False + recorded_stats.append(file_stat) + return expected, file_identity + + monkeypatch.setattr(cache_manager.cache, "get_cached_result_with_identity", lookup_with_record) + + cached_result = cache_manager.get_cached_result_with_stat( + str(file_path), + provided_stat, + version_context=version_context, + ) + + assert cached_result == expected + assert len(recorded_stats) == 1 + assert recorded_stats[0] is provided_stat + + def test_cached_scan_does_not_store_result_after_post_scan_replacement(tmp_path: Path) -> None: file_path = _make_cacheable_file(tmp_path, name="race.dat") clean_payload = b"clean:" + (b"x" * 2042)