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

748 lines
27 KiB
Rust
Raw Normal View History

#![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, PrepareStageState, 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> {
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.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,
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,
}
}
}