use super::{
Reader, Writer, read_active_binding, read_committed_terminal, read_leave_committed,
read_marker_progress, read_wide, write_active_binding, write_committed_terminal,
write_leave_committed, write_marker_progress, write_wide,
};
use crate::lifecycle::{
ActiveBinding, CommittedBindingTerminalRestore, DebtCompletionRestore, LeaveCommittedRestore,
MarkerCursorProgressRestore, OwnerCheckpointRestoreError,
};
use crate::wire::{ConversationId, DeliverySeq, ParticipantId};
use alloc::vec::Vec;
#[doc(hidden)]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ObserverProjectionCheckpointKind {
DetachedTerminal,
DiedTerminal,
ParticipantAck,
MarkerAck,
Leave,
OrdinaryFate,
RecoveredFate,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(in crate::lifecycle) enum ObserverProjectionSource {
Terminal(CommittedBindingTerminalRestore),
Ack {
binding: ActiveBinding,
previous_cursor: DeliverySeq,
},
Marker {
progress: MarkerCursorProgressRestore,
previous_cursor: DeliverySeq,
},
Leave(LeaveCommittedRestore),
Ordinary {
binding: ActiveBinding,
through_seq: DeliverySeq,
terminal: CommittedBindingTerminalRestore,
},
Recovered {
binding: ActiveBinding,
predecessor: DebtCompletionRestore,
},
}
impl ObserverProjectionSource {
pub(in crate::lifecycle) const fn kind(self) -> ObserverProjectionCheckpointKind {
match self {
Self::Terminal(value) => {
if matches!(
value.cause,
crate::wire::CloseCause::CleanDeregister
| crate::wire::CloseCause::Superseded
| crate::wire::CloseCause::ServerShutdown
) {
ObserverProjectionCheckpointKind::DetachedTerminal
} else {
ObserverProjectionCheckpointKind::DiedTerminal
}
}
Self::Ack { .. } => ObserverProjectionCheckpointKind::ParticipantAck,
Self::Marker { .. } => ObserverProjectionCheckpointKind::MarkerAck,
Self::Leave(_) => ObserverProjectionCheckpointKind::Leave,
Self::Ordinary { .. } => ObserverProjectionCheckpointKind::OrdinaryFate,
Self::Recovered { .. } => ObserverProjectionCheckpointKind::RecoveredFate,
}
}
pub(in crate::lifecycle) const fn participant_id(self) -> ParticipantId {
match self {
Self::Terminal(value) => value.binding.participant_id,
Self::Ack { binding, .. }
| Self::Ordinary { binding, .. }
| Self::Recovered { binding, .. } => binding.participant_id,
Self::Marker { progress, .. } => progress.participant_id,
Self::Leave(value) => value.participant_id,
}
}
pub(in crate::lifecycle) const fn binding_epoch(self) -> Option<crate::wire::BindingEpoch> {
match self {
Self::Terminal(value) => Some(value.binding.binding_epoch),
Self::Ack { binding, .. }
| Self::Ordinary { binding, .. }
| Self::Recovered { binding, .. } => Some(binding.binding_epoch),
Self::Marker { progress, .. } => Some(progress.binding_epoch),
Self::Leave(value) => value.ended_binding_epoch,
}
}
fn validate(
self,
conversation_id: ConversationId,
progress: DeliverySeq,
) -> Result<(), OwnerCheckpointRestoreError> {
use OwnerCheckpointRestoreError as Error;
let valid = match self {
Self::Terminal(value) => {
let terminal = value.restore().map_err(Error::Provenance)?;
terminal.conversation_id() == conversation_id && terminal.delivery_seq() == progress
}
Self::Ack {
binding,
previous_cursor,
} => binding.conversation_id == conversation_id && progress > previous_cursor,
Self::Marker {
progress: marker,
previous_cursor,
} => {
marker.conversation_id == conversation_id
&& marker.through_seq == progress
&& marker.marker_delivery_seq == progress
&& previous_cursor < progress
&& marker.participant_id == marker.delivery.participant_id
&& marker.binding_epoch == marker.delivery.binding_epoch
&& marker.marker_delivery_seq == marker.delivery.marker_delivery_seq
}
Self::Leave(value) => {
let leave = value.restore().map_err(Error::Provenance)?;
leave.conversation_id() == conversation_id && leave.left_delivery_seq() == progress
}
Self::Ordinary {
binding,
through_seq: _,
terminal,
} => {
let terminal = terminal.restore().map_err(Error::Provenance)?;
matches!(
terminal,
crate::lifecycle::CommittedBindingTerminal::Died(_)
) && terminal.conversation_id() == conversation_id
&& terminal.participant_id() == binding.participant_id
&& terminal.binding_epoch() == binding.binding_epoch
&& binding.conversation_id == conversation_id
}
Self::Recovered {
binding,
predecessor,
} => {
predecessor.restore().map_err(Error::Provenance)?;
binding.conversation_id == conversation_id
&& !matches!(predecessor, DebtCompletionRestore::Clear)
}
};
if valid {
Ok(())
} else {
Err(Error::Field("observer_projection.native_association"))
}
}
}
pub(in crate::lifecycle) fn encode_observer_source(
conversation_id: ConversationId,
progress: DeliverySeq,
source: ObserverProjectionSource,
) -> Result<Vec<u8>, OwnerCheckpointRestoreError> {
source.validate(conversation_id, progress)?;
let mut w = Writer::default();
w.bytes.extend_from_slice(b"LOPS\x01");
w.put_u64(conversation_id);
w.put_u64(progress);
match source {
ObserverProjectionSource::Terminal(terminal) => {
w.put_u8(1);
write_committed_terminal(&mut w, terminal);
}
ObserverProjectionSource::Ack {
binding,
previous_cursor,
} => {
w.put_u8(2);
write_active_binding(&mut w, binding);
w.put_u64(previous_cursor);
}
ObserverProjectionSource::Marker {
progress,
previous_cursor,
} => {
w.put_u8(3);
write_marker_progress(&mut w, &progress);
w.put_u64(previous_cursor);
}
ObserverProjectionSource::Leave(leave) => {
w.put_u8(4);
write_leave_committed(&mut w, leave);
}
ObserverProjectionSource::Ordinary {
binding,
through_seq,
terminal,
} => {
w.put_u8(5);
write_active_binding(&mut w, binding);
w.put_u64(through_seq);
write_committed_terminal(&mut w, terminal);
}
ObserverProjectionSource::Recovered {
binding,
predecessor,
} => {
w.put_u8(6);
write_active_binding(&mut w, binding);
write_debt_completion(&mut w, &predecessor);
}
}
Ok(w.bytes)
}
pub(in crate::lifecycle) fn decode_observer_source(
bytes: &[u8],
) -> Result<(ConversationId, DeliverySeq, ObserverProjectionSource), OwnerCheckpointRestoreError> {
use OwnerCheckpointRestoreError as Error;
let read = || -> Result<_, super::CapsuleDecodeError> {
if bytes.get(..5) != Some(b"LOPS\x01".as_slice()) {
return Err(super::CapsuleDecodeError::InvalidMagic);
}
let mut r = Reader::new(&bytes[5..], 9);
let conversation_id = r.take_u64()?;
let progress = r.take_u64()?;
let source = match r.take_u8()? {
1 => ObserverProjectionSource::Terminal(read_committed_terminal(&mut r)?),
2 => ObserverProjectionSource::Ack {
binding: read_active_binding(&mut r)?,
previous_cursor: r.take_u64()?,
},
3 => ObserverProjectionSource::Marker {
progress: read_marker_progress(&mut r)?,
previous_cursor: r.take_u64()?,
},
4 => ObserverProjectionSource::Leave(read_leave_committed(&mut r)?),
5 => ObserverProjectionSource::Ordinary {
binding: read_active_binding(&mut r)?,
through_seq: r.take_u64()?,
terminal: read_committed_terminal(&mut r)?,
},
6 => ObserverProjectionSource::Recovered {
binding: read_active_binding(&mut r)?,
predecessor: read_debt_completion(&mut r)?,
},
_ => return Err(r.invalid()),
};
r.finish()?;
Ok((conversation_id, progress, source))
};
let (conversation_id, progress, source) = read().map_err(Error::Framing)?;
if encode_observer_source(conversation_id, progress, source)? != bytes {
return Err(Error::Field("observer_projection.canonical_source"));
}
Ok((conversation_id, progress, source))
}
fn write_debt_completion(w: &mut Writer, completion: &DebtCompletionRestore) {
match *completion {
DebtCompletionRestore::Clear => w.put_u8(1),
DebtCompletionRestore::ObserverProjection { debt, through_seq } => {
w.put_u8(2);
write_wide(w, debt);
w.put_u64(through_seq);
}
DebtCompletionRestore::PhysicalCompaction {
debt,
from_floor,
through_seq,
} => {
w.put_u8(3);
write_wide(w, debt);
w.put_u64(from_floor);
w.put_u64(through_seq);
}
}
}
fn read_debt_completion(
r: &mut Reader<'_>,
) -> Result<DebtCompletionRestore, super::CapsuleDecodeError> {
match r.take_u8()? {
1 => Ok(DebtCompletionRestore::Clear),
2 => Ok(DebtCompletionRestore::ObserverProjection {
debt: read_wide(r)?,
through_seq: r.take_u64()?,
}),
3 => Ok(DebtCompletionRestore::PhysicalCompaction {
debt: read_wide(r)?,
from_floor: r.take_u64()?,
through_seq: r.take_u64()?,
}),
_ => Err(r.invalid()),
}
}