Add the transactional prepared-artifact catalog

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

@@ -0,0 +1,617 @@
+#![doc = "Bounded transactional access to Lazaret's shared prepared-artifact catalog."]
+
+use std::fmt;
+use std::fs::{File, OpenOptions};
+use std::io;
+use std::path::{Path, PathBuf};
+use std::thread;
+use std::time::{Duration, Instant};
+
+use lazaret_core::codec::{self, CodecError};
+use lazaret_core::identity::{PrepareOperationId, PreparedArtifactId};
+use lazaret_core::records::{
+    ArtifactLease, ArtifactState, PrepareOperation, PreparedArtifact, StorageCatalog,
+    StorageCatalogMetadata,
+};
+use lazaret_linux::filesystem::{ExclusiveFileLock, try_exclusive_flock};
+use redb::{
+    Database, ReadTransaction, ReadableDatabase, ReadableTable, TableDefinition, WriteTransaction,
+};
+
+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, Debug)]
+pub struct Catalog {
+    database_path: PathBuf,
+    lock_path: PathBuf,
+    lock_wait: Duration,
+}
+
+impl Catalog {
+    #[must_use]
+    pub fn new(database_path: impl Into<PathBuf>, lock_path: impl Into<PathBuf>) -> Self {
+        Self {
+            database_path: database_path.into(),
+            lock_path: lock_path.into(),
+            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_path.exists() {
+                return Err(CatalogError::AlreadyInitialized);
+            }
+            let database = Database::create(&self.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> {
+        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(())
+        })
+    }
+
+    /// 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 {
+            return Err(CatalogError::InvalidArtifactState);
+        }
+        if artifact.prepared_artifact_id != expected_operation.expected_artifact_id
+            || artifact.source_identity != expected_operation.source_identity
+        {
+            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(())
+        })
+    }
+
+    /// 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.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.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.write(|transaction| {
+            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 = Database::open(&self.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 = Database::open(&self.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 = acquire_lock(&self.lock_path, self.lock_wait)?;
+        operation()
+    }
+}
+
+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 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,
+    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::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,
+        }
+    }
+}