pipecircuit/delay/

ledger.rs

use crate::storage::warp_table::{
    Capacity, WarpId, WarpTable, WarpTableCreationError, WarpTableInsertOutcome,
    WarpTableRemoveOutcome, WarpTableRow,
};
use crate::{TimerArmDecision, TimerArmOutcome, TimerArmRefusal, TimerIn, TimerToken, WakeAfter};

use super::Delayed;

#[derive(WarpTableRow)]
struct PendingDelayedMessage<Message> {
    message: Message,
}

type PendingDelayedMessageCreationError<Message> = WarpTableCreationError<
    <PendingDelayedMessage<Message> as WarpTableRow>::ColumnsCreationError,
    <PendingDelayedMessage<Message> as WarpTableRow>::IndexesCreationError,
>;

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum DelayLedgerCreationFailure {
    StorageUnavailable,
}

#[derive(Debug)]
pub struct DelayLedgerCreationError<Message> {
    failure: DelayLedgerCreationFailure,
    _source: PendingDelayedMessageCreationError<Message>,
}

impl<Message> DelayLedgerCreationError<Message> {
    pub const fn failure(&self) -> DelayLedgerCreationFailure {
        self.failure
    }
}

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

pub struct DelayLedger<Message> {
    pending: WarpTable<PendingDelayedMessage<Message>>,
}

impl<Message> DelayLedger<Message> {
    pub fn try_new(capacity: Capacity) -> Result<Self, DelayLedgerCreationError<Message>> {
        let pending = WarpTable::try_new(capacity).map_err(|source| DelayLedgerCreationError {
            failure: DelayLedgerCreationFailure::StorageUnavailable,
            _source: source,
        })?;
        Ok(Self { pending })
    }

    pub const fn capacity(&self) -> Capacity {
        self.pending.capacity()
    }

    pub const fn len(&self) -> usize {
        self.pending.len()
    }

    pub const fn is_empty(&self) -> bool {
        self.pending.is_empty()
    }

    pub fn admit(&mut self, delayed: Delayed<Message>) -> DelayAdmissionOutcome<Message> {
        let (message, wake_after) = delayed.into_parts();
        match self.pending.insert(PendingDelayedMessage { message }) {
            WarpTableInsertOutcome::Inserted { id, .. } => DelayAdmissionOutcome::Admitted {
                admission: DelayAdmission {
                    token: TimerToken::from_nonzero(id.value()),
                    wake_after,
                },
            },
            WarpTableInsertOutcome::Full {
                rejected: PendingDelayedMessage { message },
            } => DelayAdmissionOutcome::Refused {
                reason: DelayUnavailable::AtCapacity {
                    maximum_pending: self.capacity().value().get(),
                },
                rejected: Delayed::new(message, wake_after),
            },
            WarpTableInsertOutcome::IdentifiersExhausted {
                rejected: PendingDelayedMessage { message },
            } => DelayAdmissionOutcome::Refused {
                reason: DelayUnavailable::IdentifiersExhausted,
                rejected: Delayed::new(message, wake_after),
            },
            WarpTableInsertOutcome::UniqueConflict {
                conflict,
                rejected: _,
            } => match conflict {},
        }
    }

    pub fn refuse(&mut self, refusal: DelayArmRefusal) -> DelayRefusalOutcome<Message> {
        match self.pending.remove(WarpId::new(refusal.token.value())) {
            WarpTableRemoveOutcome::Removed {
                removed: PendingDelayedMessage { message },
                ..
            } => DelayRefusalOutcome::Refused {
                reason: DelayUnavailable::WakeRefused {
                    reason: refusal.reason,
                },
                rejected: Delayed::new(message, refusal.wake_after),
            },
            WarpTableRemoveOutcome::Absent { .. } => DelayRefusalOutcome::Stale { refusal },
        }
    }

    pub fn withdraw(&mut self, admission: DelayAdmission) -> DelayWithdrawalOutcome<Message> {
        match self.pending.remove(WarpId::new(admission.token.value())) {
            WarpTableRemoveOutcome::Removed {
                removed: PendingDelayedMessage { message },
                ..
            } => DelayWithdrawalOutcome::Withdrawn {
                rejected: Delayed::new(message, admission.wake_after),
            },
            WarpTableRemoveOutcome::Absent { .. } => DelayWithdrawalOutcome::Absent { admission },
        }
    }

    pub fn release(&mut self, token: TimerToken) -> DelayReleaseOutcome<Message> {
        match self.pending.remove(WarpId::new(token.value())) {
            WarpTableRemoveOutcome::Removed {
                removed: PendingDelayedMessage { message },
                ..
            } => DelayReleaseOutcome::Released { message },
            WarpTableRemoveOutcome::Absent { .. } => DelayReleaseOutcome::Stale { token },
        }
    }
}

#[must_use = "admitted delayed work or exact refusal must be handled by the fitting"]
pub enum DelayAdmissionOutcome<Message> {
    Admitted {
        admission: DelayAdmission,
    },
    Refused {
        reason: DelayUnavailable,
        rejected: Delayed<Message>,
    },
}

#[must_use = "delayed-message admission must be armed before returning"]
pub struct DelayAdmission {
    token: TimerToken,
    wake_after: WakeAfter,
}

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

    pub fn settle_timer_arm(
        self,
        outcome: TimerArmOutcome<TimerToken>,
    ) -> Result<DelayArmOutcome, DelayArmMismatch> {
        let decision = outcome.into_decision();
        let matches_admission = match decision {
            TimerArmDecision::Armed { token } => token == self.token,
            TimerArmDecision::Refused { token, after, .. } => {
                token == self.token && after == self.wake_after
            }
        };
        if !matches_admission {
            return Err(DelayArmMismatch {
                admission: self,
                decision,
            });
        }
        match decision {
            TimerArmDecision::Armed { .. } => Ok(DelayArmOutcome::Armed),
            TimerArmDecision::Refused { reason, .. } => Ok(DelayArmOutcome::Refused {
                refusal: DelayArmRefusal {
                    token: self.token,
                    wake_after: self.wake_after,
                    reason,
                },
            }),
        }
    }
}

#[must_use = "armed delayed work or exact timer refusal must be handled by the fitting"]
pub enum DelayArmOutcome {
    Armed,
    Refused { refusal: DelayArmRefusal },
}

pub struct DelayArmMismatch {
    admission: DelayAdmission,
    decision: TimerArmDecision<TimerToken>,
}

impl DelayArmMismatch {
    pub fn into_parts(self) -> (DelayAdmission, TimerArmDecision<TimerToken>) {
        (self.admission, self.decision)
    }
}

#[must_use = "timer refusal must be returned to the delay ledger"]
pub struct DelayArmRefusal {
    token: TimerToken,
    wake_after: WakeAfter,
    reason: TimerArmRefusal,
}

impl DelayArmRefusal {
    pub const fn reason(&self) -> TimerArmRefusal {
        self.reason
    }
}

#[must_use = "resolved or stale timer refusal must be handled by the fitting"]
pub enum DelayRefusalOutcome<Message> {
    Refused {
        reason: DelayUnavailable,
        rejected: Delayed<Message>,
    },
    Stale {
        refusal: DelayArmRefusal,
    },
}

#[must_use = "released or stale timer tokens must be handled by the fitting"]
pub enum DelayReleaseOutcome<Message> {
    Released { message: Message },
    Stale { token: TimerToken },
}

#[must_use = "withdrawn delayed work or an absent admission must be handled by the fitting"]
pub enum DelayWithdrawalOutcome<Message> {
    Withdrawn { rejected: Delayed<Message> },
    Absent { admission: DelayAdmission },
}