fix nix flake cache with chroot stores

The flake cache was not properly checking store paths when custom stores
were used (e.g., when using --store flag or CLAN_TEST_STORE). This
caused cache validation to fail even when the store paths existed.

This fix:
- Updates store path detection to properly identify any path with
  "/store/" in it and a proper nix store item format (hash-name)
- Normalizes store paths to use the current store when checking
  if they exist (CLAN_TEST_STORE or nix config)
- Uses CLAN_TEST_STORE environment variable for test stores, which
  matches how nix --store flag works

Added comprehensive tests to verify the fix works with custom stores.
This commit is contained in:
Jörg Thalheim
2025-06-17 15:05:54 +02:00
parent c303ed8347
commit 34cc793af2
2 changed files with 168 additions and 7 deletions

View File

@@ -9,6 +9,7 @@ from tempfile import NamedTemporaryFile
from typing import Any
from clan_lib.errors import ClanError
from clan_lib.nix import nix_config
log = logging.getLogger(__name__)
@@ -319,9 +320,7 @@ class FlakeCacheEntry:
# strings need to be checked if they are store paths
# if they are, we store them as a dict with the outPath key
# this is to mirror nix behavior, where the outPath of an attrset is used if no further key is specified
elif isinstance(value, str) and value.startswith(
os.environ.get("NIX_STORE_DIR", "/nix/store")
):
elif isinstance(value, str) and self._is_store_path(value):
assert selectors == []
self.value = {"outPath": FlakeCacheEntry(value)}
@@ -335,14 +334,68 @@ class FlakeCacheEntry:
msg = f"Cannot insert {value} into cache, already have {self.value}"
raise TypeError(msg)
def _is_store_path(self, value: str) -> bool:
"""Check if a string is a nix store path."""
# A store path is any path that has "store" as one of its parent directories
# and contains a hash-prefixed name after it
path_parts = Path(value).parts
try:
store_idx = path_parts.index("store")
except ValueError:
return False
# Check if there's at least one more component after "store"
if store_idx + 1 < len(path_parts):
# Check if the component after store looks like a nix store item
# (starts with a hash)
store_item = path_parts[store_idx + 1]
# Basic check: nix store items typically start with a hash
return len(store_item) > 32 and "-" in store_item
return False
def _normalize_store_path(self, store_path: str) -> Path | None:
"""
Normalize a store path to use the current NIX_STORE_DIR.
Returns None if the path is not a valid store path.
"""
# Extract the store item (hash-name) from the path
path_parts = Path(store_path).parts
# Find the index of "store" in the path
try:
store_idx = path_parts.index("store")
except ValueError:
return None
if store_idx + 1 < len(path_parts):
store_item = path_parts[store_idx + 1]
# Get the current store path
# Check if we're using a test store first
test_store = os.environ.get("CLAN_TEST_STORE")
if test_store:
# In test mode, the store is at CLAN_TEST_STORE/nix/store
current_store = str(Path(test_store) / "nix" / "store")
else:
# Otherwise use nix config
config = nix_config()
current_store = config.get("store", "/nix/store")
return Path(current_store) / store_item if current_store else None
return None
def is_cached(self, selectors: list[Selector]) -> bool:
selector: Selector
# for store paths we have to check if they still exist, otherwise they have to be rebuild and are thus not cached
if isinstance(self.value, str) and self.value.startswith(
os.environ.get("NIX_STORE_DIR", "/nix/store")
):
return Path(self.value).exists()
if isinstance(self.value, str) and self._is_store_path(self.value):
normalized_path = self._normalize_store_path(self.value)
if normalized_path:
return normalized_path.exists()
return False
# if self.value is not dict but we request more selectors, we assume we are cached and an error will be thrown in the select function
if isinstance(self.value, str | float | int | None):

View File

@@ -0,0 +1,108 @@
"""Test flake cache with chroot stores."""
import os
import tempfile
from pathlib import Path
import pytest
from clan_lib.flake.flake import FlakeCache, FlakeCacheEntry
def test_flake_cache_with_chroot_store(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Test that flake cache works correctly with chroot stores."""
# Create a mock store path
normal_store = "/nix/store"
chroot_store = str(tmp_path / "nix" / "store")
# Create the chroot store directory
Path(chroot_store).mkdir(parents=True)
# Create a fake derivation in the chroot store with proper nix store format
fake_drv = "abcd1234abcd1234abcd1234abcd1234-test-package"
fake_store_path = f"{chroot_store}/{fake_drv}"
Path(fake_store_path).mkdir(parents=True)
# Test 1: Cache entry with normal store path
cache = FlakeCache()
# Insert a store path that doesn't exist in the normal store
non_existent_path = f"{normal_store}/{fake_drv}"
cache.insert({"test": {"package": non_existent_path}}, "")
# Without chroot store, this should be uncached (path doesn't exist)
assert not cache.is_cached("test.package")
# Test 2: Set CLAN_TEST_STORE to chroot store parent
# CLAN_TEST_STORE should point to the parent of nix/store
monkeypatch.setenv("CLAN_TEST_STORE", str(tmp_path))
# Create a new cache with the chroot store
cache2 = FlakeCache()
# Insert the same path but now it should use chroot store
cache2.insert({"test": {"package": fake_store_path}}, "")
# This should be cached because the path exists in chroot store
assert cache2.is_cached("test.package")
# Test 3: Cache persistence with chroot store
cache_file = tmp_path / "cache.json"
cache2.save_to_file(cache_file)
# Load cache in a new instance
cache3 = FlakeCache()
cache3.load_from_file(cache_file)
# Should still be cached with chroot store
assert cache3.is_cached("test.package")
# Test 4: Cache validation fails when chroot store changes
monkeypatch.setenv("CLAN_TEST_STORE", "/different")
# Same cache should now be invalid
assert not cache3.is_cached("test.package")
def test_flake_cache_entry_store_path_validation() -> None:
"""Test that FlakeCacheEntry correctly validates store paths."""
# Test with default store
entry = FlakeCacheEntry()
# Insert a non-existent store path with proper format
fake_path = "/nix/store/abcd1234abcd1234abcd1234abcd1234-fake-package"
entry.insert(fake_path, [])
# Should not be cached because path doesn't exist
assert not entry.is_cached([])
# Test with environment variable
with tempfile.TemporaryDirectory() as tmpdir:
# Create nix/store structure
store_dir = Path(tmpdir) / "nix" / "store"
store_dir.mkdir(parents=True)
# Create a fake store path with proper format
fake_drv = "test1234test1234test1234test1234-package"
fake_path_obj = store_dir / fake_drv
fake_path_obj.mkdir()
fake_path = str(fake_path_obj)
# Set CLAN_TEST_STORE to parent of store dir
old_env = os.environ.get("CLAN_TEST_STORE")
try:
os.environ["CLAN_TEST_STORE"] = str(tmpdir)
entry2 = FlakeCacheEntry()
entry2.insert(str(fake_path), [])
# Should be cached because path exists
assert entry2.is_cached([])
finally:
if old_env is None:
os.environ.pop("CLAN_TEST_STORE", None)
else:
os.environ["CLAN_TEST_STORE"] = old_env