pipecircuit 0.0.19

#[cfg(feature = "storage")]
mod ledger;

use core::{
    cmp::Ordering,
    fmt,
    hash::{Hash, Hasher},
    marker::PhantomData,
    num::NonZeroU64,
};

use crate::{
    Pipe, StateMachine, StepInputOf, TimerArmDecision, TimerArmOutcome, TimerCancellationDecision,
    TimerCancellationOutcome, TimerIn, TimerToken, WakeAfter,
};

#[cfg(feature = "storage")]
pub use ledger::{
    ContinuationAdmissionOutcome, ContinuationExpiration, ContinuationExpirationOutcome,
    ContinuationLedger, ContinuationLedgerCreationError, ContinuationLedgerCreationFailure,
    ContinuationRefusalOutcome, ContinuationResolutionOutcome, ContinuationRetractionOutcome,
};

pub type TimeoutAfter = WakeAfter;

pub struct Continued<Worker>(PhantomData<fn() -> Worker>);

impl<Worker> Pipe for Continued<Worker>
where
    Worker: Pipe,
{
    type In<'message> = Owed<Worker::In<'message>>;
    type Out<'message> = ContinuationResult<'message, Worker>;
}

#[must_use = "owed work must be sent through a continued pipe"]
pub struct Owed<Work> {
    work: Work,
    timeout_after: TimeoutAfter,
}

impl<Work> Owed<Work> {
    pub const fn new(work: Work, timeout_after: TimeoutAfter) -> Self {
        Self {
            work,
            timeout_after,
        }
    }

    pub const fn work(&self) -> &Work {
        &self.work
    }

    pub const fn timeout_after(&self) -> TimeoutAfter {
        self.timeout_after
    }

    fn into_parts(self) -> (Work, TimeoutAfter) {
        (self.work, self.timeout_after)
    }
}

impl Owed<()> {
    pub const fn wait(timeout_after: TimeoutAfter) -> Self {
        Self::new((), timeout_after)
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ContinuationUnavailable {
    AtCapacity { maximum_pending: u32 },
    IdentifiersExhausted,
    TimerRefused { reason: crate::TimerArmRefusal },
}

#[must_use = "continuation failures must be handled"]
pub enum ContinuationFailure<Work> {
    Expired,
    Unavailable {
        reason: ContinuationUnavailable,
        rejected: Owed<Work>,
    },
}

pub type ContinuationResult<'message, Worker> =
    Result<<Worker as Pipe>::Out<'message>, ContinuationFailure<<Worker as Pipe>::In<'message>>>;

pub struct ContinuationId<Worker> {
    value: NonZeroU64,
    worker: PhantomData<fn() -> Worker>,
}

impl<Worker> ContinuationId<Worker> {
    #[cfg(feature = "storage")]
    const fn new(value: NonZeroU64) -> Self {
        Self {
            value,
            worker: PhantomData,
        }
    }

    pub const fn value(self) -> NonZeroU64 {
        self.value
    }

    const fn timer_token(self) -> TimerToken {
        TimerToken::from_nonzero(self.value)
    }
}

impl<Worker> Clone for ContinuationId<Worker> {
    fn clone(&self) -> Self {
        *self
    }
}

impl<Worker> Copy for ContinuationId<Worker> {}

impl<Worker> fmt::Debug for ContinuationId<Worker> {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        self.value.fmt(formatter)
    }
}

impl<Worker> PartialEq for ContinuationId<Worker> {
    fn eq(&self, other: &Self) -> bool {
        self.value == other.value
    }
}

impl<Worker> Eq for ContinuationId<Worker> {}

impl<Worker> PartialOrd for ContinuationId<Worker> {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

impl<Worker> Ord for ContinuationId<Worker> {
    fn cmp(&self, other: &Self) -> Ordering {
        self.value.cmp(&other.value)
    }
}

impl<Worker> Hash for ContinuationId<Worker> {
    fn hash<H: Hasher>(&self, state: &mut H) {
        self.value.hash(state);
    }
}

#[must_use = "admitted continuations must be armed before their work is released"]
pub struct ContinuationAdmission<'message, Worker>
where
    Worker: Pipe,
{
    id: ContinuationId<Worker>,
    owed: Owed<Worker::In<'message>>,
}

impl<'message, Worker> ContinuationAdmission<'message, Worker>
where
    Worker: Pipe,
{
    pub const fn id(&self) -> ContinuationId<Worker> {
        self.id
    }

    pub const fn timeout_after(&self) -> TimeoutAfter {
        self.owed.timeout_after()
    }

    pub const fn timer_token(&self) -> TimerToken {
        self.id.timer_token()
    }

    pub const fn timer_input(&self) -> TimerIn<TimerToken> {
        TimerIn::Arm {
            token: self.timer_token(),
            after: self.timeout_after(),
        }
    }

    pub fn settle_timer_arm(
        self,
        outcome: TimerArmOutcome<TimerToken>,
    ) -> Result<ContinuationArmOutcome<'message, Worker>, ContinuationArmMismatch<'message, Worker>>
    {
        let decision = outcome.into_decision();
        let returned_token = match decision {
            TimerArmDecision::Armed { token } | TimerArmDecision::Refused { token, .. } => token,
        };
        if returned_token != self.timer_token() {
            return Err(ContinuationArmMismatch {
                admission: self,
                decision,
            });
        }
        match decision {
            TimerArmDecision::Armed { .. } => Ok(ContinuationArmOutcome::Armed {
                work: ContinuationWork {
                    id: self.id,
                    owed: self.owed,
                },
            }),
            TimerArmDecision::Refused { reason, .. } => Ok(ContinuationArmOutcome::Refused {
                refusal: ContinuationArmRefusal {
                    id: self.id,
                    reason,
                    owed: self.owed,
                },
            }),
        }
    }
}

#[must_use = "armed work or exact timer refusal must be handled by the fitting"]
pub enum ContinuationArmOutcome<'message, Worker>
where
    Worker: Pipe,
{
    Armed {
        work: ContinuationWork<'message, Worker>,
    },
    Refused {
        refusal: ContinuationArmRefusal<'message, Worker>,
    },
}

pub struct ContinuationArmMismatch<'message, Worker>
where
    Worker: Pipe,
{
    admission: ContinuationAdmission<'message, Worker>,
    decision: TimerArmDecision<TimerToken>,
}

impl<'message, Worker> ContinuationArmMismatch<'message, Worker>
where
    Worker: Pipe,
{
    pub fn into_parts(
        self,
    ) -> (
        ContinuationAdmission<'message, Worker>,
        TimerArmDecision<TimerToken>,
    ) {
        (self.admission, self.decision)
    }
}

#[must_use = "timer refusal must be returned to the continuation ledger"]
pub struct ContinuationArmRefusal<'message, Worker>
where
    Worker: Pipe,
{
    id: ContinuationId<Worker>,
    reason: crate::TimerArmRefusal,
    owed: Owed<Worker::In<'message>>,
}

impl<'message, Worker> ContinuationArmRefusal<'message, Worker>
where
    Worker: Pipe,
{
    pub const fn id(&self) -> ContinuationId<Worker> {
        self.id
    }

    pub const fn reason(&self) -> crate::TimerArmRefusal {
        self.reason
    }

    pub fn into_owed(self) -> Owed<Worker::In<'message>> {
        self.owed
    }
}

#[must_use = "continuation work must be carried by the selected fitting"]
pub struct ContinuationWork<'message, Worker>
where
    Worker: Pipe,
{
    id: ContinuationId<Worker>,
    owed: Owed<Worker::In<'message>>,
}

impl<'message, Worker> ContinuationWork<'message, Worker>
where
    Worker: Pipe,
{
    pub const fn id(&self) -> ContinuationId<Worker> {
        self.id
    }

    pub const fn input(&self) -> &Worker::In<'message> {
        self.owed.work()
    }

    pub const fn timeout_after(&self) -> TimeoutAfter {
        self.owed.timeout_after()
    }

    pub const fn timer_token(&self) -> TimerToken {
        self.id.timer_token()
    }

    pub const fn timer_input(&self) -> TimerIn<TimerToken> {
        TimerIn::Cancel {
            token: self.timer_token(),
        }
    }

    pub fn settle_timer_cancellation(
        self,
        outcome: TimerCancellationOutcome<TimerToken>,
    ) -> Result<
        ContinuationWorkCancellationOutcome<'message, Worker>,
        ContinuationWorkCancellationMismatch<'message, Worker>,
    > {
        let decision = outcome.into_decision();
        let returned_token = match decision {
            TimerCancellationDecision::Cancelled { token }
            | TimerCancellationDecision::Refused { token, .. } => token,
        };
        if returned_token != self.timer_token() {
            return Err(ContinuationWorkCancellationMismatch {
                work: self,
                decision,
            });
        }
        match decision {
            TimerCancellationDecision::Cancelled { .. } => {
                Ok(ContinuationWorkCancellationOutcome::Cancelled {
                    work: CancelledContinuationWork(self),
                })
            }
            TimerCancellationDecision::Refused { .. } => {
                Ok(ContinuationWorkCancellationOutcome::Stale { work: self })
            }
        }
    }

    /// Consume the input now and retain its move-only reply obligation across
    /// steps. Beginning work relinquishes the ability to retract it as unstarted.
    pub fn begin(self) -> (Worker::In<'message>, ContinuationReply<'message, Worker>) {
        let (input, _started_timeout) = self.owed.into_parts();
        (
            input,
            ContinuationReply {
                id: self.id,
                output: PhantomData,
            },
        )
    }

    pub fn fulfill(
        self,
        output: Worker::Out<'message>,
    ) -> ContinuationCompletion<'message, Worker> {
        let (_input, reply) = self.begin();
        reply.fulfill(output)
    }

    pub fn fulfill_with(
        self,
        fulfill: impl FnOnce(Worker::In<'message>) -> Worker::Out<'message>,
    ) -> ContinuationCompletion<'message, Worker> {
        let (input, reply) = self.begin();
        reply.fulfill(fulfill(input))
    }
}

/// The one reply still owed after a worker consumes its input. Expiry at the
/// requester does not cancel the worker's effects; its eventual completion is
/// classified by the original ledger and deadline.
#[must_use = "started work retains its reply obligation until completion"]
pub struct ContinuationReply<'message, Worker: Pipe> {
    id: ContinuationId<Worker>,
    output: PhantomData<fn() -> Worker::Out<'message>>,
}

impl<'message, Worker: Pipe> ContinuationReply<'message, Worker> {
    pub fn fulfill(
        self,
        output: Worker::Out<'message>,
    ) -> ContinuationCompletion<'message, Worker> {
        ContinuationCompletion {
            id: self.id,
            output,
        }
    }
}

#[must_use = "cancelled or stale unstarted work must be handled by the fitting"]
pub enum ContinuationWorkCancellationOutcome<'message, Worker>
where
    Worker: Pipe,
{
    Cancelled {
        work: CancelledContinuationWork<'message, Worker>,
    },
    Stale {
        work: ContinuationWork<'message, Worker>,
    },
}

pub struct ContinuationWorkCancellationMismatch<'message, Worker>
where
    Worker: Pipe,
{
    work: ContinuationWork<'message, Worker>,
    decision: TimerCancellationDecision<TimerToken>,
}

impl<'message, Worker> ContinuationWorkCancellationMismatch<'message, Worker>
where
    Worker: Pipe,
{
    pub fn into_parts(
        self,
    ) -> (
        ContinuationWork<'message, Worker>,
        TimerCancellationDecision<TimerToken>,
    ) {
        (self.work, self.decision)
    }
}

#[must_use = "cancelled unstarted work must be retracted through its ledger"]
pub struct CancelledContinuationWork<'message, Worker>(ContinuationWork<'message, Worker>)
where
    Worker: Pipe;

impl<'message, Worker> CancelledContinuationWork<'message, Worker>
where
    Worker: Pipe,
{
    pub const fn id(&self) -> ContinuationId<Worker> {
        self.0.id()
    }

    pub const fn owed(&self) -> &Owed<Worker::In<'message>> {
        &self.0.owed
    }
}

impl<'message, Worker> StepInputOf<Worker> for ContinuationWork<'message, Worker>
where
    Worker: StateMachine + Pipe,
    Worker::In<'message>: StepInputOf<Worker, Outcome = Worker::Out<'message>>,
{
    type Outcome = ContinuationCompletion<'message, Worker>;

    fn step(self, worker: &mut Worker) -> Self::Outcome {
        self.fulfill_with(|input| worker.step(input))
    }
}

#[must_use = "continuation completions must be settled by the fitting"]
pub struct ContinuationCompletion<'message, Worker>
where
    Worker: Pipe,
{
    id: ContinuationId<Worker>,
    output: Worker::Out<'message>,
}

impl<'message, Worker> ContinuationCompletion<'message, Worker>
where
    Worker: Pipe,
{
    pub const fn id(&self) -> ContinuationId<Worker> {
        self.id
    }

    pub const fn output(&self) -> &Worker::Out<'message> {
        &self.output
    }

    pub const fn timer_token(&self) -> TimerToken {
        self.id.timer_token()
    }

    pub const fn timer_input(&self) -> TimerIn<TimerToken> {
        TimerIn::Cancel {
            token: self.timer_token(),
        }
    }

    pub fn settle_timer_cancellation(
        self,
        outcome: TimerCancellationOutcome<TimerToken>,
    ) -> Result<
        ContinuationCancellationOutcome<'message, Worker>,
        ContinuationCancellationMismatch<'message, Worker>,
    > {
        let decision = outcome.into_decision();
        let returned_token = match decision {
            TimerCancellationDecision::Cancelled { token }
            | TimerCancellationDecision::Refused { token, .. } => token,
        };
        if returned_token != self.timer_token() {
            return Err(ContinuationCancellationMismatch {
                completion: self,
                decision,
            });
        }
        match decision {
            TimerCancellationDecision::Cancelled { .. } => {
                Ok(ContinuationCancellationOutcome::Cancelled {
                    completion: CancelledContinuationCompletion(self),
                })
            }
            TimerCancellationDecision::Refused { .. } => {
                Ok(ContinuationCancellationOutcome::Stale { completion: self })
            }
        }
    }
}

#[must_use = "cancelled or stale completion must be handled by the fitting"]
pub enum ContinuationCancellationOutcome<'message, Worker>
where
    Worker: Pipe,
{
    Cancelled {
        completion: CancelledContinuationCompletion<'message, Worker>,
    },
    Stale {
        completion: ContinuationCompletion<'message, Worker>,
    },
}

pub struct ContinuationCancellationMismatch<'message, Worker>
where
    Worker: Pipe,
{
    completion: ContinuationCompletion<'message, Worker>,
    decision: TimerCancellationDecision<TimerToken>,
}

impl<'message, Worker> ContinuationCancellationMismatch<'message, Worker>
where
    Worker: Pipe,
{
    pub fn into_parts(
        self,
    ) -> (
        ContinuationCompletion<'message, Worker>,
        TimerCancellationDecision<TimerToken>,
    ) {
        (self.completion, self.decision)
    }
}

#[must_use = "cancelled completions must be resolved through the continuation ledger"]
pub struct CancelledContinuationCompletion<'message, Worker>(
    ContinuationCompletion<'message, Worker>,
)
where
    Worker: Pipe;

#[cfg(all(kani, feature = "storage"))]
mod proofs;
#[cfg(test)]
mod tests;

#[cfg(all(test, feature = "storage"))]
mod retraction_tests;

#[cfg(all(test, feature = "storage"))]
mod deferred_tests;