use super::*; #[test] fn high_expansion_layer_is_disk_spooled_and_published() { let fixture = Fixture::new(); let payload = vec![0u8; 32 * 1024 * 1024]; let tar = layer(&[Entry { path: "expanded", kind: EntryType::Regular, data: &payload, target: None, mode: 0o600, }]); let compressed = compress(&tar, OCI_GZIP); assert!(compressed.len() * 100 < payload.len()); let request = fixture.request(30, &[(compressed, OCI_GZIP)]); let artifact = fixture.store.prepare(&request).unwrap(); assert_eq!( std::fs::metadata(fixture.artifact_root(&artifact).join("expanded")) .unwrap() .len(), payload.len() as u64 ); assert!( fixture .root .join("stages") .read_dir() .unwrap() .next() .is_none() ); } #[test] fn concurrent_publication_reuses_one_immutable_artifact() { let fixture = Fixture::new(); let bytes = layer(&[Entry { path: "payload", kind: EntryType::Regular, data: b"same", target: None, mode: 0o644, }]); let first = fixture.request(40, &[(bytes.clone(), OCI_TAR)]); let mut second = PrepareRequest { operation_id: PrepareOperationId::new([41; 32]), ..first.clone() }; second.layout_path = first.layout_path.clone(); let store = Arc::new(fixture.store.clone()); let barrier = Arc::new(Barrier::new(3)); let first_handle = { let store = Arc::clone(&store); let barrier = Arc::clone(&barrier); thread::spawn(move || { barrier.wait(); store.prepare(&first) }) }; let second_handle = { let store = Arc::clone(&store); let barrier = Arc::clone(&barrier); thread::spawn(move || { barrier.wait(); store.prepare(&second) }) }; barrier.wait(); let first = first_handle.join().unwrap().unwrap(); let second = second_handle.join().unwrap().unwrap(); assert_eq!(first, second); let snapshot = fixture.catalog.snapshot().unwrap(); assert_eq!(snapshot.prepared_artifacts, vec![first]); assert!(snapshot.prepare_operations.is_empty()); } #[test] fn startup_recovery_preserves_an_independently_locked_owned_stage() { let fixture = Fixture::new(); let operation_id = PrepareOperationId::new([42; 32]); let stage_relative = PathBuf::from("stages").join(hex(operation_id.as_bytes())); let stage_path = fixture.root.join(&stage_relative); std::fs::create_dir(&stage_path).unwrap(); let metadata = stage_path.metadata().unwrap(); let intent = PrepareOperation { operation_id, expected_artifact_id: PreparedArtifactId::new([43; 32]), source_identity: SourceIdentity::new(b"synthetic-live-owner".to_vec()), selected_manifest_digest: SelectedManifestDigest::new([44; 32]), policy_id: PolicyId::new([45; 32]), id_map_digest: IdMapDigest::new([46; 32]), stage_path: HostPath::new(stage_relative.as_os_str().as_bytes().to_vec()), stage_state: PrepareStageState::Intent, device: None, inode: None, }; let owned = PrepareOperation { stage_state: PrepareStageState::Owned, device: Some(DeviceId::new(metadata.dev())), inode: Some(InodeId::new(metadata.ino())), ..intent.clone() }; fixture.catalog.record_prepare(&intent).unwrap(); fixture.catalog.bind_prepare_stage(&intent, &owned).unwrap(); let live_owner = try_exclusive_flock(File::open(&stage_path).unwrap()).unwrap(); let (_, recovery_store) = fixture.independent_store(); assert_eq!( recovery_store.recover_preparations().unwrap(), vec![(operation_id, PrepareRecovery::PreservedLiveOwnedStage,)] ); assert_eq!( fixture.catalog.snapshot().unwrap().prepare_operations, vec![owned] ); assert!(stage_path.is_dir()); drop(live_owner); assert_eq!( recovery_store.recover_preparations().unwrap(), vec![(operation_id, PrepareRecovery::RemovedOwnedStage)] ); assert!( fixture .catalog .snapshot() .unwrap() .prepare_operations .is_empty() ); assert!(!stage_path.exists()); } #[test] fn publication_during_reclaim_preserves_replacement_and_deletes_old_inode() { let fixture = Fixture::new(); let bytes = layer(&[Entry { path: "payload", kind: EntryType::Regular, data: b"same", target: None, mode: 0o644, }]); let first_request = fixture.request(50, &[(bytes, OCI_TAR)]); let first = fixture.store.prepare(&first_request).unwrap(); let reclaiming = fixture.catalog.begin_reclaim(&first).unwrap(); let replacement_request = PrepareRequest { operation_id: PrepareOperationId::new([51; 32]), ..first_request }; let barrier = Arc::new(Barrier::new(3)); let store = Arc::new(fixture.store.clone()); let reclaim_handle = { let barrier = Arc::clone(&barrier); let store = Arc::clone(&store); thread::spawn(move || { barrier.wait(); store.resume_reclaim(&reclaiming) }) }; let publish_handle = { let barrier = Arc::clone(&barrier); let store = Arc::clone(&store); thread::spawn(move || { barrier.wait(); store.prepare(&replacement_request) }) }; barrier.wait(); let reclaim = reclaim_handle.join().unwrap().unwrap(); let replacement = publish_handle.join().unwrap().unwrap(); assert!(matches!( reclaim, ReclaimOutcome::Removed | ReclaimOutcome::ReplacementPreserved )); assert_ne!(replacement.inode, first.inode); assert_eq!( fixture.catalog.snapshot().unwrap().prepared_artifacts, vec![replacement] ); assert!( fixture .root .join("quarantine") .read_dir() .unwrap() .next() .is_none() ); } #[test] fn replacement_winning_before_reclaimer_resumes_still_cleans_old_quarantine() { let fixture = Fixture::new(); let bytes = layer(&[Entry { path: "payload", kind: EntryType::Regular, data: b"same", target: None, mode: 0o644, }]); let first_request = fixture.request(52, &[(bytes, OCI_TAR)]); let first = fixture.store.prepare(&first_request).unwrap(); let reclaiming = fixture.catalog.begin_reclaim(&first).unwrap(); let replacement_request = PrepareRequest { operation_id: PrepareOperationId::new([53; 32]), ..first_request }; let replacement = fixture.store.prepare(&replacement_request).unwrap(); assert_ne!(replacement.inode, first.inode); assert_eq!( fixture.store.resume_reclaim(&reclaiming).unwrap(), ReclaimOutcome::ReplacementPreserved ); assert_eq!( fixture.catalog.snapshot().unwrap().prepared_artifacts, vec![replacement.clone()] ); assert!(fixture.artifact_root(&replacement).exists()); assert!( fixture .root .join("quarantine") .read_dir() .unwrap() .next() .is_none() ); } #[test] fn replacement_reclaim_start_does_not_orphan_prior_quarantine() { let fixture = Fixture::new(); let bytes = layer(&[Entry { path: "payload", kind: EntryType::Regular, data: b"same", target: None, mode: 0o644, }]); let first_request = fixture.request(55, &[(bytes, OCI_TAR)]); let first = fixture.store.prepare(&first_request).unwrap(); let first_reclaiming = fixture.catalog.begin_reclaim(&first).unwrap(); let replacement_request = PrepareRequest { operation_id: PrepareOperationId::new([56; 32]), ..first_request }; let replacement = fixture.store.prepare(&replacement_request).unwrap(); let replacement_reclaiming = fixture.catalog.begin_reclaim(&replacement).unwrap(); assert_eq!( fixture.store.resume_reclaim(&first_reclaiming).unwrap(), ReclaimOutcome::ReplacementPreserved ); assert!(fixture.artifact_root(&replacement).is_dir()); assert!( fixture .root .join("quarantine") .read_dir() .unwrap() .next() .is_none() ); assert_eq!( fixture .store .resume_reclaim(&replacement_reclaiming) .unwrap(), ReclaimOutcome::Removed ); assert!( fixture .catalog .snapshot() .unwrap() .prepared_artifacts .is_empty() ); assert!(!fixture.artifact_root(&replacement).exists()); assert!( fixture .root .join("quarantine") .read_dir() .unwrap() .next() .is_none() ); } #[test] fn reclaim_refuses_preexisting_foreign_quarantine_path() { let fixture = Fixture::new(); let bytes = layer(&[Entry { path: "payload", kind: EntryType::Regular, data: b"same", target: None, mode: 0o644, }]); let artifact = fixture .store .prepare(&fixture.request(54, &[(bytes, OCI_TAR)])) .unwrap(); let foreign = fixture.root.join("quarantine").join(format!( "{}-{}-{}", hex(artifact.prepared_artifact_id.as_bytes()), artifact.device.get(), artifact.inode.get() )); std::fs::create_dir(&foreign).unwrap(); assert!(fixture.store.reclaim(&artifact).is_err()); assert!(foreign.is_dir()); assert!(fixture.artifact_root(&artifact).is_dir()); } #[test] fn recovery_deletes_only_exact_owned_stage_and_preserves_foreign_stage() { let fixture = Fixture::new(); let operation_id = PrepareOperationId::new([60; 32]); let stage = Path::new("stages").join(hex(operation_id.as_bytes())); let intent = PrepareOperation { operation_id, expected_artifact_id: PreparedArtifactId::new([61; 32]), source_identity: SourceIdentity::new(b"source".to_vec()), selected_manifest_digest: SelectedManifestDigest::new([62; 32]), policy_id: PolicyId::new([63; 32]), id_map_digest: IdMapDigest::new([64; 32]), stage_path: HostPath::new(stage.as_os_str().as_encoded_bytes().to_vec()), stage_state: PrepareStageState::Intent, device: None, inode: None, }; fixture.catalog.record_prepare(&intent).unwrap(); std::fs::create_dir(fixture.root.join(&stage)).unwrap(); std::fs::write(fixture.root.join(&stage).join("foreign"), b"keep").unwrap(); assert!(matches!( fixture.store.recover_prepare(&intent), Err(PreparedLowerError::Recovery { .. }) )); assert!(fixture.root.join(&stage).join("foreign").exists()); std::fs::remove_dir_all(fixture.root.join(&stage)).unwrap(); assert_eq!( fixture.store.recover_prepare(&intent).unwrap(), PrepareRecovery::ClearedUnmaterializedIntent ); let owned_id = PrepareOperationId::new([65; 32]); let owned_stage = Path::new("stages").join(hex(owned_id.as_bytes())); let owned_intent = PrepareOperation { operation_id: owned_id, stage_path: HostPath::new(owned_stage.as_os_str().as_encoded_bytes().to_vec()), ..intent }; fixture.catalog.record_prepare(&owned_intent).unwrap(); std::fs::create_dir(fixture.root.join(&owned_stage)).unwrap(); std::fs::write(fixture.root.join(&owned_stage).join("partial"), b"partial").unwrap(); let metadata = std::fs::metadata(fixture.root.join(&owned_stage)).unwrap(); let owned = PrepareOperation { stage_state: PrepareStageState::Owned, device: Some(DeviceId::new(metadata.dev())), inode: Some(InodeId::new(metadata.ino())), ..owned_intent.clone() }; fixture .catalog .bind_prepare_stage(&owned_intent, &owned) .unwrap(); assert_eq!( fixture.store.recover_prepare(&owned).unwrap(), PrepareRecovery::RemovedOwnedStage ); assert!(!fixture.root.join(&owned_stage).exists()); } #[test] fn reclaim_refuses_forged_identity_and_leases() { let fixture = Fixture::new(); let bytes = layer(&[Entry { path: "payload", kind: EntryType::Regular, data: b"same", target: None, mode: 0o644, }]); let artifact = fixture .store .prepare(&fixture.request(70, &[(bytes, OCI_TAR)])) .unwrap(); let lease = lazaret_core::records::ArtifactLease { prepared_artifact_id: artifact.prepared_artifact_id, domain_id: lazaret_core::identity::DomainId::new([71; 32]), generation_id: lazaret_core::identity::GenerationId::new(1), }; fixture.catalog.acquire_lease(&lease).unwrap(); assert!(matches!( fixture.store.reclaim(&artifact), Err(error @ PreparedLowerError::Catalog { .. }) if matches!(error.downcast_ref::(), Some(CatalogError::LeasesPresent)) )); fixture.catalog.release_lease(&lease).unwrap(); let mut forged = artifact.clone(); forged.inode = InodeId::new(artifact.inode.get() + 1); assert!(fixture.store.reclaim(&forged).is_err()); assert!(fixture.artifact_root(&artifact).exists()); } #[test] fn selected_gc_is_idempotent_only_at_the_durable_reclaim_boundary() { let fixture = Fixture::new(); let artifact = fixture .store .prepare(&fixture.request( 71, &[( layer(&[Entry { path: "payload", kind: EntryType::Regular, data: b"gc", target: None, mode: 0o644, }]), OCI_TAR, )], )) .unwrap(); let id = artifact.prepared_artifact_id; assert!(matches!( fixture.store.gc_artifacts(&[id, id]), Err(PreparedLowerError::Reclaim { .. }) )); let lease = lazaret_core::records::ArtifactLease { prepared_artifact_id: id, domain_id: lazaret_core::identity::DomainId::new([72; 32]), generation_id: lazaret_core::identity::GenerationId::new(1), }; fixture.catalog.acquire_lease(&lease).unwrap(); assert!(fixture.store.gc_artifacts(&[id]).is_err()); fixture.catalog.release_lease(&lease).unwrap(); let result = fixture.store.gc_artifacts(&[id]).unwrap(); assert_eq!(result[0].prepared_artifact_id, id); assert_eq!(result[0].outcome, ReclaimOutcome::Removed); assert!(fixture.store.gc_artifacts(&[id]).is_err()); } #[test] fn concurrent_unleased_gc_serializes_observation_with_reclaim() { let fixture = Fixture::new(); let artifact = fixture .store .prepare(&fixture.request( 75, &[( layer(&[Entry { path: "payload", kind: EntryType::Regular, data: b"concurrent-gc", target: None, mode: 0o644, }]), OCI_TAR, )], )) .unwrap(); let (_, first) = fixture.independent_store(); let (_, second) = fixture.independent_store(); let barrier = Arc::new(Barrier::new(3)); let first_barrier = Arc::clone(&barrier); let first = thread::spawn(move || { first_barrier.wait(); first.gc_unleased_artifacts() }); let second_barrier = Arc::clone(&barrier); let second = thread::spawn(move || { second_barrier.wait(); second.gc_unleased_artifacts() }); barrier.wait(); let results = [ first.join().unwrap().unwrap(), second.join().unwrap().unwrap(), ]; assert_eq!(results.iter().map(Vec::len).sum::(), 1); assert_eq!( results.iter().flatten().next().unwrap().outcome, ReclaimOutcome::Removed ); assert!( fixture .catalog .snapshot() .unwrap() .prepared_artifacts .is_empty() ); assert!(!fixture.artifact_root(&artifact).exists()); } #[test] fn selected_artifact_excludes_gc_until_the_claim_is_released() { let fixture = Fixture::new(); let artifact = fixture .store .prepare(&fixture.request( 72, &[( layer(&[Entry { path: "payload", kind: EntryType::Regular, data: b"selected", target: None, mode: 0o644, }]), OCI_TAR, )], )) .unwrap(); let selected = fixture.store.select_artifact(&artifact).unwrap().unwrap(); let (_, gc_store) = fixture.independent_store(); let retained = gc_store .gc_artifacts(&[artifact.prepared_artifact_id]) .unwrap(); assert_eq!(retained[0].outcome, ReclaimOutcome::RetainedBySelection); assert_eq!( fixture.catalog.snapshot().unwrap().prepared_artifacts, vec![artifact.clone()] ); assert!(fixture.artifact_root(&artifact).exists()); drop(selected); let removed = gc_store .gc_artifacts(&[artifact.prepared_artifact_id]) .unwrap(); assert_eq!(removed[0].outcome, ReclaimOutcome::Removed); } #[test] fn selection_rebinds_to_an_equivalent_replacement() { let fixture = Fixture::new(); let bytes = layer(&[Entry { path: "payload", kind: EntryType::Regular, data: b"replacement", target: None, mode: 0o644, }]); let request = fixture.request(73, &[(bytes, OCI_TAR)]); let first = fixture.store.prepare(&request).unwrap(); assert_eq!( fixture.store.reclaim(&first).unwrap(), ReclaimOutcome::Removed ); let replacement = fixture .store .prepare(&PrepareRequest { operation_id: PrepareOperationId::new([74; 32]), ..request }) .unwrap(); let selection = fixture.store.select_artifact(&first).unwrap().unwrap(); assert_eq!(selection.artifact(), &replacement); assert_ne!(selection.artifact().inode, first.inode); } #[test] fn destination_layout_is_materialized_and_canonical_in_artifact_identity() { let fixture = Fixture::new(); let bytes = layer_with_owner(&[], 0, 0); let mut first = fixture.request(72, &[(bytes.clone(), OCI_TAR)]); first.id_map = id_map_with_container_root(); first.destinations = vec![ PreparedDestination { path: "/root".into(), kind: PreparedDestinationKind::Directory, }, PreparedDestination { path: "/usr".into(), kind: PreparedDestinationKind::Directory, }, PreparedDestination { path: "/var".into(), kind: PreparedDestinationKind::Directory, }, PreparedDestination { path: "/etc".into(), kind: PreparedDestinationKind::Directory, }, PreparedDestination { path: "/opt".into(), kind: PreparedDestinationKind::Directory, }, PreparedDestination { path: "/etc/resolv.conf".into(), kind: PreparedDestinationKind::RegularFile, }, PreparedDestination { path: "/etc/hosts".into(), kind: PreparedDestinationKind::RegularFile, }, PreparedDestination { path: "/etc/hostname".into(), kind: PreparedDestinationKind::RegularFile, }, ]; let artifact = fixture.store.prepare(&first).unwrap(); let root = fixture.artifact_root(&artifact); for destination in ["root", "usr", "var", "etc", "opt"] { assert!(root.join(destination).is_dir()); } for destination in ["etc/resolv.conf", "etc/hosts", "etc/hostname"] { assert!(root.join(destination).is_file()); } let mut reordered = fixture.request(73, &[(bytes.clone(), OCI_TAR)]); reordered.id_map = id_map_with_container_root(); reordered.destinations = first.destinations.iter().rev().cloned().collect(); assert_eq!(fixture.store.prepare(&reordered).unwrap(), artifact); let mut unequal = fixture.request(74, &[(bytes, OCI_TAR)]); unequal.id_map = id_map_with_container_root(); unequal.destinations = vec![PreparedDestination { path: "/opt/other".into(), kind: PreparedDestinationKind::Directory, }]; assert_ne!( fixture .store .prepare(&unequal) .unwrap() .prepared_artifact_id, artifact.prepared_artifact_id ); } #[test] fn destination_type_conflict_refuses_before_publication() { let fixture = Fixture::new(); let bytes = layer_with_owner( &[Entry { path: "opt/lazaret-state", kind: EntryType::Regular, data: b"conflict", target: None, mode: 0o644, }], 0, 0, ); let mut request = fixture.request(75, &[(bytes, OCI_TAR)]); request.id_map = id_map_with_container_root(); request.destinations = vec![PreparedDestination { path: "/opt/lazaret-state".into(), kind: PreparedDestinationKind::Directory, }]; assert!(fixture.store.prepare(&request).is_err()); assert!( fixture .catalog .snapshot() .unwrap() .prepared_artifacts .is_empty() ); } fn id_map_with_container_root() -> IdMap { IdMap { uid: vec![IdMapRange { container_id: 0, namespace_id: rustix::process::geteuid().as_raw(), size: 1, }], gid: vec![IdMapRange { container_id: 0, namespace_id: rustix::process::getegid().as_raw(), size: 1, }], } } pub(super) fn hex(bytes: &[u8]) -> String { use std::fmt::Write as _; bytes.iter().fold( String::with_capacity(bytes.len() * 2), |mut output, byte| { write!(output, "{byte:02x}").expect("writing to String cannot fail"); output }, ) }