pipecircuit 0.0.18

//! One retained command across uncertain delivery. A receipt is never replaced
//! by reconnecting, querying its outcome, or explicitly retrying the command.
//! Commands require value semantics: clones preserve meaning, and shared
//! interior mutation must not change the retained operation. The application
//! protocol bounds the bytes of this single owned command.
use super::*;
use crate::{StateMachine, StepInputOf};
#[cfg(kani)]
mod proofs;

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum MutationStatus {
    Ready,
    AwaitingReply,
    Uncertain,
    Confirmed,
}

pub struct PendingMutation<Command> {
    receipt: Receipt,
    command: Command,
    guarantee: Guarantee,
    request: Option<RequestId>,
    status: MutationStatus,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum MutationReceiveError {
    Unexpected,
    WrongReceipt,
    InsufficientGuarantee,
}

/// The caller keeps the reply and its exact domain outcome, including a domain
/// refusal. Only a matching confirmation can settle this machine.
pub struct ReceiveMutation<'a, Value, Outcome, Refusal> {
    pub id: RequestId,
    pub reply: &'a Result<TableReply<Value, Outcome, Refusal>, StorageFailure>,
}

impl<Command> StateMachine for PendingMutation<Command> {}
impl<Command> PendingMutation<Command> {
    /// Refusal returns the complete non-mutation request without changing it.
    pub fn new<Query>(
        request: TableRequest<Query, Command>,
    ) -> Result<Self, TableRequest<Query, Command>> {
        match request {
            TableRequest::Mutate {
                receipt,
                command,
                guarantee,
            } => Ok(Self {
                receipt,
                command,
                guarantee,
                request: None,
                status: MutationStatus::Ready,
            }),
            other @ (TableRequest::Read { .. }
            | TableRequest::Inquire { .. }
            | TableRequest::Retire { .. }) => Err(other),
        }
    }
    pub const fn receipt(&self) -> Receipt {
        self.receipt
    }
    pub const fn status(&self) -> MutationStatus {
        self.status
    }
    pub const fn command(&self) -> &Command {
        &self.command
    }
    pub const fn guarantee(&self) -> Guarantee {
        self.guarantee
    }

    /// Retain the original while the carrier takes an owned copy. Cloning the
    /// domain command is explicit here; the tracker adds no allocation itself.
    /// After an interruption, retry is a caller choice, never an automatic
    /// interpretation of Unknown as "did not execute".
    pub fn retry<Query, View>(
        &mut self,
        id: RequestId,
    ) -> Result<ContinuousRequest<Query, Command, View>, MutationStatus>
    where
        Command: Clone,
    {
        match self.status {
            MutationStatus::Ready | MutationStatus::Uncertain => {}
            MutationStatus::AwaitingReply | MutationStatus::Confirmed => return Err(self.status),
        }
        let command = self.command.clone();
        self.request = Some(id);
        self.status = MutationStatus::AwaitingReply;
        Ok(ContinuousRequest::Command {
            id,
            request: TableRequest::Mutate {
                receipt: self.receipt,
                command,
                guarantee: self.guarantee,
            },
        })
    }
    /// Query the original receipt after an uncertain delivery. This does not
    /// mint a new operation or execute the command again.
    pub fn inquire<Query, View>(
        &mut self,
        id: RequestId,
    ) -> Result<ContinuousRequest<Query, Command, View>, MutationStatus> {
        if self.status != MutationStatus::Uncertain {
            return Err(self.status);
        }
        self.request = Some(id);
        self.status = MutationStatus::AwaitingReply;
        Ok(ContinuousRequest::Command {
            id,
            request: TableRequest::Inquire {
                receipt: self.receipt,
                guarantee: self.guarantee,
            },
        })
    }
    /// Call for lost delivery, including a failed send. Transport failure is
    /// not evidence that the owner did or did not execute the command.
    pub fn disconnect(&mut self) -> MutationStatus {
        if self.status == MutationStatus::AwaitingReply {
            self.status = MutationStatus::Uncertain;
            self.request = None;
        }
        self.status
    }
    /// Explicitly relinquish tracking, preserving the exact owned operation.
    pub fn into_request<Query>(self) -> TableRequest<Query, Command> {
        TableRequest::Mutate {
            receipt: self.receipt,
            command: self.command,
            guarantee: self.guarantee,
        }
    }
}

impl<Command, Value, Outcome, Refusal> StepInputOf<PendingMutation<Command>>
    for ReceiveMutation<'_, Value, Outcome, Refusal>
{
    type Outcome = Result<MutationStatus, MutationReceiveError>;
    fn step(self, pending: &mut PendingMutation<Command>) -> Self::Outcome {
        if pending.status != MutationStatus::AwaitingReply || pending.request != Some(self.id) {
            return Err(MutationReceiveError::Unexpected);
        }
        let status = match self.reply {
            Ok(TableReply::Mutation(answer)) => match answer {
                MutationAnswer::Pending { receipt, .. } => {
                    if *receipt != pending.receipt {
                        return Err(MutationReceiveError::WrongReceipt);
                    }
                    MutationStatus::AwaitingReply
                }
                MutationAnswer::Confirmed {
                    receipt,
                    established,
                    ..
                } => {
                    if *receipt != pending.receipt {
                        return Err(MutationReceiveError::WrongReceipt);
                    }
                    let sufficient = match pending.guarantee {
                        Guarantee::Accepted => true,
                        Guarantee::LocallyDurable => matches!(
                            established,
                            Guarantee::LocallyDurable | Guarantee::ReplicatedDurable
                        ),
                        Guarantee::ReplicatedDurable => {
                            *established == Guarantee::ReplicatedDurable
                        }
                    };
                    if !sufficient {
                        return Err(MutationReceiveError::InsufficientGuarantee);
                    }
                    MutationStatus::Confirmed
                }
                MutationAnswer::Unknown | MutationAnswer::Unavailable(_) => {
                    MutationStatus::Uncertain
                }
            },
            Err(_) => MutationStatus::Uncertain,
            Ok(TableReply::Read(_) | TableReply::Retired(_)) => {
                return Err(MutationReceiveError::Unexpected);
            }
        };
        pending.status = status;
        if status != MutationStatus::AwaitingReply {
            pending.request = None;
        }
        Ok(status)
    }
}