use super::{ClientParticipantAggregate, ClientResponseCorrelation};
use crate::wire::{
AttachBound, DetachCommitted, DetachEnvelope, DetachInProgress, LeaveCommitted,
TerminalizedDetachCell,
};
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum DetachReplayStatus {
Parked,
InFlight,
Superseded,
LeaveSuperseded,
Terminal(DetachReplayTerminal),
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum DetachReplayTerminal {
DetachCommitted(DetachCommitted),
DetachInProgress(DetachInProgress),
TerminalizedDetachCell(TerminalizedDetachCell),
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) enum DetachReplayState {
Empty,
Recorded {
request: DetachEnvelope,
status: DetachReplayStatus,
},
}
#[derive(Debug, PartialEq, Eq)]
pub struct SdkDetachReplayAggregate {
pub(super) state: DetachReplayState,
}
impl SdkDetachReplayAggregate {
pub(super) const fn new() -> Self {
Self {
state: DetachReplayState::Empty,
}
}
#[must_use]
pub const fn request(&self) -> Option<&DetachEnvelope> {
match &self.state {
DetachReplayState::Empty => None,
DetachReplayState::Recorded { request, .. } => Some(request),
}
}
#[must_use]
pub const fn status(&self) -> Option<&DetachReplayStatus> {
match &self.state {
DetachReplayState::Empty => None,
DetachReplayState::Recorded { status, .. } => Some(status),
}
}
pub(super) const fn mark_initial_attempt_started(&mut self) {
if let DetachReplayState::Recorded { status, .. } = &mut self.state {
if matches!(status, DetachReplayStatus::Parked) {
*status = DetachReplayStatus::InFlight;
}
}
}
pub(super) fn can_replace_with(&self, request: &DetachEnvelope) -> bool {
match &self.state {
DetachReplayState::Recorded {
request: retained,
status,
} => {
retained.conversation_id == request.conversation_id
&& retained.participant_id == request.participant_id
&& request.capability_generation > retained.capability_generation
&& matches!(
status,
DetachReplayStatus::Superseded | DetachReplayStatus::Terminal(_)
)
}
DetachReplayState::Empty => false,
}
}
pub(super) fn apply_attach(&mut self, attach: &AttachBound) -> bool {
let DetachReplayState::Recorded { request, status } = &mut self.state else {
return false;
};
if attach.conversation_id() == request.conversation_id
&& attach.participant_id() == request.participant_id
&& attach.request_generation() == request.capability_generation
&& attach.capability_generation() > request.capability_generation
{
*status = DetachReplayStatus::Superseded;
true
} else {
false
}
}
pub(super) fn apply_leave(&mut self, leave: &LeaveCommitted) -> bool {
let DetachReplayState::Recorded { request, status } = &mut self.state else {
return false;
};
if leave.conversation_id() == request.conversation_id
&& leave.participant_id() == request.participant_id
&& leave.presented_generation() == request.capability_generation
{
*status = DetachReplayStatus::LeaveSuperseded;
true
} else {
false
}
}
pub(super) fn apply_retired(
&mut self,
conversation_id: u64,
participant_id: u64,
retired_generation: crate::wire::Generation,
) -> bool {
let DetachReplayState::Recorded { request, status } = &mut self.state else {
return false;
};
if request.conversation_id == conversation_id
&& request.participant_id == participant_id
&& retired_generation >= request.capability_generation
{
*status = DetachReplayStatus::LeaveSuperseded;
true
} else {
false
}
}
pub(super) fn apply_detach_committed(&mut self, value: &DetachCommitted) -> bool {
let DetachReplayState::Recorded { request, status } = &mut self.state else {
return false;
};
if detach_committed_matches(request, value) {
*status =
DetachReplayStatus::Terminal(DetachReplayTerminal::DetachCommitted(value.clone()));
true
} else {
false
}
}
pub(super) fn apply_detach_in_progress(&mut self, value: &DetachInProgress) -> bool {
let DetachReplayState::Recorded { request, status } = &mut self.state else {
return false;
};
if detach_in_progress_matches(request, value) {
*status =
DetachReplayStatus::Terminal(DetachReplayTerminal::DetachInProgress(value.clone()));
true
} else {
false
}
}
pub(super) fn apply_terminalized_detach_cell(
&mut self,
value: &TerminalizedDetachCell,
) -> bool {
let DetachReplayState::Recorded { request, status } = &mut self.state else {
return false;
};
if terminalized_matches(request, value) {
*status = DetachReplayStatus::Terminal(DetachReplayTerminal::TerminalizedDetachCell(
value.clone(),
));
true
} else {
false
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum DetachReplayRefusalReason {
AlreadyRecorded,
InvalidStatus,
ForeignInput,
LostAuthorityPending,
}
#[derive(Debug, PartialEq, Eq)]
pub struct DetachReplayApplied {
aggregate: ClientParticipantAggregate,
}
impl DetachReplayApplied {
#[must_use]
pub fn into_aggregate(self) -> ClientParticipantAggregate {
self.aggregate
}
}
#[derive(Debug, PartialEq, Eq)]
pub struct DetachReplayRefusal<T> {
aggregate: ClientParticipantAggregate,
input: T,
reason: DetachReplayRefusalReason,
}
impl<T> DetachReplayRefusal<T> {
#[must_use]
pub const fn reason(&self) -> DetachReplayRefusalReason {
self.reason
}
#[must_use]
pub fn into_parts(self) -> (ClientParticipantAggregate, T) {
(self.aggregate, self.input)
}
}
#[derive(Debug, PartialEq, Eq)]
pub struct DetachTransportAttempt {
request: DetachEnvelope,
authorization: u64,
}
impl DetachTransportAttempt {
#[must_use]
pub const fn request(&self) -> &DetachEnvelope {
&self.request
}
#[must_use]
pub const fn into_request(self) -> (DetachEnvelope, ClientResponseCorrelation) {
(
self.request,
ClientResponseCorrelation {
authorization: self.authorization,
},
)
}
}
#[derive(Debug, PartialEq, Eq)]
pub enum DetachTransportAttemptDecision {
Started {
aggregate: ClientParticipantAggregate,
attempt: DetachTransportAttempt,
},
Refused(DetachReplayRefusal<()>),
}
#[must_use]
pub fn transport_attempt_started(
mut aggregate: ClientParticipantAggregate,
) -> DetachTransportAttemptDecision {
let request = match &aggregate.detach_replay.state {
DetachReplayState::Recorded {
request,
status: DetachReplayStatus::Parked,
} => request.clone(),
DetachReplayState::Empty | DetachReplayState::Recorded { .. } => {
return DetachTransportAttemptDecision::Refused(DetachReplayRefusal {
aggregate,
input: (),
reason: DetachReplayRefusalReason::InvalidStatus,
});
}
};
let expected_matches = aggregate.expected.as_ref().is_some_and(|expected| {
expected.authorization != 0
&& matches!(&expected.request, crate::wire::ClientRequest::Detach(value)
if value.conversation_id == request.conversation_id
&& value.participant_id == request.participant_id
&& value.capability_generation == request.capability_generation
&& value.detach_attempt_token == request.detach_attempt_token)
});
if !expected_matches {
return DetachTransportAttemptDecision::Refused(DetachReplayRefusal {
aggregate,
input: (),
reason: DetachReplayRefusalReason::InvalidStatus,
});
}
let authorization = aggregate
.expected
.as_ref()
.map_or(0, |expected| expected.authorization);
if let Some(expected) = aggregate.expected.as_mut() {
expected.issued = true;
}
if let DetachReplayState::Recorded { status, .. } = &mut aggregate.detach_replay.state {
*status = DetachReplayStatus::InFlight;
}
DetachTransportAttemptDecision::Started {
aggregate,
attempt: DetachTransportAttempt {
request,
authorization,
},
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum DetachTransportFate {
ResponseUnavailable,
}
#[derive(Debug, PartialEq, Eq)]
pub enum DetachTransportFateDecision {
Parked(DetachReplayApplied),
Refused(DetachReplayRefusal<(ClientResponseCorrelation, DetachTransportFate)>),
}
#[must_use]
pub fn transport_fate(
mut aggregate: ClientParticipantAggregate,
correlation: ClientResponseCorrelation,
fate: DetachTransportFate,
) -> DetachTransportFateDecision {
if aggregate.operation_loss_pending() {
return DetachTransportFateDecision::Refused(DetachReplayRefusal {
aggregate,
input: (correlation, fate),
reason: DetachReplayRefusalReason::LostAuthorityPending,
});
}
if detach_authority_matches(&aggregate, &correlation) {
if let DetachReplayState::Recorded { status, .. } = &mut aggregate.detach_replay.state {
*status = DetachReplayStatus::Parked;
}
if let Some(expected) = aggregate.expected.as_mut() {
expected.issued = false;
}
return DetachTransportFateDecision::Parked(DetachReplayApplied { aggregate });
}
DetachTransportFateDecision::Refused(DetachReplayRefusal {
aggregate,
input: (correlation, fate),
reason: DetachReplayRefusalReason::InvalidStatus,
})
}
#[derive(Debug, PartialEq, Eq)]
pub enum ApplyAttachDecision {
Superseded(DetachReplayApplied),
Refused(DetachReplayRefusal<(AttachBound, ClientResponseCorrelation)>),
}
#[must_use]
pub fn apply_attach(
mut aggregate: ClientParticipantAggregate,
attach: AttachBound,
correlation: ClientResponseCorrelation,
) -> ApplyAttachDecision {
if aggregate.operation_loss_pending() {
return ApplyAttachDecision::Refused(DetachReplayRefusal {
aggregate,
input: (attach, correlation),
reason: DetachReplayRefusalReason::LostAuthorityPending,
});
}
if detach_authority_matches(&aggregate, &correlation)
&& aggregate.detach_replay.apply_attach(&attach)
{
aggregate.expected = None;
ApplyAttachDecision::Superseded(DetachReplayApplied { aggregate })
} else {
ApplyAttachDecision::Refused(DetachReplayRefusal {
aggregate,
input: (attach, correlation),
reason: DetachReplayRefusalReason::ForeignInput,
})
}
}
#[derive(Debug, PartialEq, Eq)]
pub enum ApplyLeaveDecision {
Superseded(DetachReplayApplied),
Refused(DetachReplayRefusal<(LeaveCommitted, ClientResponseCorrelation)>),
}
#[must_use]
pub fn apply_leave_durable(
mut aggregate: ClientParticipantAggregate,
leave: LeaveCommitted,
correlation: ClientResponseCorrelation,
) -> ApplyLeaveDecision {
if aggregate.operation_loss_pending() {
return ApplyLeaveDecision::Refused(DetachReplayRefusal {
aggregate,
input: (leave, correlation),
reason: DetachReplayRefusalReason::LostAuthorityPending,
});
}
if detach_authority_matches(&aggregate, &correlation)
&& aggregate.detach_replay.apply_leave(&leave)
{
aggregate.expected = None;
ApplyLeaveDecision::Superseded(DetachReplayApplied { aggregate })
} else {
ApplyLeaveDecision::Refused(DetachReplayRefusal {
aggregate,
input: (leave, correlation),
reason: DetachReplayRefusalReason::ForeignInput,
})
}
}
#[derive(Debug, PartialEq, Eq)]
pub enum DetachReplayOutcome {
DetachCommitted(DetachCommitted),
DetachInProgress(DetachInProgress),
TerminalizedDetachCell(TerminalizedDetachCell),
}
#[derive(Debug, PartialEq, Eq)]
pub enum ApplyDetachOutcomeDecision {
Terminal(DetachReplayApplied),
Refused(DetachReplayRefusal<(DetachReplayOutcome, ClientResponseCorrelation)>),
}
#[must_use]
pub fn apply_detach_outcome(
mut aggregate: ClientParticipantAggregate,
outcome: DetachReplayOutcome,
correlation: ClientResponseCorrelation,
) -> ApplyDetachOutcomeDecision {
if aggregate.operation_loss_pending() {
return ApplyDetachOutcomeDecision::Refused(DetachReplayRefusal {
aggregate,
input: (outcome, correlation),
reason: DetachReplayRefusalReason::LostAuthorityPending,
});
}
if !detach_authority_matches(&aggregate, &correlation) {
return ApplyDetachOutcomeDecision::Refused(DetachReplayRefusal {
aggregate,
input: (outcome, correlation),
reason: DetachReplayRefusalReason::InvalidStatus,
});
}
let applied = match &outcome {
DetachReplayOutcome::DetachCommitted(value) => {
aggregate.detach_replay.apply_detach_committed(value)
}
DetachReplayOutcome::DetachInProgress(value) => {
aggregate.detach_replay.apply_detach_in_progress(value)
}
DetachReplayOutcome::TerminalizedDetachCell(value) => aggregate
.detach_replay
.apply_terminalized_detach_cell(value),
};
if applied {
aggregate.expected = None;
ApplyDetachOutcomeDecision::Terminal(DetachReplayApplied { aggregate })
} else {
ApplyDetachOutcomeDecision::Refused(DetachReplayRefusal {
aggregate,
input: (outcome, correlation),
reason: DetachReplayRefusalReason::ForeignInput,
})
}
}
fn detach_authority_matches(
aggregate: &ClientParticipantAggregate,
correlation: &ClientResponseCorrelation,
) -> bool {
let Some(expected) = aggregate.expected.as_ref() else {
return false;
};
if !expected.issued || expected.authorization != correlation.authorization {
return false;
}
let DetachReplayState::Recorded {
request,
status: DetachReplayStatus::InFlight,
} = &aggregate.detach_replay.state
else {
return false;
};
matches!(&expected.request, crate::wire::ClientRequest::Detach(value)
if value.conversation_id == request.conversation_id
&& value.participant_id == request.participant_id
&& value.capability_generation == request.capability_generation
&& value.detach_attempt_token == request.detach_attempt_token)
}
fn detach_committed_matches(request: &DetachEnvelope, value: &DetachCommitted) -> bool {
value.conversation_id() == request.conversation_id
&& value.participant_id() == request.participant_id
&& value.capability_generation() == request.capability_generation
&& value.detach_attempt_token() == request.detach_attempt_token
}
fn detach_in_progress_matches(request: &DetachEnvelope, value: &DetachInProgress) -> bool {
let expected_generation = request.capability_generation;
let presented_generation = value.presented_generation;
let expected_token = request.detach_attempt_token;
let presented_token = value.presented_token;
value.conversation_id == request.conversation_id
&& value.participant_id == request.participant_id
&& presented_generation == expected_generation
&& presented_token == expected_token
}
fn terminalized_matches(request: &DetachEnvelope, value: &TerminalizedDetachCell) -> bool {
value.conversation_id() == request.conversation_id
&& value.participant_id() == request.participant_id
&& value.capability_generation() == request.capability_generation
&& value.detach_attempt_token() == request.detach_attempt_token
}