use crate::wire::{
BindingEpoch, ConversationId, DeliverySeq, Generation, MarkerAck, MarkerAckCommitted,
MarkerAckEnvelope, MarkerAckResponse, ParticipantId,
};
use super::{
super::{
BindingRequiredLookupResult, BindingState, LiveMember, ObserverProgressProjection,
ParticipantBindingRequest, PresentedIdentity, lookup_binding_required,
membership::{LiveMemberCursorUpdate, LiveMemberCursorUpdateError},
},
marker_proof::{
MarkerProofDecision, MarkerProofInput, MarkerProofPermit, MarkerProofState,
select_marker_proof,
},
};
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct MarkerAckCommit {
outcome: MarkerAckCommitted,
proof: MarkerProofPermit,
cursor_update: LiveMemberCursorUpdate,
}
impl MarkerAckCommit {
#[must_use]
pub const fn outcome(&self) -> &MarkerAckCommitted {
&self.outcome
}
#[must_use]
pub const fn observer_progress_projection(&self) -> ObserverProgressProjection {
let request = self.outcome.request();
ObserverProgressProjection::new(request.conversation_id, request.marker_delivery_seq)
}
#[must_use]
pub const fn canonical_request(&self) -> MarkerAck {
let request = self.outcome.request();
MarkerAck {
conversation_id: request.conversation_id,
participant_id: request.participant_id,
capability_generation: request.capability_generation,
marker_delivery_seq: request.marker_delivery_seq,
}
}
#[must_use]
pub const fn receiving_binding_epoch(&self) -> BindingEpoch {
self.proof.proof_binding_epoch()
}
#[must_use]
pub const fn offered_marker_delivery_seq(&self) -> DeliverySeq {
self.proof.expected_marker_delivery_seq()
}
#[must_use]
pub const fn delivered_binding_epoch(&self) -> BindingEpoch {
self.proof.proof_binding_epoch()
}
#[must_use]
pub const fn from_cursor(&self) -> DeliverySeq {
self.cursor_update.previous_cursor()
}
#[must_use]
pub const fn resulting_cursor(&self) -> DeliverySeq {
self.cursor_update.resulting_cursor()
}
#[must_use]
pub const fn proof(&self) -> &MarkerProofPermit {
&self.proof
}
pub fn apply_to<F>(
self,
member: &mut LiveMember<F>,
) -> Result<MarkerAckCommitted, MarkerAckCommitError> {
member
.apply_cursor_update(self.cursor_update)
.map_err(MarkerAckCommitError::from_member_error)?;
Ok(self.outcome)
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum MarkerAckCommitError {
Conversation {
expected: ConversationId,
actual: ConversationId,
},
Participant {
expected: ParticipantId,
actual: ParticipantId,
},
Generation {
expected: Generation,
actual: Generation,
},
NonAdvancing {
from_cursor: DeliverySeq,
resulting_cursor: DeliverySeq,
},
CursorPrestate {
expected_from_cursor: DeliverySeq,
resulting_cursor: DeliverySeq,
actual_cursor: DeliverySeq,
},
}
impl MarkerAckCommitError {
const fn from_member_error(error: LiveMemberCursorUpdateError) -> Self {
match error {
LiveMemberCursorUpdateError::Conversation { expected, actual } => {
Self::Conversation { expected, actual }
}
LiveMemberCursorUpdateError::Participant { expected, actual } => {
Self::Participant { expected, actual }
}
LiveMemberCursorUpdateError::Generation { expected, actual } => {
Self::Generation { expected, actual }
}
LiveMemberCursorUpdateError::NonAdvancing {
from_cursor,
resulting_cursor,
} => Self::NonAdvancing {
from_cursor,
resulting_cursor,
},
LiveMemberCursorUpdateError::CursorPrestate {
expected_from_cursor,
resulting_cursor,
actual_cursor,
} => Self::CursorPrestate {
expected_from_cursor,
resulting_cursor,
actual_cursor,
},
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum MarkerAckDecision {
Respond(MarkerAckResponse),
Commit(MarkerAckCommit),
}
#[must_use]
pub fn apply_marker_ack<EF, V, LF>(
presented_identity: PresentedIdentity<'_, EF, V, LF>,
binding: &BindingState,
receiving_binding_epoch: BindingEpoch,
request: &MarkerAck,
marker_state: &MarkerProofState,
) -> MarkerAckDecision {
let lookup_request = ParticipantBindingRequest::MarkerAck(request.clone());
let (member, active_binding) = match lookup_binding_required(
presented_identity,
binding,
Some(receiving_binding_epoch),
&lookup_request,
) {
BindingRequiredLookupResult::Retired(outcome) => {
return MarkerAckDecision::Respond(MarkerAckResponse::from_retired(outcome));
}
BindingRequiredLookupResult::ParticipantUnknown(outcome) => {
return MarkerAckDecision::Respond(MarkerAckResponse::from_participant_unknown(
outcome,
));
}
BindingRequiredLookupResult::StaleAuthority(outcome) => {
return MarkerAckDecision::Respond(MarkerAckResponse::from_stale_authority(outcome));
}
BindingRequiredLookupResult::NoBinding(outcome) => {
return MarkerAckDecision::Respond(MarkerAckResponse::from_no_binding(outcome));
}
BindingRequiredLookupResult::Authorized { member, binding } => (member, binding),
};
let exact_state = MarkerProofState::new(
member.cursor(),
marker_state.accepted_marker_at_cursor(),
marker_state.expected_marker_delivery_seq(),
active_binding.binding_epoch,
marker_state.delivered_to_proof_epoch(),
);
match select_marker_proof(&exact_state, MarkerProofInput::marker_ack(request)) {
MarkerProofDecision::AckNoOp(outcome) => {
MarkerAckDecision::Respond(MarkerAckResponse::from_ack_no_op(outcome))
}
MarkerProofDecision::MarkerMismatch(outcome) => {
MarkerAckDecision::Respond(MarkerAckResponse::from_marker_mismatch(outcome))
}
MarkerProofDecision::MarkerNotDelivered(outcome) => {
MarkerAckDecision::Respond(MarkerAckResponse::from_marker_not_delivered(outcome))
}
MarkerProofDecision::Permit(proof) => {
let envelope = marker_ack_envelope(request);
MarkerAckDecision::Commit(MarkerAckCommit {
outcome: MarkerAckCommitted::new(envelope),
proof,
cursor_update: LiveMemberCursorUpdate::new(
request.conversation_id,
request.participant_id,
request.capability_generation,
member.cursor(),
request.marker_delivery_seq,
),
})
}
}
}
const fn marker_ack_envelope(request: &MarkerAck) -> MarkerAckEnvelope {
MarkerAckEnvelope {
conversation_id: request.conversation_id,
participant_id: request.participant_id,
capability_generation: request.capability_generation,
marker_delivery_seq: request.marker_delivery_seq,
}
}