Coverage for src / invariant / store / chain.py: 100.00%

1"""ChainStore: Composite artifact store chaining MemoryStore and DiskStore.""" 

2 

3from typing import Any 

4 

5from invariant.store.base import ArtifactStore 

6from invariant.store.disk import DiskStore 

7from invariant.store.memory import MemoryStore 

8 

9 

10class ChainStore(ArtifactStore): 

11 """Composite artifact store that chains MemoryStore (L1) and DiskStore (L2). 

12 

13 Provides a two-tier caching strategy: 

14 - L1 (MemoryStore): Fast, session-scoped cache checked first 

15 - L2 (DiskStore): Persistent filesystem cache checked if L1 misses 

16 

17 On L2 hit, the artifact is promoted to L1 for faster subsequent access. 

18 On put, artifacts are written to both L1 and L2. 

19 """ 

20 

21 def __init__( 

22 self, 

23 l1: MemoryStore | None = None, 

24 l2: DiskStore | None = None, 

25 ) -> None: 

26 """Initialize ChainStore. 

27 

28 Args: 

29 l1: MemoryStore instance for L1 cache. If None, creates a new one. 

30 l2: DiskStore instance for L2 cache. If None, creates a new one. 

31 """ 

32 super().__init__() 

33 self.l1 = l1 or MemoryStore(cache="lru") 

34 self.l2 = l2 or DiskStore() 

35 

36 def exists(self, op_name: str, digest: str) -> bool: 

37 """Check if an artifact exists in L1 or L2. 

38 

39 Args: 

40 op_name: The name of the operation that produced the artifact. 

41 digest: The SHA-256 hash (64 character hex string) of the manifest. 

42 

43 Returns: 

44 True if artifact exists in either store, False otherwise. 

45 """ 

46 # Check L1 first (fast path) 

47 if self.l1.exists(op_name, digest): 

48 self.stats.hits += 1 

49 return True 

50 # Check L2 (slower, but persistent) 

51 if self.l2.exists(op_name, digest): 

52 self.stats.hits += 1 

53 return True 

54 self.stats.misses += 1 

55 return False 

56 

57 def get(self, op_name: str, digest: str) -> Any: 

58 """Retrieve an artifact from L1 or L2. 

59 

60 If found in L2 but not L1, promotes the artifact to L1 for faster 

61 subsequent access. 

62 

63 Args: 

64 op_name: The name of the operation that produced the artifact. 

65 digest: The SHA-256 hash (64 character hex string) of the manifest. 

66 

67 Returns: 

68 The deserialized artifact (native type or ICacheable domain type). 

69 

70 Raises: 

71 KeyError: If artifact does not exist in either store. 

72 """ 

73 # Try L1 first (fast path) 

74 if self.l1.exists(op_name, digest): 

75 return self.l1.get(op_name, digest) 

76 

77 # Try L2 (slower, but persistent) 

78 if self.l2.exists(op_name, digest): 

79 # Promote to L1 for faster subsequent access 

80 artifact = self.l2.get(op_name, digest) 

81 self.l1.put(op_name, digest, artifact) 

82 return artifact 

83 

84 # Not found in either store 

85 raise KeyError( 

86 f"Artifact with op_name '{op_name}' and digest '{digest}' not found in L1 or L2" 

87 ) 

88 

89 def put(self, op_name: str, digest: str, artifact: Any) -> None: 

90 """Store an artifact in both L1 and L2. 

91 

92 Args: 

93 op_name: The name of the operation that produced the artifact. 

94 digest: The SHA-256 hash (64 character hex string) of the manifest. 

95 artifact: The artifact to store (must be cacheable). 

96 """ 

97 # Write to both stores 

98 self.l1.put(op_name, digest, artifact) 

99 self.l2.put(op_name, digest, artifact) 

100 self.stats.puts += 1