Coverage for src/footman/_gc.py: 95%

1"""Cache garbage collection — the detached child a run spawns at most daily. 

2 

3The cache is derived state top to bottom (completion manifests rebuild on 

4any execution-path run; timing history regrows), which is what makes 

5collecting it casually safe: the worst possible outcome of any deletion is 

6a rebuild. Two rules, in order: 

7 

81. **The directory is gone.** Manifests bake in the ``cwd`` they describe; 

9 if that path no longer exists, the pair (manifest + timing history) is 

10 leftovers from a deleted project — collected at any age. 

112. **The pair is idle.** Untouched for `IDLE_DAYS`, nobody even TAB-completes 

12 there any more (background refreshes keep a visited manifest's mtime 

13 fresh) — collected. Manifests from before the ``cwd`` key rely on this 

14 rule alone. 

15 

16The invoking directory's own pair is never touched, and every failure is 

17silent — a concurrently-reading completion child on Windows may hold a file 

18open, and a collector must never be louder than what it collects. 

19""" 

20 

21from __future__ import annotations 

22 

23import contextlib 

24import json 

25import sys 

26import time 

27from pathlib import Path 

28 

29IDLE_DAYS = 90 

30STAMP = "gc.stamp" 

31 

32 

33def collect(cache_dir: Path, skip_stem: str = "") -> int: 

34 """Apply the rules to *cache_dir*; returns the number of files removed. 

35 

36 *skip_stem* is the invoking directory's manifest stem (its hash) — that 

37 pair is in active use and never touched. 

38 """ 

39 now = time.time() 

40 removed = 0 

41 

42 def unlink(path: Path) -> None: 

43 nonlocal removed 

44 try: 

45 path.unlink() 

46 removed += 1 

47 except OSError: 

48 pass 

49 

50 for manifest_path in cache_dir.glob("*.json"): 

51 name = manifest_path.name 

52 if name.endswith(".times.json"): 

53 continue # handled with its manifest, or as an orphan below 

54 stem = name[: -len(".json")] 

55 if stem == skip_stem: 

56 continue 

57 times_path = cache_dir / f"{stem}.times.json" 

58 try: 

59 data = json.loads(manifest_path.read_text("utf-8")) 

60 except (OSError, ValueError): 

61 data = None 

62 cwd = data.get("cwd") if isinstance(data, dict) else None 

63 if isinstance(cwd, str) and cwd and not Path(cwd).exists(): 

64 doomed = True # rule 1: leftovers of a deleted project 

65 else: 

66 doomed = _idle(now, manifest_path, times_path) # rule 2 

67 if doomed: 

68 unlink(manifest_path) 

69 unlink(times_path) 

70 

71 # Timing files whose manifest twin is already gone: age them alone. 

72 for times_path in cache_dir.glob("*.times.json"): 

73 stem = times_path.name[: -len(".times.json")] 

74 if stem == skip_stem or (cache_dir / f"{stem}.json").exists(): 

75 continue 

76 if _idle(now, times_path): 

77 unlink(times_path) 

78 

79 return removed 

80 

81 

82def _idle(now: float, *paths: Path) -> bool: 

83 """Whether the newest of *paths* is older than the idle window.""" 

84 newest = 0.0 

85 for path in paths: 

86 with contextlib.suppress(OSError): 

87 newest = max(newest, path.stat().st_mtime) 

88 return newest > 0 and (now - newest) > IDLE_DAYS * 86400 

89 

90 

91def main() -> None: 

92 """Entry for the detached child: argv is (cache_dir, skip_stem).""" 

93 if len(sys.argv) < 2: 93 ↛ 94line 93 didn't jump to line 94 because the condition on line 93 was never true

94 return 

95 skip = sys.argv[2] if len(sys.argv) > 2 else "" 

96 collect(Path(sys.argv[1]), skip)