liminal_server/services/site/

storage.rs

//! One snapshot operation keeps the target, approval, receipt and outcome.

use haematite::tree::Hash;
use liminal::durability::{DurableStore, RoutedBatchItem, bridge::block_on};
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use std::io::Read;
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use super::super::{Lifecycle, LifecycleError, Registry, ServiceRecord, store};
use super::{Construction, SiteError, SiteLimits};

/// Operator policy and runtime effects, never serialized or included in Debug.
pub struct SiteBindings {
    pub(in super::super) store: Arc<dyn DurableStore>,
    pub(in super::super) limits: SiteLimits,
    pub(in super::super) templates: Option<super::super::templates::TemplatePolicy>,
    pub(in super::super) effects: Arc<dyn super::effects::SiteEffects>,
    pub(in super::super) retained: Arc<crate::agents::retained::Retained>,
}

#[derive(Clone, Default, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub(in super::super) struct Snapshot {
    pub records: BTreeMap<String, ServiceRecord>,
    #[serde(default)]
    pub construction: Construction,
}

fn path(dir: &Path) -> PathBuf {
    dir.join("construction").join("registry.json")
}

fn read_bounded(path: &Path, bound: usize) -> Result<Vec<u8>, LifecycleError> {
    if !std::fs::symlink_metadata(path)
        .map_err(|error| store(path, &error))?
        .is_file()
    {
        return Err(LifecycleError::Store(
            "construction_snapshot_not_regular".into(),
        ));
    }
    let canonical = path.canonicalize().map_err(|error| store(path, &error))?;
    let file = super::super::templates::open_command(&canonical)
        .map_err(|_| LifecycleError::Store("construction_snapshot_read_failed".into()))?;
    if !file
        .metadata()
        .map_err(|error| store(path, &error))?
        .is_file()
    {
        return Err(LifecycleError::Store(
            "construction_snapshot_not_regular".into(),
        ));
    }
    let limit =
        u64::try_from(bound).map_err(|_| LifecycleError::Store("snapshot_bytes_invalid".into()))?;
    let mut bytes = Vec::new();
    file.take(limit.saturating_add(1))
        .read_to_end(&mut bytes)
        .map_err(|error| store(path, &error))?;
    if bytes.len() > bound {
        return Err(LifecycleError::Store(
            "snapshot_bytes_bound_exceeded".into(),
        ));
    }
    Ok(bytes)
}

pub(in super::super) fn load_bounded(
    dir: &Path,
    limits: &SiteLimits,
) -> Result<Snapshot, LifecycleError> {
    let archive = path(dir);
    match std::fs::symlink_metadata(&archive) {
        Ok(metadata) => {
            if !metadata.is_file() {
                return Err(LifecycleError::Store(
                    "construction_snapshot_not_regular".into(),
                ));
            }
            return serde_json::from_slice(&read_bounded(&archive, limits.snapshot_bytes.get())?)
                .map_err(|_| LifecycleError::Store("construction_snapshot_invalid".into()));
        }
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
        Err(error) => return Err(store(&archive, &error)),
    }
    let mut snapshot = Snapshot::default();
    let mut total = 0_usize;
    for entry in std::fs::read_dir(dir).map_err(|error| store(dir, &error))? {
        let path = entry.map_err(|error| store(dir, &error))?.path();
        if path.extension().is_none_or(|extension| extension != "json") {
            continue;
        }
        if snapshot.records.len() >= limits.sites.get() {
            return Err(LifecycleError::Store(
                "migration_records_bound_exceeded".into(),
            ));
        }
        let bytes = read_bounded(&path, limits.snapshot_bytes.get().saturating_sub(total))?;
        total = total
            .checked_add(bytes.len())
            .ok_or_else(|| LifecycleError::Store("snapshot_bytes_bound_exceeded".into()))?;
        let record: ServiceRecord = serde_json::from_slice(&bytes)
            .map_err(|_| LifecycleError::Store("legacy_service_record_invalid".into()))?;
        if record.id.is_empty() || snapshot.records.insert(record.id.clone(), record).is_some() {
            return Err(LifecycleError::Store("legacy_service_id_repeated".into()));
        }
    }
    write(dir, &snapshot, Some(limits.snapshot_bytes.get()))?;
    Ok(snapshot)
}

/// Migrates old installs by reading each service once, then committing the
/// complete new snapshot before returning. Old files are kept as evidence,
/// and are never replayed after the snapshot exists.
pub(in super::super) fn load(dir: &Path) -> Result<Snapshot, LifecycleError> {
    let snapshot_path = path(dir);
    match std::fs::File::open(&snapshot_path) {
        Ok(file) => {
            return serde_json::from_reader(file)
                .map_err(|_| LifecycleError::Store("construction_snapshot_invalid".into()));
        }
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
        Err(error) => return Err(store(&snapshot_path, &error)),
    }
    let mut snapshot = Snapshot::default();
    for entry in std::fs::read_dir(dir).map_err(|error| store(dir, &error))? {
        let path = entry.map_err(|error| store(dir, &error))?.path();
        if path.extension().is_none_or(|extension| extension != "json") {
            continue;
        }
        let record: ServiceRecord = serde_json::from_reader(
            std::fs::File::open(&path).map_err(|error| store(&path, &error))?,
        )
        .map_err(|_| LifecycleError::Store("legacy_service_record_invalid".into()))?;
        if record.id.is_empty() || snapshot.records.insert(record.id.clone(), record).is_some() {
            return Err(LifecycleError::Store("legacy_service_id_repeated".into()));
        }
    }
    write(dir, &snapshot, None)?;
    Ok(snapshot)
}

pub(in super::super) fn keep(
    lifecycle: &Lifecycle,
    registry: &mut Registry,
    record: &ServiceRecord,
) -> Result<(), LifecycleError> {
    let mut snapshot = Snapshot {
        records: registry.records.clone(),
        construction: registry.construction.clone(),
    };
    snapshot.records.insert(record.id.clone(), record.clone());
    commit(lifecycle, registry, &snapshot)
}

pub(in super::super) fn commit(
    lifecycle: &Lifecycle,
    registry: &mut Registry,
    snapshot: &Snapshot,
) -> Result<(), LifecycleError> {
    let bound = lifecycle
        .site_config
        .get()
        .map(|bindings| bindings.limits.snapshot_bytes.get());
    let bytes = serde_json::to_vec(snapshot)
        .map_err(|_| LifecycleError::Store("snapshot_encoding_failed".into()))?;
    if bound.is_some_and(|limit| bytes.len() > limit) {
        return Err(LifecycleError::Store(
            "snapshot_bytes_bound_exceeded".into(),
        ));
    }
    if let Some(bindings) = lifecycle.site_config.get() {
        if bindings
            .store
            .max_request_bytes()
            .is_some_and(|limit| bytes.len() > limit)
        {
            return Err(LifecycleError::Store(
                "store_request_bytes_bound_exceeded".into(),
            ));
        }
        let hash = Hash::of(&bytes);
        let result = block_on(bindings.store.apply_routed_batch(
            ROUTE,
            vec![RoutedBatchItem {
                key: KEY.into(),
                expected: registry.construction_hash,
                value: bytes,
                ttl: None,
            }],
        ))
        .map_err(|_| LifecycleError::Store("construction_batch_failed".into()))
        .and_then(|result| {
            result.map_err(|_| LifecycleError::Store("construction_batch_failed".into()))
        })
        .and_then(|()| {
            block_on(bindings.store.flush())
                .map_err(|_| LifecycleError::Store("construction_flush_failed".into()))
                .and_then(|result| {
                    result.map_err(|_| LifecycleError::Store("construction_flush_failed".into()))
                })
        });
        if let Err(error) = result {
            registry.poisoned = true;
            return Err(error);
        }
        registry.construction_hash = Some(hash);
    } else if let Err(error) = write(&lifecycle.dir, snapshot, None) {
        registry.poisoned = true;
        return Err(error);
    }
    registry.records = snapshot.records.clone();
    registry.construction = snapshot.construction.clone();
    lifecycle.changed.notify_all();
    Ok(())
}

fn write(dir: &Path, snapshot: &Snapshot, bound: Option<usize>) -> Result<(), LifecycleError> {
    let bytes = serde_json::to_vec(snapshot)
        .map_err(|_| LifecycleError::Store("snapshot_encoding_failed".into()))?;
    if bound.is_some_and(|limit| bytes.len() > limit) {
        return Err(LifecycleError::Store(
            "snapshot_bytes_bound_exceeded".into(),
        ));
    }
    let destination = path(dir);
    let parent = destination
        .parent()
        .ok_or_else(|| LifecycleError::Store("snapshot_parent_absent".into()))?;
    std::fs::create_dir_all(parent).map_err(|error| store(parent, &error))?;
    let temporary = parent.join("registry.new");
    let mut file = std::fs::File::create(&temporary).map_err(|error| store(&temporary, &error))?;
    file.write_all(&bytes)
        .and_then(|()| file.sync_all())
        .map_err(|error| store(&temporary, &error))?;
    std::fs::rename(&temporary, &destination).map_err(|error| store(&destination, &error))?;
    std::fs::File::open(parent)
        .and_then(|directory| directory.sync_all())
        .map_err(|error| store(parent, &error))
}

impl Lifecycle {
    /// Returns one bounded current snapshot for an approval owner's atomic commit.
    pub(in super::super) fn snapshot(registry: &Registry) -> Snapshot {
        Snapshot {
            records: registry.records.clone(),
            construction: registry.construction.clone(),
        }
    }

    /// Validates every restored construction count before admission.
    ///
    /// # Errors
    /// Oversized or inconsistent restored history.
    pub fn validate_site_storage(&self) -> Result<(), SiteError> {
        let limits = &self.site_bindings()?.limits;
        let registry = self.lock()?;
        let snapshot = Self::snapshot(&registry);
        if serde_json::to_vec(&snapshot)
            .map_err(|_| SiteError::Store)?
            .len()
            > limits.snapshot_bytes.get()
        {
            return Err(SiteError::Bound("snapshot_bytes"));
        }
        if snapshot.construction.drafts.len() > limits.drafts.get()
            || snapshot.construction.outcomes.len() > limits.drafts.get()
        {
            return Err(SiteError::Bound("drafts"));
        }
        if registry
            .records
            .values()
            .filter(|record| record.purpose.is_some())
            .count()
            > limits.sites.get()
        {
            return Err(SiteError::Bound("sites"));
        }
        for (id, record) in &registry.records {
            if id != &record.id || !crate::config::templates::is_segment(id) {
                return Err(SiteError::Invalid("stored service id"));
            }
            if record.purpose.is_some() && record.transitions.len() > limits.transitions.get() {
                return Err(SiteError::Bound("transitions"));
            }
            if record.plan.steps.len() > limits.steps.get()
                || record.plan.transitions.len() > limits.transitions.get()
            {
                return Err(SiteError::Bound("plan history"));
            }
            record.plan.validate(limits)?;
            if record.builder_bound
                && (record.state != super::super::ServiceState::Building
                    || record.builder.is_none()
                    || record.builder_grants.is_none())
            {
                return Err(SiteError::Invalid("stored builder binding"));
            }
            if record.approvals.len() > limits.transitions.get()
                || record.approved_by.len() > limits.approvers.get()
            {
                return Err(SiteError::Bound("stored service approvals"));
            }
        }
        validate_templates(&registry, self.site_bindings()?)?;
        drop(registry);
        Ok(())
    }
}

fn validate_templates(registry: &Registry, bindings: &SiteBindings) -> Result<(), SiteError> {
    if registry
        .construction
        .outcomes
        .keys()
        .any(|id| !registry.construction.drafts.contains_key(id))
    {
        return Err(SiteError::Invalid("stored outcome has no preparation"));
    }
    if registry.construction.templates.is_empty() {
        return Ok(());
    }
    let policy = bindings
        .templates
        .as_ref()
        .ok_or(SiteError::Invalid("[auth.lys.templates] roots"))?;
    if registry.construction.templates.len() > policy.records.get() {
        return Err(SiteError::Bound("template records"));
    }
    for (name, template) in &registry.construction.templates {
        policy.validate(&template.proposal)?;
        let service = registry
            .records
            .get(&template.proposal.service)
            .ok_or(SiteError::Invalid("stored template site"))?;
        if name != &template.proposal.name
            || template.proposal.subjects != service.subjects
            || !template.proposal.command.is_absolute()
            || template.proposal.digest.len() != 64
            || !template
                .proposal
                .digest
                .bytes()
                .all(|byte| byte.is_ascii_hexdigit())
        {
            return Err(SiteError::Invalid("stored template contract or digest"));
        }
        if template.transitions.len() > bindings.limits.transitions.get()
            || template.approved_by.is_empty()
            || template.approved_by.len() > bindings.limits.approvers.get()
        {
            return Err(SiteError::Bound("stored template approvals"));
        }
    }
    Ok(())
}

const ROUTE: &str = "liminal/construction/v1";
const KEY: &[u8] = b"construction";

/// Imports old file records only when the new durable key is absent. Once kept,
/// the conditional key is authoritative and a stale archive is never replayed.
pub(in super::super) fn restore(
    lifecycle: &Lifecycle,
    registry: &mut Registry,
) -> Result<(), SiteError> {
    let bindings = lifecycle.site_bindings()?;
    let bytes = block_on(bindings.store.get_routed(ROUTE, KEY))
        .map_err(|_| SiteError::Store)?
        .map_err(|_| SiteError::Store)?;
    if let Some(bytes) = bytes {
        if bytes.len() > bindings.limits.snapshot_bytes.get() {
            return Err(SiteError::Bound("snapshot_bytes"));
        }
        let snapshot: Snapshot = serde_json::from_slice(&bytes).map_err(|_| SiteError::Store)?;
        registry.construction_hash = Some(Hash::of(&bytes));
        registry.records = snapshot.records;
        registry.construction = snapshot.construction;
    } else {
        let snapshot = Lifecycle::snapshot(registry);
        commit(lifecycle, registry, &snapshot)?;
    }
    Ok(())
}