1use std::collections::BTreeSet;
70use std::sync::Arc;
71
72use async_trait::async_trait;
73use tracing::warn;
74
75use super::nar_refs::NarRefIndex;
76use super::nar_stream::{self, NarSource, NarStream};
77use super::{NarResidency, StorageBackend};
78use crate::StoreError;
79
80struct TierNarSource {
105 tier: Arc<dyn StorageBackend>,
106 path: String,
107}
108
109impl TierNarSource {
110 fn new(tier: &Arc<dyn StorageBackend>, path: &str) -> Self {
111 Self { tier: Arc::clone(tier), path: path.to_string() }
112 }
113}
114
115#[async_trait]
116impl NarSource for TierNarSource {
117 async fn open(&self) -> Result<NarStream, StoreError> {
118 self.tier.get_nar_stream(&self.path).await?.ok_or_else(|| {
119 StoreError::PathNotFound(format!(
123 "{}: vanished from the source tier mid-promotion",
124 self.path
125 ))
126 })
127 }
128}
129
130#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, serde::Serialize, serde::Deserialize)]
133#[serde(rename_all = "kebab-case")]
134pub enum WritePolicy {
135 #[default]
137 WriteThrough,
138 WriteBack,
141 WriteAround,
143}
144
145#[derive(Debug, Clone, Copy, PartialEq, Eq)]
149pub enum TieredTier {
150 MockParityProven,
154 LiveClusterProven,
157}
158
159pub const TIERED_BACKEND_TIER: TieredTier = TieredTier::MockParityProven;
163
164pub struct TieredBackend {
170 l1: Arc<dyn StorageBackend>,
171 l2: Arc<dyn StorageBackend>,
172 l3: Arc<dyn StorageBackend>,
173 write_policy: WritePolicy,
174}
175
176impl TieredBackend {
177 #[must_use]
179 pub fn new(
180 l1: Arc<dyn StorageBackend>,
181 l2: Arc<dyn StorageBackend>,
182 l3: Arc<dyn StorageBackend>,
183 ) -> Self {
184 Self::with_write_policy(l1, l2, l3, WritePolicy::default())
185 }
186
187 #[must_use]
189 pub fn with_write_policy(
190 l1: Arc<dyn StorageBackend>,
191 l2: Arc<dyn StorageBackend>,
192 l3: Arc<dyn StorageBackend>,
193 write_policy: WritePolicy,
194 ) -> Self {
195 Self { l1, l2, l3, write_policy }
196 }
197
198 #[must_use]
200 pub fn write_policy(&self) -> WritePolicy {
201 self.write_policy
202 }
203
204 fn tiers(&self) -> [(&'static str, &Arc<dyn StorageBackend>); 3] {
206 [("l1", &self.l1), ("l2", &self.l2), ("l3", &self.l3)]
207 }
208
209 async fn warm_narinfo(tier: &Arc<dyn StorageBackend>, hash: &str, content: &str) {
212 if let Err(e) = tier.put_narinfo(hash, content).await {
213 warn!(hash = %hash, error = %e, "tiered: best-effort narinfo warm failed");
214 }
215 }
216
217 async fn warm_nar_from(tier: &Arc<dyn StorageBackend>, path: &str, src: &dyn NarSource) {
224 if let Err(e) = tier.put_nar_stream(path, src).await {
225 warn!(path = %path, error = %e, "tiered: best-effort NAR warm failed");
226 }
227 }
228
229 async fn warm_nar_from_tier(
232 tier: &Arc<dyn StorageBackend>,
233 from: &Arc<dyn StorageBackend>,
234 path: &str,
235 ) {
236 Self::warm_nar_from(tier, path, &TierNarSource::new(from, path)).await;
237 }
238
239 fn note_tier_read_failure(tier: &'static str, key: &str, e: StoreError) -> StoreError {
249 tracing::error!(
250 tier = tier,
251 key = %key,
252 error = %e,
253 "tiered: READ FAILED on a tier — falling through to the next tier; \
254 this tier is degraded and needs attention",
255 );
256 e
257 }
258
259 fn durable_write_outcome(
274 kind: &'static str,
275 key: &str,
276 l2: Result<(), StoreError>,
277 l3: Result<(), StoreError>,
278 ) -> Result<(), StoreError> {
279 match (l2, l3) {
280 (Ok(()), Ok(())) => Ok(()),
281 (Err(e), Ok(())) => {
282 warn!(
283 kind = kind, key = %key, tier = "l2", error = %e,
284 "tiered: durable write failed on ONE tier; the other durable tier \
285 accepted it, so the content is still serveable — redundancy lost",
286 );
287 Ok(())
288 }
289 (Ok(()), Err(e)) => {
290 warn!(
291 kind = kind, key = %key, tier = "l3", error = %e,
292 "tiered: durable write failed on ONE tier; the other durable tier \
293 accepted it, so the content is still serveable — redundancy lost",
294 );
295 Ok(())
296 }
297 (Err(e2), Err(e3)) => {
298 tracing::error!(
299 kind = kind, key = %key, l2_error = %e2, l3_error = %e3,
300 "tiered: durable write failed on EVERY durable tier — nothing was stored",
301 );
302 Err(e2)
304 }
305 }
306 }
307}
308
309#[async_trait]
310impl StorageBackend for TieredBackend {
311 async fn get_narinfo(&self, hash: &str) -> Result<Option<String>, StoreError> {
312 let mut broken: Option<StoreError> = None;
313
314 match self.l1.get_narinfo(hash).await {
316 Ok(Some(v)) => return Ok(Some(v)),
317 Ok(None) => {}
318 Err(e) => broken = Some(Self::note_tier_read_failure("l1", hash, e)),
319 }
320 match self.l2.get_narinfo(hash).await {
322 Ok(Some(v)) => {
323 Self::warm_narinfo(&self.l1, hash, &v).await;
324 return Ok(Some(v));
325 }
326 Ok(None) => {}
327 Err(e) => broken = Some(Self::note_tier_read_failure("l2", hash, e)),
328 }
329 match self.l3.get_narinfo(hash).await {
331 Ok(Some(v)) => {
332 Self::warm_narinfo(&self.l2, hash, &v).await;
333 Self::warm_narinfo(&self.l1, hash, &v).await;
334 return Ok(Some(v));
335 }
336 Ok(None) => {}
337 Err(e) => broken = Some(Self::note_tier_read_failure("l3", hash, e)),
338 }
339
340 match broken {
341 Some(e) => Err(e),
342 None => Ok(None),
343 }
344 }
345
346 async fn get_nar(&self, path: &str) -> Result<Option<Vec<u8>>, StoreError> {
347 match self.get_nar_stream(path).await? {
351 Some(s) => Ok(Some(nar_stream::collect_nar(s, None).await?)),
352 None => Ok(None),
353 }
354 }
355
356 fn nar_residency(&self) -> NarResidency {
361 self.l1
362 .nar_residency()
363 .weaker(self.l2.nar_residency())
364 .weaker(self.l3.nar_residency())
365 }
366
367 async fn get_nar_stream(&self, path: &str) -> Result<Option<NarStream>, StoreError> {
376 let mut broken: Option<StoreError> = None;
377
378 match self.l1.get_nar_stream(path).await {
379 Ok(Some(s)) => return Ok(Some(s)),
380 Ok(None) => {}
381 Err(e) => broken = Some(Self::note_tier_read_failure("l1", path, e)),
382 }
383 match self.l2.get_nar_stream(path).await {
384 Ok(Some(s)) => {
385 Self::warm_nar_from_tier(&self.l1, &self.l2, path).await;
386 return Ok(Some(s));
387 }
388 Ok(None) => {}
389 Err(e) => broken = Some(Self::note_tier_read_failure("l2", path, e)),
390 }
391 match self.l3.get_nar_stream(path).await {
392 Ok(Some(s)) => {
393 Self::warm_nar_from_tier(&self.l2, &self.l3, path).await;
394 Self::warm_nar_from_tier(&self.l1, &self.l3, path).await;
395 return Ok(Some(s));
396 }
397 Ok(None) => {}
398 Err(e) => broken = Some(Self::note_tier_read_failure("l3", path, e)),
399 }
400
401 match broken {
402 Some(e) => Err(e),
403 None => Ok(None),
404 }
405 }
406
407 async fn put_narinfo_record(&self, hash: &str, content: &str) -> Result<(), StoreError> {
408 if self.write_policy == WritePolicy::WriteBack {
409 Self::warm_narinfo(&self.l1, hash, content).await;
410 }
411
412 let l2 = self.l2.put_narinfo_record(hash, content).await;
413 let l3 = self.l3.put_narinfo_record(hash, content).await;
414 Self::durable_write_outcome("narinfo", hash, l2, l3)?;
415
416 if self.write_policy == WritePolicy::WriteThrough {
417 Self::warm_narinfo(&self.l1, hash, content).await;
418 }
419 Ok(())
420 }
421
422 async fn delete_narinfo_record(&self, hash: &str) -> Result<(), StoreError> {
439 let mut failed: Option<StoreError> = None;
440 for (name, tier) in self.tiers() {
441 if let Err(e) = tier.delete_narinfo_record(hash).await {
442 tracing::error!(
443 hash = %hash, tier = name, error = %e,
444 "tiered: narinfo delete FAILED on a tier — reads fall through, so this \
445 narinfo is still servable; refusing to report the delete as complete",
446 );
447 failed = Some(e);
448 }
449 }
450 failed.map_or(Ok(()), Err)
451 }
452
453 async fn delete_nar_record(&self, nar_path: &str) -> Result<(), StoreError> {
459 let mut failed: Option<StoreError> = None;
460 for (name, tier) in self.tiers() {
461 if let Err(e) = tier.delete_nar_record(nar_path).await {
462 warn!(
463 path = %nar_path, tier = name, error = %e,
464 "tiered: NAR delete failed on a tier — a copy survives there",
465 );
466 failed = Some(e);
467 }
468 }
469 failed.map_or(Ok(()), Err)
470 }
471
472 fn nar_ref_index(&self) -> &dyn NarRefIndex {
473 self
474 }
475
476 async fn put_nar(&self, path: &str, data: &[u8]) -> Result<(), StoreError> {
477 self.put_nar_stream(path, &nar_stream::BytesNarSource::from(data)).await
478 }
479
480 async fn put_nar_stream(&self, path: &str, src: &dyn NarSource) -> Result<(), StoreError> {
497 if self.write_policy == WritePolicy::WriteBack {
498 Self::warm_nar_from(&self.l1, path, src).await;
499 }
500
501 let l2 = self.l2.put_nar_stream(path, src).await;
502 let l3 = self.l3.put_nar_stream(path, src).await;
503 Self::durable_write_outcome("nar", path, l2, l3)?;
504
505 if self.write_policy == WritePolicy::WriteThrough {
506 Self::warm_nar_from(&self.l1, path, src).await;
507 }
508 Ok(())
509 }
510
511 async fn list_narinfos(&self) -> Result<Vec<String>, StoreError> {
512 let mut set = BTreeSet::new();
515 set.extend(self.l2.list_narinfos().await?);
516 set.extend(self.l3.list_narinfos().await?);
517 Ok(set.into_iter().collect())
518 }
519
520 async fn wipe_all(&self) -> Result<usize, StoreError> {
526 let mut cleared = 0usize;
527 for (name, tier) in self.tiers() {
528 match tier.wipe_all().await {
529 Ok(n) => cleared = cleared.max(n),
530 Err(e) => warn!(tier = name, error = %e, "tiered: best-effort wipe failed"),
531 }
532 }
533 Ok(cleared)
534 }
535}
536
537#[async_trait]
553impl NarRefIndex for TieredBackend {
554 async fn record(&self, nar_path: &str, hash: &str) -> Result<(), StoreError> {
559 let l2 = self.l2.nar_ref_index().record(nar_path, hash).await;
560 let l3 = self.l3.nar_ref_index().record(nar_path, hash).await;
561 Self::durable_write_outcome("nar-ref", nar_path, l2, l3)?;
562 if let Err(e) = self.l1.nar_ref_index().record(nar_path, hash).await {
563 warn!(path = %nar_path, error = %e, "tiered: best-effort nar-ref warm failed");
564 }
565 Ok(())
566 }
567
568 async fn forget(&self, nar_path: &str, hash: &str) -> Result<(), StoreError> {
574 for (name, tier) in self.tiers() {
575 if let Err(e) = tier.nar_ref_index().forget(nar_path, hash).await {
576 warn!(
577 path = %nar_path, tier = name, error = %e,
578 "tiered: best-effort nar-ref forget failed — the edge survives, so the \
579 NAR is retained rather than stranded",
580 );
581 }
582 }
583 Ok(())
584 }
585
586 async fn referrers(&self, nar_path: &str) -> Result<Vec<String>, StoreError> {
587 let mut set = BTreeSet::new();
588 let mut broken: Option<StoreError> = None;
589 for (name, tier) in self.tiers() {
590 match tier.nar_ref_index().referrers(nar_path).await {
591 Ok(hashes) => set.extend(hashes),
592 Err(e) => {
593 broken = Some(Self::note_tier_read_failure(name, nar_path, e));
594 }
595 }
596 }
597 match broken {
598 Some(e) => Err(e),
599 None => Ok(set.into_iter().collect()),
600 }
601 }
602}
603
604#[cfg(test)]
611mod tests {
612 use super::*;
613 use crate::storage::nar_refs::MemNarRefIndex;
614 use crate::storage::LocalStorage;
615 use std::collections::HashMap;
616 use std::sync::Mutex;
617
618 #[derive(Default)]
622 struct MemBackend {
623 narinfo: Mutex<HashMap<String, String>>,
624 nar: Mutex<HashMap<String, Vec<u8>>>,
625 refs: MemNarRefIndex,
629 writes_fail: Mutex<bool>,
630 reads_fail: Mutex<bool>,
631 deletes_fail: Mutex<bool>,
634 }
635
636 impl MemBackend {
637 fn has_narinfo(&self, hash: &str) -> bool {
638 self.narinfo.lock().unwrap().contains_key(hash)
639 }
640 fn has_nar(&self, path: &str) -> bool {
641 self.nar.lock().unwrap().contains_key(path)
642 }
643 fn clear(&self) {
644 self.narinfo.lock().unwrap().clear();
645 self.nar.lock().unwrap().clear();
646 }
647 fn set_writes_fail(&self, v: bool) {
648 *self.writes_fail.lock().unwrap() = v;
649 }
650 fn set_reads_fail(&self, v: bool) {
654 *self.reads_fail.lock().unwrap() = v;
655 }
656 fn fail_if_configured(&self) -> Result<(), StoreError> {
657 if *self.writes_fail.lock().unwrap() {
658 Err(StoreError::NotImplemented("mock writes disabled"))
659 } else {
660 Ok(())
661 }
662 }
663 fn fail_reads_if_configured(&self) -> Result<(), StoreError> {
664 if *self.reads_fail.lock().unwrap() {
665 Err(StoreError::SchemaMissing("mock: relation does not exist".to_string()))
666 } else {
667 Ok(())
668 }
669 }
670 fn set_deletes_fail(&self, v: bool) {
671 *self.deletes_fail.lock().unwrap() = v;
672 }
673 fn fail_deletes_if_configured(&self) -> Result<(), StoreError> {
674 if *self.deletes_fail.lock().unwrap() {
675 Err(StoreError::NotImplemented("mock deletes disabled"))
676 } else {
677 Ok(())
678 }
679 }
680 }
681
682 #[async_trait]
683 impl StorageBackend for MemBackend {
684 async fn get_narinfo(&self, hash: &str) -> Result<Option<String>, StoreError> {
685 self.fail_reads_if_configured()?;
686 Ok(self.narinfo.lock().unwrap().get(hash).cloned())
687 }
688 async fn put_narinfo_record(&self, hash: &str, content: &str) -> Result<(), StoreError> {
689 self.fail_if_configured()?;
690 self.narinfo.lock().unwrap().insert(hash.to_string(), content.to_string());
691 Ok(())
692 }
693 async fn delete_narinfo_record(&self, hash: &str) -> Result<(), StoreError> {
694 self.fail_deletes_if_configured()?;
695 self.narinfo.lock().unwrap().remove(hash);
696 Ok(())
697 }
698 async fn delete_nar_record(&self, nar_path: &str) -> Result<(), StoreError> {
699 self.fail_deletes_if_configured()?;
700 self.nar.lock().unwrap().remove(nar_path);
701 Ok(())
702 }
703 fn nar_ref_index(&self) -> &dyn NarRefIndex {
704 &self.refs
705 }
706 async fn get_nar(&self, path: &str) -> Result<Option<Vec<u8>>, StoreError> {
707 self.fail_reads_if_configured()?;
708 Ok(self.nar.lock().unwrap().get(path).cloned())
709 }
710 async fn put_nar(&self, path: &str, data: &[u8]) -> Result<(), StoreError> {
711 self.fail_if_configured()?;
712 self.nar.lock().unwrap().insert(path.to_string(), data.to_vec());
713 Ok(())
714 }
715 fn nar_residency(&self) -> NarResidency {
718 NarResidency::WholeValue
719 }
720 async fn list_narinfos(&self) -> Result<Vec<String>, StoreError> {
721 Ok(self.narinfo.lock().unwrap().keys().cloned().collect())
722 }
723 }
724
725 const NARINFO: &str = "StorePath: /nix/store/abc-hello\nURL: nar/abc.nar.xz\nCompression: xz\nNarHash: sha256:bbb\nNarSize: 200\nReferences: \n";
726
727 const ADVERTISED_NAR: &str = "nar/abc.nar.xz";
731
732 fn mocks() -> (Arc<MemBackend>, Arc<MemBackend>, Arc<MemBackend>, TieredBackend) {
735 let l1 = Arc::new(MemBackend::default());
736 let l2 = Arc::new(MemBackend::default());
737 let l3 = Arc::new(MemBackend::default());
738 let tiered = TieredBackend::new(l1.clone(), l2.clone(), l3.clone());
739 (l1, l2, l3, tiered)
740 }
741
742 #[tokio::test]
745 async fn l1_hit_returns_without_touching_lower_tiers() {
746 let (l1, l2, l3, tiered) = mocks();
747 l1.put_narinfo("h", "hot").await.unwrap();
748 assert_eq!(tiered.get_narinfo("h").await.unwrap().unwrap(), "hot");
749 assert!(!l2.has_narinfo("h"));
751 assert!(!l3.has_narinfo("h"));
752 }
753
754 #[tokio::test]
755 async fn l2_hit_promotes_into_l1() {
756 let (l1, l2, _l3, tiered) = mocks();
757 l2.put_narinfo("h", NARINFO).await.unwrap();
758 assert!(!l1.has_narinfo("h"));
759 let got = tiered.get_narinfo("h").await.unwrap().unwrap();
760 assert_eq!(got, NARINFO);
761 assert!(l1.has_narinfo("h"), "L2 hit must promote into L1");
763 }
764
765 #[tokio::test]
766 async fn l3_hit_promotes_into_l2_and_l1() {
767 let (l1, l2, l3, tiered) = mocks();
768 l3.put_narinfo("h", NARINFO).await.unwrap();
769 let got = tiered.get_narinfo("h").await.unwrap().unwrap();
770 assert_eq!(got, NARINFO);
771 assert!(l2.has_narinfo("h"), "L3 hit must promote into L2");
772 assert!(l1.has_narinfo("h"), "L3 hit must promote into L1");
773 }
774
775 #[tokio::test]
776 async fn nar_l3_hit_promotes_into_l2_and_l1() {
777 let (l1, l2, l3, tiered) = mocks();
778 l3.put_nar("nar/x.nar.xz", b"blob").await.unwrap();
779 let got = tiered.get_nar("nar/x.nar.xz").await.unwrap().unwrap();
780 assert_eq!(got, b"blob");
781 assert!(l2.has_nar("nar/x.nar.xz"));
782 assert!(l1.has_nar("nar/x.nar.xz"));
783 }
784
785 #[tokio::test]
786 async fn miss_at_all_tiers_is_none() {
787 let (_l1, _l2, _l3, tiered) = mocks();
788 assert!(tiered.get_narinfo("ghost").await.unwrap().is_none());
789 assert!(tiered.get_nar("nar/ghost.nar.xz").await.unwrap().is_none());
790 }
791
792 #[tokio::test]
795 async fn l2_read_failure_falls_through_to_l3() {
796 let (l1, l2, l3, tiered) = mocks();
801 l3.put_narinfo("h", NARINFO).await.unwrap();
802 l3.put_nar("nar/h.nar.xz", b"blob").await.unwrap();
803 l2.set_reads_fail(true);
804
805 assert_eq!(
806 tiered.get_narinfo("h").await.unwrap().unwrap(),
807 NARINFO,
808 "a broken L2 must not hide a healthy L3",
809 );
810 assert_eq!(tiered.get_nar("nar/h.nar.xz").await.unwrap().unwrap(), b"blob");
811 assert!(l1.has_narinfo("h"), "the L3 hit still warms the working hot tier");
813 }
814
815 #[tokio::test]
816 async fn broken_l1_and_l2_still_serve_from_l3() {
817 let (l1, l2, l3, tiered) = mocks();
818 l3.put_narinfo("h", NARINFO).await.unwrap();
819 l1.set_reads_fail(true);
821 l2.set_reads_fail(true);
822 assert_eq!(tiered.get_narinfo("h").await.unwrap().unwrap(), NARINFO);
823 }
824
825 #[tokio::test]
826 async fn every_tier_broken_and_no_hit_surfaces_an_error_not_a_false_absence() {
827 let (l1, l2, l3, tiered) = mocks();
832 for t in [&l1, &l2, &l3] {
833 t.set_reads_fail(true);
834 }
835 assert!(matches!(
836 tiered.get_narinfo("h").await.unwrap_err(),
837 StoreError::SchemaMissing(_),
838 ));
839 assert!(matches!(
840 tiered.get_nar("nar/h.nar.xz").await.unwrap_err(),
841 StoreError::SchemaMissing(_),
842 ));
843 }
844
845 #[tokio::test]
846 async fn all_tiers_healthy_and_empty_is_a_clean_miss_not_an_error() {
847 let (_l1, _l2, _l3, tiered) = mocks();
850 assert!(tiered.get_narinfo("ghost").await.unwrap().is_none());
851 }
852
853 #[tokio::test]
854 async fn promotion_failure_does_not_break_a_read() {
855 let (l1, l2, _l3, tiered) = mocks();
858 l2.put_narinfo("h", NARINFO).await.unwrap();
859 l1.set_writes_fail(true); let got = tiered.get_narinfo("h").await.unwrap();
861 assert_eq!(got.unwrap(), NARINFO);
862 assert!(!l1.has_narinfo("h"), "warm failed, so L1 stays empty — but the read still succeeded");
863 }
864
865 #[tokio::test]
868 async fn write_through_populates_all_tiers() {
869 let (l1, l2, l3, tiered) = mocks();
870 tiered.put_narinfo("h", NARINFO).await.unwrap();
871 assert!(l1.has_narinfo("h"), "write-through warms L1");
872 assert!(l2.has_narinfo("h"), "write-through persists L2");
873 assert!(l3.has_narinfo("h"), "write-through persists L3");
874 }
875
876 #[tokio::test]
877 async fn write_around_skips_l1_but_persists_durable() {
878 let l1 = Arc::new(MemBackend::default());
879 let l2 = Arc::new(MemBackend::default());
880 let l3 = Arc::new(MemBackend::default());
881 let tiered = TieredBackend::with_write_policy(
882 l1.clone(), l2.clone(), l3.clone(), WritePolicy::WriteAround,
883 );
884 tiered.put_narinfo("h", NARINFO).await.unwrap();
885 assert!(!l1.has_narinfo("h"), "write-around must NOT touch L1");
886 assert!(l2.has_narinfo("h"));
887 assert!(l3.has_narinfo("h"));
888 let _ = tiered.get_narinfo("h").await.unwrap();
890 assert!(l1.has_narinfo("h"), "read-through fills L1 after a write-around");
891 }
892
893 #[tokio::test]
894 async fn write_back_populates_all_tiers_and_is_durable() {
895 let l1 = Arc::new(MemBackend::default());
896 let l2 = Arc::new(MemBackend::default());
897 let l3 = Arc::new(MemBackend::default());
898 let tiered = TieredBackend::with_write_policy(
899 l1.clone(), l2.clone(), l3.clone(), WritePolicy::WriteBack,
900 );
901 assert_eq!(tiered.write_policy(), WritePolicy::WriteBack);
902 tiered.put_nar("nar/x.nar.xz", b"blob").await.unwrap();
903 assert!(l1.has_nar("nar/x.nar.xz"));
905 assert!(l2.has_nar("nar/x.nar.xz"));
906 assert!(l3.has_nar("nar/x.nar.xz"));
907 }
908
909 #[tokio::test]
910 async fn one_broken_durable_tier_still_lands_the_write_on_the_other() {
911 let l1 = Arc::new(MemBackend::default());
916 let l2 = Arc::new(MemBackend::default());
917 let l3 = Arc::new(MemBackend::default());
918 l2.set_writes_fail(true);
919 let tiered = TieredBackend::new(l1.clone(), l2.clone(), l3.clone());
920
921 tiered.put_narinfo("h", NARINFO).await.expect("one healthy durable tier must accept");
922 tiered.put_nar("nar/h.nar.xz", b"blob").await.expect("one healthy durable tier must accept");
923
924 assert!(!l2.has_narinfo("h"), "the broken tier holds nothing");
925 assert!(l3.has_narinfo("h"), "the healthy durable tier MUST have taken the write");
926 assert!(l3.has_nar("nar/h.nar.xz"));
927 assert_eq!(tiered.get_narinfo("h").await.unwrap().unwrap(), NARINFO);
929 }
930
931 #[tokio::test]
932 async fn write_fails_only_when_every_durable_tier_rejects() {
933 let l1 = Arc::new(MemBackend::default());
936 let l2 = Arc::new(MemBackend::default());
937 let l3 = Arc::new(MemBackend::default());
938 l2.set_writes_fail(true);
939 l3.set_writes_fail(true);
940 let tiered = TieredBackend::new(l1, l2, l3);
941 let err = tiered.put_narinfo("h", NARINFO).await.unwrap_err();
942 assert!(matches!(err, StoreError::NotImplemented(_)));
943 }
944
945 #[tokio::test]
948 async fn pod_roll_losing_l1_loses_nothing() {
949 let (l1, _l2, _l3, tiered) = mocks();
950 tiered.put_narinfo("h", NARINFO).await.unwrap();
951 tiered.put_nar("nar/h.nar.xz", b"blob").await.unwrap();
952 l1.clear();
954 assert_eq!(tiered.get_narinfo("h").await.unwrap().unwrap(), NARINFO);
956 assert_eq!(tiered.get_nar("nar/h.nar.xz").await.unwrap().unwrap(), b"blob");
957 }
958
959 #[tokio::test]
962 async fn delete_fans_out_to_all_tiers() {
963 let (l1, l2, l3, tiered) = mocks();
964 tiered.put_narinfo("h", NARINFO).await.unwrap();
965 tiered.put_nar(ADVERTISED_NAR, b"blob").await.unwrap();
966 tiered.delete("h").await.unwrap();
967 for t in [&l1, &l2, &l3] {
968 assert!(!t.has_narinfo("h"));
969 assert!(!t.has_nar(ADVERTISED_NAR));
970 }
971 }
972
973 #[tokio::test]
976 async fn delete_resolves_across_tiers_instead_of_guessing() {
977 let (l1, l2, l3, tiered) = mocks();
978 tiered.put_narinfo("h", NARINFO).await.unwrap();
979 tiered.put_nar(ADVERTISED_NAR, b"blob").await.unwrap();
980 tiered.put_nar("nar/h.nar.zst", b"someone else's nar").await.unwrap();
981
982 tiered.delete("h").await.unwrap();
983
984 for t in [&l1, &l2, &l3] {
985 assert!(!t.has_nar(ADVERTISED_NAR), "the advertised NAR must go");
986 }
987 assert_eq!(
988 tiered.get_nar("nar/h.nar.zst").await.unwrap().unwrap(),
989 b"someone else's nar",
990 );
991 }
992
993 #[tokio::test]
996 async fn a_co_referenced_nar_survives_the_first_delete_on_every_tier() {
997 let (l1, l2, l3, tiered) = mocks();
998 tiered.put_narinfo("pathA", NARINFO).await.unwrap();
999 tiered.put_narinfo("pathB", NARINFO).await.unwrap();
1000 tiered.put_nar(ADVERTISED_NAR, b"shared").await.unwrap();
1001
1002 tiered.delete("pathA").await.unwrap();
1003 for t in [&l1, &l2, &l3] {
1004 assert!(t.has_nar(ADVERTISED_NAR), "pathB still advertises it");
1005 }
1006
1007 tiered.delete("pathB").await.unwrap();
1008 for t in [&l1, &l2, &l3] {
1009 assert!(!t.has_nar(ADVERTISED_NAR), "the last referrer is gone");
1010 }
1011 }
1012
1013 #[tokio::test]
1022 async fn a_failed_narinfo_delete_must_not_take_the_nar_with_it() {
1023 let (l1, l2, l3, tiered) = mocks();
1024 tiered.put_narinfo("h", NARINFO).await.unwrap();
1025 tiered.put_nar(ADVERTISED_NAR, b"blob").await.unwrap();
1026
1027 l2.set_deletes_fail(true);
1030
1031 let err = tiered.delete("h").await.expect_err("a partial delete must surface");
1032 assert!(
1033 matches!(err, StoreError::NotImplemented(_)),
1034 "expected the tier's own error, got {err:?}",
1035 );
1036
1037 assert!(l2.has_narinfo("h"), "L2 kept the narinfo — that is the premise");
1038 assert_eq!(
1039 tiered.get_narinfo("h").await.unwrap().unwrap(),
1040 NARINFO,
1041 "and a read still serves it, because reads fall through",
1042 );
1043 for t in [&l1, &l2, &l3] {
1044 assert!(
1045 t.has_nar(ADVERTISED_NAR),
1046 "the NAR must be untouched: its narinfo is still servable",
1047 );
1048 }
1049 }
1050
1051 #[tokio::test]
1052 async fn wipe_all_clears_every_tier() {
1053 let (l1, l2, l3, tiered) = mocks();
1054 tiered.put_narinfo("h", NARINFO).await.unwrap();
1056 tiered.put_nar(ADVERTISED_NAR, b"blob").await.unwrap();
1057 l2.put_narinfo("only2", "x").await.unwrap();
1059
1060 let removed = tiered.wipe_all().await.unwrap();
1061 assert!(removed >= 1, "wipe reported nothing cleared");
1062
1063 for t in [&l1, &l2, &l3] {
1064 assert!(t.list_narinfos().await.unwrap().is_empty(), "a tier survived the wipe");
1065 assert!(!t.has_narinfo("h"));
1066 assert!(!t.has_nar(ADVERTISED_NAR));
1067 }
1068 assert!(tiered.list_narinfos().await.unwrap().is_empty(), "cache not cold after wipe");
1069 assert!(tiered.get_narinfo("h").await.unwrap().is_none());
1071 }
1072
1073 #[tokio::test]
1074 async fn list_narinfos_unions_durable_tiers_deduped() {
1075 let (l1, l2, l3, tiered) = mocks();
1076 l2.put_narinfo("shared", "x").await.unwrap();
1078 l3.put_narinfo("shared", "x").await.unwrap();
1079 l2.put_narinfo("only2", "y").await.unwrap();
1080 l3.put_narinfo("only3", "z").await.unwrap();
1081 l1.put_narinfo("hot-only", "w").await.unwrap();
1083 let listed = tiered.list_narinfos().await.unwrap();
1084 assert_eq!(listed, vec!["only2".to_string(), "only3".to_string(), "shared".to_string()]);
1085 }
1086
1087 #[tokio::test]
1090 async fn read_through_from_a_real_local_storage_l3() {
1091 let dir = tempfile::tempdir().unwrap();
1092 let l1 = Arc::new(MemBackend::default());
1093 let l2 = Arc::new(MemBackend::default());
1094 let l3_disk = Arc::new(LocalStorage::new(dir.path()));
1095 l3_disk.put_narinfo("h", NARINFO).await.unwrap();
1097 l3_disk.put_nar("nar/h.nar.xz", b"disk-blob").await.unwrap();
1098
1099 let tiered = TieredBackend::new(l1.clone(), l2.clone(), l3_disk);
1100 assert_eq!(tiered.get_narinfo("h").await.unwrap().unwrap(), NARINFO);
1102 assert_eq!(tiered.get_nar("nar/h.nar.xz").await.unwrap().unwrap(), b"disk-blob");
1103 assert!(l1.has_narinfo("h"));
1104 assert!(l2.has_narinfo("h"));
1105 }
1106
1107 struct RecordingTier {
1113 name: &'static str,
1114 log: Arc<Mutex<Vec<&'static str>>>,
1115 refuse: bool,
1117 refs: MemNarRefIndex,
1118 }
1119
1120 #[async_trait]
1121 impl StorageBackend for RecordingTier {
1122 async fn get_narinfo(&self, _h: &str) -> Result<Option<String>, StoreError> {
1123 Ok(None)
1124 }
1125 async fn put_narinfo_record(&self, _h: &str, _c: &str) -> Result<(), StoreError> {
1126 Ok(())
1127 }
1128 async fn delete_narinfo_record(&self, _h: &str) -> Result<(), StoreError> {
1129 Ok(())
1130 }
1131 async fn delete_nar_record(&self, _p: &str) -> Result<(), StoreError> {
1132 Ok(())
1133 }
1134 fn nar_ref_index(&self) -> &dyn NarRefIndex {
1135 &self.refs
1136 }
1137 async fn get_nar(&self, _p: &str) -> Result<Option<Vec<u8>>, StoreError> {
1138 Ok(None)
1139 }
1140 async fn put_nar(&self, _p: &str, _d: &[u8]) -> Result<(), StoreError> {
1141 self.log.lock().unwrap().push(self.name);
1142 if self.refuse {
1143 return Err(StoreError::TooLarge { limit: 1, at_least: 2 });
1144 }
1145 Ok(())
1146 }
1147 fn nar_residency(&self) -> NarResidency {
1148 NarResidency::WholeValue
1149 }
1150 async fn list_narinfos(&self) -> Result<Vec<String>, StoreError> {
1151 Ok(vec![])
1152 }
1153 }
1154
1155 fn recording_tiers(
1156 refuse_l1: bool,
1157 ) -> (Arc<Mutex<Vec<&'static str>>>, TieredBackend) {
1158 let log = Arc::new(Mutex::new(Vec::new()));
1159 let mk = |name, refuse| {
1160 Arc::new(RecordingTier {
1161 name,
1162 log: Arc::clone(&log),
1163 refuse,
1164 refs: MemNarRefIndex::new(),
1165 }) as Arc<dyn StorageBackend>
1166 };
1167 let tiered = TieredBackend::new(mk("l1", refuse_l1), mk("l2", false), mk("l3", false));
1168 (log, tiered)
1169 }
1170
1171 #[tokio::test]
1172 async fn streamed_put_writes_l2_then_l3_then_warms_l1() {
1173 let (log, tiered) = recording_tiers(false);
1178 tiered.put_nar("nar/x.nar.xz", b"blob").await.unwrap();
1179 assert_eq!(*log.lock().unwrap(), vec!["l2", "l3", "l1"]);
1180 }
1181
1182 #[tokio::test]
1183 async fn a_refused_l1_warm_never_fails_the_write() {
1184 let (log, tiered) = recording_tiers(true);
1189 tiered
1190 .put_nar("nar/x.nar.xz", b"blob")
1191 .await
1192 .expect("a refused L1 warm must not fail the write");
1193 assert_eq!(*log.lock().unwrap(), vec!["l2", "l3", "l1"], "L1 is still attempted, last");
1194 }
1195
1196 #[tokio::test]
1197 async fn write_back_warms_l1_before_the_durable_gate() {
1198 let log = Arc::new(Mutex::new(Vec::new()));
1199 let mk = |name| {
1200 Arc::new(RecordingTier {
1201 name,
1202 log: Arc::clone(&log),
1203 refuse: false,
1204 refs: MemNarRefIndex::new(),
1205 }) as Arc<dyn StorageBackend>
1206 };
1207 let tiered = TieredBackend::with_write_policy(
1208 mk("l1"), mk("l2"), mk("l3"), WritePolicy::WriteBack,
1209 );
1210 tiered.put_nar("nar/x.nar.xz", b"blob").await.unwrap();
1211 assert_eq!(*log.lock().unwrap(), vec!["l1", "l2", "l3"]);
1212 }
1213
1214 #[tokio::test]
1215 async fn a_multi_chunk_nar_round_trips_through_a_real_disk_tier() {
1216 let dir = tempfile::tempdir().unwrap();
1219 let l1 = Arc::new(MemBackend::default());
1220 let l2: Arc<dyn StorageBackend> = Arc::new(LocalStorage::new(dir.path().join("l2")));
1221 let l3: Arc<dyn StorageBackend> = Arc::new(LocalStorage::new(dir.path().join("l3")));
1222 let tiered = TieredBackend::new(l1.clone(), l2, l3);
1223
1224 let nar: Vec<u8> = (0..crate::storage::NAR_CHUNK_BYTES + 777).map(|i| (i % 251) as u8).collect();
1225 tiered.put_nar("nar/big.nar.xz", &nar).await.unwrap();
1226 assert_eq!(tiered.get_nar("nar/big.nar.xz").await.unwrap().unwrap(), nar);
1227 }
1228
1229 #[tokio::test]
1230 async fn residency_reports_the_weakest_tier_not_the_resolver() {
1231 let dir = tempfile::tempdir().unwrap();
1235 let disk1: Arc<dyn StorageBackend> = Arc::new(LocalStorage::new(dir.path().join("a")));
1236 let disk2: Arc<dyn StorageBackend> = Arc::new(LocalStorage::new(dir.path().join("b")));
1237 let disk3: Arc<dyn StorageBackend> = Arc::new(LocalStorage::new(dir.path().join("c")));
1238 let all_disk = TieredBackend::new(disk1.clone(), disk2.clone(), disk3);
1239 assert_eq!(all_disk.nar_residency(), NarResidency::Streaming);
1240
1241 let with_double =
1242 TieredBackend::new(Arc::new(MemBackend::default()), disk1, disk2);
1243 assert_eq!(with_double.nar_residency(), NarResidency::WholeValue);
1244 }
1245
1246 #[test]
1249 fn honest_gate_tier_is_mock_parity_proven_not_live_cluster() {
1250 assert_eq!(TIERED_BACKEND_TIER, TieredTier::MockParityProven);
1255 }
1256}