#[cfg(feature = "storage")]
mod ledger;
use core::{
cmp::Ordering,
fmt,
hash::{Hash, Hasher},
marker::PhantomData,
num::NonZeroU64,
};
use crate::{
Participant, StateMachine, StepInputOf, TimerArmDecision, TimerArmOutcome,
TimerCancellationDecision, TimerCancellationOutcome, TimerIn, TimerToken, WakeAfter,
};
#[cfg(feature = "storage")]
pub use ledger::{
ContinuationAdmissionOutcome, ContinuationExpiration, ContinuationExpirationOutcome,
ContinuationLedger, ContinuationLedgerCreationError, ContinuationLedgerCreationFailure,
ContinuationRefusalOutcome, ContinuationResolutionOutcome,
};
pub type TimeoutAfter = WakeAfter;
pub struct Continued<Worker>(PhantomData<fn() -> Worker>);
impl<Worker> Participant for Continued<Worker>
where
Worker: Participant,
{
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 Participant>::Out<'message>,
ContinuationFailure<<Worker as Participant>::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: Participant,
{
id: ContinuationId<Worker>,
owed: Owed<Worker::In<'message>>,
}
impl<'message, Worker> ContinuationAdmission<'message, Worker>
where
Worker: Participant,
{
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 { .. } => {
let (input, _) = self.owed.into_parts();
Ok(ContinuationArmOutcome::Armed {
work: ContinuationWork { id: self.id, input },
})
}
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: Participant,
{
Armed {
work: ContinuationWork<'message, Worker>,
},
Refused {
refusal: ContinuationArmRefusal<'message, Worker>,
},
}
pub struct ContinuationArmMismatch<'message, Worker>
where
Worker: Participant,
{
admission: ContinuationAdmission<'message, Worker>,
decision: TimerArmDecision<TimerToken>,
}
impl<'message, Worker> ContinuationArmMismatch<'message, Worker>
where
Worker: Participant,
{
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: Participant,
{
id: ContinuationId<Worker>,
reason: crate::TimerArmRefusal,
owed: Owed<Worker::In<'message>>,
}
impl<'message, Worker> ContinuationArmRefusal<'message, Worker>
where
Worker: Participant,
{
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: Participant,
{
id: ContinuationId<Worker>,
input: Worker::In<'message>,
}
impl<'message, Worker> ContinuationWork<'message, Worker>
where
Worker: Participant,
{
pub const fn id(&self) -> ContinuationId<Worker> {
self.id
}
pub const fn input(&self) -> &Worker::In<'message> {
&self.input
}
pub fn fulfill(
self,
output: Worker::Out<'message>,
) -> ContinuationCompletion<'message, Worker> {
ContinuationCompletion {
id: self.id,
output,
}
}
pub fn fulfill_with(
self,
fulfill: impl FnOnce(Worker::In<'message>) -> Worker::Out<'message>,
) -> ContinuationCompletion<'message, Worker> {
ContinuationCompletion {
id: self.id,
output: fulfill(self.input),
}
}
}
impl<'message, Worker> StepInputOf<Worker> for ContinuationWork<'message, Worker>
where
Worker: StateMachine + Participant,
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: Participant,
{
id: ContinuationId<Worker>,
output: Worker::Out<'message>,
}
impl<'message, Worker> ContinuationCompletion<'message, Worker>
where
Worker: Participant,
{
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: Participant,
{
Cancelled {
completion: CancelledContinuationCompletion<'message, Worker>,
},
Stale {
completion: ContinuationCompletion<'message, Worker>,
},
}
pub struct ContinuationCancellationMismatch<'message, Worker>
where
Worker: Participant,
{
completion: ContinuationCompletion<'message, Worker>,
decision: TimerCancellationDecision<TimerToken>,
}
impl<'message, Worker> ContinuationCancellationMismatch<'message, Worker>
where
Worker: Participant,
{
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: Participant;
#[cfg(all(kani, feature = "storage"))]
mod proofs;
#[cfg(test)]
mod tests;