1use std::collections::BTreeSet;
56use std::sync::Arc;
57
58use async_trait::async_trait;
59use tracing::warn;
60
61use super::StorageBackend;
62use crate::CacheError;
63
64#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, serde::Serialize, serde::Deserialize)]
67#[serde(rename_all = "kebab-case")]
68pub enum WritePolicy {
69 #[default]
71 WriteThrough,
72 WriteBack,
75 WriteAround,
77}
78
79#[derive(Debug, Clone, Copy, PartialEq, Eq)]
83pub enum TieredTier {
84 MockParityProven,
88 LiveClusterProven,
91}
92
93pub const TIERED_BACKEND_TIER: TieredTier = TieredTier::MockParityProven;
97
98pub struct TieredBackend {
104 l1: Arc<dyn StorageBackend>,
105 l2: Arc<dyn StorageBackend>,
106 l3: Arc<dyn StorageBackend>,
107 write_policy: WritePolicy,
108}
109
110impl TieredBackend {
111 #[must_use]
113 pub fn new(
114 l1: Arc<dyn StorageBackend>,
115 l2: Arc<dyn StorageBackend>,
116 l3: Arc<dyn StorageBackend>,
117 ) -> Self {
118 Self::with_write_policy(l1, l2, l3, WritePolicy::default())
119 }
120
121 #[must_use]
123 pub fn with_write_policy(
124 l1: Arc<dyn StorageBackend>,
125 l2: Arc<dyn StorageBackend>,
126 l3: Arc<dyn StorageBackend>,
127 write_policy: WritePolicy,
128 ) -> Self {
129 Self { l1, l2, l3, write_policy }
130 }
131
132 #[must_use]
134 pub fn write_policy(&self) -> WritePolicy {
135 self.write_policy
136 }
137
138 async fn warm_narinfo(tier: &Arc<dyn StorageBackend>, hash: &str, content: &str) {
141 if let Err(e) = tier.put_narinfo(hash, content).await {
142 warn!(hash = %hash, error = %e, "tiered: best-effort narinfo warm failed");
143 }
144 }
145
146 async fn warm_nar(tier: &Arc<dyn StorageBackend>, path: &str, data: &[u8]) {
147 if let Err(e) = tier.put_nar(path, data).await {
148 warn!(path = %path, error = %e, "tiered: best-effort NAR warm failed");
149 }
150 }
151}
152
153#[async_trait]
154impl StorageBackend for TieredBackend {
155 async fn get_narinfo(&self, hash: &str) -> Result<Option<String>, CacheError> {
156 if let Some(v) = self.l1.get_narinfo(hash).await? {
158 return Ok(Some(v));
159 }
160 if let Some(v) = self.l2.get_narinfo(hash).await? {
162 Self::warm_narinfo(&self.l1, hash, &v).await;
163 return Ok(Some(v));
164 }
165 if let Some(v) = self.l3.get_narinfo(hash).await? {
167 Self::warm_narinfo(&self.l2, hash, &v).await;
168 Self::warm_narinfo(&self.l1, hash, &v).await;
169 return Ok(Some(v));
170 }
171 Ok(None)
172 }
173
174 async fn get_nar(&self, path: &str) -> Result<Option<Vec<u8>>, CacheError> {
175 if let Some(v) = self.l1.get_nar(path).await? {
176 return Ok(Some(v));
177 }
178 if let Some(v) = self.l2.get_nar(path).await? {
179 Self::warm_nar(&self.l1, path, &v).await;
180 return Ok(Some(v));
181 }
182 if let Some(v) = self.l3.get_nar(path).await? {
183 Self::warm_nar(&self.l2, path, &v).await;
184 Self::warm_nar(&self.l1, path, &v).await;
185 return Ok(Some(v));
186 }
187 Ok(None)
188 }
189
190 async fn put_narinfo(&self, hash: &str, content: &str) -> Result<(), CacheError> {
191 match self.write_policy {
192 WritePolicy::WriteThrough => {
193 self.l2.put_narinfo(hash, content).await?;
194 self.l3.put_narinfo(hash, content).await?;
195 Self::warm_narinfo(&self.l1, hash, content).await;
196 }
197 WritePolicy::WriteBack => {
198 Self::warm_narinfo(&self.l1, hash, content).await;
199 self.l2.put_narinfo(hash, content).await?;
200 self.l3.put_narinfo(hash, content).await?;
201 }
202 WritePolicy::WriteAround => {
203 self.l2.put_narinfo(hash, content).await?;
204 self.l3.put_narinfo(hash, content).await?;
205 }
206 }
207 Ok(())
208 }
209
210 async fn put_nar(&self, path: &str, data: &[u8]) -> Result<(), CacheError> {
211 match self.write_policy {
212 WritePolicy::WriteThrough => {
213 self.l2.put_nar(path, data).await?;
214 self.l3.put_nar(path, data).await?;
215 Self::warm_nar(&self.l1, path, data).await;
216 }
217 WritePolicy::WriteBack => {
218 Self::warm_nar(&self.l1, path, data).await;
219 self.l2.put_nar(path, data).await?;
220 self.l3.put_nar(path, data).await?;
221 }
222 WritePolicy::WriteAround => {
223 self.l2.put_nar(path, data).await?;
224 self.l3.put_nar(path, data).await?;
225 }
226 }
227 Ok(())
228 }
229
230 async fn delete(&self, hash: &str) -> Result<(), CacheError> {
231 for (name, tier) in [("l1", &self.l1), ("l2", &self.l2), ("l3", &self.l3)] {
235 if let Err(e) = tier.delete(hash).await {
236 warn!(hash = %hash, tier = name, error = %e, "tiered: best-effort delete failed");
237 }
238 }
239 Ok(())
240 }
241
242 async fn list_narinfos(&self) -> Result<Vec<String>, CacheError> {
243 let mut set = BTreeSet::new();
246 set.extend(self.l2.list_narinfos().await?);
247 set.extend(self.l3.list_narinfos().await?);
248 Ok(set.into_iter().collect())
249 }
250
251 async fn wipe_all(&self) -> Result<usize, CacheError> {
257 let mut cleared = 0usize;
258 for (name, tier) in [("l1", &self.l1), ("l2", &self.l2), ("l3", &self.l3)] {
259 match tier.wipe_all().await {
260 Ok(n) => cleared = cleared.max(n),
261 Err(e) => warn!(tier = name, error = %e, "tiered: best-effort wipe failed"),
262 }
263 }
264 Ok(cleared)
265 }
266}
267
268#[cfg(test)]
275mod tests {
276 use super::*;
277 use crate::storage::LocalStorage;
278 use std::collections::HashMap;
279 use std::sync::Mutex;
280
281 #[derive(Default)]
285 struct MemBackend {
286 narinfo: Mutex<HashMap<String, String>>,
287 nar: Mutex<HashMap<String, Vec<u8>>>,
288 writes_fail: Mutex<bool>,
289 }
290
291 impl MemBackend {
292 fn has_narinfo(&self, hash: &str) -> bool {
293 self.narinfo.lock().unwrap().contains_key(hash)
294 }
295 fn has_nar(&self, path: &str) -> bool {
296 self.nar.lock().unwrap().contains_key(path)
297 }
298 fn narinfo_count(&self) -> usize {
299 self.narinfo.lock().unwrap().len()
300 }
301 fn clear(&self) {
302 self.narinfo.lock().unwrap().clear();
303 self.nar.lock().unwrap().clear();
304 }
305 fn set_writes_fail(&self, v: bool) {
306 *self.writes_fail.lock().unwrap() = v;
307 }
308 fn fail_if_configured(&self) -> Result<(), CacheError> {
309 if *self.writes_fail.lock().unwrap() {
310 Err(CacheError::NotImplemented("mock writes disabled"))
311 } else {
312 Ok(())
313 }
314 }
315 }
316
317 #[async_trait]
318 impl StorageBackend for MemBackend {
319 async fn get_narinfo(&self, hash: &str) -> Result<Option<String>, CacheError> {
320 Ok(self.narinfo.lock().unwrap().get(hash).cloned())
321 }
322 async fn put_narinfo(&self, hash: &str, content: &str) -> Result<(), CacheError> {
323 self.fail_if_configured()?;
324 self.narinfo.lock().unwrap().insert(hash.to_string(), content.to_string());
325 Ok(())
326 }
327 async fn get_nar(&self, path: &str) -> Result<Option<Vec<u8>>, CacheError> {
328 Ok(self.nar.lock().unwrap().get(path).cloned())
329 }
330 async fn put_nar(&self, path: &str, data: &[u8]) -> Result<(), CacheError> {
331 self.fail_if_configured()?;
332 self.nar.lock().unwrap().insert(path.to_string(), data.to_vec());
333 Ok(())
334 }
335 async fn delete(&self, hash: &str) -> Result<(), CacheError> {
336 self.narinfo.lock().unwrap().remove(hash);
337 for ext in ["nar.xz", "nar.zst", "nar"] {
338 self.nar.lock().unwrap().remove(&format!("nar/{hash}.{ext}"));
339 }
340 Ok(())
341 }
342 async fn list_narinfos(&self) -> Result<Vec<String>, CacheError> {
343 Ok(self.narinfo.lock().unwrap().keys().cloned().collect())
344 }
345 }
346
347 const NARINFO: &str = "StorePath: /nix/store/abc-hello\nURL: nar/abc.nar.xz\nCompression: xz\nNarHash: sha256:bbb\nNarSize: 200\nReferences: \n";
348
349 fn mocks() -> (Arc<MemBackend>, Arc<MemBackend>, Arc<MemBackend>, TieredBackend) {
352 let l1 = Arc::new(MemBackend::default());
353 let l2 = Arc::new(MemBackend::default());
354 let l3 = Arc::new(MemBackend::default());
355 let tiered = TieredBackend::new(l1.clone(), l2.clone(), l3.clone());
356 (l1, l2, l3, tiered)
357 }
358
359 #[tokio::test]
362 async fn l1_hit_returns_without_touching_lower_tiers() {
363 let (l1, l2, l3, tiered) = mocks();
364 l1.put_narinfo("h", "hot").await.unwrap();
365 assert_eq!(tiered.get_narinfo("h").await.unwrap().unwrap(), "hot");
366 assert!(!l2.has_narinfo("h"));
368 assert!(!l3.has_narinfo("h"));
369 }
370
371 #[tokio::test]
372 async fn l2_hit_promotes_into_l1() {
373 let (l1, l2, _l3, tiered) = mocks();
374 l2.put_narinfo("h", NARINFO).await.unwrap();
375 assert!(!l1.has_narinfo("h"));
376 let got = tiered.get_narinfo("h").await.unwrap().unwrap();
377 assert_eq!(got, NARINFO);
378 assert!(l1.has_narinfo("h"), "L2 hit must promote into L1");
380 }
381
382 #[tokio::test]
383 async fn l3_hit_promotes_into_l2_and_l1() {
384 let (l1, l2, l3, tiered) = mocks();
385 l3.put_narinfo("h", NARINFO).await.unwrap();
386 let got = tiered.get_narinfo("h").await.unwrap().unwrap();
387 assert_eq!(got, NARINFO);
388 assert!(l2.has_narinfo("h"), "L3 hit must promote into L2");
389 assert!(l1.has_narinfo("h"), "L3 hit must promote into L1");
390 }
391
392 #[tokio::test]
393 async fn nar_l3_hit_promotes_into_l2_and_l1() {
394 let (l1, l2, l3, tiered) = mocks();
395 l3.put_nar("nar/x.nar.xz", b"blob").await.unwrap();
396 let got = tiered.get_nar("nar/x.nar.xz").await.unwrap().unwrap();
397 assert_eq!(got, b"blob");
398 assert!(l2.has_nar("nar/x.nar.xz"));
399 assert!(l1.has_nar("nar/x.nar.xz"));
400 }
401
402 #[tokio::test]
403 async fn miss_at_all_tiers_is_none() {
404 let (_l1, _l2, _l3, tiered) = mocks();
405 assert!(tiered.get_narinfo("ghost").await.unwrap().is_none());
406 assert!(tiered.get_nar("nar/ghost.nar.xz").await.unwrap().is_none());
407 }
408
409 #[tokio::test]
410 async fn promotion_failure_does_not_break_a_read() {
411 let (l1, l2, _l3, tiered) = mocks();
414 l2.put_narinfo("h", NARINFO).await.unwrap();
415 l1.set_writes_fail(true); let got = tiered.get_narinfo("h").await.unwrap();
417 assert_eq!(got.unwrap(), NARINFO);
418 assert!(!l1.has_narinfo("h"), "warm failed, so L1 stays empty — but the read still succeeded");
419 }
420
421 #[tokio::test]
424 async fn write_through_populates_all_tiers() {
425 let (l1, l2, l3, tiered) = mocks();
426 tiered.put_narinfo("h", NARINFO).await.unwrap();
427 assert!(l1.has_narinfo("h"), "write-through warms L1");
428 assert!(l2.has_narinfo("h"), "write-through persists L2");
429 assert!(l3.has_narinfo("h"), "write-through persists L3");
430 }
431
432 #[tokio::test]
433 async fn write_around_skips_l1_but_persists_durable() {
434 let l1 = Arc::new(MemBackend::default());
435 let l2 = Arc::new(MemBackend::default());
436 let l3 = Arc::new(MemBackend::default());
437 let tiered = TieredBackend::with_write_policy(
438 l1.clone(), l2.clone(), l3.clone(), WritePolicy::WriteAround,
439 );
440 tiered.put_narinfo("h", NARINFO).await.unwrap();
441 assert!(!l1.has_narinfo("h"), "write-around must NOT touch L1");
442 assert!(l2.has_narinfo("h"));
443 assert!(l3.has_narinfo("h"));
444 let _ = tiered.get_narinfo("h").await.unwrap();
446 assert!(l1.has_narinfo("h"), "read-through fills L1 after a write-around");
447 }
448
449 #[tokio::test]
450 async fn write_back_populates_all_tiers_and_is_durable() {
451 let l1 = Arc::new(MemBackend::default());
452 let l2 = Arc::new(MemBackend::default());
453 let l3 = Arc::new(MemBackend::default());
454 let tiered = TieredBackend::with_write_policy(
455 l1.clone(), l2.clone(), l3.clone(), WritePolicy::WriteBack,
456 );
457 assert_eq!(tiered.write_policy(), WritePolicy::WriteBack);
458 tiered.put_nar("nar/x.nar.xz", b"blob").await.unwrap();
459 assert!(l1.has_nar("nar/x.nar.xz"));
461 assert!(l2.has_nar("nar/x.nar.xz"));
462 assert!(l3.has_nar("nar/x.nar.xz"));
463 }
464
465 #[tokio::test]
466 async fn durable_write_failure_propagates() {
467 let l1 = Arc::new(MemBackend::default());
470 let l2 = Arc::new(MemBackend::default());
471 let l3 = Arc::new(MemBackend::default());
472 l2.set_writes_fail(true);
473 let tiered = TieredBackend::new(l1.clone(), l2, l3);
474 let err = tiered.put_narinfo("h", NARINFO).await.unwrap_err();
475 assert!(matches!(err, CacheError::NotImplemented(_)));
476 }
477
478 #[tokio::test]
481 async fn pod_roll_losing_l1_loses_nothing() {
482 let (l1, _l2, _l3, tiered) = mocks();
483 tiered.put_narinfo("h", NARINFO).await.unwrap();
484 tiered.put_nar("nar/h.nar.xz", b"blob").await.unwrap();
485 l1.clear();
487 assert_eq!(tiered.get_narinfo("h").await.unwrap().unwrap(), NARINFO);
489 assert_eq!(tiered.get_nar("nar/h.nar.xz").await.unwrap().unwrap(), b"blob");
490 }
491
492 #[tokio::test]
495 async fn delete_fans_out_to_all_tiers() {
496 let (l1, l2, l3, tiered) = mocks();
497 tiered.put_narinfo("h", NARINFO).await.unwrap();
498 tiered.put_nar("nar/h.nar.xz", b"blob").await.unwrap();
499 tiered.delete("h").await.unwrap();
500 for t in [&l1, &l2, &l3] {
501 assert!(!t.has_narinfo("h"));
502 assert!(!t.has_nar("nar/h.nar.xz"));
503 }
504 }
505
506 #[tokio::test]
507 async fn wipe_all_clears_every_tier() {
508 let (l1, l2, l3, tiered) = mocks();
509 tiered.put_narinfo("h", NARINFO).await.unwrap();
511 tiered.put_nar("nar/h.nar.xz", b"blob").await.unwrap();
512 l2.put_narinfo("only2", "x").await.unwrap();
514
515 let removed = tiered.wipe_all().await.unwrap();
516 assert!(removed >= 1, "wipe reported nothing cleared");
517
518 for t in [&l1, &l2, &l3] {
519 assert!(t.list_narinfos().await.unwrap().is_empty(), "a tier survived the wipe");
520 assert!(!t.has_narinfo("h"));
521 assert!(!t.has_nar("nar/h.nar.xz"));
522 }
523 assert!(tiered.list_narinfos().await.unwrap().is_empty(), "cache not cold after wipe");
524 assert!(tiered.get_narinfo("h").await.unwrap().is_none());
526 }
527
528 #[tokio::test]
529 async fn list_narinfos_unions_durable_tiers_deduped() {
530 let (l1, l2, l3, tiered) = mocks();
531 l2.put_narinfo("shared", "x").await.unwrap();
533 l3.put_narinfo("shared", "x").await.unwrap();
534 l2.put_narinfo("only2", "y").await.unwrap();
535 l3.put_narinfo("only3", "z").await.unwrap();
536 l1.put_narinfo("hot-only", "w").await.unwrap();
538 let listed = tiered.list_narinfos().await.unwrap();
539 assert_eq!(listed, vec!["only2".to_string(), "only3".to_string(), "shared".to_string()]);
540 }
541
542 #[tokio::test]
545 async fn read_through_from_a_real_local_storage_l3() {
546 let dir = tempfile::tempdir().unwrap();
547 let l1 = Arc::new(MemBackend::default());
548 let l2 = Arc::new(MemBackend::default());
549 let l3_disk = Arc::new(LocalStorage::new(dir.path()));
550 l3_disk.put_narinfo("h", NARINFO).await.unwrap();
552 l3_disk.put_nar("nar/h.nar.xz", b"disk-blob").await.unwrap();
553
554 let tiered = TieredBackend::new(l1.clone(), l2.clone(), l3_disk);
555 assert_eq!(tiered.get_narinfo("h").await.unwrap().unwrap(), NARINFO);
557 assert_eq!(tiered.get_nar("nar/h.nar.xz").await.unwrap().unwrap(), b"disk-blob");
558 assert!(l1.has_narinfo("h"));
559 assert!(l2.has_narinfo("h"));
560 }
561
562 #[test]
565 fn honest_gate_tier_is_mock_parity_proven_not_live_cluster() {
566 assert_eq!(TIERED_BACKEND_TIER, TieredTier::MockParityProven);
571 }
572}