use std::fs::OpenOptions; use std::sync::{Arc, Barrier}; use std::thread; use std::time::Duration; use lazaret_core::codec::encode; use lazaret_core::identity::{ DeviceId, DomainId, GenerationId, HostPath, IdMapDigest, InodeId, PolicyId, PrepareOperationId, PreparedArtifactId, SelectedManifestDigest, SourceIdentity, }; use lazaret_core::records::{ ArtifactLease, ArtifactState, PrepareOperation, PrepareStageState, PreparedArtifact, }; use lazaret_linux::filesystem::try_exclusive_flock; use lazaret_storage_catalog::{Catalog, CatalogError, CatalogRow}; use redb::{Database, TableDefinition}; use serde::Serialize; use tempfile::TempDir; const METADATA: TableDefinition = TableDefinition::new("catalog_metadata"); struct Fixture { directory: TempDir, catalog: Catalog, } impl Fixture { fn new() -> Self { let directory = TempDir::new().unwrap(); let catalog = Catalog::new( directory.path().join("catalog.redb"), directory.path().join("storage-catalog.lock"), ); catalog.initialize().unwrap(); Self { directory, catalog } } fn artifact(seed: u8) -> PreparedArtifact { PreparedArtifact { prepared_artifact_id: PreparedArtifactId::new([seed; 32]), source_identity: SourceIdentity::new(vec![seed, 1]), selected_manifest_digest: SelectedManifestDigest::new([seed; 32]), policy_id: PolicyId::new([seed.wrapping_add(1); 32]), id_map_digest: IdMapDigest::new([seed.wrapping_add(2); 32]), path: HostPath::new(format!("artifacts/{seed}").into_bytes()), device: DeviceId::new(u64::from(seed)), inode: InodeId::new(u64::from(seed) + 100), state: ArtifactState::Available, } } fn prepare(artifact: &PreparedArtifact, seed: u8) -> PrepareOperation { PrepareOperation { operation_id: PrepareOperationId::new([seed; 32]), expected_artifact_id: artifact.prepared_artifact_id, source_identity: artifact.source_identity.clone(), selected_manifest_digest: artifact.selected_manifest_digest, policy_id: artifact.policy_id, id_map_digest: artifact.id_map_digest, stage_path: HostPath::new(format!("stages/{seed}").into_bytes()), stage_state: PrepareStageState::Owned, device: Some(artifact.device), inode: Some(artifact.inode), } } fn publish(&self, seed: u8) -> PreparedArtifact { let artifact = Self::artifact(seed); let operation = Self::prepare(&artifact, seed); let intent = PrepareOperation { stage_state: PrepareStageState::Intent, device: None, inode: None, ..operation.clone() }; self.catalog.record_prepare(&intent).unwrap(); self.catalog .bind_prepare_stage(&intent, &operation) .unwrap(); self.catalog.publish(&operation, &artifact).unwrap(); artifact } } fn lease(artifact: &PreparedArtifact, domain: u8, generation: u64) -> ArtifactLease { ArtifactLease { prepared_artifact_id: artifact.prepared_artifact_id, domain_id: DomainId::new([domain; 32]), generation_id: GenerationId::new(generation), } } #[test] fn catalog_reopens_with_committed_rows() { let fixture = Fixture::new(); let artifact = fixture.publish(1); let first_lease = lease(&artifact, 10, 1); fixture.catalog.acquire_lease(&first_lease).unwrap(); let reopened = Catalog::new( fixture.directory.path().join("catalog.redb"), fixture.directory.path().join("storage-catalog.lock"), ); let snapshot = reopened.snapshot().unwrap(); assert_eq!(snapshot.prepared_artifacts, vec![artifact]); assert_eq!(snapshot.artifact_leases, vec![first_lease]); assert!(snapshot.prepare_operations.is_empty()); } #[test] fn publication_is_one_exact_transaction() { let fixture = Fixture::new(); let artifact = Fixture::artifact(2); let operation = Fixture::prepare(&artifact, 2); let intent = PrepareOperation { stage_state: PrepareStageState::Intent, device: None, inode: None, ..operation.clone() }; fixture.catalog.record_prepare(&intent).unwrap(); fixture .catalog .bind_prepare_stage(&intent, &operation) .unwrap(); let mut forged = operation.clone(); forged.inode = Some(InodeId::new(999)); assert!(matches!( fixture.catalog.publish(&forged, &artifact), Err(CatalogError::OwnershipMismatch { row: CatalogRow::PrepareOperation }) )); let before = fixture.catalog.snapshot().unwrap(); assert_eq!(before.prepare_operations, vec![operation.clone()]); assert!(before.prepared_artifacts.is_empty()); fixture.catalog.publish(&operation, &artifact).unwrap(); let after = fixture.catalog.snapshot().unwrap(); assert!(after.prepare_operations.is_empty()); assert_eq!(after.prepared_artifacts, vec![artifact]); } #[test] fn shared_artifact_waits_for_every_lease() { let fixture = Fixture::new(); let artifact = fixture.publish(3); let first = lease(&artifact, 1, 1); let second = lease(&artifact, 2, 9); fixture.catalog.acquire_lease(&first).unwrap(); fixture.catalog.acquire_lease(&second).unwrap(); assert!(matches!( fixture.catalog.begin_reclaim(&artifact), Err(CatalogError::LeasesPresent) )); fixture.catalog.release_lease(&first).unwrap(); assert!(matches!( fixture.catalog.begin_reclaim(&artifact), Err(CatalogError::LeasesPresent) )); fixture.catalog.release_lease(&second).unwrap(); let reclaiming = fixture.catalog.begin_reclaim(&artifact).unwrap(); assert_eq!(reclaiming.state, ArtifactState::Reclaiming); fixture.catalog.finish_reclaim(&reclaiming).unwrap(); assert!( fixture .catalog .snapshot() .unwrap() .prepared_artifacts .is_empty() ); } #[test] fn release_evidence_is_checked_while_lease_remains_locked() { let fixture = Fixture::new(); let artifact = fixture.publish(31); let exact = lease(&artifact, 1, 3); fixture.catalog.acquire_lease(&exact).unwrap(); let refused = fixture.catalog.release_lease_checked(&exact, || { Err(CatalogError::Io(std::io::Error::other( "injected release-evidence mismatch", ))) }); assert!(matches!(refused, Err(CatalogError::Io(_)))); assert_eq!( fixture.catalog.snapshot().unwrap().artifact_leases, vec![exact] ); fixture .catalog .release_lease_checked(&exact, || Ok(())) .unwrap(); assert!( fixture .catalog .snapshot() .unwrap() .artifact_leases .is_empty() ); } #[test] fn lease_acquisition_binds_the_exact_artifact_row() { let fixture = Fixture::new(); let artifact = fixture.publish(32); let exact = lease(&artifact, 2, 4); let mut forged = artifact.clone(); forged.inode = InodeId::new(9_999); assert!(matches!( fixture.catalog.acquire_lease_checked(&forged, &exact), Err(CatalogError::OwnershipMismatch { row: CatalogRow::PreparedArtifact }) )); assert!( fixture .catalog .snapshot() .unwrap() .artifact_leases .is_empty() ); fixture .catalog .acquire_lease_checked(&artifact, &exact) .unwrap(); } #[test] fn one_domain_generation_cannot_bind_two_artifacts() { let fixture = Fixture::new(); let first_artifact = fixture.publish(5); let second_artifact = fixture.publish(6); let first = lease(&first_artifact, 4, 11); let conflicting = lease(&second_artifact, 4, 11); fixture.catalog.acquire_lease(&first).unwrap(); assert!(matches!( fixture.catalog.acquire_lease(&conflicting), Err(CatalogError::AlreadyExists { row: CatalogRow::ArtifactLease }) )); assert_eq!( fixture.catalog.snapshot().unwrap().artifact_leases, vec![first] ); } #[test] fn lease_and_reclaim_have_exactly_one_transaction_winner() { for seed in 10..30 { let fixture = Fixture::new(); let artifact = fixture.publish(seed); let candidate = lease(&artifact, seed, u64::from(seed)); let barrier = Arc::new(Barrier::new(3)); let lease_catalog = fixture.catalog.clone(); let lease_barrier = Arc::clone(&barrier); let lease_handle = thread::spawn(move || { lease_barrier.wait(); lease_catalog.acquire_lease(&candidate) }); let reclaim_catalog = fixture.catalog.clone(); let reclaim_barrier = Arc::clone(&barrier); let reclaim_artifact = artifact.clone(); let reclaim_handle = thread::spawn(move || { reclaim_barrier.wait(); reclaim_catalog.begin_reclaim(&reclaim_artifact) }); barrier.wait(); let lease_result = lease_handle.join().unwrap(); let reclaim_result = reclaim_handle.join().unwrap(); assert_ne!(lease_result.is_ok(), reclaim_result.is_ok()); if lease_result.is_ok() { assert!(matches!(reclaim_result, Err(CatalogError::LeasesPresent))); } else { assert!(matches!( lease_result, Err(CatalogError::ArtifactUnavailable) )); } } } #[test] fn exact_row_deletion_refuses_forged_ownership() { let fixture = Fixture::new(); let artifact = fixture.publish(4); let first = lease(&artifact, 1, 7); fixture.catalog.acquire_lease(&first).unwrap(); let forged_lease = lease(&artifact, 2, 7); assert!(matches!( fixture.catalog.release_lease(&forged_lease), Err(CatalogError::NotFound { row: CatalogRow::ArtifactLease }) )); assert_eq!( fixture.catalog.snapshot().unwrap().artifact_leases, vec![first] ); fixture.catalog.release_lease(&first).unwrap(); let reclaiming = fixture.catalog.begin_reclaim(&artifact).unwrap(); let mut forged_artifact = reclaiming.clone(); forged_artifact.inode = InodeId::new(999); assert!(matches!( fixture.catalog.finish_reclaim(&forged_artifact), Err(CatalogError::OwnershipMismatch { row: CatalogRow::PreparedArtifact }) )); assert_eq!( fixture.catalog.snapshot().unwrap().prepared_artifacts, vec![reclaiming] ); } #[derive(Serialize)] struct UnknownMetadata<'a> { schema_id: &'a str, schema_version: u32, } impl lazaret_core::records::SchemaType for UnknownMetadata<'_> { const NAME: &'static str = "test::UnknownMetadata"; } #[test] fn unknown_version_refuses_before_rows() { let fixture = Fixture::new(); let database = Database::open(fixture.directory.path().join("catalog.redb")).unwrap(); let transaction = database.begin_write().unwrap(); { let mut metadata = transaction.open_table(METADATA).unwrap(); let encoded = encode(&UnknownMetadata { schema_id: "lazaret-storage-catalog-v1", schema_version: 2, }) .unwrap(); metadata.insert(0, encoded.as_slice()).unwrap(); } transaction.commit().unwrap(); drop(database); assert!(matches!( fixture.catalog.snapshot(), Err(CatalogError::UnknownSchema) )); } #[test] fn lock_wait_is_bounded() { let fixture = Fixture::new(); let lock_file = OpenOptions::new() .read(true) .write(true) .open(fixture.directory.path().join("storage-catalog.lock")) .unwrap(); let _held = try_exclusive_flock(lock_file).unwrap(); let impatient = fixture .catalog .clone() .with_lock_wait(Duration::from_millis(10)); assert!(matches!( impatient.snapshot(), Err(CatalogError::LockTimeout) )); }