haematite-cli 0.21.7

//! Dormant contract admission, retaining the caller's admission ownership.

use super::{
    Captured, Failure, SERVICE, Storage, branch_name, discards, reads, receipt_key, version,
};
use crate::serve::contracts::{Contract, Receipt, ResolvedMutation};
use crate::serve::protocol::{Change, Selection};

#[path = "admission_reads.rs"]
mod dependencies;

/// Contract admission is valid only while the caller retains admission.
/// Preparing and publishing the logical change remain separate steps.
#[derive(Debug)]
pub enum Admission {
    Replayed(Receipt),
    Admitted(Captured),
}

impl Storage {
    /// Decide a resolved request without publishing or retaining another version.
    /// Receipt reconciliation precedes every contract and origin check.
    ///
    /// # Errors
    /// Refuses reused identities, changed dependencies, unconfirmed heads and
    /// mutation kinds whose dependencies cannot yet be proved. Propagates reads.
    pub fn decide_resolved(&mut self, request: &ResolvedMutation) -> Result<Admission, Failure> {
        if let Some(receipt) = self.resolved_replay(request)? {
            return Ok(Admission::Replayed(receipt));
        }
        let name = branch_name(request.branch());
        let record = self
            .refs
            .get(&name)
            .ok_or_else(|| Failure::new("branch_missing", request.branch()))?;
        let installed = version(record);
        if let Contract::ExactVersion { expected } = request.contract()
            && *expected != installed
        {
            return Err(Failure::new(
                "version_conflict",
                "expected branch generation or sequence differs",
            ));
        }
        if self.pending_discards.contains(&name) {
            return Err(Failure::new(
                "discard_pending",
                "branch removal requires receipt reconciliation",
            ));
        }
        let head = self.capture(&Selection {
            branch: request.branch().to_owned(),
            version: None,
        })?;
        if head.version != installed {
            return Err(Failure::new(
                "durability_unknown",
                "installed head requires receipt reconciliation before admission",
            ));
        }
        if head.shards.len() != self.reader.shard_count()
            || head
                .shards
                .iter()
                .enumerate()
                .any(|(id, shard)| id != shard.shard_id)
        {
            return Err(Failure::new(
                "incomplete_roots",
                "head does not name every physical shard in order",
            ));
        }
        if let Contract::Dependencies { origin, reads } = request.contract() {
            dependencies::require_supported(request.change())?;
            if origin.generation != head.version.generation {
                return Err(Failure::new(
                    "generation_mismatch",
                    "dependency origin names another branch generation",
                ));
            }
            if origin.sequence > head.version.sequence {
                return Err(Failure::new(
                    "version_missing",
                    "dependency origin is ahead of the branch",
                ));
            }
            let at = self.capture(&Selection {
                branch: request.branch().to_owned(),
                version: Some(*origin),
            })?;
            dependencies::validate(&self.reader, &at, &head, reads, request.change())?;
        }
        Ok(Admission::Admitted(head))
    }

    fn resolved_replay(&mut self, request: &ResolvedMutation) -> Result<Option<Receipt>, Failure> {
        let service = self
            .refs
            .get(SERVICE)
            .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
        let at = Captured {
            branch: SERVICE.to_owned(),
            version: version(service),
            shards: service.shards.clone(),
        };
        if let Some(bytes) = reads::get_bytes(
            &self.reader,
            &at,
            &discards::key(request.destination(), request.request_id()),
        )? {
            let receipt = Receipt::decode(&bytes)?;
            receipt.replay(request)?;
            self.confirm_replay(SERVICE, at.version)?;
            return Ok(Some(receipt));
        }
        let name = branch_name(request.destination());
        if self.pending_discards.contains(&name) {
            return Err(Failure::new(
                "discard_pending",
                "branch removal requires receipt reconciliation",
            ));
        }
        let Some(record) = self.refs.get(&name) else {
            return Ok(None);
        };
        let at = Captured {
            branch: request.destination().to_owned(),
            version: version(record),
            shards: record.shards.clone(),
        };
        let key = receipt_key(
            request.destination(),
            at.version.generation,
            request.request_id(),
        );
        let Some(bytes) = reads::get_bytes(&self.reader, &at, &key)? else {
            if matches!(request.change(), Change::Fork(_)) {
                return Err(Failure::new(
                    "branch_exists",
                    "destination exists without this fork receipt",
                ));
            }
            return Ok(None);
        };
        let receipt = Receipt::decode(&bytes)?;
        receipt.replay(request)?;
        if receipt.version().generation != at.version.generation
            || receipt.version().sequence > at.version.sequence
        {
            return Err(Failure::new(
                "receipt_corrupt",
                "receipt publication version differs from its stored branch",
            ));
        }
        self.confirm_replay(&name, at.version)?;
        Ok(Some(receipt))
    }

    fn confirm_replay(
        &mut self,
        name: &str,
        at: crate::serve::protocol::Version,
    ) -> Result<(), Failure> {
        self.refs
            .confirm_version(name, at.generation, at.sequence)
            .map_err(|error| Failure::new("durability_unknown", error.to_string()))?;
        self.publish_record(name)
    }
}

#[cfg(test)]
#[path = "admission_tests.rs"]
mod tests;