use alloc::vec::Vec;
use crate::algebra::WideResourceVector;
use crate::outcome::ParticipantStateCorruptReason;
use crate::wire::{
AttachSecret, BindingEpoch, CloseCause, ConversationId, DeliverySeq, DetachAttemptToken,
Generation, LeaveAttemptToken, LeaveCommitted, ObserverEpoch, ParticipantId, TransactionOrder,
};
use super::{
ActiveBinding, AdmissionOrder, BindingOrigin, BindingState, BoundParticipantCursor,
ClaimFrontiers, ClaimFrontiersRestore, ClosureDebt, ClosureState, CommittedBindingTerminal,
DebtCompletion, DetachCell, DetachedCredentialRecovery, DetachedMarkerRelease, EmptyDetach,
EnrollmentFingerprint, Event, FencedAttachCommit, FrontierBinding, IdentityState,
LeaveFingerprint, LiveMember, LiveMemberRestore, MarkerDelivery, NonzeroDebtCursorEpisode,
ObserverProjection, OrderLedger, OrdinaryBindingAuthority, OrdinaryBindingFate,
ParticipantCursorProgress, PendingFinalization, PendingRecoveredCursorRelease,
PhysicalCompaction, RecoveredBindingFate, RecoveredBindingFateTransition, RetiredIdentity,
SequenceLedger, StoredEdge,
binding::{restore_committed_terminal, restore_pending_finalization},
claim_frontier::{
HistoricalCausalAuthority, MarkerRecordOccurrence, MarkerRecordRequest,
ValidatedConversationHistory, ValidatedMarkerRecord,
},
cursor_facts::{CursorProgressFact, CursorProgressKey},
detach::{
restore_committed_detach, restore_pending_detach, restore_terminalized_detach,
validate_pending_pair,
},
};
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum StorageRestoreError {
CommittedBindingTerminal,
PendingFinalization,
BindingAuthority,
MembershipInvariant,
LeaveResult,
RetiredIdentity,
DetachCell,
DetachBindingPair,
CursorEpisode,
ClosureDebt,
StoredEdgeProvenance,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum ConversationStateRestoreError {
ClaimFrontier(ParticipantStateCorruptReason),
Storage(StorageRestoreError),
}
#[derive(Debug, PartialEq, Eq)]
pub(super) struct RestoredConversationState {
frontiers: ClaimFrontiers,
closure: ClosureState,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ParticipantConversationRestore<EF, V, LF, D> {
pub participants: Vec<ParticipantLifecycleRestore<EF, V, LF, D>>,
pub frontiers: ClaimFrontiersRestore,
pub sequence_ledger: SequenceLedger,
pub order_ledger: OrderLedger,
pub closure: ClosureStateRestore,
}
#[derive(Debug, PartialEq, Eq)]
pub struct ParticipantConversationState<EF, V, LF, D> {
participants: Vec<RestoredParticipantLifecycle<EF, V, LF, D>>,
frontiers: ClaimFrontiers,
closure: ClosureState,
}
impl<EF, V, LF, D> ParticipantConversationState<EF, V, LF, D> {
#[must_use]
pub fn participants(&self) -> &[RestoredParticipantLifecycle<EF, V, LF, D>] {
&self.participants
}
#[must_use]
pub const fn frontiers(&self) -> &ClaimFrontiers {
&self.frontiers
}
#[must_use]
pub const fn closure(&self) -> ClosureState {
self.closure
}
#[must_use]
#[allow(clippy::type_complexity)]
pub fn into_parts(
self,
) -> (
Vec<RestoredParticipantLifecycle<EF, V, LF, D>>,
ClaimFrontiers,
ClosureState,
) {
(self.participants, self.frontiers, self.closure)
}
}
impl RestoredConversationState {
#[must_use]
#[cfg(test)]
pub(crate) const fn frontiers(&self) -> &ClaimFrontiers {
&self.frontiers
}
#[must_use]
#[cfg(test)]
pub(crate) const fn closure(&self) -> ClosureState {
self.closure
}
#[must_use]
pub(crate) fn into_parts(self) -> (ClaimFrontiers, ClosureState) {
(self.frontiers, self.closure)
}
}
#[cfg(test)]
pub(super) fn restore_conversation_state(
frontier_restore: ClaimFrontiersRestore,
sequence_ledger: SequenceLedger,
order_ledger: OrderLedger,
closure_restore: &ClosureStateRestore,
) -> Result<RestoredConversationState, ConversationStateRestoreError> {
let history = ValidatedConversationHistory::empty();
restore_conversation_with_history(
frontier_restore,
sequence_ledger,
order_ledger,
closure_restore,
&history,
)
}
fn restore_conversation_with_history(
frontier_restore: ClaimFrontiersRestore,
sequence_ledger: SequenceLedger,
order_ledger: OrderLedger,
closure_restore: &ClosureStateRestore,
history: &ValidatedConversationHistory,
) -> Result<RestoredConversationState, ConversationStateRestoreError> {
let mut prevalidated = ClaimFrontiers::prevalidate_with_history(
frontier_restore,
sequence_ledger,
order_ledger,
history,
)
.map_err(ConversationStateRestoreError::ClaimFrontier)?;
let marker_request = closure_restore.marker_record_request();
let ordinary_request = closure_restore.ordinary_binding_request();
let closure = match (marker_request, ordinary_request) {
(Some(marker_request), None) => {
let record = prevalidated.take_marker_record(marker_request).ok_or(
ConversationStateRestoreError::Storage(StorageRestoreError::StoredEdgeProvenance),
)?;
(*closure_restore)
.restore_with_marker_record(prevalidated.conversation_id(), record)
.map_err(ConversationStateRestoreError::Storage)?
}
(None, Some(binding)) => {
let origin = history
.ordinary_origin(
binding.conversation_id,
binding.participant_id,
binding.binding_epoch,
)
.ok_or(ConversationStateRestoreError::Storage(
StorageRestoreError::StoredEdgeProvenance,
))?;
(*closure_restore)
.restore_with_binding_origin(origin)
.map_err(ConversationStateRestoreError::Storage)?
}
(None, None) => (*closure_restore)
.restore()
.map_err(ConversationStateRestoreError::Storage)?,
(Some(_), Some(_)) => {
return Err(ConversationStateRestoreError::Storage(
StorageRestoreError::StoredEdgeProvenance,
));
}
};
let current_edge = match closure {
ClosureState::Clear => None,
ClosureState::Owed { edge, .. } => Some(edge),
};
let frontiers = prevalidated
.finish(current_edge)
.map_err(ConversationStateRestoreError::ClaimFrontier)?;
Ok(RestoredConversationState { frontiers, closure })
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct CommittedBindingTerminalRestore {
pub binding: ActiveBinding,
pub cause: CloseCause,
pub transaction_order: TransactionOrder,
pub delivery_seq: DeliverySeq,
}
impl CommittedBindingTerminalRestore {
pub fn restore(self) -> Result<CommittedBindingTerminal, StorageRestoreError> {
restore_committed_terminal(
self.binding,
self.cause,
self.transaction_order,
self.delivery_seq,
)
.ok_or(StorageRestoreError::CommittedBindingTerminal)
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct PendingFinalizationRestore {
pub binding: ActiveBinding,
pub cause: CloseCause,
pub transaction_order: TransactionOrder,
}
impl PendingFinalizationRestore {
pub fn restore(self) -> Result<PendingFinalization, StorageRestoreError> {
restore_pending_finalization(self.binding, self.cause, self.transaction_order)
.ok_or(StorageRestoreError::PendingFinalization)
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum BindingFateTerminalRestore {
Committed(CommittedBindingTerminalRestore),
Pending(PendingFinalizationRestore),
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum RestoredBindingFateTerminal {
Committed(CommittedBindingTerminal),
Pending(PendingFinalization),
}
impl BindingFateTerminalRestore {
pub fn restore(self) -> Result<RestoredBindingFateTerminal, StorageRestoreError> {
match self {
Self::Committed(value) => value.restore().map(RestoredBindingFateTerminal::Committed),
Self::Pending(value) => value.restore().map(RestoredBindingFateTerminal::Pending),
}
}
}
impl RestoredBindingFateTerminal {
const fn participant_id(self) -> ParticipantId {
match self {
Self::Committed(value) => value.participant_id(),
Self::Pending(value) => value.participant_id(),
}
}
const fn conversation_id(self) -> ConversationId {
match self {
Self::Committed(value) => value.conversation_id(),
Self::Pending(value) => value.conversation_id(),
}
}
const fn binding_epoch(self) -> BindingEpoch {
match self {
Self::Committed(value) => value.binding_epoch(),
Self::Pending(value) => value.binding_epoch(),
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum BindingStateRestore {
Detached,
Bound(ActiveBinding),
PendingFinalization(PendingFinalizationRestore),
}
impl BindingStateRestore {
fn restore_for<EF>(self, member: &LiveMember<EF>) -> Result<BindingState, StorageRestoreError> {
let state = match self {
Self::Detached => BindingState::Detached,
Self::Bound(binding) => BindingState::Bound(binding),
Self::PendingFinalization(raw) => BindingState::PendingFinalization(raw.restore()?),
};
let authority_matches = match state {
BindingState::Detached => true,
BindingState::Bound(binding) => {
binding.participant_id == member.participant_id()
&& binding.conversation_id == member.conversation_id()
&& binding.binding_epoch.capability_generation == member.generation()
}
BindingState::PendingFinalization(pending) => {
pending.participant_id() == member.participant_id()
&& pending.conversation_id() == member.conversation_id()
&& pending.binding_epoch().capability_generation == member.generation()
}
};
if authority_matches {
Ok(state)
} else {
Err(StorageRestoreError::BindingAuthority)
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct LiveIdentityRestore<EF> {
pub participant_id: ParticipantId,
pub conversation_id: ConversationId,
pub generation: Generation,
pub attach_secret: AttachSecret,
pub cursor: DeliverySeq,
pub enrollment_fingerprint: EnrollmentFingerprint<EF>,
pub latest_terminal: Option<CommittedBindingTerminalRestore>,
}
impl<EF> LiveIdentityRestore<EF> {
fn restore(self) -> Result<LiveMember<EF>, StorageRestoreError> {
let latest_terminal = self
.latest_terminal
.map(CommittedBindingTerminalRestore::restore)
.transpose()?;
LiveMember::restore(LiveMemberRestore {
participant_id: self.participant_id,
conversation_id: self.conversation_id,
generation: self.generation,
attach_secret: self.attach_secret,
cursor: self.cursor,
enrollment_fingerprint: self.enrollment_fingerprint,
latest_terminal,
})
.map_err(|_| StorageRestoreError::MembershipInvariant)
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct LeaveCommittedRestore {
pub conversation_id: ConversationId,
pub leave_attempt_token: LeaveAttemptToken,
pub participant_id: ParticipantId,
pub retired_generation: Generation,
pub ended_binding_epoch: Option<BindingEpoch>,
pub prior_terminal_delivery_seq: Option<DeliverySeq>,
pub left_delivery_seq: DeliverySeq,
}
impl LeaveCommittedRestore {
pub fn restore(self) -> Result<LeaveCommitted, StorageRestoreError> {
LeaveCommitted::new(
self.conversation_id,
self.leave_attempt_token,
self.participant_id,
self.retired_generation,
self.ended_binding_epoch,
self.prior_terminal_delivery_seq,
self.left_delivery_seq,
)
.ok_or(StorageRestoreError::LeaveResult)
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RetiredIdentityRestore<EF, V, LF> {
pub participant_id: ParticipantId,
pub conversation_id: ConversationId,
pub retired_generation: Generation,
pub enrollment_fingerprint: EnrollmentFingerprint<EF>,
pub leave_attempt_token: LeaveAttemptToken,
pub leave_request_verifier: V,
pub leave_fingerprint: LeaveFingerprint<LF>,
pub left_transaction_order: TransactionOrder,
pub committed_result: LeaveCommittedRestore,
}
impl<EF, V, LF> RetiredIdentityRestore<EF, V, LF> {
fn restore(self) -> Result<RetiredIdentity<EF, V, LF>, StorageRestoreError> {
let result = self.committed_result.restore()?;
RetiredIdentity::restore(
self.participant_id,
self.conversation_id,
self.retired_generation,
self.enrollment_fingerprint,
self.leave_attempt_token,
self.leave_request_verifier,
self.leave_fingerprint,
self.left_transaction_order,
result,
)
.map_err(|_| StorageRestoreError::RetiredIdentity)
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum DetachCellRestore<V> {
Empty,
Pending {
token: DetachAttemptToken,
participant_id: ParticipantId,
request_generation: Generation,
request_verifier: V,
committed_binding_epoch: BindingEpoch,
admission_order: AdmissionOrder,
refused_epoch: ObserverEpoch,
},
Committed {
token: DetachAttemptToken,
participant_id: ParticipantId,
request_generation: Generation,
request_verifier: V,
committed_binding_epoch: BindingEpoch,
detached_delivery_seq: DeliverySeq,
},
Terminalized {
token: DetachAttemptToken,
participant_id: ParticipantId,
request_generation: Generation,
request_verifier: V,
committed_binding_epoch: BindingEpoch,
},
}
impl<V> DetachCellRestore<V> {
fn restore(self) -> Result<DetachCell<V>, StorageRestoreError> {
match self {
Self::Empty => Ok(DetachCell::Empty(EmptyDetach)),
Self::Pending {
token,
participant_id,
request_generation,
request_verifier,
committed_binding_epoch,
admission_order,
refused_epoch,
} => restore_pending_detach(
token,
participant_id,
request_generation,
request_verifier,
committed_binding_epoch,
admission_order,
refused_epoch,
)
.map(DetachCell::Pending)
.ok_or(StorageRestoreError::DetachCell),
Self::Committed {
token,
participant_id,
request_generation,
request_verifier,
committed_binding_epoch,
detached_delivery_seq,
} => restore_committed_detach(
token,
participant_id,
request_generation,
request_verifier,
committed_binding_epoch,
detached_delivery_seq,
)
.map(DetachCell::Committed)
.ok_or(StorageRestoreError::DetachCell),
Self::Terminalized {
token,
participant_id,
request_generation,
request_verifier,
committed_binding_epoch,
} => restore_terminalized_detach(
token,
participant_id,
request_generation,
request_verifier,
committed_binding_epoch,
)
.map(DetachCell::Terminalized)
.ok_or(StorageRestoreError::DetachCell),
}
}
}
#[allow(
clippy::large_enum_variant,
reason = "the live storage capsule remains inline so its atomic slots cannot be restored separately"
)]
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum ParticipantLifecycleRestore<EF, V, LF, D> {
Live {
identity: LiveIdentityRestore<EF>,
binding: BindingStateRestore,
binding_origin: Option<BindingOrigin>,
detach_cell: DetachCellRestore<D>,
},
Retired(RetiredIdentityRestore<EF, V, LF>),
}
#[allow(
clippy::large_enum_variant,
reason = "the validated live capsule remains inline as one atomic lifecycle result"
)]
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum RestoredParticipantLifecycle<EF, V, LF, D> {
Live {
member: LiveMember<EF>,
binding: BindingState,
binding_origin: Option<BindingOrigin>,
detach_cell: DetachCell<D>,
},
Retired(RetiredIdentity<EF, V, LF>),
}
impl<EF, V, LF, D> ParticipantLifecycleRestore<EF, V, LF, D> {
pub fn restore(
self,
) -> Result<RestoredParticipantLifecycle<EF, V, LF, D>, StorageRestoreError> {
match self {
Self::Retired(identity) => identity
.restore()
.map(RestoredParticipantLifecycle::Retired),
Self::Live {
identity,
binding,
binding_origin,
detach_cell,
} => {
let member = identity.restore()?;
let binding = binding.restore_for(&member)?;
let binding_origin = binding_origin
.map(|origin| validate_binding_origin(origin, &member, binding))
.transpose()?;
let origin_required = !matches!(binding, BindingState::Detached)
|| member.latest_terminal().is_some();
if origin_required != binding_origin.is_some() {
return Err(StorageRestoreError::BindingAuthority);
}
let detach_cell = detach_cell.restore()?;
validate_live_pair(&member, binding, &detach_cell)?;
Ok(RestoredParticipantLifecycle::Live {
member,
binding,
binding_origin,
detach_cell,
})
}
}
}
}
fn validate_binding_origin<EF>(
origin: BindingOrigin,
member: &LiveMember<EF>,
binding_state: BindingState,
) -> Result<BindingOrigin, StorageRestoreError> {
let expected_epoch = match binding_state {
BindingState::Bound(current) => Some(current.binding_epoch),
BindingState::PendingFinalization(pending) => Some(pending.binding_epoch()),
BindingState::Detached => member
.latest_terminal()
.map(CommittedBindingTerminal::binding_epoch),
};
let attached = origin.attached();
if origin.participant_id() != member.participant_id()
|| origin.conversation_id() != member.conversation_id()
|| expected_epoch != Some(origin.binding_epoch())
|| attached.participant_id() != member.participant_id()
|| attached.conversation_id() != member.conversation_id()
{
Err(StorageRestoreError::BindingAuthority)
} else {
Ok(origin)
}
}
impl<EF, V, LF, D> RestoredParticipantLifecycle<EF, V, LF, D> {
#[must_use]
#[allow(clippy::type_complexity)]
pub fn into_parts(
self,
) -> (
IdentityState<EF, V, LF>,
Option<BindingState>,
Option<DetachCell<D>>,
) {
match self {
Self::Live {
member,
binding,
binding_origin: _,
detach_cell,
} => (
IdentityState::Live(member),
Some(binding),
Some(detach_cell),
),
Self::Retired(identity) => (IdentityState::Retired(identity), None, None),
}
}
}
impl<EF, V, LF, D> ParticipantConversationRestore<EF, V, LF, D> {
pub fn restore(
self,
) -> Result<ParticipantConversationState<EF, V, LF, D>, ConversationStateRestoreError> {
let participants = self
.participants
.into_iter()
.map(ParticipantLifecycleRestore::restore)
.collect::<Result<Vec<_>, _>>()
.map_err(ConversationStateRestoreError::Storage)?;
validate_participant_frontier_projection(
&participants,
self.frontiers.conversation_id,
&self.frontiers.active_identities,
)?;
let history = validated_conversation_history(&participants)?;
let restored = restore_conversation_with_history(
self.frontiers,
self.sequence_ledger,
self.order_ledger,
&self.closure,
&history,
)?;
let (frontiers, closure) = restored.into_parts();
Ok(ParticipantConversationState {
participants,
frontiers,
closure,
})
}
}
fn validated_conversation_history<EF, V, LF, D>(
participants: &[RestoredParticipantLifecycle<EF, V, LF, D>],
) -> Result<ValidatedConversationHistory, ConversationStateRestoreError> {
let mut causal_authorities = Vec::new();
let mut binding_origins = Vec::new();
let mut seen = Vec::with_capacity(participants.len());
for participant in participants {
let participant_id = match participant {
RestoredParticipantLifecycle::Live {
member,
binding_origin,
..
} => {
if let Some(terminal) = member.latest_terminal() {
causal_authorities
.push(HistoricalCausalAuthority::from_committed_terminal(terminal));
}
if let Some(origin) = binding_origin {
binding_origins.push(*origin);
}
member.participant_id()
}
RestoredParticipantLifecycle::Retired(retired) => {
causal_authorities.push(HistoricalCausalAuthority::from_retired(retired));
retired.participant_id()
}
};
seen.push(participant_id);
}
seen.sort_unstable();
let has_duplicate = seen
.iter()
.zip(seen.iter().skip(1))
.any(|(current, next)| current == next);
if has_duplicate {
return Err(ConversationStateRestoreError::Storage(
StorageRestoreError::MembershipInvariant,
));
}
Ok(ValidatedConversationHistory::new(
causal_authorities,
binding_origins,
))
}
fn validate_participant_frontier_projection<EF, V, LF, D>(
participants: &[RestoredParticipantLifecycle<EF, V, LF, D>],
conversation_id: ConversationId,
frontier: &[super::FrontierParticipant],
) -> Result<(), ConversationStateRestoreError> {
let mut projected = Vec::new();
for participant in participants {
match participant {
RestoredParticipantLifecycle::Live {
member, binding, ..
} => {
if member.conversation_id() != conversation_id {
return Err(ConversationStateRestoreError::Storage(
StorageRestoreError::MembershipInvariant,
));
}
let binding = match binding {
BindingState::Bound(binding) => FrontierBinding::Bound(binding.binding_epoch),
BindingState::PendingFinalization(pending) => {
FrontierBinding::Detached(pending.binding_epoch())
}
BindingState::Detached => {
let Some(terminal) = member.latest_terminal() else {
return Err(ConversationStateRestoreError::Storage(
StorageRestoreError::BindingAuthority,
));
};
FrontierBinding::Detached(terminal.binding_epoch())
}
};
projected.push(super::FrontierParticipant::new(
member.participant_id(),
member.cursor(),
binding,
));
}
RestoredParticipantLifecycle::Retired(retired) => {
if retired.conversation_id() != conversation_id {
return Err(ConversationStateRestoreError::Storage(
StorageRestoreError::MembershipInvariant,
));
}
}
}
}
projected.sort_by_key(|participant| participant.participant_index());
if projected == frontier {
Ok(())
} else {
Err(ConversationStateRestoreError::Storage(
StorageRestoreError::MembershipInvariant,
))
}
}
#[allow(clippy::suspicious_operation_groupings)]
fn validate_live_pair<EF, D>(
member: &LiveMember<EF>,
binding: BindingState,
detach_cell: &DetachCell<D>,
) -> Result<(), StorageRestoreError> {
match detach_cell {
DetachCell::Empty(_) => Ok(()),
DetachCell::Pending(cell) => {
if cell.participant_id() != member.participant_id()
|| cell.request_generation() != member.generation()
|| validate_pending_pair(binding, cell, Some(member.conversation_id())).is_err()
{
Err(StorageRestoreError::DetachBindingPair)
} else {
Ok(())
}
}
DetachCell::Committed(cell) => {
let terminal_matches = member.latest_terminal().is_some_and(|terminal| {
terminal.participant_id() == cell.participant_id()
&& terminal.conversation_id() == member.conversation_id()
&& terminal.binding_epoch() == cell.committed_binding_epoch()
&& terminal.delivery_seq() == cell.detached_delivery_seq()
&& terminal.detached_cause()
== Some(crate::wire::DetachedCause::CleanDeregister)
});
if cell.participant_id() != member.participant_id()
|| cell.request_generation() != member.generation()
|| binding != BindingState::Detached
|| !terminal_matches
{
Err(StorageRestoreError::DetachBindingPair)
} else {
Ok(())
}
}
DetachCell::Terminalized(cell) => {
if cell.participant_id() != member.participant_id()
|| cell.request_generation().get() >= member.generation().get()
{
Err(StorageRestoreError::DetachBindingPair)
} else {
Ok(())
}
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CursorEpisodeRestore {
pub conversation_id: ConversationId,
pub debt: WideResourceVector,
pub observer_progress: DeliverySeq,
pub candidate_high_watermark: DeliverySeq,
pub current_floor: u128,
pub cap_floor: u128,
pub participants: Vec<BoundParticipantCursor>,
pub facts: Vec<(CursorProgressKey, CursorProgressFact)>,
}
impl CursorEpisodeRestore {
pub fn restore(self) -> Result<NonzeroDebtCursorEpisode, StorageRestoreError> {
let debt = ClosureDebt::new(self.debt).ok_or(StorageRestoreError::ClosureDebt)?;
NonzeroDebtCursorEpisode::restore(
self.conversation_id,
debt,
self.observer_progress,
self.candidate_high_watermark,
self.current_floor,
self.cap_floor,
self.participants,
self.facts,
)
.ok_or(StorageRestoreError::CursorEpisode)
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct OrdinaryBindingAuthorityRestore {
pub binding: ActiveBinding,
pub through_seq: DeliverySeq,
}
impl OrdinaryBindingAuthorityRestore {
fn restore_with_origin(
self,
origin: &BindingOrigin,
) -> Result<OrdinaryBindingAuthority, StorageRestoreError> {
if !origin.is_unfenced()
|| origin.conversation_id() != self.binding.conversation_id
|| origin.participant_id() != self.binding.participant_id
|| origin.binding_epoch() != self.binding.binding_epoch
{
return Err(StorageRestoreError::StoredEdgeProvenance);
}
Ok(OrdinaryBindingAuthority::new(
self.binding,
self.through_seq,
))
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct MarkerDeliveryRestore {
pub participant_id: ParticipantId,
pub binding_epoch: BindingEpoch,
pub marker_delivery_seq: DeliverySeq,
}
impl MarkerDeliveryRestore {
#[cfg(any(test, feature = "test-support"))]
pub(super) fn restore_bound(
self,
conversation_id: ConversationId,
record_authority: ValidatedMarkerRecord,
) -> Result<MarkerDelivery, StorageRestoreError> {
let restored = self.restore_with_target(
conversation_id,
&record_authority,
MarkerRecordTarget::Bound,
MarkerRecordOccurrence::Undelivered,
);
record_authority.consume();
restored
}
#[cfg(test)]
pub(super) fn restore_detached_delivered_for_test(
self,
conversation_id: ConversationId,
record_authority: ValidatedMarkerRecord,
) -> Result<MarkerDelivery, StorageRestoreError> {
let restored = self.restore_with_target(
conversation_id,
&record_authority,
MarkerRecordTarget::Detached,
MarkerRecordOccurrence::Delivered,
);
record_authority.consume();
restored
}
fn restore_detached(
self,
conversation_id: ConversationId,
record_authority: &ValidatedMarkerRecord,
) -> Result<MarkerDelivery, StorageRestoreError> {
self.restore_with_target(
conversation_id,
record_authority,
MarkerRecordTarget::Detached,
MarkerRecordOccurrence::Undelivered,
)
}
fn restore_with_target(
self,
conversation_id: ConversationId,
record_authority: &ValidatedMarkerRecord,
target: MarkerRecordTarget,
occurrence: MarkerRecordOccurrence,
) -> Result<MarkerDelivery, StorageRestoreError> {
if record_authority.conversation_id() != conversation_id
|| !target.matches(record_authority.target_binding(), self.binding_epoch)
|| record_authority.occurrence() != occurrence
{
return Err(StorageRestoreError::StoredEdgeProvenance);
}
let delivery = MarkerDelivery::from_validated_record(record_authority);
if delivery.participant_id() != self.participant_id
|| delivery.binding_epoch() != self.binding_epoch
|| delivery.marker_delivery_seq() != self.marker_delivery_seq
{
return Err(StorageRestoreError::StoredEdgeProvenance);
}
Ok(delivery)
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum MarkerRecordTarget {
Bound,
Detached,
}
impl MarkerRecordTarget {
fn matches(self, target: super::FrontierBinding, epoch: BindingEpoch) -> bool {
matches!(
(self, target),
(Self::Bound, super::FrontierBinding::Bound(actual))
| (Self::Detached, super::FrontierBinding::Detached(actual))
if actual == epoch
)
}
}
#[derive(Debug)]
enum MarkerRestoreAuthority<'a> {
Absent,
Record {
conversation_id: ConversationId,
record: &'a ValidatedMarkerRecord,
},
}
impl MarkerRestoreAuthority<'_> {
const fn require_absent(&self) -> Result<(), StorageRestoreError> {
match self {
Self::Absent => Ok(()),
Self::Record { .. } => Err(StorageRestoreError::StoredEdgeProvenance),
}
}
const fn require_record(
&self,
) -> Result<(ConversationId, &ValidatedMarkerRecord), StorageRestoreError> {
match self {
Self::Record {
conversation_id,
record,
} if record.conversation_id() == *conversation_id => Ok((*conversation_id, record)),
Self::Absent | Self::Record { .. } => Err(StorageRestoreError::StoredEdgeProvenance),
}
}
fn record_for(
&self,
expected_conversation_id: ConversationId,
) -> Result<&ValidatedMarkerRecord, StorageRestoreError> {
let (conversation_id, record) = self.require_record()?;
if conversation_id == expected_conversation_id {
Ok(record)
} else {
Err(StorageRestoreError::StoredEdgeProvenance)
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct MarkerCursorProgressRestore {
pub conversation_id: ConversationId,
pub participant_id: ParticipantId,
pub binding_epoch: BindingEpoch,
pub through_seq: DeliverySeq,
pub marker_delivery_seq: DeliverySeq,
pub delivery: MarkerDeliveryRestore,
}
impl MarkerCursorProgressRestore {
fn restore_with_debt(
self,
debt: ClosureDebt,
record_authority: &ValidatedMarkerRecord,
target: MarkerRecordTarget,
) -> Result<ParticipantCursorProgress, StorageRestoreError> {
if self.participant_id != self.delivery.participant_id
|| self.binding_epoch != self.delivery.binding_epoch
|| self.marker_delivery_seq != self.delivery.marker_delivery_seq
|| self.through_seq != self.marker_delivery_seq
{
return Err(StorageRestoreError::StoredEdgeProvenance);
}
let marker = self.delivery.restore_with_target(
self.conversation_id,
record_authority,
target,
MarkerRecordOccurrence::Delivered,
)?;
let state = marker
.delivered(
debt,
Event::marker_delivered(
self.participant_id,
self.binding_epoch,
self.marker_delivery_seq,
),
)
.map_err(|_| StorageRestoreError::StoredEdgeProvenance)?;
match state {
ClosureState::Owed {
edge: StoredEdge::ParticipantCursorProgress(progress),
..
} => Ok(progress),
ClosureState::Clear | ClosureState::Owed { .. } => {
Err(StorageRestoreError::StoredEdgeProvenance)
}
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct DetachedCredentialRecoveryRestore {
pub participant_id: ParticipantId,
pub marker_delivery_seq: DeliverySeq,
pub prior_binding_epoch: BindingEpoch,
pub resulting_floor: DeliverySeq,
pub terminal: BindingFateTerminalRestore,
pub progress: MarkerCursorProgressRestore,
}
impl DetachedCredentialRecoveryRestore {
pub fn restore_description(self) -> Result<DetachedCredentialRecovery, StorageRestoreError> {
if self.participant_id != self.progress.participant_id
|| self.marker_delivery_seq != self.progress.marker_delivery_seq
|| self.prior_binding_epoch != self.progress.binding_epoch
|| self.progress.through_seq != self.marker_delivery_seq
|| self.progress.participant_id != self.progress.delivery.participant_id
|| self.progress.binding_epoch != self.progress.delivery.binding_epoch
|| self.progress.marker_delivery_seq != self.progress.delivery.marker_delivery_seq
{
return Err(StorageRestoreError::StoredEdgeProvenance);
}
let terminal = self.terminal.restore()?;
if terminal.participant_id() != self.participant_id
|| terminal.binding_epoch() != self.prior_binding_epoch
|| terminal.conversation_id() != self.progress.conversation_id
{
return Err(StorageRestoreError::StoredEdgeProvenance);
}
Ok(DetachedCredentialRecovery::from_storage_description(
self.progress.conversation_id,
self.participant_id,
self.marker_delivery_seq,
self.prior_binding_epoch,
))
}
fn restore_with_debt(
self,
debt: ClosureDebt,
record_authority: &ValidatedMarkerRecord,
) -> Result<DetachedCredentialRecovery, StorageRestoreError> {
if self.participant_id != self.progress.participant_id
|| self.marker_delivery_seq != self.progress.marker_delivery_seq
|| self.prior_binding_epoch != self.progress.binding_epoch
{
return Err(StorageRestoreError::StoredEdgeProvenance);
}
let terminal = self.terminal.restore()?;
if terminal.participant_id() != self.participant_id
|| terminal.binding_epoch() != self.prior_binding_epoch
|| terminal.conversation_id() != self.progress.conversation_id
{
return Err(StorageRestoreError::StoredEdgeProvenance);
}
let progress = self.progress.restore_with_debt(
debt,
record_authority,
MarkerRecordTarget::Detached,
)?;
let successor = progress
.binding_fate(
debt,
Event::binding_fate_observed(
self.participant_id,
self.prior_binding_epoch,
self.resulting_floor,
),
)
.map_err(|_| StorageRestoreError::StoredEdgeProvenance)?;
match successor.into_stored_edge() {
StoredEdge::DetachedCredentialRecovery(edge) => Ok(edge),
_ => Err(StorageRestoreError::StoredEdgeProvenance),
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct DetachedMarkerReleaseRestore {
pub conversation_id: ConversationId,
pub participant_id: ParticipantId,
pub marker_delivery_seq: DeliverySeq,
pub last_dead_binding_epoch: BindingEpoch,
pub resulting_floor: DeliverySeq,
pub terminal: BindingFateTerminalRestore,
pub delivery: MarkerDeliveryRestore,
}
impl DetachedMarkerReleaseRestore {
fn restore_with_debt(
self,
debt: ClosureDebt,
record_authority: &ValidatedMarkerRecord,
) -> Result<DetachedMarkerRelease, StorageRestoreError> {
if self.participant_id != self.delivery.participant_id
|| self.marker_delivery_seq != self.delivery.marker_delivery_seq
|| self.last_dead_binding_epoch != self.delivery.binding_epoch
{
return Err(StorageRestoreError::StoredEdgeProvenance);
}
let terminal = self.terminal.restore()?;
if terminal.participant_id() != self.participant_id
|| terminal.binding_epoch() != self.last_dead_binding_epoch
|| terminal.conversation_id() != self.conversation_id
{
return Err(StorageRestoreError::StoredEdgeProvenance);
}
let marker = self
.delivery
.restore_detached(self.conversation_id, record_authority)?;
let state = marker
.binding_fate(
debt,
Event::binding_fate_observed(
self.participant_id,
self.last_dead_binding_epoch,
self.resulting_floor,
),
)
.map_err(|_| StorageRestoreError::StoredEdgeProvenance)?;
match state {
ClosureState::Owed {
edge: StoredEdge::DetachedMarkerRelease(edge),
..
} => Ok(edge),
ClosureState::Clear | ClosureState::Owed { .. } => {
Err(StorageRestoreError::StoredEdgeProvenance)
}
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct OrdinaryBindingFateRestore {
pub authority: OrdinaryBindingAuthorityRestore,
pub terminal: CommittedBindingTerminalRestore,
pub resulting_floor: DeliverySeq,
}
impl OrdinaryBindingFateRestore {
fn restore_with_origin(
self,
origin: &BindingOrigin,
) -> Result<OrdinaryBindingFate, StorageRestoreError> {
let authority = self.authority.restore_with_origin(origin)?;
let terminal = self.terminal.restore()?;
let CommittedBindingTerminal::Died(terminal) = terminal else {
return Err(StorageRestoreError::StoredEdgeProvenance);
};
authority
.binding_fate(terminal, self.resulting_floor)
.map_err(|_| StorageRestoreError::StoredEdgeProvenance)
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum DebtCompletionRestore {
Clear,
ObserverProjection {
debt: WideResourceVector,
through_seq: DeliverySeq,
},
PhysicalCompaction {
debt: WideResourceVector,
from_floor: DeliverySeq,
through_seq: DeliverySeq,
},
}
impl DebtCompletionRestore {
pub fn restore(self) -> Result<DebtCompletion, StorageRestoreError> {
match self {
Self::Clear => Ok(DebtCompletion::clear()),
Self::ObserverProjection { debt, through_seq } => {
let debt = ClosureDebt::new(debt).ok_or(StorageRestoreError::ClosureDebt)?;
Ok(DebtCompletion::observer_projection(
debt,
ObserverProjection::new(through_seq),
))
}
Self::PhysicalCompaction {
debt,
from_floor,
through_seq,
} => {
let debt = ClosureDebt::new(debt).ok_or(StorageRestoreError::ClosureDebt)?;
let edge = PhysicalCompaction::new(from_floor, through_seq)
.ok_or(StorageRestoreError::StoredEdgeProvenance)?;
Ok(DebtCompletion::physical_compaction(debt, edge))
}
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct FencedAttachCommitRestore {
pub predecessor: DetachedCredentialRecoveryRestore,
pub predecessor_debt: WideResourceVector,
pub participant_id: ParticipantId,
pub marker_delivery_seq: DeliverySeq,
pub prior_binding_epoch: BindingEpoch,
pub new_binding_epoch: BindingEpoch,
pub resulting_floor: DeliverySeq,
pub successor: DebtCompletionRestore,
}
impl FencedAttachCommitRestore {
#[cfg(test)]
pub(super) fn restore(
self,
record_authority: ValidatedMarkerRecord,
) -> Result<FencedAttachCommit, StorageRestoreError> {
let restored = self.restore_with_record(&record_authority);
record_authority.consume();
restored
}
fn restore_with_record(
self,
record_authority: &ValidatedMarkerRecord,
) -> Result<FencedAttachCommit, StorageRestoreError> {
let debt =
ClosureDebt::new(self.predecessor_debt).ok_or(StorageRestoreError::ClosureDebt)?;
let predecessor = self.predecessor.restore_with_debt(debt, record_authority)?;
FencedAttachCommit::restore_validated(
predecessor,
record_authority,
debt,
Event::fenced_recovery_committed(
self.participant_id,
self.marker_delivery_seq,
self.prior_binding_epoch,
self.new_binding_epoch,
self.resulting_floor,
),
self.successor.restore()?,
)
.ok_or(StorageRestoreError::StoredEdgeProvenance)
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct RecoveredBindingFateRestore {
pub fenced_attach: FencedAttachCommitRestore,
pub participant_id: ParticipantId,
pub binding_epoch: BindingEpoch,
pub resulting_floor: DeliverySeq,
}
impl RecoveredBindingFateRestore {
#[cfg(test)]
pub(super) fn restore(
self,
record_authority: ValidatedMarkerRecord,
) -> Result<RecoveredBindingFate, StorageRestoreError> {
let restored = self.restore_with_record(&record_authority);
record_authority.consume();
restored
}
fn restore_with_record(
self,
record_authority: &ValidatedMarkerRecord,
) -> Result<RecoveredBindingFate, StorageRestoreError> {
let commit = self.fenced_attach.restore_with_record(record_authority)?;
commit
.recovered_binding_fate(Event::binding_fate_observed(
self.participant_id,
self.binding_epoch,
self.resulting_floor,
))
.map_err(|_| StorageRestoreError::StoredEdgeProvenance)
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct PendingRecoveredCursorReleaseRestore {
pub fate: RecoveredBindingFateRestore,
pub resulting_debt: WideResourceVector,
}
impl PendingRecoveredCursorReleaseRestore {
#[cfg(test)]
pub(super) fn restore(
self,
record_authority: ValidatedMarkerRecord,
) -> Result<PendingRecoveredCursorRelease, StorageRestoreError> {
let restored = self.restore_with_record(&record_authority);
record_authority.consume();
restored
}
fn restore_with_record(
self,
record_authority: &ValidatedMarkerRecord,
) -> Result<PendingRecoveredCursorRelease, StorageRestoreError> {
let authority = self.fate.restore_with_record(record_authority)?;
let predecessor_state = authority.predecessor_state();
let resulting_debt =
ClosureDebt::new(self.resulting_debt).ok_or(StorageRestoreError::ClosureDebt)?;
let transition = match predecessor_state {
ClosureState::Owed {
debt,
edge: StoredEdge::ObserverProjection(edge),
} => edge.apply_recovered_binding_fate(debt, resulting_debt, authority),
ClosureState::Owed {
debt,
edge: StoredEdge::PhysicalCompaction(edge),
} => edge.apply_recovered_binding_fate(debt, resulting_debt, authority),
ClosureState::Clear | ClosureState::Owed { .. } => {
return Err(StorageRestoreError::StoredEdgeProvenance);
}
}
.map_err(|_| StorageRestoreError::StoredEdgeProvenance)?;
match transition {
RecoveredBindingFateTransition::PendingStorage(pending) => Ok(pending),
RecoveredBindingFateTransition::DetachedCursorRelease(_) => {
Err(StorageRestoreError::StoredEdgeProvenance)
}
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum RecoveredStorageCompletionRestore {
ObserverProjection {
through_seq: DeliverySeq,
resulting_debt: Option<WideResourceVector>,
},
PhysicalCompaction {
from_floor: DeliverySeq,
through_seq: DeliverySeq,
resulting_floor: DeliverySeq,
resulting_debt: Option<WideResourceVector>,
},
}
impl RecoveredStorageCompletionRestore {
fn restore(
self,
pending: PendingRecoveredCursorRelease,
) -> Result<ClosureState, StorageRestoreError> {
let current = pending.current_state();
match (current, self) {
(
ClosureState::Owed {
edge: StoredEdge::ObserverProjection(edge),
..
},
Self::ObserverProjection {
through_seq,
resulting_debt,
},
) => edge
.complete_after_recovered_binding_fate(
Event::projection_completed(through_seq),
optional_debt(resulting_debt)?,
pending,
)
.map_err(|_| StorageRestoreError::StoredEdgeProvenance),
(
ClosureState::Owed {
edge: StoredEdge::PhysicalCompaction(edge),
..
},
Self::PhysicalCompaction {
from_floor,
through_seq,
resulting_floor,
resulting_debt,
},
) => {
let event = Event::compaction_completed(from_floor, through_seq, resulting_floor)
.ok_or(StorageRestoreError::StoredEdgeProvenance)?;
edge.complete_after_recovered_binding_fate(
event,
optional_debt(resulting_debt)?,
pending,
)
.map_err(|_| StorageRestoreError::StoredEdgeProvenance)
}
_ => Err(StorageRestoreError::StoredEdgeProvenance),
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum DetachedCursorReleaseProvenanceRestore {
Ordinary(OrdinaryBindingFateRestore),
RecoveredDirect {
fate: RecoveredBindingFateRestore,
resulting_debt: WideResourceVector,
},
RecoveredAfterStorage {
pending: PendingRecoveredCursorReleaseRestore,
completion: RecoveredStorageCompletionRestore,
},
}
impl DetachedCursorReleaseProvenanceRestore {
const fn ordinary_binding_request(self) -> Option<ActiveBinding> {
match self {
Self::Ordinary(fate) => Some(fate.authority.binding),
Self::RecoveredDirect { .. } | Self::RecoveredAfterStorage { .. } => None,
}
}
const fn marker_record_request(self) -> Option<MarkerRecordRequest> {
match self {
Self::Ordinary(_) => None,
Self::RecoveredDirect { fate, .. } => Some(MarkerRecordRequest::recovered(
fate.participant_id,
fate.fenced_attach.marker_delivery_seq,
fate.fenced_attach.prior_binding_epoch,
fate.fenced_attach.new_binding_epoch,
)),
Self::RecoveredAfterStorage { pending, .. } => Some(MarkerRecordRequest::recovered(
pending.fate.participant_id,
pending.fate.fenced_attach.marker_delivery_seq,
pending.fate.fenced_attach.prior_binding_epoch,
pending.fate.fenced_attach.new_binding_epoch,
)),
}
}
fn restore_state(
self,
debt: ClosureDebt,
marker_authority: &MarkerRestoreAuthority<'_>,
ordinary_origin: Option<&BindingOrigin>,
) -> Result<ClosureState, StorageRestoreError> {
match self {
Self::Ordinary(provenance) => {
marker_authority.require_absent()?;
let origin = ordinary_origin.ok_or(StorageRestoreError::StoredEdgeProvenance)?;
Ok(provenance
.restore_with_origin(origin)?
.into_direct_state(debt))
}
Self::RecoveredDirect {
fate,
resulting_debt,
} => {
let (_, record_authority) = marker_authority.require_record()?;
let authority = fate.restore_with_record(record_authority)?;
let predecessor = authority.predecessor_state();
let resulting_debt =
ClosureDebt::new(resulting_debt).ok_or(StorageRestoreError::ClosureDebt)?;
let transition = match predecessor {
ClosureState::Owed {
debt: predecessor_debt,
edge: StoredEdge::PhysicalCompaction(edge),
} => edge.apply_recovered_binding_fate(
predecessor_debt,
resulting_debt,
authority,
),
ClosureState::Clear | ClosureState::Owed { .. } => {
return Err(StorageRestoreError::StoredEdgeProvenance);
}
}
.map_err(|_| StorageRestoreError::StoredEdgeProvenance)?;
match transition {
RecoveredBindingFateTransition::DetachedCursorRelease(release)
if release.debt() == debt =>
{
Ok(release.into_state())
}
RecoveredBindingFateTransition::PendingStorage(_)
| RecoveredBindingFateTransition::DetachedCursorRelease(_) => {
Err(StorageRestoreError::StoredEdgeProvenance)
}
}
}
Self::RecoveredAfterStorage {
pending,
completion,
} => {
let (_, record_authority) = marker_authority.require_record()?;
let state = completion.restore(pending.restore_with_record(record_authority)?)?;
match state {
ClosureState::Owed {
debt: restored_debt,
edge: StoredEdge::DetachedCursorRelease(_),
} if restored_debt == debt => Ok(state),
ClosureState::Clear | ClosureState::Owed { .. } => {
Err(StorageRestoreError::StoredEdgeProvenance)
}
}
}
}
}
}
#[allow(clippy::large_enum_variant)]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum StoredEdgeRestore {
ObserverProjection {
through_seq: DeliverySeq,
},
PhysicalCompaction {
from_floor: DeliverySeq,
through_seq: DeliverySeq,
},
MarkerDelivery(MarkerDeliveryRestore),
ParticipantCursorProgressContinuous {
participant_id: ParticipantId,
binding_epoch: BindingEpoch,
through_seq: DeliverySeq,
authority: OrdinaryBindingAuthorityRestore,
},
ParticipantCursorProgressMarker(MarkerCursorProgressRestore),
DetachedCredentialRecovery(DetachedCredentialRecoveryRestore),
DetachedMarkerRelease(DetachedMarkerReleaseRestore),
DetachedCursorRelease {
participant_id: ParticipantId,
last_dead_binding_epoch: BindingEpoch,
provenance: DetachedCursorReleaseProvenanceRestore,
},
}
impl StoredEdgeRestore {
const fn ordinary_binding_request(self) -> Option<ActiveBinding> {
match self {
Self::ParticipantCursorProgressContinuous { authority, .. } => Some(authority.binding),
Self::DetachedCursorRelease { provenance, .. } => provenance.ordinary_binding_request(),
Self::ObserverProjection { .. }
| Self::PhysicalCompaction { .. }
| Self::MarkerDelivery(_)
| Self::ParticipantCursorProgressMarker(_)
| Self::DetachedCredentialRecovery(_)
| Self::DetachedMarkerRelease(_) => None,
}
}
const fn marker_record_request(self) -> Option<MarkerRecordRequest> {
match self {
Self::MarkerDelivery(value) => Some(MarkerRecordRequest::planned(
value.participant_id,
value.marker_delivery_seq,
super::FrontierBinding::Bound(value.binding_epoch),
)),
Self::ParticipantCursorProgressMarker(value) => Some(MarkerRecordRequest::delivered(
value.participant_id,
value.marker_delivery_seq,
super::FrontierBinding::Bound(value.binding_epoch),
)),
Self::DetachedCredentialRecovery(value) => Some(MarkerRecordRequest::delivered(
value.participant_id,
value.marker_delivery_seq,
super::FrontierBinding::Detached(value.prior_binding_epoch),
)),
Self::DetachedMarkerRelease(value) => Some(MarkerRecordRequest::planned(
value.participant_id,
value.marker_delivery_seq,
super::FrontierBinding::Detached(value.last_dead_binding_epoch),
)),
Self::DetachedCursorRelease { provenance, .. } => provenance.marker_record_request(),
Self::ObserverProjection { .. }
| Self::PhysicalCompaction { .. }
| Self::ParticipantCursorProgressContinuous { .. } => None,
}
}
fn restore_with_debt(
self,
debt: ClosureDebt,
marker_authority: &MarkerRestoreAuthority<'_>,
ordinary_origin: Option<&BindingOrigin>,
) -> Result<StoredEdge, StorageRestoreError> {
match self {
Self::ObserverProjection { through_seq } => {
marker_authority.require_absent()?;
Ok(StoredEdge::ObserverProjection(ObserverProjection::new(
through_seq,
)))
}
Self::PhysicalCompaction {
from_floor,
through_seq,
} => {
marker_authority.require_absent()?;
PhysicalCompaction::new(from_floor, through_seq)
.map(StoredEdge::PhysicalCompaction)
.ok_or(StorageRestoreError::StoredEdgeProvenance)
}
Self::MarkerDelivery(value) => {
let (conversation_id, record_authority) = marker_authority.require_record()?;
value
.restore_with_target(
conversation_id,
record_authority,
MarkerRecordTarget::Bound,
MarkerRecordOccurrence::Undelivered,
)
.map(StoredEdge::MarkerDelivery)
}
Self::ParticipantCursorProgressContinuous {
participant_id,
binding_epoch,
through_seq,
authority,
} => {
marker_authority.require_absent()?;
let origin = ordinary_origin.ok_or(StorageRestoreError::StoredEdgeProvenance)?;
ParticipantCursorProgress::restore_continuous(
authority.restore_with_origin(origin)?,
participant_id,
binding_epoch,
through_seq,
)
.map(StoredEdge::ParticipantCursorProgress)
.ok_or(StorageRestoreError::StoredEdgeProvenance)
}
Self::ParticipantCursorProgressMarker(value) => {
let record_authority = marker_authority.record_for(value.conversation_id)?;
value
.restore_with_debt(debt, record_authority, MarkerRecordTarget::Bound)
.map(StoredEdge::ParticipantCursorProgress)
}
Self::DetachedCredentialRecovery(value) => {
let record_authority =
marker_authority.record_for(value.progress.conversation_id)?;
value
.restore_with_debt(debt, record_authority)
.map(StoredEdge::DetachedCredentialRecovery)
}
Self::DetachedMarkerRelease(value) => {
let record_authority = marker_authority.record_for(value.conversation_id)?;
value
.restore_with_debt(debt, record_authority)
.map(StoredEdge::DetachedMarkerRelease)
}
Self::DetachedCursorRelease {
participant_id,
last_dead_binding_epoch,
provenance,
} => {
let state = provenance.restore_state(debt, marker_authority, ordinary_origin)?;
match state {
ClosureState::Owed {
edge: StoredEdge::DetachedCursorRelease(edge),
..
} if edge.participant_id() == participant_id
&& edge.last_dead_binding_epoch() == last_dead_binding_epoch =>
{
Ok(StoredEdge::DetachedCursorRelease(edge))
}
ClosureState::Clear | ClosureState::Owed { .. } => {
Err(StorageRestoreError::StoredEdgeProvenance)
}
}
}
}
}
}
#[allow(clippy::large_enum_variant)]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ClosureStateRestore {
Clear,
Owed {
debt: WideResourceVector,
edge: StoredEdgeRestore,
},
}
impl ClosureStateRestore {
const fn ordinary_binding_request(self) -> Option<ActiveBinding> {
match self {
Self::Clear => None,
Self::Owed { edge, .. } => edge.ordinary_binding_request(),
}
}
const fn marker_record_request(self) -> Option<MarkerRecordRequest> {
match self {
Self::Clear => None,
Self::Owed { edge, .. } => edge.marker_record_request(),
}
}
pub fn restore(self) -> Result<ClosureState, StorageRestoreError> {
self.restore_with_authority(&MarkerRestoreAuthority::Absent, None)
}
pub(super) fn restore_with_marker_record(
self,
conversation_id: ConversationId,
record: ValidatedMarkerRecord,
) -> Result<ClosureState, StorageRestoreError> {
let restored = self.restore_with_authority(
&MarkerRestoreAuthority::Record {
conversation_id,
record: &record,
},
None,
);
record.consume();
restored
}
fn restore_with_binding_origin(
self,
origin: &BindingOrigin,
) -> Result<ClosureState, StorageRestoreError> {
self.restore_with_authority(&MarkerRestoreAuthority::Absent, Some(origin))
}
fn restore_with_authority(
self,
marker_authority: &MarkerRestoreAuthority<'_>,
ordinary_origin: Option<&BindingOrigin>,
) -> Result<ClosureState, StorageRestoreError> {
match self {
Self::Clear => {
marker_authority.require_absent()?;
Ok(ClosureState::Clear)
}
Self::Owed { debt, edge } => {
let debt = ClosureDebt::new(debt).ok_or(StorageRestoreError::ClosureDebt)?;
Ok(ClosureState::Owed {
debt,
edge: edge.restore_with_debt(debt, marker_authority, ordinary_origin)?,
})
}
}
}
}
fn optional_debt(
raw: Option<WideResourceVector>,
) -> Result<Option<ClosureDebt>, StorageRestoreError> {
raw.map(|value| ClosureDebt::new(value).ok_or(StorageRestoreError::ClosureDebt))
.transpose()
}