use std::error::Error;
use std::fmt;
use std::time::Duration;
use crate::{ChunkDeliveryMode, ClassifierRevision, FailureCategory, FailureId, FailureSummary};
const MAX_BACKOFF: Duration = Duration::from_hours(24);
const MAX_RETRY: u32 = 65_535;
const MIN_RETRY_STATE: u32 = 1;
const MAX_RETRY_STATE: u32 = 256;
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
#[non_exhaustive]
pub enum FaultPhase {
Read,
Process,
Write,
Transaction,
Checkpoint,
Listener,
Backoff,
}
impl FaultPhase {
#[must_use]
pub const fn is_policy_eligible(self) -> bool {
!matches!(self, Self::Listener)
}
#[must_use]
pub const fn is_skippable(self) -> bool {
matches!(self, Self::Read | Self::Process | Self::Write)
}
#[must_use]
pub const fn allows_commit_safe_skip(self) -> bool {
matches!(self, Self::Read | Self::Process)
}
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::Read => "read",
Self::Process => "process",
Self::Write => "write",
Self::Transaction => "transaction",
Self::Checkpoint => "checkpoint",
Self::Listener => "listener",
Self::Backoff => "backoff",
}
}
#[must_use]
pub fn from_durable_name(value: &str) -> Option<Self> {
Some(match value {
"read" => Self::Read,
"process" => Self::Process,
"write" => Self::Write,
"transaction" => Self::Transaction,
"checkpoint" => Self::Checkpoint,
"listener" => Self::Listener,
"backoff" => Self::Backoff,
_ => return None,
})
}
}
impl fmt::Display for FaultPhase {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(self.as_str())
}
}
#[derive(Clone, Copy, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct RetryOrdinal(u16);
impl RetryOrdinal {
pub const INITIAL: Self = Self(0);
pub fn new(value: u32) -> Result<Self, FaultPolicyError> {
u16::try_from(value)
.map(Self)
.map_err(|_| FaultPolicyError::RetryOrdinalOutOfRange { max: MAX_RETRY })
}
#[must_use]
#[allow(
clippy::cast_lossless,
reason = "`From` is not const; the widening cast is exact"
)]
pub const fn get(self) -> u32 {
self.0 as u32
}
#[must_use]
pub const fn is_initial(self) -> bool {
self.0 == 0
}
pub fn checked_next(self) -> Result<Self, FaultPolicyError> {
self.0
.checked_add(1)
.map(Self)
.ok_or(FaultPolicyError::RetryOrdinalOutOfRange { max: MAX_RETRY })
}
}
#[derive(Clone, Copy, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct RetryLimit(u16);
impl RetryLimit {
pub const NONE: Self = Self(0);
pub fn new(value: u32) -> Result<Self, FaultPolicyError> {
u16::try_from(value)
.map(Self)
.map_err(|_| FaultPolicyError::RetryLimitOutOfRange { max: MAX_RETRY })
}
#[must_use]
#[allow(
clippy::cast_lossless,
reason = "`From` is not const; the widening cast is exact"
)]
pub const fn get(self) -> u32 {
self.0 as u32
}
#[must_use]
pub const fn is_none(self) -> bool {
self.0 == 0
}
#[must_use]
pub const fn permits(self, ordinal: RetryOrdinal) -> bool {
!ordinal.is_initial() && ordinal.0 <= self.0
}
}
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct RetryStateLimit(u16);
impl RetryStateLimit {
pub fn new(value: u32) -> Result<Self, FaultPolicyError> {
if !(MIN_RETRY_STATE..=MAX_RETRY_STATE).contains(&value) {
return Err(FaultPolicyError::RetryStateLimitOutOfRange {
min: MIN_RETRY_STATE,
max: MAX_RETRY_STATE,
});
}
u16::try_from(value)
.map(Self)
.map_err(|_| FaultPolicyError::RetryStateLimitOutOfRange {
min: MIN_RETRY_STATE,
max: MAX_RETRY_STATE,
})
}
#[must_use]
#[allow(
clippy::cast_lossless,
reason = "`From` is not const; the widening cast is exact"
)]
pub const fn get(self) -> u32 {
self.0 as u32
}
}
#[derive(Clone, Copy, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct SkipLimit(u64);
impl SkipLimit {
pub const NONE: Self = Self(0);
#[must_use]
pub const fn new(value: u64) -> Self {
Self(value)
}
#[must_use]
pub const fn get(self) -> u64 {
self.0
}
}
#[derive(Clone, Copy, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct SkipCounts {
read: u64,
process: u64,
write: u64,
}
impl SkipCounts {
pub const ZERO: Self = Self {
read: 0,
process: 0,
write: 0,
};
#[must_use]
pub const fn new(read: u64, process: u64, write: u64) -> Self {
Self {
read,
process,
write,
}
}
#[must_use]
pub const fn read(self) -> u64 {
self.read
}
#[must_use]
pub const fn process(self) -> u64 {
self.process
}
#[must_use]
pub const fn write(self) -> u64 {
self.write
}
pub fn checked_total(self) -> Result<u64, FaultPolicyError> {
self.read
.checked_add(self.process)
.and_then(|partial| partial.checked_add(self.write))
.ok_or(FaultPolicyError::SkipCountOverflow)
}
pub fn checked_add(self, other: Self) -> Result<Self, FaultPolicyError> {
let next = Self {
read: self
.read
.checked_add(other.read)
.ok_or(FaultPolicyError::SkipCountOverflow)?,
process: self
.process
.checked_add(other.process)
.ok_or(FaultPolicyError::SkipCountOverflow)?,
write: self
.write
.checked_add(other.write)
.ok_or(FaultPolicyError::SkipCountOverflow)?,
};
next.checked_total()?;
Ok(next)
}
pub fn checked_increment(self, phase: FaultPhase) -> Result<Self, FaultPolicyError> {
let mut next = self;
let counter = match phase {
FaultPhase::Read => &mut next.read,
FaultPhase::Process => &mut next.process,
FaultPhase::Write => &mut next.write,
other => return Err(FaultPolicyError::PhaseNotSkippable { phase: other }),
};
*counter = counter
.checked_add(1)
.ok_or(FaultPolicyError::SkipCountOverflow)?;
next.checked_total()?;
Ok(next)
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct FaultDescriptor {
phase: FaultPhase,
summary: FailureSummary,
retry_ordinal: RetryOrdinal,
committed_skips: SkipCounts,
transaction_open: bool,
delivery_mode: ChunkDeliveryMode,
}
impl FaultDescriptor {
#[must_use]
pub const fn new(
phase: FaultPhase,
summary: FailureSummary,
retry_ordinal: RetryOrdinal,
committed_skips: SkipCounts,
transaction_open: bool,
delivery_mode: ChunkDeliveryMode,
) -> Self {
Self {
phase,
summary,
retry_ordinal,
committed_skips,
transaction_open,
delivery_mode,
}
}
#[must_use]
pub const fn phase(self) -> FaultPhase {
self.phase
}
#[must_use]
pub const fn summary(self) -> FailureSummary {
self.summary
}
#[must_use]
pub const fn category(self) -> FailureCategory {
self.summary.category()
}
#[must_use]
pub const fn failure_id(self) -> FailureId {
self.summary.failure_id()
}
#[must_use]
pub const fn retry_ordinal(self) -> RetryOrdinal {
self.retry_ordinal
}
#[must_use]
pub const fn committed_skips(self) -> SkipCounts {
self.committed_skips
}
#[must_use]
pub const fn is_transaction_open(self) -> bool {
self.transaction_open
}
#[must_use]
pub const fn delivery_mode(self) -> ChunkDeliveryMode {
self.delivery_mode
}
}
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
#[non_exhaustive]
pub enum BackoffKind {
None,
Fixed,
Exponential,
}
impl BackoffKind {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::None => "none",
Self::Fixed => "fixed",
Self::Exponential => "exponential",
}
}
}
impl fmt::Display for BackoffKind {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(self.as_str())
}
}
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
pub struct BackoffPolicy {
kind: BackoffKind,
initial: Duration,
multiplier: u32,
maximum: Duration,
}
impl BackoffPolicy {
#[must_use]
pub const fn none() -> Self {
Self {
kind: BackoffKind::None,
initial: Duration::ZERO,
multiplier: 1,
maximum: Duration::ZERO,
}
}
pub fn fixed(delay: Duration) -> Result<Self, FaultPolicyError> {
check_delay(delay)?;
Ok(Self {
kind: BackoffKind::Fixed,
initial: delay,
multiplier: 1,
maximum: delay,
})
}
pub fn exponential(
initial: Duration,
multiplier: u32,
maximum: Duration,
) -> Result<Self, FaultPolicyError> {
check_delay(initial)?;
check_delay(maximum)?;
if multiplier == 0 {
return Err(FaultPolicyError::ZeroBackoffMultiplier);
}
if maximum < initial {
return Err(FaultPolicyError::BackoffMaximumBelowInitial);
}
Ok(Self {
kind: BackoffKind::Exponential,
initial,
multiplier,
maximum,
})
}
#[must_use]
pub const fn kind(self) -> BackoffKind {
self.kind
}
#[must_use]
pub const fn initial(self) -> Duration {
self.initial
}
#[must_use]
pub const fn multiplier(self) -> u32 {
self.multiplier
}
#[must_use]
pub const fn maximum(self) -> Duration {
self.maximum
}
#[must_use]
pub fn delay_for(self, ordinal: RetryOrdinal) -> Duration {
if ordinal.is_initial() {
return Duration::ZERO;
}
match self.kind {
BackoffKind::None => Duration::ZERO,
BackoffKind::Fixed => self.initial,
BackoffKind::Exponential => self.exponential_delay(ordinal.get()),
}
}
fn exponential_delay(self, ordinal: u32) -> Duration {
let maximum_nanos = self.maximum.as_nanos();
let mut nanos = self.initial.as_nanos();
if nanos >= maximum_nanos {
return self.maximum;
}
if self.multiplier > 1 {
let factor = u128::from(self.multiplier);
for _ in 1..ordinal {
nanos = nanos.saturating_mul(factor);
if nanos >= maximum_nanos {
return self.maximum;
}
}
}
u64::try_from(nanos).map_or(self.maximum, Duration::from_nanos)
}
}
fn check_delay(delay: Duration) -> Result<(), FaultPolicyError> {
if delay > MAX_BACKOFF {
return Err(FaultPolicyError::BackoffDelayTooLong {
max_seconds: MAX_BACKOFF.as_secs(),
});
}
Ok(())
}
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
#[non_exhaustive]
pub enum RollbackDisposition {
Rollback,
CommitSafeSkip,
}
impl RollbackDisposition {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::Rollback => "rollback",
Self::CommitSafeSkip => "commit_safe_skip",
}
}
}
#[derive(Clone, Copy, Debug, Default, Eq, Hash, PartialEq)]
pub struct FaultAction {
retryable: bool,
skip: Option<RollbackDisposition>,
}
impl FaultAction {
#[must_use]
pub const fn fail() -> Self {
Self {
retryable: false,
skip: None,
}
}
#[must_use]
pub const fn retry() -> Self {
Self {
retryable: true,
skip: None,
}
}
#[must_use]
pub const fn skip(disposition: RollbackDisposition) -> Self {
Self {
retryable: false,
skip: Some(disposition),
}
}
#[must_use]
pub const fn retry_then_skip(disposition: RollbackDisposition) -> Self {
Self {
retryable: true,
skip: Some(disposition),
}
}
#[must_use]
pub const fn is_retryable(self) -> bool {
self.retryable
}
#[must_use]
pub const fn skip_disposition(self) -> Option<RollbackDisposition> {
self.skip
}
}
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
pub struct FaultRule {
phase: FaultPhase,
category: FailureCategory,
action: FaultAction,
}
impl FaultRule {
pub fn new(
phase: FaultPhase,
category: FailureCategory,
action: FaultAction,
) -> Result<Self, FaultPolicyError> {
if !phase.is_policy_eligible() || !category.is_policy_eligible() {
return Err(FaultPolicyError::NotPolicyEligible { phase, category });
}
if action.skip.is_some() && !phase.is_skippable() {
return Err(FaultPolicyError::PhaseNotSkippable { phase });
}
if action.skip == Some(RollbackDisposition::CommitSafeSkip)
&& !phase.allows_commit_safe_skip()
{
return Err(FaultPolicyError::CommitSafeSkipPhase { phase });
}
Ok(Self {
phase,
category,
action,
})
}
#[must_use]
pub const fn phase(self) -> FaultPhase {
self.phase
}
#[must_use]
pub const fn category(self) -> FailureCategory {
self.category
}
#[must_use]
pub const fn action(self) -> FaultAction {
self.action
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct FaultClassifier {
revision: ClassifierRevision,
rules: Box<[FaultRule]>,
}
impl FaultClassifier {
pub fn new(
revision: ClassifierRevision,
rules: impl IntoIterator<Item = FaultRule>,
) -> Result<Self, FaultPolicyError> {
let mut accepted: Vec<FaultRule> = Vec::new();
for rule in rules {
if accepted
.iter()
.any(|existing| existing.phase == rule.phase && existing.category == rule.category)
{
return Err(FaultPolicyError::DuplicateRule {
phase: rule.phase,
category: rule.category,
});
}
accepted.push(rule);
}
Ok(Self {
revision,
rules: accepted.into_boxed_slice(),
})
}
#[must_use]
pub const fn revision(&self) -> &ClassifierRevision {
&self.revision
}
#[must_use]
pub fn rules(&self) -> &[FaultRule] {
&self.rules
}
#[must_use]
pub fn action_for(&self, phase: FaultPhase, category: FailureCategory) -> Option<FaultAction> {
self.rules
.iter()
.find(|rule| rule.phase == phase && rule.category == category)
.map(|rule| rule.action)
}
}
#[derive(Clone, Copy, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct FaultEvidence {
located: bool,
known_rollback: bool,
forward_checkpoint_proof: bool,
}
impl FaultEvidence {
pub const NONE: Self = Self {
located: false,
known_rollback: false,
forward_checkpoint_proof: false,
};
#[must_use]
pub const fn new(located: bool, known_rollback: bool, forward_checkpoint_proof: bool) -> Self {
Self {
located,
known_rollback,
forward_checkpoint_proof,
}
}
#[must_use]
pub const fn with_located(mut self, located: bool) -> Self {
self.located = located;
self
}
#[must_use]
pub const fn with_known_rollback(mut self, known_rollback: bool) -> Self {
self.known_rollback = known_rollback;
self
}
#[must_use]
pub const fn with_forward_checkpoint_proof(mut self, proof: bool) -> Self {
self.forward_checkpoint_proof = proof;
self
}
#[must_use]
pub const fn is_located(self) -> bool {
self.located
}
#[must_use]
pub const fn is_known_rollback(self) -> bool {
self.known_rollback
}
#[must_use]
pub const fn has_forward_checkpoint_proof(self) -> bool {
self.forward_checkpoint_proof
}
}
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
#[non_exhaustive]
pub enum FaultDecision {
Retry {
ordinal: RetryOrdinal,
delay: Duration,
},
Skip {
disposition: RollbackDisposition,
},
FailAndRollback,
Unknown,
Stop,
}
impl FaultDecision {
#[must_use]
pub const fn is_retry(self) -> bool {
matches!(self, Self::Retry { .. })
}
#[must_use]
pub const fn skip_disposition(self) -> Option<RollbackDisposition> {
match self {
Self::Skip { disposition } => Some(disposition),
_ => None,
}
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct FaultPolicy {
classifier: FaultClassifier,
retry_limit: RetryLimit,
retry_state_limit: RetryStateLimit,
skip_limit: SkipLimit,
backoff: BackoffPolicy,
}
impl FaultPolicy {
pub fn new(
classifier: FaultClassifier,
retry_limit: RetryLimit,
retry_state_limit: RetryStateLimit,
skip_limit: SkipLimit,
backoff: BackoffPolicy,
) -> Result<Self, FaultPolicyError> {
if retry_limit.is_none()
&& let Some(rule) = classifier
.rules()
.iter()
.find(|rule| rule.action().is_retryable())
{
return Err(FaultPolicyError::UnreachableRetryRule {
phase: rule.phase(),
category: rule.category(),
});
}
Ok(Self {
classifier,
retry_limit,
retry_state_limit,
skip_limit,
backoff,
})
}
#[must_use]
pub const fn classifier(&self) -> &FaultClassifier {
&self.classifier
}
#[must_use]
pub const fn retry_limit(&self) -> RetryLimit {
self.retry_limit
}
#[must_use]
pub const fn retry_state_limit(&self) -> RetryStateLimit {
self.retry_state_limit
}
#[must_use]
pub const fn skip_limit(&self) -> SkipLimit {
self.skip_limit
}
#[must_use]
pub const fn backoff(&self) -> BackoffPolicy {
self.backoff
}
#[must_use]
pub fn requires_commit_safe_skip(&self) -> bool {
self.classifier.rules().iter().any(|rule| {
rule.action().skip_disposition() == Some(RollbackDisposition::CommitSafeSkip)
})
}
pub fn validate_capabilities(
&self,
supports_atomic_skip: bool,
) -> Result<(), FaultPolicyError> {
if self.requires_commit_safe_skip() && !supports_atomic_skip {
return Err(FaultPolicyError::CommitSafeSkipUnsupported);
}
Ok(())
}
#[must_use]
pub fn decide(&self, fault: &FaultDescriptor, evidence: FaultEvidence) -> FaultDecision {
let category = fault.category();
if category == FailureCategory::UnknownCommit {
return FaultDecision::Unknown;
}
if category == FailureCategory::Cancelled {
return FaultDecision::Stop;
}
let phase = fault.phase();
if !phase.is_policy_eligible() || !category.is_policy_eligible() {
return FaultDecision::FailAndRollback;
}
let Some(action) = self.classifier.action_for(phase, category) else {
return FaultDecision::FailAndRollback;
};
if action.is_retryable()
&& let Ok(next) = fault.retry_ordinal().checked_next()
&& self.retry_limit.permits(next)
{
return FaultDecision::Retry {
ordinal: next,
delay: self.backoff.delay_for(next),
};
}
match action.skip_disposition() {
Some(disposition) => self.decide_skip(fault, disposition, evidence),
None => FaultDecision::FailAndRollback,
}
}
fn decide_skip(
&self,
fault: &FaultDescriptor,
disposition: RollbackDisposition,
evidence: FaultEvidence,
) -> FaultDecision {
let phase = fault.phase();
if !evidence.is_located() {
return FaultDecision::FailAndRollback;
}
let phase_evidence = match phase {
FaultPhase::Read => evidence.has_forward_checkpoint_proof(),
FaultPhase::Process => true,
FaultPhase::Write => evidence.is_known_rollback(),
_ => false,
};
if !phase_evidence {
return FaultDecision::FailAndRollback;
}
if disposition == RollbackDisposition::CommitSafeSkip
&& !(phase.allows_commit_safe_skip()
&& evidence.is_known_rollback()
&& evidence.has_forward_checkpoint_proof())
{
return FaultDecision::FailAndRollback;
}
match fault.committed_skips().checked_increment(phase) {
Ok(next) => match next.checked_total() {
Ok(total) if total <= self.skip_limit.get() => FaultDecision::Skip { disposition },
_ => FaultDecision::FailAndRollback,
},
Err(_) => FaultDecision::FailAndRollback,
}
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[non_exhaustive]
pub enum FaultPolicyError {
RetryLimitOutOfRange {
max: u32,
},
RetryOrdinalOutOfRange {
max: u32,
},
RetryStateLimitOutOfRange {
min: u32,
max: u32,
},
BackoffDelayTooLong {
max_seconds: u64,
},
ZeroBackoffMultiplier,
BackoffMaximumBelowInitial,
NotPolicyEligible {
phase: FaultPhase,
category: FailureCategory,
},
PhaseNotSkippable {
phase: FaultPhase,
},
CommitSafeSkipPhase {
phase: FaultPhase,
},
CommitSafeSkipUnsupported,
DuplicateRule {
phase: FaultPhase,
category: FailureCategory,
},
UnreachableRetryRule {
phase: FaultPhase,
category: FailureCategory,
},
SkipCountOverflow,
}
impl fmt::Display for FaultPolicyError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::RetryLimitOutOfRange { max } => {
write!(formatter, "retry limit exceeds {max}")
}
Self::RetryOrdinalOutOfRange { max } => {
write!(formatter, "retry ordinal exceeds {max}")
}
Self::RetryStateLimitOutOfRange { min, max } => {
write!(formatter, "retry state limit must be within {min}..={max}")
}
Self::BackoffDelayTooLong { max_seconds } => {
write!(formatter, "backoff delay exceeds {max_seconds} seconds")
}
Self::ZeroBackoffMultiplier => {
formatter.write_str("exponential backoff requires a nonzero multiplier")
}
Self::BackoffMaximumBelowInitial => {
formatter.write_str("exponential backoff maximum is below its initial delay")
}
Self::NotPolicyEligible { phase, category } => write!(
formatter,
"{phase} {category:?} faults are never retried or skipped"
),
Self::PhaseNotSkippable { phase } => {
write!(formatter, "{phase} faults cannot commit a skip")
}
Self::CommitSafeSkipPhase { phase } => {
write!(formatter, "{phase} faults cannot commit a skip safely")
}
Self::CommitSafeSkipUnsupported => {
formatter.write_str("the selected resource cannot commit a skip atomically")
}
Self::DuplicateRule { phase, category } => write!(
formatter,
"duplicate classifier rule for {phase} {category:?}"
),
Self::UnreachableRetryRule { phase, category } => write!(
formatter,
"{phase} {category:?} retry rule requires a nonzero retry limit"
),
Self::SkipCountOverflow => formatter.write_str("skip counters overflowed"),
}
}
}
impl Error for FaultPolicyError {}