lazaret/crates/lazaret-storage-catalog/src/lib.rs

#![doc = "Bounded transactional access to Lazaret's shared prepared-artifact catalog."]

use std::fmt;
use std::fs::{File, OpenOptions};
use std::io;
use std::os::fd::AsFd;
use std::os::unix::ffi::OsStringExt;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::thread;
use std::time::{Duration, Instant};

use lazaret_core::codec::{self, CodecError};
use lazaret_core::configuration::ConfinedPathPlan;
use lazaret_core::identity::{MountIdentity, PrepareOperationId, PreparedArtifactId};
use lazaret_core::records::{
    ArtifactLease, ArtifactState, PrepareOperation, PrepareStageState, PreparedArtifact,
    StorageCatalog, StorageCatalogMetadata,
};
use lazaret_linux::filesystem::{
    ExclusiveFileLock, mount_identity, require_mount_identity, try_exclusive_flock,
};
use redb::{
    Database, ReadTransaction, ReadableDatabase, ReadableTable, TableDefinition, WriteTransaction,
};
use rustix::fs::{Mode, OFlags, ResolveFlags, openat2};

const METADATA: TableDefinition<u8, &[u8]> = TableDefinition::new("catalog_metadata");
const ARTIFACTS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("prepared_artifacts");
const LEASES: TableDefinition<&[u8], &[u8]> = TableDefinition::new("artifact_leases");
const PREPARES: TableDefinition<&[u8], &[u8]> = TableDefinition::new("prepare_operations");
const METADATA_KEY: u8 = 0;
const DEFAULT_LOCK_WAIT: Duration = Duration::from_secs(5);
const LOCK_RETRY_INTERVAL: Duration = Duration::from_millis(2);

#[derive(Clone)]
pub struct Catalog {
    location: CatalogLocation,
    lock_wait: Duration,
}

#[derive(Clone)]
enum CatalogLocation {
    Ambient {
        database_path: PathBuf,
        lock_path: PathBuf,
    },
    StorageRoot(Arc<StorageRootCatalog>),
}

struct StorageRootCatalog {
    root: File,
    root_label: PathBuf,
    identity: MountIdentity,
}

impl fmt::Debug for Catalog {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        let location = match &self.location {
            CatalogLocation::Ambient { .. } => "ambient",
            CatalogLocation::StorageRoot(_) => "storage-root",
        };
        formatter
            .debug_struct("Catalog")
            .field("location", &location)
            .field("lock_wait", &self.lock_wait)
            .finish()
    }
}

impl Catalog {
    #[must_use]
    pub fn new(database_path: impl Into<PathBuf>, lock_path: impl Into<PathBuf>) -> Self {
        Self {
            location: CatalogLocation::Ambient {
                database_path: database_path.into(),
                lock_path: lock_path.into(),
            },
            lock_wait: DEFAULT_LOCK_WAIT,
        }
    }

    /// Binds the fixed catalog children to the validated Storage Root authority.
    ///
    /// # Errors
    /// Returns an error when the exact Storage Root cannot be retained and identified.
    pub fn under_storage_root(storage_root: &ConfinedPathPlan) -> Result<Self, CatalogError> {
        let root = File::from(
            storage_root
                .open_authority()
                .map_err(|error| CatalogError::Io(io::Error::other(error.to_string())))?,
        );
        let identity = mount_identity(&root).map_err(CatalogError::Io)?;
        let root_label = PathBuf::from(std::ffi::OsString::from_vec(
            storage_root.canonical_label().as_bytes().to_vec(),
        ));
        Ok(Self {
            location: CatalogLocation::StorageRoot(Arc::new(StorageRootCatalog {
                root,
                root_label,
                identity,
            })),
            lock_wait: DEFAULT_LOCK_WAIT,
        })
    }

    #[must_use]
    pub fn with_lock_wait(mut self, lock_wait: Duration) -> Self {
        self.lock_wait = lock_wait;
        self
    }

    /// Creates all catalog tables and commits the schema identity atomically.
    ///
    /// # Errors
    ///
    /// Returns an error if locking, creation, or the initialization transaction fails.
    pub fn initialize(&self) -> Result<(), CatalogError> {
        self.with_lock(|| {
            if self.database_exists()? {
                return Err(CatalogError::AlreadyInitialized);
            }
            let database_path = self.database_path_for_handoff()?;
            let database = Database::create(database_path).map_err(CatalogError::redb)?;
            let transaction = database.begin_write().map_err(CatalogError::redb)?;
            {
                let mut metadata = transaction
                    .open_table(METADATA)
                    .map_err(CatalogError::redb)?;
                let encoded = encode(&StorageCatalogMetadata::v1())?;
                metadata
                    .insert(METADATA_KEY, encoded.as_slice())
                    .map_err(CatalogError::redb)?;
                transaction
                    .open_table(ARTIFACTS)
                    .map_err(CatalogError::redb)?;
                transaction.open_table(LEASES).map_err(CatalogError::redb)?;
                transaction
                    .open_table(PREPARES)
                    .map_err(CatalogError::redb)?;
            }
            transaction.commit().map_err(CatalogError::redb)
        })
    }

    /// Reads a consistent snapshot in one catalog transaction.
    ///
    /// # Errors
    ///
    /// Returns an error for lock refusal, database failure, unknown schema metadata, or corrupt rows.
    pub fn snapshot(&self) -> Result<StorageCatalog, CatalogError> {
        self.read(|transaction| {
            let prepared_artifacts = read_rows(transaction, ARTIFACTS)?;
            let artifact_leases = read_rows(transaction, LEASES)?;
            let prepare_operations = read_rows(transaction, PREPARES)?;
            Ok(StorageCatalog {
                metadata: StorageCatalogMetadata::v1(),
                prepared_artifacts,
                artifact_leases,
                prepare_operations,
            })
        })
    }

    /// Records durable preparation ownership before a staging tree is created.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation identity already exists or the transaction fails.
    pub fn record_prepare(&self, operation: &PrepareOperation) -> Result<(), CatalogError> {
        if operation.stage_state != PrepareStageState::Intent
            || !operation.has_valid_evidence_shape()
        {
            return Err(CatalogError::InvalidPrepareState);
        }
        self.write(|transaction| {
            let key = operation_key(operation.operation_id);
            let mut table = transaction
                .open_table(PREPARES)
                .map_err(CatalogError::redb)?;
            ensure_absent(&table, &key, CatalogRow::PrepareOperation)?;
            let encoded = encode(operation)?;
            table
                .insert(key.as_slice(), encoded.as_slice())
                .map_err(CatalogError::redb)?;
            Ok(())
        })
    }

    /// Binds exact physical stage ownership to an already durable preparation intent.
    ///
    /// # Errors
    ///
    /// Returns an error for nonmatching identity, invalid phase/evidence, or transaction failure.
    pub fn bind_prepare_stage(
        &self,
        expected_intent: &PrepareOperation,
        owned_stage: &PrepareOperation,
    ) -> Result<(), CatalogError> {
        if expected_intent.stage_state != PrepareStageState::Intent
            || owned_stage.stage_state != PrepareStageState::Owned
            || !expected_intent.has_valid_evidence_shape()
            || !owned_stage.has_valid_evidence_shape()
            || expected_intent.operation_id != owned_stage.operation_id
            || expected_intent.expected_artifact_id != owned_stage.expected_artifact_id
            || expected_intent.source_identity != owned_stage.source_identity
            || expected_intent.selected_manifest_digest != owned_stage.selected_manifest_digest
            || expected_intent.policy_id != owned_stage.policy_id
            || expected_intent.id_map_digest != owned_stage.id_map_digest
            || expected_intent.stage_path != owned_stage.stage_path
        {
            return Err(CatalogError::InvalidPrepareState);
        }
        self.write(|transaction| {
            let key = operation_key(expected_intent.operation_id);
            let mut table = transaction
                .open_table(PREPARES)
                .map_err(CatalogError::redb)?;
            require_exact(&table, &key, expected_intent, CatalogRow::PrepareOperation)?;
            let encoded = encode(owned_stage)?;
            table
                .insert(key.as_slice(), encoded.as_slice())
                .map_err(CatalogError::redb)?;
            Ok(())
        })
    }

    /// Removes only an exactly matching preparation ownership row.
    ///
    /// # Errors
    ///
    /// Returns an error when the row is absent, differs from `expected`, or the transaction fails.
    pub fn remove_prepare(&self, expected: &PrepareOperation) -> Result<(), CatalogError> {
        self.write(|transaction| {
            let key = operation_key(expected.operation_id);
            let mut table = transaction
                .open_table(PREPARES)
                .map_err(CatalogError::redb)?;
            require_exact(&table, &key, expected, CatalogRow::PrepareOperation)?;
            table.remove(key.as_slice()).map_err(CatalogError::redb)?;
            Ok(())
        })
    }

    /// Atomically publishes an artifact and removes its exact preparation ownership row.
    ///
    /// # Errors
    ///
    /// Returns an error for mismatched ownership, invalid publication state, identity mismatch,
    /// duplicate artifact identity, or transaction failure.
    pub fn publish(
        &self,
        expected_operation: &PrepareOperation,
        artifact: &PreparedArtifact,
    ) -> Result<(), CatalogError> {
        if artifact.state != ArtifactState::Available
            || expected_operation.stage_state != PrepareStageState::Owned
            || !expected_operation.has_valid_evidence_shape()
        {
            return Err(CatalogError::InvalidArtifactState);
        }
        if artifact.prepared_artifact_id != expected_operation.expected_artifact_id
            || artifact.source_identity != expected_operation.source_identity
            || artifact.selected_manifest_digest != expected_operation.selected_manifest_digest
            || artifact.policy_id != expected_operation.policy_id
            || artifact.id_map_digest != expected_operation.id_map_digest
            || expected_operation.device != Some(artifact.device)
            || expected_operation.inode != Some(artifact.inode)
        {
            return Err(CatalogError::OwnershipMismatch {
                row: CatalogRow::PrepareOperation,
            });
        }

        self.write(|transaction| {
            let prepare_key = operation_key(expected_operation.operation_id);
            let artifact_key = artifact_key(artifact.prepared_artifact_id);
            {
                let prepares = transaction
                    .open_table(PREPARES)
                    .map_err(CatalogError::redb)?;
                require_exact(
                    &prepares,
                    &prepare_key,
                    expected_operation,
                    CatalogRow::PrepareOperation,
                )?;
                let artifacts = transaction
                    .open_table(ARTIFACTS)
                    .map_err(CatalogError::redb)?;
                ensure_absent(&artifacts, &artifact_key, CatalogRow::PreparedArtifact)?;
            }
            {
                let encoded = encode(artifact)?;
                let mut artifacts = transaction
                    .open_table(ARTIFACTS)
                    .map_err(CatalogError::redb)?;
                artifacts
                    .insert(artifact_key.as_slice(), encoded.as_slice())
                    .map_err(CatalogError::redb)?;
            }
            let mut prepares = transaction
                .open_table(PREPARES)
                .map_err(CatalogError::redb)?;
            prepares
                .remove(prepare_key.as_slice())
                .map_err(CatalogError::redb)?;
            Ok(())
        })
    }

    /// Atomically replaces an exact reclaiming generation and consumes its prepare ownership.
    ///
    /// # Errors
    ///
    /// Returns an error for mismatched old/new ownership, invalid state, leases, or transaction
    /// failure.
    pub fn publish_replacement(
        &self,
        expected_operation: &PrepareOperation,
        expected_reclaiming: &PreparedArtifact,
        replacement: &PreparedArtifact,
    ) -> Result<(), CatalogError> {
        if expected_operation.stage_state != PrepareStageState::Owned
            || !expected_operation.has_valid_evidence_shape()
            || expected_reclaiming.state != ArtifactState::Reclaiming
            || replacement.state != ArtifactState::Available
            || replacement.prepared_artifact_id != expected_reclaiming.prepared_artifact_id
            || replacement.prepared_artifact_id != expected_operation.expected_artifact_id
            || replacement.source_identity != expected_operation.source_identity
            || replacement.selected_manifest_digest != expected_operation.selected_manifest_digest
            || replacement.policy_id != expected_operation.policy_id
            || replacement.id_map_digest != expected_operation.id_map_digest
            || expected_operation.device != Some(replacement.device)
            || expected_operation.inode != Some(replacement.inode)
            || replacement.source_identity != expected_reclaiming.source_identity
            || replacement.selected_manifest_digest != expected_reclaiming.selected_manifest_digest
            || replacement.policy_id != expected_reclaiming.policy_id
            || replacement.id_map_digest != expected_reclaiming.id_map_digest
            || replacement.path != expected_reclaiming.path
        {
            return Err(CatalogError::InvalidPrepareState);
        }
        self.write(|transaction| {
            let prepare_key = operation_key(expected_operation.operation_id);
            let artifact_key = artifact_key(replacement.prepared_artifact_id);
            {
                let prepares = transaction
                    .open_table(PREPARES)
                    .map_err(CatalogError::redb)?;
                require_exact(
                    &prepares,
                    &prepare_key,
                    expected_operation,
                    CatalogRow::PrepareOperation,
                )?;
                let artifacts = transaction
                    .open_table(ARTIFACTS)
                    .map_err(CatalogError::redb)?;
                require_exact(
                    &artifacts,
                    &artifact_key,
                    expected_reclaiming,
                    CatalogRow::PreparedArtifact,
                )?;
            }
            if artifact_has_lease(transaction, replacement.prepared_artifact_id)? {
                return Err(CatalogError::LeasesPresent);
            }
            let encoded = encode(replacement)?;
            let mut artifacts = transaction
                .open_table(ARTIFACTS)
                .map_err(CatalogError::redb)?;
            artifacts
                .insert(artifact_key.as_slice(), encoded.as_slice())
                .map_err(CatalogError::redb)?;
            drop(artifacts);
            let mut prepares = transaction
                .open_table(PREPARES)
                .map_err(CatalogError::redb)?;
            prepares
                .remove(prepare_key.as_slice())
                .map_err(CatalogError::redb)?;
            Ok(())
        })
    }

    /// Inserts one exact Domain-generation lease for an available artifact.
    ///
    /// # Errors
    ///
    /// Returns an error if the artifact is absent or reclaiming, the lease exists, or the
    /// transaction fails.
    pub fn acquire_lease(&self, lease: &ArtifactLease) -> Result<(), CatalogError> {
        self.acquire_lease_with(lease, None)
    }

    /// Inserts one lease only for an exactly matching available artifact row.
    ///
    /// # Errors
    /// Returns an error for artifact identity drift, absence, reclaim, lease conflict, or a
    /// transaction failure.
    pub fn acquire_lease_checked(
        &self,
        expected_artifact: &PreparedArtifact,
        lease: &ArtifactLease,
    ) -> Result<(), CatalogError> {
        if expected_artifact.prepared_artifact_id != lease.prepared_artifact_id {
            return Err(CatalogError::OwnershipMismatch {
                row: CatalogRow::PreparedArtifact,
            });
        }
        self.acquire_lease_with(lease, Some(expected_artifact))
    }

    /// Inserts an exact lease or accepts the same already-committed lease checkpoint.
    ///
    /// # Errors
    /// Returns an error for artifact or lease ownership drift, absence, reclaim, or a transaction
    /// failure.
    pub fn acquire_or_resume_lease_checked(
        &self,
        expected_artifact: &PreparedArtifact,
        lease: &ArtifactLease,
    ) -> Result<(), CatalogError> {
        if expected_artifact.prepared_artifact_id != lease.prepared_artifact_id {
            return Err(CatalogError::OwnershipMismatch {
                row: CatalogRow::PreparedArtifact,
            });
        }
        self.write(|transaction| {
            let artifact_key = artifact_key(lease.prepared_artifact_id);
            let artifacts = transaction
                .open_table(ARTIFACTS)
                .map_err(CatalogError::redb)?;
            let artifact: PreparedArtifact =
                require_row(&artifacts, &artifact_key, CatalogRow::PreparedArtifact)?;
            if artifact != *expected_artifact {
                return Err(CatalogError::OwnershipMismatch {
                    row: CatalogRow::PreparedArtifact,
                });
            }
            if artifact.state != ArtifactState::Available {
                return Err(CatalogError::ArtifactUnavailable);
            }
            drop(artifacts);

            let key = lease_key(lease);
            let mut leases = transaction.open_table(LEASES).map_err(CatalogError::redb)?;
            if leases
                .get(key.as_slice())
                .map_err(CatalogError::redb)?
                .is_some()
            {
                return require_exact(&leases, &key, lease, CatalogRow::ArtifactLease);
            }
            let encoded = encode(lease)?;
            leases
                .insert(key.as_slice(), encoded.as_slice())
                .map_err(CatalogError::redb)?;
            Ok(())
        })
    }

    fn acquire_lease_with(
        &self,
        lease: &ArtifactLease,
        expected_artifact: Option<&PreparedArtifact>,
    ) -> Result<(), CatalogError> {
        self.write(|transaction| {
            let artifact_key = artifact_key(lease.prepared_artifact_id);
            let artifacts = transaction
                .open_table(ARTIFACTS)
                .map_err(CatalogError::redb)?;
            let artifact: PreparedArtifact =
                require_row(&artifacts, &artifact_key, CatalogRow::PreparedArtifact)?;
            if expected_artifact.is_some_and(|expected| *expected != artifact) {
                return Err(CatalogError::OwnershipMismatch {
                    row: CatalogRow::PreparedArtifact,
                });
            }
            if artifact.state != ArtifactState::Available {
                return Err(CatalogError::ArtifactUnavailable);
            }
            drop(artifacts);

            let lease_key = lease_key(lease);
            let mut leases = transaction.open_table(LEASES).map_err(CatalogError::redb)?;
            ensure_absent(&leases, &lease_key, CatalogRow::ArtifactLease)?;
            let encoded = encode(lease)?;
            leases
                .insert(lease_key.as_slice(), encoded.as_slice())
                .map_err(CatalogError::redb)?;
            Ok(())
        })
    }

    /// Removes only one exactly matching Domain-generation lease.
    ///
    /// # Errors
    ///
    /// Returns an error when the exact row is absent, forged, or the transaction fails.
    pub fn release_lease(&self, expected: &ArtifactLease) -> Result<(), CatalogError> {
        self.release_lease_checked(expected, || Ok(()))
    }

    /// Removes one exact lease only after caller-owned Domain and mount evidence is rechecked
    /// while the catalog lock is held.
    ///
    /// # Errors
    /// Returns the evidence refusal or any exact-row and transaction failure without removal.
    pub fn release_lease_checked(
        &self,
        expected: &ArtifactLease,
        verify_release: impl FnOnce() -> Result<(), CatalogError>,
    ) -> Result<(), CatalogError> {
        self.write(|transaction| {
            verify_release()?;
            let key = lease_key(expected);
            let mut table = transaction.open_table(LEASES).map_err(CatalogError::redb)?;
            require_exact(&table, &key, expected, CatalogRow::ArtifactLease)?;
            table.remove(key.as_slice()).map_err(CatalogError::redb)?;
            Ok(())
        })
    }

    /// Atomically excludes new leases by changing an exact zero-lease artifact to `Reclaiming`.
    ///
    /// # Errors
    ///
    /// Returns an error for forged ownership, a non-available artifact, any remaining lease, or
    /// transaction failure.
    pub fn begin_reclaim(
        &self,
        expected: &PreparedArtifact,
    ) -> Result<PreparedArtifact, CatalogError> {
        if expected.state != ArtifactState::Available {
            return Err(CatalogError::InvalidArtifactState);
        }
        self.write(|transaction| {
            let key = artifact_key(expected.prepared_artifact_id);
            {
                let artifacts = transaction
                    .open_table(ARTIFACTS)
                    .map_err(CatalogError::redb)?;
                require_exact(&artifacts, &key, expected, CatalogRow::PreparedArtifact)?;
            }
            if artifact_has_lease(transaction, expected.prepared_artifact_id)? {
                return Err(CatalogError::LeasesPresent);
            }

            let reclaiming = PreparedArtifact {
                state: ArtifactState::Reclaiming,
                ..expected.clone()
            };
            let encoded = encode(&reclaiming)?;
            let mut artifacts = transaction
                .open_table(ARTIFACTS)
                .map_err(CatalogError::redb)?;
            artifacts
                .insert(key.as_slice(), encoded.as_slice())
                .map_err(CatalogError::redb)?;
            Ok(reclaiming)
        })
    }

    /// Removes one exact `Reclaiming` row after physical quarantine deletion succeeds.
    ///
    /// # Errors
    ///
    /// Returns an error for forged ownership, wrong state, a remaining lease, or transaction
    /// failure.
    pub fn finish_reclaim(&self, expected: &PreparedArtifact) -> Result<(), CatalogError> {
        if expected.state != ArtifactState::Reclaiming {
            return Err(CatalogError::InvalidArtifactState);
        }
        self.write(|transaction| {
            let key = artifact_key(expected.prepared_artifact_id);
            {
                let artifacts = transaction
                    .open_table(ARTIFACTS)
                    .map_err(CatalogError::redb)?;
                require_exact(&artifacts, &key, expected, CatalogRow::PreparedArtifact)?;
            }
            if artifact_has_lease(transaction, expected.prepared_artifact_id)? {
                return Err(CatalogError::LeasesPresent);
            }
            let mut artifacts = transaction
                .open_table(ARTIFACTS)
                .map_err(CatalogError::redb)?;
            artifacts
                .remove(key.as_slice())
                .map_err(CatalogError::redb)?;
            Ok(())
        })
    }

    fn read<T>(
        &self,
        operation: impl FnOnce(&ReadTransaction) -> Result<T, CatalogError>,
    ) -> Result<T, CatalogError> {
        self.with_lock(|| {
            let database_path = self.database_path_for_handoff()?;
            let database = Database::open(database_path).map_err(CatalogError::redb)?;
            let transaction = database.begin_read().map_err(CatalogError::redb)?;
            validate_read_metadata(&transaction)?;
            operation(&transaction)
        })
    }

    fn write<T>(
        &self,
        operation: impl FnOnce(&WriteTransaction) -> Result<T, CatalogError>,
    ) -> Result<T, CatalogError> {
        self.with_lock(|| {
            let database_path = self.database_path_for_handoff()?;
            let database = Database::open(database_path).map_err(CatalogError::redb)?;
            let transaction = database.begin_write().map_err(CatalogError::redb)?;
            validate_write_metadata(&transaction)?;
            let result = operation(&transaction)?;
            transaction.commit().map_err(CatalogError::redb)?;
            Ok(result)
        })
    }

    fn with_lock<T>(
        &self,
        operation: impl FnOnce() -> Result<T, CatalogError>,
    ) -> Result<T, CatalogError> {
        let _lock = match &self.location {
            CatalogLocation::Ambient { lock_path, .. } => acquire_lock(lock_path, self.lock_wait)?,
            CatalogLocation::StorageRoot(location) => {
                acquire_storage_root_lock(location, self.lock_wait)?
            }
        };
        operation()
    }

    fn database_exists(&self) -> Result<bool, CatalogError> {
        match &self.location {
            CatalogLocation::Ambient { database_path, .. } => Ok(database_path.exists()),
            CatalogLocation::StorageRoot(location) => match open_storage_root_child(
                location,
                "catalog.redb",
                OFlags::RDONLY | OFlags::NOFOLLOW,
                Mode::empty(),
            ) {
                Ok(_) => Ok(true),
                Err(CatalogError::Io(error)) if error.kind() == io::ErrorKind::NotFound => {
                    Ok(false)
                }
                Err(error) => Err(error),
            },
        }
    }

    fn database_path_for_handoff(&self) -> Result<PathBuf, CatalogError> {
        match &self.location {
            CatalogLocation::Ambient { database_path, .. } => Ok(database_path.clone()),
            CatalogLocation::StorageRoot(location) => {
                require_mount_identity(&location.root_label, location.identity)
                    .map_err(CatalogError::Io)?;
                Ok(location.root_label.join("catalog.redb"))
            }
        }
    }
}

fn acquire_lock(path: &Path, wait: Duration) -> Result<ExclusiveFileLock, CatalogError> {
    let deadline = Instant::now() + wait;
    loop {
        let file = open_lock(path)?;
        match try_exclusive_flock(file) {
            Ok(lock) => return Ok(lock),
            Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
                let now = Instant::now();
                if now >= deadline {
                    return Err(CatalogError::LockTimeout);
                }
                thread::sleep(LOCK_RETRY_INTERVAL.min(deadline.saturating_duration_since(now)));
            }
            Err(error) => return Err(CatalogError::Io(error)),
        }
    }
}

fn acquire_storage_root_lock(
    location: &StorageRootCatalog,
    wait: Duration,
) -> Result<ExclusiveFileLock, CatalogError> {
    let deadline = Instant::now() + wait;
    loop {
        let file = File::from(open_storage_root_child(
            location,
            "storage-catalog.lock",
            OFlags::CREATE | OFlags::RDWR | OFlags::NOFOLLOW,
            Mode::from_raw_mode(0o600),
        )?);
        match try_exclusive_flock(file) {
            Ok(lock) => return Ok(lock),
            Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
                let now = Instant::now();
                if now >= deadline {
                    return Err(CatalogError::LockTimeout);
                }
                thread::sleep(LOCK_RETRY_INTERVAL.min(deadline.saturating_duration_since(now)));
            }
            Err(error) => return Err(CatalogError::Io(error)),
        }
    }
}

fn open_storage_root_child(
    location: &StorageRootCatalog,
    name: &str,
    flags: OFlags,
    mode: Mode,
) -> Result<std::os::fd::OwnedFd, CatalogError> {
    if mount_identity(&location.root).map_err(CatalogError::Io)? != location.identity {
        return Err(CatalogError::Io(io::Error::other(
            "retained Storage Root identity changed",
        )));
    }
    openat2(
        location.root.as_fd(),
        name,
        flags | OFlags::CLOEXEC,
        mode,
        ResolveFlags::BENEATH
            | ResolveFlags::NO_MAGICLINKS
            | ResolveFlags::NO_SYMLINKS
            | ResolveFlags::NO_XDEV,
    )
    .map_err(|error| CatalogError::Io(error.into()))
}

fn open_lock(path: &Path) -> Result<File, CatalogError> {
    OpenOptions::new()
        .create(true)
        .read(true)
        .truncate(false)
        .write(true)
        .open(path)
        .map_err(CatalogError::Io)
}

fn validate_read_metadata(transaction: &ReadTransaction) -> Result<(), CatalogError> {
    let table = transaction
        .open_table(METADATA)
        .map_err(CatalogError::redb)?;
    validate_metadata_table(&table)
}

fn validate_write_metadata(transaction: &WriteTransaction) -> Result<(), CatalogError> {
    let table = transaction
        .open_table(METADATA)
        .map_err(CatalogError::redb)?;
    validate_metadata_table(&table)
}

fn validate_metadata_table(
    table: &impl ReadableTable<u8, &'static [u8]>,
) -> Result<(), CatalogError> {
    let encoded = table
        .get(METADATA_KEY)
        .map_err(CatalogError::redb)?
        .ok_or(CatalogError::MissingMetadata)?;
    codec::decode::<StorageCatalogMetadata>(encoded.value())
        .map(|_| ())
        .map_err(|_| CatalogError::UnknownSchema)
}

fn artifact_has_lease(
    transaction: &WriteTransaction,
    artifact_id: PreparedArtifactId,
) -> Result<bool, CatalogError> {
    let table = transaction.open_table(LEASES).map_err(CatalogError::redb)?;
    for row in table.iter().map_err(CatalogError::redb)? {
        let (key, value) = row.map_err(CatalogError::redb)?;
        let lease: ArtifactLease = decode_row(value.value())?;
        if lease_key(&lease).as_slice() != key.value() {
            return Err(CatalogError::OwnershipMismatch {
                row: CatalogRow::ArtifactLease,
            });
        }
        if lease.prepared_artifact_id == artifact_id {
            return Ok(true);
        }
    }
    Ok(false)
}

fn read_rows<T>(
    transaction: &ReadTransaction,
    definition: TableDefinition<&[u8], &[u8]>,
) -> Result<Vec<T>, CatalogError>
where
    T: CatalogRecord,
{
    let table = transaction
        .open_table(definition)
        .map_err(CatalogError::redb)?;
    table
        .iter()
        .map_err(CatalogError::redb)?
        .map(|row| {
            let (key, value) = row.map_err(CatalogError::redb)?;
            let record: T = decode_row(value.value())?;
            if record.catalog_key().as_slice() != key.value() {
                return Err(CatalogError::OwnershipMismatch { row: T::ROW });
            }
            Ok(record)
        })
        .collect()
}

fn require_row<T>(
    table: &impl ReadableTable<&'static [u8], &'static [u8]>,
    key: &[u8],
    row: CatalogRow,
) -> Result<T, CatalogError>
where
    T: CatalogRecord,
{
    let encoded = table
        .get(key)
        .map_err(CatalogError::redb)?
        .ok_or(CatalogError::NotFound { row })?;
    let record: T = decode_row(encoded.value())?;
    if record.catalog_key() == key {
        Ok(record)
    } else {
        Err(CatalogError::OwnershipMismatch { row: T::ROW })
    }
}

fn require_exact<T>(
    table: &impl ReadableTable<&'static [u8], &'static [u8]>,
    key: &[u8],
    expected: &T,
    row: CatalogRow,
) -> Result<(), CatalogError>
where
    T: CatalogRecord,
{
    let stored: T = require_row(table, key, row)?;
    if stored == *expected {
        Ok(())
    } else {
        Err(CatalogError::OwnershipMismatch { row })
    }
}

fn ensure_absent(
    table: &impl ReadableTable<&'static [u8], &'static [u8]>,
    key: &[u8],
    row: CatalogRow,
) -> Result<(), CatalogError> {
    if table.get(key).map_err(CatalogError::redb)?.is_some() {
        Err(CatalogError::AlreadyExists { row })
    } else {
        Ok(())
    }
}

fn encode<T: serde::Serialize>(value: &T) -> Result<Vec<u8>, CatalogError> {
    codec::encode(value).map_err(CatalogError::Codec)
}

fn decode_row<T>(encoded: &[u8]) -> Result<T, CatalogError>
where
    T: serde::de::DeserializeOwned + serde::Serialize,
{
    codec::decode(encoded).map_err(CatalogError::Codec)
}

fn artifact_key(id: PreparedArtifactId) -> [u8; 32] {
    *id.as_bytes()
}

fn operation_key(id: PrepareOperationId) -> [u8; 32] {
    *id.as_bytes()
}

fn lease_key(lease: &ArtifactLease) -> [u8; 40] {
    let mut key = [0; 40];
    key[..32].copy_from_slice(lease.domain_id.as_bytes());
    key[32..].copy_from_slice(&lease.generation_id.get().to_be_bytes());
    key
}

trait CatalogRecord: Eq + serde::de::DeserializeOwned + serde::Serialize {
    const ROW: CatalogRow;

    fn catalog_key(&self) -> Vec<u8>;
}

impl CatalogRecord for PreparedArtifact {
    const ROW: CatalogRow = CatalogRow::PreparedArtifact;

    fn catalog_key(&self) -> Vec<u8> {
        artifact_key(self.prepared_artifact_id).to_vec()
    }
}

impl CatalogRecord for ArtifactLease {
    const ROW: CatalogRow = CatalogRow::ArtifactLease;

    fn catalog_key(&self) -> Vec<u8> {
        lease_key(self).to_vec()
    }
}

impl CatalogRecord for PrepareOperation {
    const ROW: CatalogRow = CatalogRow::PrepareOperation;

    fn catalog_key(&self) -> Vec<u8> {
        operation_key(self.operation_id).to_vec()
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum CatalogRow {
    PreparedArtifact,
    ArtifactLease,
    PrepareOperation,
}

#[derive(Debug)]
pub enum CatalogError {
    Io(io::Error),
    Database(String),
    Codec(CodecError),
    LockTimeout,
    AlreadyInitialized,
    MissingMetadata,
    UnknownSchema,
    NotFound { row: CatalogRow },
    AlreadyExists { row: CatalogRow },
    OwnershipMismatch { row: CatalogRow },
    InvalidArtifactState,
    InvalidPrepareState,
    ArtifactUnavailable,
    LeasesPresent,
}

impl CatalogError {
    fn redb(error: impl fmt::Display) -> Self {
        Self::Database(error.to_string())
    }
}

impl fmt::Display for CatalogError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Io(error) => write!(formatter, "catalog I/O failed: {error}"),
            Self::Database(error) => write!(formatter, "catalog database failed: {error}"),
            Self::Codec(error) => write!(formatter, "catalog row encoding failed: {error}"),
            Self::LockTimeout => formatter.write_str("storage catalog lock wait expired"),
            Self::AlreadyInitialized => formatter.write_str("storage catalog already exists"),
            Self::MissingMetadata => formatter.write_str("storage catalog metadata is absent"),
            Self::UnknownSchema => formatter.write_str("unknown storage catalog schema or version"),
            Self::NotFound { row } => write!(formatter, "{row:?} row is absent"),
            Self::AlreadyExists { row } => write!(formatter, "{row:?} row already exists"),
            Self::OwnershipMismatch { row } => {
                write!(formatter, "{row:?} row does not match expected ownership")
            }
            Self::InvalidArtifactState => {
                formatter.write_str("artifact state is invalid for operation")
            }
            Self::InvalidPrepareState => {
                formatter.write_str("prepare phase or physical evidence is invalid")
            }
            Self::ArtifactUnavailable => {
                formatter.write_str("artifact is not available for leasing")
            }
            Self::LeasesPresent => formatter.write_str("artifact still has generation leases"),
        }
    }
}

impl std::error::Error for CatalogError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Io(error) => Some(error),
            Self::Codec(error) => Some(error),
            _ => None,
        }
    }
}