pipecircuit 0.0.23

//! Owned questions and mutation intents. Evaluation belongs to the owner;
//! neither a query nor a command contains a borrow into a previous answer.
use super::{
    Authority, Change, Error, Guarantee, OperationId, PersistentRow, ReadRequest, Revision,
};
use crate::StepInputOf;
use crate::storage::warp_table::WarpTable;
use alloc::vec::Vec;

/// A program's closed table vocabulary. Implementations must be deterministic
/// functions of their arguments, with no shared interior mutation or I/O.
/// The schema identifies both the command representation and its meaning.
pub trait TableProtocol<Row: PersistentRow> {
    const SCHEMA: [u8; 16];
    const VERSION: u32;
    type Query;
    type Value;
    type Mutation;
    type Outcome;
    type Refusal;

    fn query(query: &Self::Query, table: &WarpTable<Row>) -> Result<Self::Value, Self::Refusal>;

    /// Prepare one atomic operation against the current authoritative state.
    /// No change is published unless the entire preparation and admission pass.
    /// A domain refusal discards all prepared changes and retains the refusal.
    fn prepare(
        command: &Self::Mutation,
        table: &WarpTable<Row>,
        changes: &mut ChangeSet<Row>,
    ) -> Result<Self::Outcome, Self::Refusal>;
}

/// One synchronous preparation turn. The physical adapter encodes the retained
/// result before submission; the internal expected revision prevents publication
/// if another mutation has executed between preparation and submission.
pub struct PrepareCommand<'a, Row: PersistentRow, P: TableProtocol<Row>> {
    pub operation: OperationId,
    pub guarantee: Guarantee,
    pub command: &'a P::Mutation,
}
pub struct PreparedCommand<Row: PersistentRow, Outcome, Refusal> {
    pub expected: Revision,
    pub changes: Vec<Change<Row, Row::Id>>,
    pub outcome: Result<Outcome, Refusal>,
}
impl<Row: PersistentRow, P: TableProtocol<Row>> StepInputOf<Authority<Row>>
    for PrepareCommand<'_, Row, P>
{
    type Outcome = Result<PreparedCommand<Row, P::Outcome, P::Refusal>, Error>;
    fn step(self, machine: &mut Authority<Row>) -> Self::Outcome {
        machine.admit_command(self.operation, self.guarantee)?;
        let mut changes = ChangeSet::new(machine.limits().changes)?;
        let table = machine
            .read(ReadRequest {
                guarantee: Guarantee::Accepted,
            })
            .into_table()?;
        let outcome = P::prepare(self.command, table, &mut changes);
        let mut changes = changes.finish()?;
        if outcome.is_err() {
            changes.clear();
        }
        Ok(PreparedCommand {
            expected: machine.revision(),
            changes,
            outcome,
        })
    }
}

/// Bounded staging owned by one command preparation. Successful submission
/// transfers its allocation into the retained operation record.
pub struct ChangeSet<Row: PersistentRow> {
    changes: Vec<Change<Row, Row::Id>>,
    limit: usize,
    refused: bool,
}
impl<Row: PersistentRow> ChangeSet<Row> {
    pub fn new(limit: usize) -> Result<Self, Error> {
        if limit == 0 {
            return Err(Error::InvalidBounds);
        }
        let mut changes = Vec::new();
        changes
            .try_reserve_exact(limit)
            .map_err(|_| Error::Allocation)?;
        Ok(Self {
            changes,
            limit,
            refused: false,
        })
    }

    /// Even if an author ignores this refusal, `finish` refuses the whole set.
    pub fn push(&mut self, change: Change<Row, Row::Id>) -> Result<(), Change<Row, Row::Id>> {
        if self.changes.len() == self.limit {
            self.refused = true;
            return Err(change);
        }
        self.changes.push(change);
        Ok(())
    }

    pub fn finish(self) -> Result<Vec<Change<Row, Row::Id>>, Error> {
        if self.refused {
            Err(Error::Full)
        } else {
            Ok(self.changes)
        }
    }
}