use crate::algebra::{ResourceVector, WideResourceVector};
use crate::wire::{BindingEpoch, ConversationId, DeliverySeq, ParticipantId, ParticipantIndex};
use super::{
ActiveBinding, CommittedDiedTerminal, ObserverProgressProjection,
claim_frontier::{ValidatedMarkerCandidate, ValidatedMarkerRecord},
};
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ClosureDebt(WideResourceVector);
impl ClosureDebt {
#[must_use]
pub const fn new(value: WideResourceVector) -> Option<Self> {
if value.is_zero() {
None
} else {
Some(Self(value))
}
}
#[must_use]
pub const fn value(self) -> WideResourceVector {
self.0
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ObserverProjection {
through_seq: DeliverySeq,
}
impl ObserverProjection {
#[must_use]
pub const fn new(through_seq: DeliverySeq) -> Self {
Self { through_seq }
}
#[must_use]
pub const fn through_seq(self) -> DeliverySeq {
self.through_seq
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct PhysicalCompaction {
from_floor: DeliverySeq,
through_seq: DeliverySeq,
}
impl PhysicalCompaction {
#[must_use]
pub const fn new(from_floor: DeliverySeq, through_seq: DeliverySeq) -> Option<Self> {
if from_floor <= through_seq {
Some(Self {
from_floor,
through_seq,
})
} else {
None
}
}
#[must_use]
pub const fn from_floor(self) -> DeliverySeq {
self.from_floor
}
#[must_use]
pub const fn through_seq(self) -> DeliverySeq {
self.through_seq
}
}
#[allow(clippy::struct_field_names)]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct MarkerDelivery {
conversation_id: ConversationId,
participant_id: ParticipantId,
binding_epoch: BindingEpoch,
marker_delivery_seq: DeliverySeq,
}
impl MarkerDelivery {
#[must_use]
pub(super) const fn successor_from_validated_candidate(
candidate: ValidatedMarkerCandidate,
) -> StoredEdge {
let conversation_id = candidate.conversation_id();
let participant_id = candidate.participant_id();
let marker_delivery_seq = candidate.delivery_seq();
let successor = match candidate.target_binding() {
super::FrontierBinding::Bound(binding_epoch) => StoredEdge::MarkerDelivery(Self {
conversation_id,
participant_id,
binding_epoch,
marker_delivery_seq,
}),
super::FrontierBinding::Detached(last_dead_binding_epoch) => {
StoredEdge::DetachedMarkerRelease(DetachedMarkerRelease {
participant_id,
marker_delivery_seq,
last_dead_binding_epoch,
})
}
};
candidate.consume();
successor
}
#[must_use]
pub(super) const fn from_validated_record(record: &ValidatedMarkerRecord) -> Self {
Self {
conversation_id: record.conversation_id(),
participant_id: record.participant_id(),
binding_epoch: record.binding_epoch(),
marker_delivery_seq: record.delivery_seq(),
}
}
#[must_use]
pub const fn conversation_id(self) -> ConversationId {
self.conversation_id
}
#[must_use]
pub const fn participant_id(self) -> ParticipantId {
self.participant_id
}
#[must_use]
pub const fn binding_epoch(self) -> BindingEpoch {
self.binding_epoch
}
#[must_use]
pub const fn marker_delivery_seq(self) -> DeliverySeq {
self.marker_delivery_seq
}
}
#[cfg(test)]
#[allow(clippy::expect_used, clippy::too_many_lines)]
pub fn validated_marker_record_for_test(
conversation_id: crate::wire::ConversationId,
participant_id: ParticipantId,
target_binding: super::claim_frontier::FrontierBinding,
marker_delivery_seq: DeliverySeq,
cursor: DeliverySeq,
) -> ValidatedMarkerRecord {
use alloc::{vec, vec::Vec};
use crate::outcome::CandidatePhase;
use super::{
AdmissionOrder, OrderClaims, OrderHigh, OrderLedger, RecoverySequenceReserve,
SequenceClaims, SequenceLedger,
claim_frontier::{
BindingTerminalOwner, ClaimFrontiers, ClaimFrontiersRestore, FrontierBinding,
FrontierParticipant, ImmutableSequenceCandidate, MarkerProvenance, MarkerRecordRequest,
MovableOrderClaim, MovableSequenceClaim, OrderClaimFrontierRestore, OrderDirectOwner,
RetainedCausalRecord, RetainedCausalRecordKind, SequenceClaimFrontierRestore,
SequenceDirectOwner, SequenceProductRangesRestore, TerminalProductRangeRestore,
},
};
let identity_slot_limit = participant_id
.checked_add(1)
.expect("test participant must fit a half-open identity domain");
let exit_seq = marker_delivery_seq
.checked_add(1)
.expect("test marker must leave an exit-claim suffix");
let binding_epoch = match target_binding {
FrontierBinding::Bound(epoch) | FrontierBinding::Detached(epoch) => epoch,
};
let terminal_owner = BindingTerminalOwner {
participant_index: participant_id,
binding_epoch,
};
let (sequence_claims, sequence_movable, products, order_claims, order_movable) =
match target_binding {
FrontierBinding::Bound(_) => {
let terminal_seq = exit_seq
.checked_add(1)
.expect("test marker must leave a terminal-claim suffix");
let product_seq = terminal_seq
.checked_add(1)
.expect("test marker must leave a terminal-product suffix");
(
SequenceClaims::new(1, 1, 0, RecoverySequenceReserve::None),
vec![
MovableSequenceClaim {
delivery_seq: exit_seq,
owner: SequenceDirectOwner::MembershipExit {
participant_index: participant_id,
},
},
MovableSequenceClaim {
delivery_seq: terminal_seq,
owner: SequenceDirectOwner::BindingTerminal(terminal_owner),
},
],
SequenceProductRangesRestore {
live_times_terminal: vec![TerminalProductRangeRestore {
start: product_seq,
length: 1,
terminal: terminal_owner,
}],
live_times_replacement_terminal: None,
other_live_times_exit: vec![],
},
OrderClaims::new(1, 1, false, false)
.expect("bound test claims have no torn recovery pair"),
vec![
MovableOrderClaim {
transaction_order: 1,
owner: OrderDirectOwner::ActiveBindingTerminal(terminal_owner),
},
MovableOrderClaim {
transaction_order: 2,
owner: OrderDirectOwner::MembershipExit {
participant_index: participant_id,
},
},
],
)
}
FrontierBinding::Detached(_) => (
SequenceClaims::new(1, 0, 0, RecoverySequenceReserve::None),
vec![MovableSequenceClaim {
delivery_seq: exit_seq,
owner: SequenceDirectOwner::MembershipExit {
participant_index: participant_id,
},
}],
SequenceProductRangesRestore::default(),
OrderClaims::new(0, 1, false, false)
.expect("detached test claims have no torn recovery pair"),
vec![MovableOrderClaim {
transaction_order: 1,
owner: OrderDirectOwner::MembershipExit {
participant_index: participant_id,
},
}],
),
};
let sequence_ledger = SequenceLedger::try_new(marker_delivery_seq, sequence_claims)
.expect("test sequence frontier is within the numeric suffix");
let order_ledger = OrderLedger::try_new(OrderHigh::Allocated(0), order_claims)
.expect("test order frontier is within the numeric suffix");
let admission_order = AdmissionOrder::new(0, CandidatePhase::CompactionMarker, participant_id);
let mut prevalidated = ClaimFrontiers::prevalidate(
ClaimFrontiersRestore {
conversation_id,
active_identities: vec![FrontierParticipant::new(
participant_id,
cursor,
target_binding,
)],
identity_slot_limit,
retained_floor: u128::from(marker_delivery_seq),
retained_record_limit: 1,
retained_records: vec![RetainedCausalRecord {
delivery_seq: marker_delivery_seq,
admission_order,
kind: RetainedCausalRecordKind::CompactionMarker {
participant_index: participant_id,
provenance: MarkerProvenance::NonProductM,
},
}],
active_marker_anchors: vec![marker_delivery_seq],
historical_marker_deliveries: vec![],
historical_causal_facts: vec![],
sequence: SequenceClaimFrontierRestore {
movable_claims: sequence_movable,
immutable_candidates: Vec::<ImmutableSequenceCandidate>::new(),
products,
recovery: None,
},
order: OrderClaimFrontierRestore {
movable_claims: order_movable,
immutable_candidates: vec![],
recovery: None,
},
recovery_marker_delivery_seq: None,
},
sequence_ledger,
order_ledger,
)
.expect("complete test claim frontier must prevalidate");
let record = prevalidated
.take_marker_record(MarkerRecordRequest::planned(
participant_id,
marker_delivery_seq,
target_binding,
))
.expect("prevalidated test frontier retains its exact marker");
if cursor >= marker_delivery_seq {
record.delivered_for_test()
} else {
record
}
}
#[cfg(test)]
pub fn marker_delivery_for_test(
participant_id: ParticipantId,
binding_epoch: BindingEpoch,
marker_delivery_seq: DeliverySeq,
) -> Result<MarkerDelivery, super::storage::StorageRestoreError> {
let record = validated_marker_record_for_test(
1,
participant_id,
super::claim_frontier::FrontierBinding::Bound(binding_epoch),
marker_delivery_seq,
marker_delivery_seq.saturating_sub(1),
);
super::storage::MarkerDeliveryRestore {
participant_id,
binding_epoch,
marker_delivery_seq,
}
.restore_bound(1, record)
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct CursorProgressContinuous {
participant_id: ParticipantId,
binding_epoch: BindingEpoch,
through_seq: DeliverySeq,
}
impl CursorProgressContinuous {
#[cfg(test)]
#[must_use]
pub(crate) const fn new(
participant_id: ParticipantId,
binding_epoch: BindingEpoch,
through_seq: DeliverySeq,
) -> Self {
Self {
participant_id,
binding_epoch,
through_seq,
}
}
#[must_use]
pub const fn participant_id(self) -> ParticipantId {
self.participant_id
}
#[must_use]
pub const fn binding_epoch(self) -> BindingEpoch {
self.binding_epoch
}
#[must_use]
pub const fn through_seq(self) -> DeliverySeq {
self.through_seq
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct CursorProgressMarker {
conversation_id: ConversationId,
participant_id: ParticipantId,
binding_epoch: BindingEpoch,
through_seq: DeliverySeq,
marker_delivery_seq: DeliverySeq,
}
impl CursorProgressMarker {
#[must_use]
pub const fn conversation_id(self) -> ConversationId {
self.conversation_id
}
#[must_use]
pub const fn participant_id(self) -> ParticipantId {
self.participant_id
}
#[must_use]
pub const fn binding_epoch(self) -> BindingEpoch {
self.binding_epoch
}
#[must_use]
pub const fn through_seq(self) -> DeliverySeq {
self.through_seq
}
#[must_use]
pub const fn marker_delivery_seq(self) -> DeliverySeq {
self.marker_delivery_seq
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ParticipantCursorProgress {
Continuous(CursorProgressContinuous),
Marker(CursorProgressMarker),
}
impl ParticipantCursorProgress {
#[cfg(test)]
#[must_use]
pub(crate) const fn continuous(
participant_id: ParticipantId,
binding_epoch: BindingEpoch,
through_seq: DeliverySeq,
) -> Self {
Self::Continuous(CursorProgressContinuous::new(
participant_id,
binding_epoch,
through_seq,
))
}
pub(super) fn restore_continuous(
authority: OrdinaryBindingAuthority,
participant_id: ParticipantId,
binding_epoch: BindingEpoch,
through_seq: DeliverySeq,
) -> Option<Self> {
if authority.binding.participant_id != participant_id
|| authority.binding.binding_epoch != binding_epoch
|| authority.through_seq != through_seq
{
return None;
}
Some(Self::Continuous(CursorProgressContinuous {
participant_id,
binding_epoch,
through_seq,
}))
}
#[must_use]
pub const fn participant_id(self) -> ParticipantId {
match self {
Self::Continuous(value) => value.participant_id,
Self::Marker(value) => value.participant_id,
}
}
#[must_use]
pub const fn binding_epoch(self) -> BindingEpoch {
match self {
Self::Continuous(value) => value.binding_epoch,
Self::Marker(value) => value.binding_epoch,
}
}
#[must_use]
pub const fn through_seq(self) -> DeliverySeq {
match self {
Self::Continuous(value) => value.through_seq,
Self::Marker(value) => value.through_seq,
}
}
#[must_use]
pub const fn marker_delivery_seq(self) -> Option<DeliverySeq> {
match self {
Self::Continuous(_) => None,
Self::Marker(value) => Some(value.marker_delivery_seq),
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct DetachedCredentialRecovery {
conversation_id: ConversationId,
participant_id: ParticipantId,
marker_delivery_seq: DeliverySeq,
prior_binding_epoch: BindingEpoch,
}
impl DetachedCredentialRecovery {
#[must_use]
pub const fn conversation_id(self) -> ConversationId {
self.conversation_id
}
#[must_use]
pub const fn participant_id(self) -> ParticipantId {
self.participant_id
}
#[must_use]
pub const fn marker_delivery_seq(self) -> DeliverySeq {
self.marker_delivery_seq
}
#[must_use]
pub const fn prior_binding_epoch(self) -> BindingEpoch {
self.prior_binding_epoch
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct DetachedMarkerRelease {
participant_id: ParticipantId,
marker_delivery_seq: DeliverySeq,
last_dead_binding_epoch: BindingEpoch,
}
impl DetachedMarkerRelease {
#[must_use]
pub const fn participant_id(self) -> ParticipantId {
self.participant_id
}
#[must_use]
pub const fn marker_delivery_seq(self) -> DeliverySeq {
self.marker_delivery_seq
}
#[must_use]
pub const fn last_dead_binding_epoch(self) -> BindingEpoch {
self.last_dead_binding_epoch
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct DetachedCursorRelease {
participant_id: ParticipantId,
last_dead_binding_epoch: BindingEpoch,
}
impl DetachedCursorRelease {
#[must_use]
pub const fn participant_id(self) -> ParticipantId {
self.participant_id
}
#[must_use]
pub const fn last_dead_binding_epoch(self) -> BindingEpoch {
self.last_dead_binding_epoch
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum StoredEdge {
ObserverProjection(ObserverProjection),
PhysicalCompaction(PhysicalCompaction),
MarkerDelivery(MarkerDelivery),
ParticipantCursorProgress(ParticipantCursorProgress),
DetachedCredentialRecovery(DetachedCredentialRecovery),
DetachedMarkerRelease(DetachedMarkerRelease),
DetachedCursorRelease(DetachedCursorRelease),
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ClosureState {
Clear,
Owed {
debt: ClosureDebt,
edge: StoredEdge,
},
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct OrdinaryDetachedAttachAdmission {
_private: (),
}
impl ClosureState {
pub const fn ordinary_detached_attach_admission(
self,
) -> Result<OrdinaryDetachedAttachAdmission, Self> {
match self {
Self::Clear => Ok(OrdinaryDetachedAttachAdmission { _private: () }),
Self::Owed { .. } => Err(self),
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct DebtCompletion(ClosureState);
impl DebtCompletion {
#[must_use]
pub const fn clear() -> Self {
Self(ClosureState::Clear)
}
#[must_use]
pub const fn observer_projection(debt: ClosureDebt, edge: ObserverProjection) -> Self {
Self(ClosureState::Owed {
debt,
edge: StoredEdge::ObserverProjection(edge),
})
}
#[must_use]
pub const fn physical_compaction(debt: ClosureDebt, edge: PhysicalCompaction) -> Self {
Self(ClosureState::Owed {
debt,
edge: StoredEdge::PhysicalCompaction(edge),
})
}
#[must_use]
pub const fn into_state(self) -> ClosureState {
self.0
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct OrdinaryBindingAuthority {
binding: ActiveBinding,
through_seq: DeliverySeq,
}
impl OrdinaryBindingAuthority {
pub(crate) const fn new(binding: ActiveBinding, through_seq: DeliverySeq) -> Self {
Self {
binding,
through_seq,
}
}
#[must_use]
pub const fn binding(self) -> ActiveBinding {
self.binding
}
#[must_use]
pub const fn through_seq(self) -> DeliverySeq {
self.through_seq
}
#[allow(
dead_code,
reason = "the crate-owned participant-ack operation advances this sealed authority"
)]
pub(crate) fn cursor_progressed(self, event: Event) -> Result<Self, Self> {
let EventKind::CursorProgressed {
participant_id,
binding_epoch,
progress:
CursorProgressEvent::Normal {
previous_cursor,
through_seq,
},
..
} = event.0
else {
return Err(self);
};
if participant_id != self.binding.participant_id
|| binding_epoch != self.binding.binding_epoch
|| previous_cursor != self.through_seq
{
return Err(self);
}
Ok(Self {
through_seq,
..self
})
}
pub(crate) fn binding_fate(
self,
terminal: CommittedDiedTerminal,
resulting_floor: DeliverySeq,
) -> Result<OrdinaryBindingFate, Self> {
if terminal.participant_id() != self.binding.participant_id
|| terminal.conversation_id() != self.binding.conversation_id
|| terminal.binding_epoch() != self.binding.binding_epoch
{
return Err(self);
}
Ok(OrdinaryBindingFate {
conversation_id: self.binding.conversation_id,
through_seq: self.through_seq,
resulting_floor,
release: DetachedCursorRelease {
participant_id: self.binding.participant_id,
last_dead_binding_epoch: self.binding.binding_epoch,
},
})
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct OrdinaryBindingFate {
conversation_id: ConversationId,
through_seq: DeliverySeq,
resulting_floor: DeliverySeq,
release: DetachedCursorRelease,
}
impl OrdinaryBindingFate {
#[must_use]
pub const fn conversation_id(self) -> ConversationId {
self.conversation_id
}
#[must_use]
pub const fn through_seq(self) -> DeliverySeq {
self.through_seq
}
#[must_use]
pub const fn participant_id(self) -> ParticipantId {
self.release.participant_id
}
#[must_use]
pub const fn last_dead_binding_epoch(self) -> BindingEpoch {
self.release.last_dead_binding_epoch
}
#[must_use]
pub const fn resulting_floor(self) -> DeliverySeq {
self.resulting_floor
}
#[must_use]
pub const fn observer_progress_projection(&self) -> ObserverProgressProjection {
ObserverProgressProjection::new(self.conversation_id, self.resulting_floor)
}
#[must_use]
pub const fn into_direct_state(self, debt: ClosureDebt) -> ClosureState {
owed(debt, StoredEdge::DetachedCursorRelease(self.release))
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct FencedAttachCommit {
conversation_id: ConversationId,
participant_id: ParticipantId,
marker_delivery_seq: DeliverySeq,
prior_binding_epoch: BindingEpoch,
new_binding_epoch: BindingEpoch,
next_state: ClosureState,
}
impl FencedAttachCommit {
#[must_use]
pub const fn conversation_id(self) -> ConversationId {
self.conversation_id
}
#[must_use]
pub const fn participant_id(self) -> ParticipantId {
self.participant_id
}
#[must_use]
pub const fn marker_delivery_seq(self) -> DeliverySeq {
self.marker_delivery_seq
}
#[must_use]
pub const fn prior_binding_epoch(self) -> BindingEpoch {
self.prior_binding_epoch
}
#[must_use]
pub const fn new_binding_epoch(self) -> BindingEpoch {
self.new_binding_epoch
}
#[must_use]
pub const fn next_state(self) -> ClosureState {
self.next_state
}
pub fn recovered_binding_fate(
self,
event: Event,
) -> Result<RecoveredBindingFate, ClosureState> {
let EventKind::BindingFateObserved {
participant_id,
binding_epoch,
resulting_floor,
} = event.0
else {
return Err(self.next_state);
};
if participant_id != self.participant_id || binding_epoch != self.new_binding_epoch {
return Err(self.next_state);
}
let ClosureState::Owed { debt, edge } = self.next_state else {
return Err(self.next_state);
};
let predecessor = match edge {
StoredEdge::ObserverProjection(value) => {
RecoveredStorageEdge::ObserverProjection(value)
}
StoredEdge::PhysicalCompaction(value) => {
RecoveredStorageEdge::PhysicalCompaction(value)
}
_ => return Err(self.next_state),
};
Ok(RecoveredBindingFate {
conversation_id: self.conversation_id,
predecessor_debt: debt,
predecessor,
resulting_floor,
release: DetachedCursorRelease {
participant_id,
last_dead_binding_epoch: binding_epoch,
},
})
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum RecoveredStorageEdge {
ObserverProjection(ObserverProjection),
PhysicalCompaction(PhysicalCompaction),
}
impl RecoveredStorageEdge {
const fn into_stored_edge(self) -> StoredEdge {
match self {
Self::ObserverProjection(value) => StoredEdge::ObserverProjection(value),
Self::PhysicalCompaction(value) => StoredEdge::PhysicalCompaction(value),
}
}
}
#[derive(Debug, PartialEq, Eq)]
pub struct RecoveredBindingFate {
conversation_id: ConversationId,
predecessor_debt: ClosureDebt,
predecessor: RecoveredStorageEdge,
resulting_floor: DeliverySeq,
release: DetachedCursorRelease,
}
impl RecoveredBindingFate {
#[must_use]
pub const fn conversation_id(&self) -> ConversationId {
self.conversation_id
}
#[must_use]
pub const fn predecessor_state(&self) -> ClosureState {
owed(self.predecessor_debt, self.predecessor.into_stored_edge())
}
#[must_use]
pub const fn participant_id(&self) -> ParticipantId {
self.release.participant_id
}
#[must_use]
pub const fn last_dead_binding_epoch(&self) -> BindingEpoch {
self.release.last_dead_binding_epoch
}
#[must_use]
pub const fn resulting_floor(&self) -> DeliverySeq {
self.resulting_floor
}
#[must_use]
pub const fn observer_progress_projection(&self) -> ObserverProgressProjection {
ObserverProgressProjection::new(self.conversation_id, self.resulting_floor)
}
}
#[derive(Debug, PartialEq, Eq)]
pub struct PendingRecoveredCursorRelease {
debt: ClosureDebt,
predecessor: RecoveredStorageEdge,
release: DetachedCursorRelease,
}
impl PendingRecoveredCursorRelease {
#[must_use]
pub const fn current_state(&self) -> ClosureState {
owed(self.debt, self.predecessor.into_stored_edge())
}
#[must_use]
pub const fn participant_id(&self) -> ParticipantId {
self.release.participant_id
}
#[must_use]
pub const fn last_dead_binding_epoch(&self) -> BindingEpoch {
self.release.last_dead_binding_epoch
}
}
#[derive(Debug, PartialEq, Eq)]
pub struct RecoveredCursorRelease {
debt: ClosureDebt,
release: DetachedCursorRelease,
}
impl RecoveredCursorRelease {
#[must_use]
pub const fn debt(&self) -> ClosureDebt {
self.debt
}
#[must_use]
pub const fn edge(&self) -> DetachedCursorRelease {
self.release
}
#[must_use]
pub const fn into_state(self) -> ClosureState {
owed(self.debt, StoredEdge::DetachedCursorRelease(self.release))
}
}
#[derive(Debug, PartialEq, Eq)]
pub enum RecoveredBindingFateTransition {
PendingStorage(PendingRecoveredCursorRelease),
DetachedCursorRelease(RecoveredCursorRelease),
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum SuccessorUse {
ObserverCompletion,
ObserverMarkerAppend,
ObserverLeave,
PhysicalCompletion,
PhysicalCover,
CursorGreaterAck,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ProjectionCompactionSuccessor {
predecessor: StoredEdge,
event: Event,
use_kind: SuccessorUse,
state: ClosureState,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum CursorFateSuccessor {
DetachedCredentialRecovery(DetachedCredentialRecovery),
DetachedCursorRelease(DetachedCursorRelease),
}
impl CursorFateSuccessor {
#[must_use]
pub const fn into_stored_edge(self) -> StoredEdge {
match self {
Self::DetachedCredentialRecovery(value) => {
StoredEdge::DetachedCredentialRecovery(value)
}
Self::DetachedCursorRelease(value) => StoredEdge::DetachedCursorRelease(value),
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum DetachedAttachRefusal {
RecoveryFence,
MarkerNotDelivered,
MarkerMismatch,
DeliveredMarkerAwaitingAck,
EpisodeChurnLimit,
StaleAuthority,
NoBinding,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum DetachedClaimTarget {
CredentialRecovery {
marker_delivery_seq: DeliverySeq,
binding_epoch: BindingEpoch,
},
MarkerRelease {
marker_delivery_seq: DeliverySeq,
binding_epoch: BindingEpoch,
},
CursorRelease {
binding_epoch: BindingEpoch,
},
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct KClaimBackedDetachedLeave {
participant_id: ParticipantId,
target: DetachedClaimTarget,
actual_charge: ResourceVector,
}
impl KClaimBackedDetachedLeave {
#[must_use]
pub const fn participant_id(self) -> ParticipantId {
self.participant_id
}
#[must_use]
pub const fn actual_charge(self) -> ResourceVector {
self.actual_charge
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Event(EventKind);
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum EventKind {
ProjectionCompleted {
through_seq: DeliverySeq,
},
CompactionCompleted {
from_floor: DeliverySeq,
through_seq: DeliverySeq,
resulting_floor: DeliverySeq,
},
MarkerAppended {
marker_delivery_seq: DeliverySeq,
resulting_projection_through: DeliverySeq,
},
MarkerDelivered {
participant_id: ParticipantId,
binding_epoch: BindingEpoch,
marker_delivery_seq: DeliverySeq,
},
CursorProgressed {
participant_index: ParticipantIndex,
participant_id: ParticipantId,
binding_epoch: BindingEpoch,
progress: CursorProgressEvent,
resulting_floor: DeliverySeq,
},
BindingFateObserved {
participant_id: ParticipantId,
binding_epoch: BindingEpoch,
resulting_floor: DeliverySeq,
},
LeaveCommitted {
participant_id: ParticipantId,
authority: LeaveAuthority,
resulting_floor: DeliverySeq,
},
FencedRecoveryCommitted {
participant_id: ParticipantId,
marker_delivery_seq: DeliverySeq,
prior_binding_epoch: BindingEpoch,
new_binding_epoch: BindingEpoch,
resulting_floor: DeliverySeq,
},
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum CursorProgressEvent {
Normal {
previous_cursor: DeliverySeq,
through_seq: DeliverySeq,
},
Marker {
marker_delivery_seq: DeliverySeq,
},
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum LeaveAuthority {
Live(BindingEpoch),
Detached,
}
impl Event {
#[must_use]
pub const fn projection_completed(through_seq: DeliverySeq) -> Self {
Self(EventKind::ProjectionCompleted { through_seq })
}
#[must_use]
pub const fn compaction_completed(
from_floor: DeliverySeq,
through_seq: DeliverySeq,
resulting_floor: DeliverySeq,
) -> Option<Self> {
if from_floor <= through_seq && resulting_floor > through_seq {
Some(Self(EventKind::CompactionCompleted {
from_floor,
through_seq,
resulting_floor,
}))
} else {
None
}
}
#[must_use]
pub const fn marker_appended(
marker_delivery_seq: DeliverySeq,
resulting_projection_through: DeliverySeq,
) -> Self {
Self(EventKind::MarkerAppended {
marker_delivery_seq,
resulting_projection_through,
})
}
#[must_use]
pub const fn marker_delivered(
participant_id: ParticipantId,
binding_epoch: BindingEpoch,
marker_delivery_seq: DeliverySeq,
) -> Self {
Self(EventKind::MarkerDelivered {
participant_id,
binding_epoch,
marker_delivery_seq,
})
}
#[must_use]
pub const fn cursor_progressed(
participant_id: ParticipantId,
binding_epoch: BindingEpoch,
previous_cursor: DeliverySeq,
through_seq: DeliverySeq,
resulting_floor: DeliverySeq,
) -> Option<Self> {
if through_seq > previous_cursor {
Some(Self(EventKind::CursorProgressed {
participant_index: participant_id,
participant_id,
binding_epoch,
progress: CursorProgressEvent::Normal {
previous_cursor,
through_seq,
},
resulting_floor,
}))
} else {
None
}
}
#[must_use]
pub const fn marker_acknowledged(
participant_id: ParticipantId,
binding_epoch: BindingEpoch,
marker_delivery_seq: DeliverySeq,
resulting_floor: DeliverySeq,
) -> Self {
Self(EventKind::CursorProgressed {
participant_index: participant_id,
participant_id,
binding_epoch,
progress: CursorProgressEvent::Marker {
marker_delivery_seq,
},
resulting_floor,
})
}
#[must_use]
pub const fn binding_fate_observed(
participant_id: ParticipantId,
binding_epoch: BindingEpoch,
resulting_floor: DeliverySeq,
) -> Self {
Self(EventKind::BindingFateObserved {
participant_id,
binding_epoch,
resulting_floor,
})
}
#[must_use]
pub const fn live_leave_committed(
participant_id: ParticipantId,
binding_epoch: BindingEpoch,
resulting_floor: DeliverySeq,
) -> Self {
Self(EventKind::LeaveCommitted {
participant_id,
authority: LeaveAuthority::Live(binding_epoch),
resulting_floor,
})
}
#[must_use]
pub const fn detached_leave_committed(
participant_id: ParticipantId,
resulting_floor: DeliverySeq,
) -> Self {
Self(EventKind::LeaveCommitted {
participant_id,
authority: LeaveAuthority::Detached,
resulting_floor,
})
}
#[must_use]
pub const fn fenced_recovery_committed(
participant_id: ParticipantId,
marker_delivery_seq: DeliverySeq,
prior_binding_epoch: BindingEpoch,
new_binding_epoch: BindingEpoch,
resulting_floor: DeliverySeq,
) -> Self {
Self(EventKind::FencedRecoveryCommitted {
participant_id,
marker_delivery_seq,
prior_binding_epoch,
new_binding_epoch,
resulting_floor,
})
}
}
impl ObserverProjection {
#[must_use]
#[allow(
dead_code,
reason = "the crate-owned binding-fate operation invokes this sealed OP transition"
)]
pub const fn apply_ordinary_binding_fate(
self,
resulting_debt: ClosureDebt,
authority: OrdinaryBindingFate,
) -> PendingRecoveredCursorRelease {
PendingRecoveredCursorRelease {
debt: resulting_debt,
predecessor: RecoveredStorageEdge::ObserverProjection(self),
release: authority.release,
}
}
pub fn apply_recovered_binding_fate(
self,
debt: ClosureDebt,
resulting_debt: ClosureDebt,
authority: RecoveredBindingFate,
) -> Result<RecoveredBindingFateTransition, RecoveredBindingFate> {
if authority.predecessor != RecoveredStorageEdge::ObserverProjection(self)
|| authority.predecessor_debt != debt
{
return Err(authority);
}
Ok(RecoveredBindingFateTransition::PendingStorage(
PendingRecoveredCursorRelease {
debt: resulting_debt,
predecessor: RecoveredStorageEdge::ObserverProjection(self),
release: authority.release,
},
))
}
pub fn complete_after_recovered_binding_fate(
self,
event: Event,
resulting_debt: Option<ClosureDebt>,
pending: PendingRecoveredCursorRelease,
) -> Result<ClosureState, PendingRecoveredCursorRelease> {
self.complete_after_binding_fate(event, resulting_debt, pending)
}
pub fn complete_after_ordinary_binding_fate(
self,
event: Event,
resulting_debt: Option<ClosureDebt>,
pending: PendingRecoveredCursorRelease,
) -> Result<ClosureState, PendingRecoveredCursorRelease> {
self.complete_after_binding_fate(event, resulting_debt, pending)
}
pub(crate) fn complete_after_binding_fate(
self,
event: Event,
resulting_debt: Option<ClosureDebt>,
pending: PendingRecoveredCursorRelease,
) -> Result<ClosureState, PendingRecoveredCursorRelease> {
if pending.predecessor != RecoveredStorageEdge::ObserverProjection(self)
|| projection_completion_boundary(self, event).is_none()
{
return Err(pending);
}
Ok(preserve_or_clear(
resulting_debt,
StoredEdge::DetachedCursorRelease(pending.release),
))
}
#[must_use]
pub const fn clear_after_completion(
&self,
event: &Event,
) -> Option<ProjectionCompactionSuccessor> {
if projection_completion_boundary(*self, *event).is_some() {
Some(ProjectionCompactionSuccessor {
predecessor: StoredEdge::ObserverProjection(*self),
event: *event,
use_kind: SuccessorUse::ObserverCompletion,
state: ClosureState::Clear,
})
} else {
None
}
}
#[must_use]
pub const fn strict_after_completion(
&self,
event: &Event,
debt: ClosureDebt,
edge: StoredEdge,
successor_boundary: DeliverySeq,
) -> Option<ProjectionCompactionSuccessor> {
let Some(completed_through) = projection_completion_boundary(*self, *event) else {
return None;
};
if successor_boundary <= completed_through
|| !strict_edge_matches_boundary(edge, successor_boundary)
{
return None;
}
Some(ProjectionCompactionSuccessor {
predecessor: StoredEdge::ObserverProjection(*self),
event: *event,
use_kind: SuccessorUse::ObserverCompletion,
state: ClosureState::Owed { debt, edge },
})
}
pub fn complete(
self,
debt: ClosureDebt,
event: Event,
successor: ProjectionCompactionSuccessor,
) -> Result<ClosureState, ClosureState> {
let original = owed(debt, StoredEdge::ObserverProjection(self));
if successor.predecessor == StoredEdge::ObserverProjection(self)
&& successor.event == event
&& successor.use_kind == SuccessorUse::ObserverCompletion
&& projection_completion_boundary(self, event).is_some()
{
Ok(successor.state)
} else {
Err(original)
}
}
#[must_use]
pub const fn later_projection_after_marker(
&self,
event: &Event,
debt: ClosureDebt,
successor: Self,
) -> Option<ProjectionCompactionSuccessor> {
let EventKind::MarkerAppended {
marker_delivery_seq,
resulting_projection_through,
} = event.0
else {
return None;
};
if marker_delivery_seq <= self.through_seq
|| resulting_projection_through < marker_delivery_seq
|| successor.through_seq != resulting_projection_through
{
return None;
}
Some(ProjectionCompactionSuccessor {
predecessor: StoredEdge::ObserverProjection(*self),
event: *event,
use_kind: SuccessorUse::ObserverMarkerAppend,
state: owed(debt, StoredEdge::ObserverProjection(successor)),
})
}
pub fn marker_appended(
self,
debt: ClosureDebt,
event: Event,
successor: ProjectionCompactionSuccessor,
) -> Result<ClosureState, ClosureState> {
let original = owed(debt, StoredEdge::ObserverProjection(self));
if successor.predecessor == StoredEdge::ObserverProjection(self)
&& successor.event == event
&& successor.use_kind == SuccessorUse::ObserverMarkerAppend
{
Ok(successor.state)
} else {
Err(original)
}
}
#[must_use]
pub const fn later_projection_after_leave(
&self,
event: &Event,
debt: ClosureDebt,
successor: Self,
) -> Option<ProjectionCompactionSuccessor> {
let EventKind::LeaveCommitted {
resulting_floor, ..
} = event.0
else {
return None;
};
if successor.through_seq <= self.through_seq || successor.through_seq < resulting_floor {
return None;
}
Some(ProjectionCompactionSuccessor {
predecessor: StoredEdge::ObserverProjection(*self),
event: *event,
use_kind: SuccessorUse::ObserverLeave,
state: owed(debt, StoredEdge::ObserverProjection(successor)),
})
}
pub fn leave_with_later_projection(
self,
debt: ClosureDebt,
event: Event,
successor: ProjectionCompactionSuccessor,
) -> Result<ClosureState, ClosureState> {
let original = owed(debt, StoredEdge::ObserverProjection(self));
if successor.predecessor == StoredEdge::ObserverProjection(self)
&& successor.event == event
&& successor.use_kind == SuccessorUse::ObserverLeave
{
Ok(successor.state)
} else {
Err(original)
}
}
pub const fn independent_event(
self,
debt: ClosureDebt,
event: Event,
resulting_debt: Option<ClosureDebt>,
) -> Result<ClosureState, ClosureState> {
let original = owed(debt, StoredEdge::ObserverProjection(self));
if !matches!(
event.0,
EventKind::CursorProgressed { .. }
| EventKind::BindingFateObserved { .. }
| EventKind::LeaveCommitted { .. }
) {
return Err(original);
}
Ok(preserve_or_clear(
resulting_debt,
StoredEdge::ObserverProjection(self),
))
}
pub const fn charged_binding_change(
self,
debt: ClosureDebt,
episode_churn_used: u64,
delta_cycles: u64,
episode_churn_limit: u64,
resulting_debt: Option<ClosureDebt>,
) -> Result<ClosureState, (ClosureState, DetachedAttachRefusal)> {
let original = owed(debt, StoredEdge::ObserverProjection(self));
if !charged_churn_fits(episode_churn_used, delta_cycles, episode_churn_limit) {
return Err((original, DetachedAttachRefusal::EpisodeChurnLimit));
}
Ok(preserve_or_clear(
resulting_debt,
StoredEdge::ObserverProjection(self),
))
}
}
impl PhysicalCompaction {
#[must_use]
#[allow(
dead_code,
reason = "the crate-owned binding-fate replay boundary invokes this sealed PC transition"
)]
pub(crate) const fn apply_ordinary_binding_fate(
self,
resulting_debt: ClosureDebt,
authority: OrdinaryBindingFate,
) -> RecoveredBindingFateTransition {
if authority.resulting_floor > self.through_seq {
RecoveredBindingFateTransition::DetachedCursorRelease(RecoveredCursorRelease {
debt: resulting_debt,
release: authority.release,
})
} else {
RecoveredBindingFateTransition::PendingStorage(PendingRecoveredCursorRelease {
debt: resulting_debt,
predecessor: RecoveredStorageEdge::PhysicalCompaction(self),
release: authority.release,
})
}
}
pub fn apply_recovered_binding_fate(
self,
debt: ClosureDebt,
resulting_debt: ClosureDebt,
authority: RecoveredBindingFate,
) -> Result<RecoveredBindingFateTransition, RecoveredBindingFate> {
if authority.predecessor != RecoveredStorageEdge::PhysicalCompaction(self)
|| authority.predecessor_debt != debt
{
return Err(authority);
}
if authority.resulting_floor > self.through_seq {
Ok(RecoveredBindingFateTransition::DetachedCursorRelease(
RecoveredCursorRelease {
debt: resulting_debt,
release: authority.release,
},
))
} else {
Ok(RecoveredBindingFateTransition::PendingStorage(
PendingRecoveredCursorRelease {
debt: resulting_debt,
predecessor: RecoveredStorageEdge::PhysicalCompaction(self),
release: authority.release,
},
))
}
}
pub fn complete_after_recovered_binding_fate(
self,
event: Event,
resulting_debt: Option<ClosureDebt>,
pending: PendingRecoveredCursorRelease,
) -> Result<ClosureState, PendingRecoveredCursorRelease> {
self.complete_after_binding_fate(event, resulting_debt, pending)
}
pub(crate) fn complete_after_binding_fate(
self,
event: Event,
resulting_debt: Option<ClosureDebt>,
pending: PendingRecoveredCursorRelease,
) -> Result<ClosureState, PendingRecoveredCursorRelease> {
if pending.predecessor != RecoveredStorageEdge::PhysicalCompaction(self)
|| physical_completion_floor(self, event).is_none()
{
return Err(pending);
}
Ok(preserve_or_clear(
resulting_debt,
StoredEdge::DetachedCursorRelease(pending.release),
))
}
#[must_use]
pub const fn clear_after_completion(
&self,
event: &Event,
) -> Option<ProjectionCompactionSuccessor> {
if physical_completion_floor(*self, *event).is_some() {
Some(ProjectionCompactionSuccessor {
predecessor: StoredEdge::PhysicalCompaction(*self),
event: *event,
use_kind: SuccessorUse::PhysicalCompletion,
state: ClosureState::Clear,
})
} else {
None
}
}
#[must_use]
pub const fn strict_after_completion(
&self,
event: &Event,
debt: ClosureDebt,
edge: StoredEdge,
successor_boundary: DeliverySeq,
) -> Option<ProjectionCompactionSuccessor> {
let Some(resulting_floor) = physical_completion_floor(*self, *event) else {
return None;
};
if successor_boundary < resulting_floor
|| !strict_edge_matches_boundary(edge, successor_boundary)
{
return None;
}
Some(ProjectionCompactionSuccessor {
predecessor: StoredEdge::PhysicalCompaction(*self),
event: *event,
use_kind: SuccessorUse::PhysicalCompletion,
state: ClosureState::Owed { debt, edge },
})
}
pub fn complete(
self,
debt: ClosureDebt,
event: Event,
successor: ProjectionCompactionSuccessor,
) -> Result<ClosureState, ClosureState> {
let original = owed(debt, StoredEdge::PhysicalCompaction(self));
if successor.predecessor == StoredEdge::PhysicalCompaction(self)
&& successor.event == event
&& successor.use_kind == SuccessorUse::PhysicalCompletion
&& physical_completion_floor(self, event).is_some()
{
Ok(successor.state)
} else {
Err(original)
}
}
pub const fn marker_appended(
self,
debt: ClosureDebt,
event: Event,
) -> Result<ClosureState, ClosureState> {
let original = owed(debt, StoredEdge::PhysicalCompaction(self));
let EventKind::MarkerAppended {
marker_delivery_seq,
resulting_projection_through,
} = event.0
else {
return Err(original);
};
if marker_delivery_seq <= self.through_seq
|| resulting_projection_through < marker_delivery_seq
{
return Err(original);
}
Ok(original)
}
pub const fn preserve_progress(
self,
debt: ClosureDebt,
event: Event,
resulting_debt: ClosureDebt,
) -> Result<ClosureState, ClosureState> {
let original = owed(debt, StoredEdge::PhysicalCompaction(self));
let Some(resulting_floor) = progress_event_floor(event) else {
return Err(original);
};
if resulting_floor > self.through_seq {
return Err(original);
}
Ok(owed(resulting_debt, StoredEdge::PhysicalCompaction(self)))
}
#[must_use]
pub const fn clear_after_progress(
&self,
event: &Event,
) -> Option<ProjectionCompactionSuccessor> {
let Some(resulting_floor) = progress_event_floor(*event) else {
return None;
};
if resulting_floor <= self.through_seq {
return None;
}
Some(ProjectionCompactionSuccessor {
predecessor: StoredEdge::PhysicalCompaction(*self),
event: *event,
use_kind: SuccessorUse::PhysicalCover,
state: ClosureState::Clear,
})
}
#[must_use]
pub const fn strict_after_progress(
&self,
event: &Event,
debt: ClosureDebt,
edge: StoredEdge,
successor_boundary: DeliverySeq,
) -> Option<ProjectionCompactionSuccessor> {
let Some(resulting_floor) = progress_event_floor(*event) else {
return None;
};
if resulting_floor <= self.through_seq
|| successor_boundary < resulting_floor
|| !strict_edge_matches_boundary(edge, successor_boundary)
{
return None;
}
Some(ProjectionCompactionSuccessor {
predecessor: StoredEdge::PhysicalCompaction(*self),
event: *event,
use_kind: SuccessorUse::PhysicalCover,
state: ClosureState::Owed { debt, edge },
})
}
pub fn covered_by_progress(
self,
debt: ClosureDebt,
event: Event,
successor: ProjectionCompactionSuccessor,
) -> Result<ClosureState, ClosureState> {
let original = owed(debt, StoredEdge::PhysicalCompaction(self));
if successor.predecessor == StoredEdge::PhysicalCompaction(self)
&& successor.event == event
&& successor.use_kind == SuccessorUse::PhysicalCover
{
Ok(successor.state)
} else {
Err(original)
}
}
#[must_use]
pub const fn unchanged(self, debt: ClosureDebt) -> ClosureState {
owed(debt, StoredEdge::PhysicalCompaction(self))
}
pub const fn charged_binding_change_preserving(
self,
debt: ClosureDebt,
episode_churn_used: u64,
delta_cycles: u64,
episode_churn_limit: u64,
resulting_floor: DeliverySeq,
resulting_debt: ClosureDebt,
) -> Result<ClosureState, (ClosureState, DetachedAttachRefusal)> {
let original = owed(debt, StoredEdge::PhysicalCompaction(self));
if !charged_churn_fits(episode_churn_used, delta_cycles, episode_churn_limit) {
return Err((original, DetachedAttachRefusal::EpisodeChurnLimit));
}
if resulting_floor > self.through_seq {
return Err((original, DetachedAttachRefusal::StaleAuthority));
}
Ok(owed(resulting_debt, StoredEdge::PhysicalCompaction(self)))
}
#[allow(clippy::too_many_arguments)]
pub const fn charged_binding_change_covering(
self,
debt: ClosureDebt,
episode_churn_used: u64,
delta_cycles: u64,
episode_churn_limit: u64,
resulting_floor: DeliverySeq,
resulting_debt: ClosureDebt,
edge: StoredEdge,
successor_boundary: DeliverySeq,
) -> Result<ClosureState, (ClosureState, DetachedAttachRefusal)> {
let original = owed(debt, StoredEdge::PhysicalCompaction(self));
if !charged_churn_fits(episode_churn_used, delta_cycles, episode_churn_limit) {
return Err((original, DetachedAttachRefusal::EpisodeChurnLimit));
}
if resulting_floor <= self.through_seq
|| successor_boundary < resulting_floor
|| !strict_edge_matches_boundary(edge, successor_boundary)
{
return Err((original, DetachedAttachRefusal::StaleAuthority));
}
Ok(owed(resulting_debt, edge))
}
}
impl MarkerDelivery {
pub fn delivered_progress(self, event: Event) -> Result<ParticipantCursorProgress, Self> {
let EventKind::MarkerDelivered {
participant_id,
binding_epoch,
marker_delivery_seq,
} = event.0
else {
return Err(self);
};
if participant_id != self.participant_id
|| binding_epoch != self.binding_epoch
|| marker_delivery_seq != self.marker_delivery_seq
{
return Err(self);
}
Ok(ParticipantCursorProgress::Marker(CursorProgressMarker {
conversation_id: self.conversation_id,
participant_id,
binding_epoch,
through_seq: marker_delivery_seq,
marker_delivery_seq,
}))
}
pub fn delivered(self, debt: ClosureDebt, event: Event) -> Result<ClosureState, ClosureState> {
let original = owed(debt, StoredEdge::MarkerDelivery(self));
let Ok(progress) = self.delivered_progress(event) else {
return Err(original);
};
Ok(owed(debt, StoredEdge::ParticipantCursorProgress(progress)))
}
pub const fn lower_progress(
self,
debt: ClosureDebt,
event: Event,
resulting_debt: Option<ClosureDebt>,
) -> Result<ClosureState, ClosureState> {
let original = owed(debt, StoredEdge::MarkerDelivery(self));
let is_lower = match event.0 {
EventKind::CursorProgressed {
progress: CursorProgressEvent::Normal { through_seq, .. },
..
}
| EventKind::ProjectionCompleted { through_seq } => {
through_seq < self.marker_delivery_seq
}
EventKind::CompactionCompleted {
through_seq,
resulting_floor,
..
} => {
through_seq < self.marker_delivery_seq
&& resulting_floor <= self.marker_delivery_seq
}
_ => false,
};
if !is_lower {
return Err(original);
}
Ok(preserve_or_clear(
resulting_debt,
StoredEdge::MarkerDelivery(self),
))
}
pub fn binding_fate(
self,
debt: ClosureDebt,
event: Event,
) -> Result<ClosureState, ClosureState> {
let original = owed(debt, StoredEdge::MarkerDelivery(self));
let EventKind::BindingFateObserved {
participant_id,
binding_epoch,
..
} = event.0
else {
return Err(original);
};
if participant_id != self.participant_id || binding_epoch != self.binding_epoch {
return Err(original);
}
Ok(owed(
debt,
StoredEdge::DetachedMarkerRelease(DetachedMarkerRelease {
participant_id,
marker_delivery_seq: self.marker_delivery_seq,
last_dead_binding_epoch: binding_epoch,
}),
))
}
pub const fn retarget(
self,
new_binding_epoch: BindingEpoch,
episode_churn_used: u64,
delta_cycles: u64,
episode_churn_limit: u64,
) -> Result<Self, (Self, DetachedAttachRefusal)> {
if !charged_churn_fits(episode_churn_used, delta_cycles, episode_churn_limit) {
return Err((self, DetachedAttachRefusal::EpisodeChurnLimit));
}
if !is_next_generation(self.binding_epoch, new_binding_epoch) {
return Err((self, DetachedAttachRefusal::StaleAuthority));
}
Ok(Self {
binding_epoch: new_binding_epoch,
..self
})
}
pub fn leave(
self,
debt: ClosureDebt,
event: Event,
successor: DebtCompletion,
) -> Result<ClosureState, ClosureState> {
let original = owed(debt, StoredEdge::MarkerDelivery(self));
let EventKind::LeaveCommitted {
participant_id,
authority: LeaveAuthority::Live(binding_epoch),
..
} = event.0
else {
return Err(original);
};
if participant_id != self.participant_id || binding_epoch != self.binding_epoch {
return Err(original);
}
Ok(successor.into_state())
}
}
impl ParticipantCursorProgress {
pub fn complete_ack(
self,
debt: ClosureDebt,
event: Event,
successor: DebtCompletion,
) -> Result<ClosureState, ClosureState> {
let original = owed(debt, StoredEdge::ParticipantCursorProgress(self));
let exact = match (self, event.0) {
(
Self::Continuous(value),
EventKind::CursorProgressed {
participant_id,
binding_epoch,
progress: CursorProgressEvent::Normal { through_seq, .. },
..
},
) => {
participant_id == value.participant_id
&& binding_epoch == value.binding_epoch
&& through_seq == value.through_seq
}
(
Self::Marker(value),
EventKind::CursorProgressed {
participant_id,
binding_epoch,
progress: CursorProgressEvent::Normal { through_seq, .. },
..
},
) => {
participant_id == value.participant_id
&& binding_epoch == value.binding_epoch
&& through_seq == value.through_seq
}
(
Self::Marker(value),
EventKind::CursorProgressed {
participant_id,
binding_epoch,
progress:
CursorProgressEvent::Marker {
marker_delivery_seq,
},
..
},
) => {
participant_id == value.participant_id
&& binding_epoch == value.binding_epoch
&& marker_delivery_seq == value.marker_delivery_seq
}
_ => false,
};
if exact {
Ok(successor.into_state())
} else {
Err(original)
}
}
pub fn lesser_ack(
self,
debt: ClosureDebt,
event: Event,
resulting_debt: ClosureDebt,
) -> Result<ClosureState, ClosureState> {
let original = owed(debt, StoredEdge::ParticipantCursorProgress(self));
let EventKind::CursorProgressed {
participant_id,
binding_epoch,
progress: CursorProgressEvent::Normal { through_seq, .. },
..
} = event.0
else {
return Err(original);
};
if participant_id != self.participant_id()
|| binding_epoch != self.binding_epoch()
|| through_seq >= self.through_seq()
{
return Err(original);
}
Ok(owed(
resulting_debt,
StoredEdge::ParticipantCursorProgress(self),
))
}
#[must_use]
pub fn clear_after_greater_ack(&self, event: &Event) -> Option<ProjectionCompactionSuccessor> {
if !greater_ack_matches(*self, *event) {
return None;
}
Some(ProjectionCompactionSuccessor {
predecessor: StoredEdge::ParticipantCursorProgress(*self),
event: *event,
use_kind: SuccessorUse::CursorGreaterAck,
state: ClosureState::Clear,
})
}
#[must_use]
pub fn strict_after_greater_ack(
&self,
event: &Event,
debt: ClosureDebt,
edge: StoredEdge,
successor_boundary: DeliverySeq,
) -> Option<ProjectionCompactionSuccessor> {
if !greater_ack_matches(*self, *event)
|| successor_boundary <= cursor_event_boundary(*event)
|| !strict_edge_matches_boundary(edge, successor_boundary)
{
return None;
}
Some(ProjectionCompactionSuccessor {
predecessor: StoredEdge::ParticipantCursorProgress(*self),
event: *event,
use_kind: SuccessorUse::CursorGreaterAck,
state: ClosureState::Owed { debt, edge },
})
}
pub fn greater_ack(
self,
debt: ClosureDebt,
event: Event,
successor: ProjectionCompactionSuccessor,
) -> Result<ClosureState, ClosureState> {
let original = owed(debt, StoredEdge::ParticipantCursorProgress(self));
if successor.predecessor == StoredEdge::ParticipantCursorProgress(self)
&& successor.event == event
&& successor.use_kind == SuccessorUse::CursorGreaterAck
&& greater_ack_matches(self, event)
{
Ok(successor.state)
} else {
Err(original)
}
}
#[must_use]
pub const fn unchanged(self, debt: ClosureDebt) -> ClosureState {
owed(debt, StoredEdge::ParticipantCursorProgress(self))
}
pub const fn storage_progress(
self,
debt: ClosureDebt,
event: Event,
resulting_debt: Option<ClosureDebt>,
) -> Result<ClosureState, ClosureState> {
let original = owed(debt, StoredEdge::ParticipantCursorProgress(self));
let valid = match event.0 {
EventKind::ProjectionCompleted { .. } => true,
EventKind::CompactionCompleted {
through_seq,
resulting_floor,
..
} => match self.marker_delivery_seq() {
None => true,
Some(marker) => through_seq < marker && resulting_floor <= marker,
},
_ => false,
};
if !valid {
return Err(original);
}
Ok(preserve_or_clear(
resulting_debt,
StoredEdge::ParticipantCursorProgress(self),
))
}
pub fn binding_fate(
self,
debt: ClosureDebt,
event: Event,
) -> Result<CursorFateSuccessor, ClosureState> {
let original = owed(debt, StoredEdge::ParticipantCursorProgress(self));
let Self::Marker(value) = self else {
return Err(original);
};
let EventKind::BindingFateObserved {
participant_id,
binding_epoch,
..
} = event.0
else {
return Err(original);
};
if participant_id != value.participant_id || binding_epoch != value.binding_epoch {
return Err(original);
}
Ok(CursorFateSuccessor::DetachedCredentialRecovery(
DetachedCredentialRecovery {
conversation_id: value.conversation_id,
participant_id: value.participant_id,
marker_delivery_seq: value.marker_delivery_seq,
prior_binding_epoch: value.binding_epoch,
},
))
}
pub const fn retarget(
self,
new_binding_epoch: BindingEpoch,
episode_churn_used: u64,
delta_cycles: u64,
episode_churn_limit: u64,
) -> Result<Self, (Self, DetachedAttachRefusal)> {
let Self::Continuous(value) = self else {
return Err((self, DetachedAttachRefusal::DeliveredMarkerAwaitingAck));
};
if !charged_churn_fits(episode_churn_used, delta_cycles, episode_churn_limit) {
return Err((self, DetachedAttachRefusal::EpisodeChurnLimit));
}
if !is_next_generation(value.binding_epoch, new_binding_epoch) {
return Err((self, DetachedAttachRefusal::StaleAuthority));
}
Ok(Self::Continuous(CursorProgressContinuous {
binding_epoch: new_binding_epoch,
..value
}))
}
pub fn leave(
self,
debt: ClosureDebt,
event: Event,
successor: DebtCompletion,
) -> Result<ClosureState, ClosureState> {
let original = owed(debt, StoredEdge::ParticipantCursorProgress(self));
let EventKind::LeaveCommitted {
participant_id,
authority: LeaveAuthority::Live(binding_epoch),
..
} = event.0
else {
return Err(original);
};
if participant_id != self.participant_id() || binding_epoch != self.binding_epoch() {
return Err(original);
}
Ok(successor.into_state())
}
}
impl DetachedCredentialRecovery {
#[must_use]
pub const fn validate_leave_claim(
&self,
participant_id: ParticipantId,
actual_charge: ResourceVector,
remaining_k: ResourceVector,
exit_claims: u64,
) -> Option<KClaimBackedDetachedLeave> {
validate_detached_claim(
self.participant_id,
DetachedClaimTarget::CredentialRecovery {
marker_delivery_seq: self.marker_delivery_seq,
binding_epoch: self.prior_binding_epoch,
},
participant_id,
actual_charge,
remaining_k,
exit_claims,
)
}
pub fn fenced_attach(
self,
debt: ClosureDebt,
event: Event,
successor: DebtCompletion,
) -> Result<FencedAttachCommit, ClosureState> {
let original = owed(debt, StoredEdge::DetachedCredentialRecovery(self));
let EventKind::FencedRecoveryCommitted {
participant_id,
marker_delivery_seq,
prior_binding_epoch,
new_binding_epoch,
..
} = event.0
else {
return Err(original);
};
if participant_id != self.participant_id
|| marker_delivery_seq != self.marker_delivery_seq
|| prior_binding_epoch != self.prior_binding_epoch
|| !is_next_generation(prior_binding_epoch, new_binding_epoch)
{
return Err(original);
}
Ok(FencedAttachCommit {
conversation_id: self.conversation_id,
participant_id,
marker_delivery_seq,
prior_binding_epoch,
new_binding_epoch,
next_state: successor.into_state(),
})
}
pub fn detached_leave(
self,
debt: ClosureDebt,
event: Event,
evidence: KClaimBackedDetachedLeave,
successor: DebtCompletion,
) -> Result<ClosureState, ClosureState> {
let original = owed(debt, StoredEdge::DetachedCredentialRecovery(self));
let EventKind::LeaveCommitted {
participant_id,
authority: LeaveAuthority::Detached,
..
} = event.0
else {
return Err(original);
};
let target = DetachedClaimTarget::CredentialRecovery {
marker_delivery_seq: self.marker_delivery_seq,
binding_epoch: self.prior_binding_epoch,
};
if participant_id != self.participant_id
|| evidence.participant_id != self.participant_id
|| evidence.target != target
{
return Err(original);
}
Ok(successor.into_state())
}
#[must_use]
pub const fn ordinary_attach_refusal(self) -> DetachedAttachRefusal {
let _ = self;
DetachedAttachRefusal::RecoveryFence
}
#[must_use]
pub const fn marker_attach_refusal(
self,
presented_marker: DeliverySeq,
) -> Option<DetachedAttachRefusal> {
if presented_marker == self.marker_delivery_seq {
None
} else {
Some(DetachedAttachRefusal::MarkerMismatch)
}
}
#[must_use]
pub const fn authority_superseded(self) -> (Self, DetachedAttachRefusal) {
(self, DetachedAttachRefusal::StaleAuthority)
}
pub const fn unrelated_event(
self,
debt: ClosureDebt,
event: Event,
resulting_debt: Option<ClosureDebt>,
) -> Result<ClosureState, ClosureState> {
unrelated_detached_event(
StoredEdge::DetachedCredentialRecovery(self),
self.participant_id,
debt,
event,
resulting_debt,
)
}
}
mod sealed {
pub trait Sealed {}
}
pub trait LeaveOnlyEdge: sealed::Sealed + Sized + Copy {
fn participant_id(self) -> ParticipantId;
fn validate_leave_claim(
&self,
participant_id: ParticipantId,
actual_charge: ResourceVector,
remaining_k: ResourceVector,
exit_claims: u64,
) -> Option<KClaimBackedDetachedLeave>;
fn leave(
self,
debt: ClosureDebt,
event: Event,
evidence: KClaimBackedDetachedLeave,
successor: DebtCompletion,
) -> Result<ClosureState, ClosureState>;
fn repeat_fate(self, event: Event) -> Result<Self, Self>;
fn authority_superseded(self) -> (Self, DetachedAttachRefusal) {
(self, DetachedAttachRefusal::StaleAuthority)
}
fn binding_required_refusal(self) -> DetachedAttachRefusal {
let _ = self;
DetachedAttachRefusal::NoBinding
}
fn unrelated_event(
self,
debt: ClosureDebt,
event: Event,
resulting_debt: Option<ClosureDebt>,
) -> Result<ClosureState, ClosureState>;
}
impl sealed::Sealed for DetachedMarkerRelease {}
impl sealed::Sealed for DetachedCursorRelease {}
impl LeaveOnlyEdge for DetachedMarkerRelease {
fn participant_id(self) -> ParticipantId {
self.participant_id
}
fn validate_leave_claim(
&self,
participant_id: ParticipantId,
actual_charge: ResourceVector,
remaining_k: ResourceVector,
exit_claims: u64,
) -> Option<KClaimBackedDetachedLeave> {
validate_detached_claim(
self.participant_id,
DetachedClaimTarget::MarkerRelease {
marker_delivery_seq: self.marker_delivery_seq,
binding_epoch: self.last_dead_binding_epoch,
},
participant_id,
actual_charge,
remaining_k,
exit_claims,
)
}
fn leave(
self,
debt: ClosureDebt,
event: Event,
evidence: KClaimBackedDetachedLeave,
successor: DebtCompletion,
) -> Result<ClosureState, ClosureState> {
let original = owed(debt, StoredEdge::DetachedMarkerRelease(self));
let EventKind::LeaveCommitted {
participant_id,
authority: LeaveAuthority::Detached,
..
} = event.0
else {
return Err(original);
};
let target = DetachedClaimTarget::MarkerRelease {
marker_delivery_seq: self.marker_delivery_seq,
binding_epoch: self.last_dead_binding_epoch,
};
if participant_id != self.participant_id
|| evidence.participant_id != self.participant_id
|| evidence.target != target
{
return Err(original);
}
Ok(successor.into_state())
}
fn repeat_fate(self, event: Event) -> Result<Self, Self> {
let EventKind::BindingFateObserved {
participant_id,
binding_epoch,
..
} = event.0
else {
return Err(self);
};
if participant_id == self.participant_id && binding_epoch == self.last_dead_binding_epoch {
Ok(self)
} else {
Err(self)
}
}
fn unrelated_event(
self,
debt: ClosureDebt,
event: Event,
resulting_debt: Option<ClosureDebt>,
) -> Result<ClosureState, ClosureState> {
unrelated_detached_event(
StoredEdge::DetachedMarkerRelease(self),
self.participant_id,
debt,
event,
resulting_debt,
)
}
}
impl DetachedMarkerRelease {
#[must_use]
pub const fn ordinary_attach_refusal(self) -> DetachedAttachRefusal {
let _ = self;
DetachedAttachRefusal::RecoveryFence
}
#[must_use]
pub const fn marker_attach_refusal(
self,
presented_marker: DeliverySeq,
) -> DetachedAttachRefusal {
if presented_marker == self.marker_delivery_seq {
DetachedAttachRefusal::MarkerNotDelivered
} else {
DetachedAttachRefusal::MarkerMismatch
}
}
}
impl LeaveOnlyEdge for DetachedCursorRelease {
fn participant_id(self) -> ParticipantId {
self.participant_id
}
fn validate_leave_claim(
&self,
participant_id: ParticipantId,
actual_charge: ResourceVector,
remaining_k: ResourceVector,
exit_claims: u64,
) -> Option<KClaimBackedDetachedLeave> {
validate_detached_claim(
self.participant_id,
DetachedClaimTarget::CursorRelease {
binding_epoch: self.last_dead_binding_epoch,
},
participant_id,
actual_charge,
remaining_k,
exit_claims,
)
}
fn leave(
self,
debt: ClosureDebt,
event: Event,
evidence: KClaimBackedDetachedLeave,
successor: DebtCompletion,
) -> Result<ClosureState, ClosureState> {
let original = owed(debt, StoredEdge::DetachedCursorRelease(self));
let EventKind::LeaveCommitted {
participant_id,
authority: LeaveAuthority::Detached,
..
} = event.0
else {
return Err(original);
};
let target = DetachedClaimTarget::CursorRelease {
binding_epoch: self.last_dead_binding_epoch,
};
if participant_id != self.participant_id
|| evidence.participant_id != self.participant_id
|| evidence.target != target
{
return Err(original);
}
Ok(successor.into_state())
}
fn repeat_fate(self, event: Event) -> Result<Self, Self> {
let EventKind::BindingFateObserved {
participant_id,
binding_epoch,
..
} = event.0
else {
return Err(self);
};
if participant_id == self.participant_id && binding_epoch == self.last_dead_binding_epoch {
Ok(self)
} else {
Err(self)
}
}
fn unrelated_event(
self,
debt: ClosureDebt,
event: Event,
resulting_debt: Option<ClosureDebt>,
) -> Result<ClosureState, ClosureState> {
unrelated_detached_event(
StoredEdge::DetachedCursorRelease(self),
self.participant_id,
debt,
event,
resulting_debt,
)
}
}
impl DetachedCursorRelease {
#[must_use]
pub const fn ordinary_attach_refusal(self) -> DetachedAttachRefusal {
let _ = self;
DetachedAttachRefusal::RecoveryFence
}
#[must_use]
pub const fn marker_attach_refusal(self) -> DetachedAttachRefusal {
let _ = self;
DetachedAttachRefusal::MarkerMismatch
}
}
const fn owed(debt: ClosureDebt, edge: StoredEdge) -> ClosureState {
ClosureState::Owed { debt, edge }
}
const fn preserve_or_clear(resulting_debt: Option<ClosureDebt>, edge: StoredEdge) -> ClosureState {
match resulting_debt {
Some(debt) => owed(debt, edge),
None => ClosureState::Clear,
}
}
const fn projection_completion_boundary(
edge: ObserverProjection,
event: Event,
) -> Option<DeliverySeq> {
let EventKind::ProjectionCompleted { through_seq } = event.0 else {
return None;
};
if through_seq == edge.through_seq {
Some(through_seq)
} else {
None
}
}
const fn physical_completion_floor(edge: PhysicalCompaction, event: Event) -> Option<DeliverySeq> {
match event.0 {
EventKind::CompactionCompleted {
from_floor,
through_seq,
resulting_floor,
} if from_floor == edge.from_floor
&& through_seq == edge.through_seq
&& resulting_floor > edge.through_seq =>
{
Some(resulting_floor)
}
_ => None,
}
}
const fn progress_event_floor(event: Event) -> Option<DeliverySeq> {
match event.0 {
EventKind::CursorProgressed {
resulting_floor, ..
}
| EventKind::BindingFateObserved {
resulting_floor, ..
}
| EventKind::LeaveCommitted {
resulting_floor, ..
} => Some(resulting_floor),
_ => None,
}
}
const fn cursor_event_boundary(event: Event) -> DeliverySeq {
match event.0 {
EventKind::CursorProgressed {
progress: CursorProgressEvent::Normal { through_seq, .. },
..
} => through_seq,
_ => 0,
}
}
fn greater_ack_matches(edge: ParticipantCursorProgress, event: Event) -> bool {
let EventKind::CursorProgressed {
participant_id,
binding_epoch,
progress:
CursorProgressEvent::Normal {
previous_cursor,
through_seq,
},
..
} = event.0
else {
return false;
};
participant_id == edge.participant_id()
&& binding_epoch == edge.binding_epoch()
&& previous_cursor < edge.through_seq()
&& through_seq > edge.through_seq()
}
const fn strict_edge_matches_boundary(edge: StoredEdge, boundary: DeliverySeq) -> bool {
match edge {
StoredEdge::ObserverProjection(value) => value.through_seq == boundary,
StoredEdge::PhysicalCompaction(value) => value.through_seq == boundary,
StoredEdge::MarkerDelivery(value) => value.marker_delivery_seq == boundary,
StoredEdge::ParticipantCursorProgress(value) => value.through_seq() == boundary,
StoredEdge::DetachedCredentialRecovery(_) => false,
StoredEdge::DetachedMarkerRelease(value) => value.marker_delivery_seq == boundary,
StoredEdge::DetachedCursorRelease(_) => true,
}
}
const fn charged_churn_fits(used: u64, delta: u64, limit: u64) -> bool {
delta > 0 && widen_u64(used) + widen_u64(delta) <= widen_u64(limit)
}
#[allow(clippy::cast_lossless)]
const fn widen_u64(value: u64) -> u128 {
value as u128
}
const fn is_next_generation(old: BindingEpoch, new: BindingEpoch) -> bool {
match old.capability_generation.get().checked_add(1) {
Some(expected) => new.capability_generation.get() == expected,
None => false,
}
}
const fn validate_detached_claim(
owner: ParticipantId,
target: DetachedClaimTarget,
participant_id: ParticipantId,
actual_charge: ResourceVector,
remaining_k: ResourceVector,
exit_claims: u64,
) -> Option<KClaimBackedDetachedLeave> {
if participant_id != owner
|| actual_charge.entries == 0
|| actual_charge.bytes == 0
|| actual_charge.entries > remaining_k.entries
|| actual_charge.bytes > remaining_k.bytes
|| exit_claims == 0
{
return None;
}
Some(KClaimBackedDetachedLeave {
participant_id,
target,
actual_charge,
})
}
const fn event_participant(event: Event) -> Option<ParticipantId> {
match event.0 {
EventKind::MarkerDelivered { participant_id, .. }
| EventKind::CursorProgressed { participant_id, .. }
| EventKind::BindingFateObserved { participant_id, .. }
| EventKind::LeaveCommitted { participant_id, .. }
| EventKind::FencedRecoveryCommitted { participant_id, .. } => Some(participant_id),
EventKind::ProjectionCompleted { .. }
| EventKind::CompactionCompleted { .. }
| EventKind::MarkerAppended { .. } => None,
}
}
const fn unrelated_detached_event(
edge: StoredEdge,
owner: ParticipantId,
debt: ClosureDebt,
event: Event,
resulting_debt: Option<ClosureDebt>,
) -> Result<ClosureState, ClosureState> {
let original = owed(debt, edge);
let Some(participant_id) = event_participant(event) else {
return Err(original);
};
if participant_id == owner {
return Err(original);
}
Ok(preserve_or_clear(resulting_debt, edge))
}