use crate::wire::{AckCommitted, BindingEpoch, DeliverySeq, ParticipantId};
use super::{
ClosureState, CursorEpisodeBuildError, FrontierBinding, LiveFrontierOwner, LiveMember,
NonzeroDebtCursorEpisode, NonzeroParticipantAckCommit, NonzeroParticipantAckCommitError,
};
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ObligationDebtOwnerError {
Episode(CursorEpisodeBuildError),
NonzeroAck(NonzeroParticipantAckCommitError),
NonzeroAckState,
}
#[derive(Debug, PartialEq, Eq)]
pub struct CoupledObligationDebtOwner {
frontier: LiveFrontierOwner,
episode: NonzeroDebtCursorEpisode,
}
#[derive(Debug, PartialEq, Eq)]
pub enum ObligationDebtDispatchState {
Clear(LiveFrontierOwner),
Owed(CoupledObligationDebtOwner),
}
#[derive(Debug)]
pub struct ObligationDebtDispatchTransition {
prior_episode: Option<NonzeroDebtCursorEpisode>,
}
impl ObligationDebtDispatchState {
pub fn from_frontier(
frontier: LiveFrontierOwner,
observer_progress: DeliverySeq,
) -> Result<Self, ObligationDebtOwnerError> {
match frontier.closure_accounting().state() {
ClosureState::Clear => Ok(Self::Clear(frontier)),
ClosureState::Owed { debt, .. } => {
let episode = NonzeroDebtCursorEpisode::from_claim_frontiers(
frontier.frontiers(),
debt,
observer_progress,
)
.map_err(ObligationDebtOwnerError::Episode)?;
Ok(Self::Owed(CoupledObligationDebtOwner { frontier, episode }))
}
}
}
#[must_use]
pub const fn frontier(&self) -> &LiveFrontierOwner {
match self {
Self::Clear(frontier) => frontier,
Self::Owed(coupled) => &coupled.frontier,
}
}
#[must_use]
pub const fn episode(&self) -> Option<&NonzeroDebtCursorEpisode> {
match self {
Self::Clear(_) => None,
Self::Owed(coupled) => Some(&coupled.episode),
}
}
#[must_use]
pub fn frontier_participant(
&self,
participant_id: ParticipantId,
) -> Option<(FrontierBinding, DeliverySeq)> {
self.frontier()
.frontiers()
.active_identities()
.participants()
.iter()
.find(|participant| participant.participant_index() == participant_id)
.map(|participant| (participant.binding(), participant.cursor()))
}
#[must_use]
pub fn participant(
&self,
participant_id: ParticipantId,
) -> Option<(FrontierBinding, DeliverySeq)> {
self.episode()?.participant_binding(participant_id)
}
#[must_use]
pub fn begin_transition(self) -> (LiveFrontierOwner, ObligationDebtDispatchTransition) {
match self {
Self::Clear(frontier) => (
frontier,
ObligationDebtDispatchTransition {
prior_episode: None,
},
),
Self::Owed(coupled) => (
coupled.frontier,
ObligationDebtDispatchTransition {
prior_episode: Some(coupled.episode),
},
),
}
}
}
impl ObligationDebtDispatchTransition {
pub fn complete(
self,
frontier: LiveFrontierOwner,
observer_progress: DeliverySeq,
) -> Result<ObligationDebtDispatchState, ObligationDebtOwnerError> {
match frontier.closure_accounting().state() {
ClosureState::Clear => Ok(ObligationDebtDispatchState::Clear(frontier)),
ClosureState::Owed { debt, .. } => {
let episode = self
.prior_episode
.map_or_else(
|| {
NonzeroDebtCursorEpisode::from_claim_frontiers(
frontier.frontiers(),
debt,
observer_progress,
)
},
|episode| {
episode.reconcile_claim_frontiers(
frontier.frontiers(),
debt,
observer_progress,
)
},
)
.map_err(ObligationDebtOwnerError::Episode)?;
Ok(ObligationDebtDispatchState::Owed(
CoupledObligationDebtOwner { frontier, episode },
))
}
}
}
pub fn complete_nonzero_ack<F>(
mut self,
frontier: LiveFrontierOwner,
commit: NonzeroParticipantAckCommit,
member: &mut LiveMember<F>,
observer_progress: DeliverySeq,
) -> Result<(ObligationDebtDispatchState, AckCommitted), ObligationDebtOwnerError> {
let ClosureState::Owed { debt, .. } = frontier.closure_accounting().state() else {
return Err(ObligationDebtOwnerError::NonzeroAckState);
};
let mut episode = self
.prior_episode
.take()
.ok_or(ObligationDebtOwnerError::NonzeroAckState)?;
let outcome = commit
.apply_to(member, &mut episode)
.map_err(ObligationDebtOwnerError::NonzeroAck)?;
if episode.debt() != debt {
return Err(ObligationDebtOwnerError::NonzeroAckState);
}
let reconciled = episode
.clone()
.reconcile_claim_frontiers(frontier.frontiers(), debt, observer_progress)
.map_err(ObligationDebtOwnerError::Episode)?;
if reconciled != episode {
return Err(ObligationDebtOwnerError::NonzeroAckState);
}
Ok((
ObligationDebtDispatchState::Owed(CoupledObligationDebtOwner { frontier, episode }),
outcome,
))
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum DebtDispatchDeferral {
NoObligation,
NoCurrentBinding,
BeyondDebtHighWatermark,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum DebtDispatchInvariant {
CoupledState,
ParticipantAuthority,
OutboxSelection,
ScalarDivergence,
}
#[derive(Debug, PartialEq, Eq)]
pub enum ObligationDebtDispatchDecision<T> {
Permit(T),
Defer(DebtDispatchDeferral),
Invariant(DebtDispatchInvariant),
}
pub fn decide_obligation_debt_dispatch<T>(
state: &ObligationDebtDispatchState,
participant_id: ParticipantId,
binding_epoch: BindingEpoch,
dispatch_after: DeliverySeq,
select_next: impl FnOnce(ParticipantId, BindingEpoch, DeliverySeq) -> Option<(DeliverySeq, T)>,
) -> ObligationDebtDispatchDecision<T> {
let frontier_participant = state
.frontier()
.frontiers()
.active_identities()
.participants()
.iter()
.find(|participant| participant.participant_index() == participant_id)
.copied();
let Some(frontier_participant) = frontier_participant else {
return ObligationDebtDispatchDecision::Invariant(
DebtDispatchInvariant::ParticipantAuthority,
);
};
let FrontierBinding::Bound(frontier_epoch) = frontier_participant.binding() else {
return ObligationDebtDispatchDecision::Defer(DebtDispatchDeferral::NoCurrentBinding);
};
if frontier_epoch != binding_epoch || frontier_participant.cursor() > dispatch_after {
return ObligationDebtDispatchDecision::Invariant(
DebtDispatchInvariant::ParticipantAuthority,
);
}
let owed_high_watermark = match state {
ObligationDebtDispatchState::Clear(frontier) => {
if !matches!(frontier.closure_accounting().state(), ClosureState::Clear) {
return ObligationDebtDispatchDecision::Invariant(
DebtDispatchInvariant::CoupledState,
);
}
None
}
ObligationDebtDispatchState::Owed(coupled) => {
let ClosureState::Owed { debt, .. } = coupled.frontier.closure_accounting().state()
else {
return ObligationDebtDispatchDecision::Invariant(
DebtDispatchInvariant::CoupledState,
);
};
if coupled.episode.debt() != debt {
return ObligationDebtDispatchDecision::Invariant(
DebtDispatchInvariant::CoupledState,
);
}
if coupled.episode.participant_binding(participant_id)
!= Some((
FrontierBinding::Bound(binding_epoch),
frontier_participant.cursor(),
))
{
return ObligationDebtDispatchDecision::Invariant(
DebtDispatchInvariant::ParticipantAuthority,
);
}
Some(coupled.episode.candidate_high_watermark())
}
};
classify_selected_obligation(
dispatch_after,
owed_high_watermark,
select_next(participant_id, binding_epoch, dispatch_after),
)
}
fn classify_selected_obligation<T>(
dispatch_after: DeliverySeq,
owed_high_watermark: Option<DeliverySeq>,
selected: Option<(DeliverySeq, T)>,
) -> ObligationDebtDispatchDecision<T> {
let Some((delivery_seq, selected)) = selected else {
return ObligationDebtDispatchDecision::Defer(DebtDispatchDeferral::NoObligation);
};
if delivery_seq <= dispatch_after {
return ObligationDebtDispatchDecision::Invariant(DebtDispatchInvariant::OutboxSelection);
}
if owed_high_watermark.is_some_and(|high_watermark| delivery_seq > high_watermark) {
return ObligationDebtDispatchDecision::Defer(
DebtDispatchDeferral::BeyondDebtHighWatermark,
);
}
ObligationDebtDispatchDecision::Permit(selected)
}
#[cfg(test)]
mod tests {
use alloc::vec;
use crate::{
algebra::WideResourceVector,
wire::{BindingEpoch, ConnectionIncarnation, Generation},
};
use super::{
super::{BoundParticipantCursor, ClosureDebt, NonzeroDebtCursorEpisode},
DebtDispatchDeferral, DebtDispatchInvariant, ObligationDebtDispatchDecision,
classify_selected_obligation,
};
fn binding_epoch() -> BindingEpoch {
BindingEpoch::new(
ConnectionIncarnation::new(1, 1),
Generation::new(1).unwrap_or(Generation::ONE),
)
}
#[test]
fn nonzero_debt_permits_testified_below_floor_and_defers_above_high_watermark() {
let cursor = 0;
let high_watermark = 100;
let physical_floor = 25;
let capacity_floor = 25;
let below_floor_endpoint = 10;
let above_high_watermark = high_watermark + 1;
let episode = NonzeroDebtCursorEpisode::new(
1,
ClosureDebt::new(WideResourceVector::new(1, 1))
.unwrap_or_else(|| unreachable!("fixture debt is nonzero")),
high_watermark,
high_watermark,
physical_floor,
capacity_floor,
vec![BoundParticipantCursor::new(0, binding_epoch(), cursor)],
)
.unwrap_or_else(|error| unreachable!("fixture episode must be valid: {error:?}"));
assert_eq!(episode.candidate_high_watermark(), high_watermark);
assert_eq!(episode.observer_progress(), high_watermark);
assert_eq!(episode.cap_floor(), capacity_floor);
assert_eq!(episode.retained_suffix_start(), Some(25));
assert!(!episode.retains(below_floor_endpoint));
assert_eq!(
classify_selected_obligation(
cursor,
Some(episode.candidate_high_watermark()),
Some((below_floor_endpoint, below_floor_endpoint)),
),
ObligationDebtDispatchDecision::Permit(below_floor_endpoint)
);
assert_eq!(
classify_selected_obligation(
cursor,
Some(episode.candidate_high_watermark()),
Some((above_high_watermark, above_high_watermark)),
),
ObligationDebtDispatchDecision::Defer(DebtDispatchDeferral::BeyondDebtHighWatermark)
);
assert_eq!(
classify_selected_obligation::<u64>(
cursor,
Some(episode.candidate_high_watermark()),
None,
),
ObligationDebtDispatchDecision::Defer(DebtDispatchDeferral::NoObligation)
);
}
#[test]
fn debt_dispatch_invariant_never_falls_back_or_fabricates_wire_refusal() {
let cursor = 7;
let selected_payload = "durable publication";
let decision =
classify_selected_obligation(cursor, Some(100), Some((cursor, selected_payload)));
assert_eq!(
decision,
ObligationDebtDispatchDecision::Invariant(DebtDispatchInvariant::OutboxSelection)
);
assert!(!matches!(
decision,
ObligationDebtDispatchDecision::Permit(_)
));
assert!(!matches!(
decision,
ObligationDebtDispatchDecision::Defer(_)
));
}
}