feat(symlink): add deep validation for symlink cache invalidation

Detects symlink changes at any depth, not just at root level. Uses two-tier validation:
- Fingerprint check for new symlinks
- Deep mapping validation for removed/retargeted symlinks
This commit is contained in:
Will Miao
2026-01-26 09:30:10 +08:00
parent c9e305397c
commit d063d48417
2 changed files with 93 additions and 5 deletions

View File

@@ -293,15 +293,19 @@ class Config:
)
self._rebuild_preview_roots()
# Only rescan if target roots have changed.
# This is stable across file additions/deletions.
current_fingerprint = self._build_symlink_fingerprint()
cached_fingerprint = self._cached_fingerprint
if cached_fingerprint and current_fingerprint == cached_fingerprint:
# Check 1: First-level symlinks unchanged (catches new symlinks at root)
fingerprint_valid = cached_fingerprint and current_fingerprint == cached_fingerprint
# Check 2: All cached mappings still valid (catches changes at any depth)
mappings_valid = self._validate_cached_mappings() if fingerprint_valid else False
if fingerprint_valid and mappings_valid:
return
logger.info("Symlink root paths changed; rescanning symbolic links")
logger.info("Symlink configuration changed; rescanning symbolic links")
self.rebuild_symlink_cache()
logger.info(
@@ -348,6 +352,36 @@ class Config:
logger.info("Symlink cache loaded with %d mappings", len(self._path_mappings))
return True
def _validate_cached_mappings(self) -> bool:
"""Verify all cached symlink mappings are still valid.
Returns True if all mappings are valid, False if rescan is needed.
This catches removed or retargeted symlinks at ANY depth.
"""
for target, link in self._path_mappings.items():
# Convert normalized paths back to OS paths
link_path = link.replace('/', os.sep)
# Check if symlink still exists
if not self._is_link(link_path):
logger.debug("Cached symlink no longer exists: %s", link_path)
return False
# Check if target is still the same
try:
actual_target = self._normalize_path(os.path.realpath(link_path))
if actual_target != target:
logger.debug(
"Symlink target changed: %s -> %s (cached: %s)",
link_path, actual_target, target
)
return False
except OSError:
logger.debug("Cannot resolve symlink: %s", link_path)
return False
return True
def _save_symlink_cache(self) -> None:
cache_path = self._get_symlink_cache_path()
payload = {

View File

@@ -217,3 +217,57 @@ def test_new_symlink_triggers_rescan(monkeypatch: pytest.MonkeyPatch, tmp_path):
second_cfg = config_module.Config()
normalized_external = _normalize(str(external_dir))
assert normalized_external in second_cfg._path_mappings
def test_removed_deep_symlink_triggers_rescan(monkeypatch: pytest.MonkeyPatch, tmp_path):
"""Removing a deep symlink should trigger cache invalidation."""
loras_dir, settings_dir = _setup_paths(monkeypatch, tmp_path)
# Create nested structure with deep symlink
subdir = loras_dir / "anime"
subdir.mkdir()
external_dir = tmp_path / "external"
external_dir.mkdir()
deep_symlink = subdir / "styles"
deep_symlink.symlink_to(external_dir, target_is_directory=True)
# Initial scan finds the deep symlink
first_cfg = config_module.Config()
normalized_external = _normalize(str(external_dir))
assert normalized_external in first_cfg._path_mappings
# Remove the deep symlink
deep_symlink.unlink()
# Second config should detect invalid cached mapping and rescan
second_cfg = config_module.Config()
assert normalized_external not in second_cfg._path_mappings
def test_retargeted_deep_symlink_triggers_rescan(monkeypatch: pytest.MonkeyPatch, tmp_path):
"""Changing a deep symlink's target should trigger cache invalidation."""
loras_dir, settings_dir = _setup_paths(monkeypatch, tmp_path)
# Create nested structure
subdir = loras_dir / "anime"
subdir.mkdir()
target_v1 = tmp_path / "external_v1"
target_v1.mkdir()
target_v2 = tmp_path / "external_v2"
target_v2.mkdir()
deep_symlink = subdir / "styles"
deep_symlink.symlink_to(target_v1, target_is_directory=True)
# Initial scan
first_cfg = config_module.Config()
assert _normalize(str(target_v1)) in first_cfg._path_mappings
# Retarget the symlink
deep_symlink.unlink()
deep_symlink.symlink_to(target_v2, target_is_directory=True)
# Second config should detect changed target and rescan
second_cfg = config_module.Config()
assert _normalize(str(target_v2)) in second_cfg._path_mappings
assert _normalize(str(target_v1)) not in second_cfg._path_mappings