use alloc::collections::BTreeSet;
use alloc::format;
use alloc::string::String;
use alloc::vec::Vec;
use crate::wire::{ConversationId, CredentialDigest, ParticipantId};
use super::checkpoint_codec::{
CLOSURE_SECTION_TAG, DigestDescription, FRONTIERS_SECTION_TAG, PARTICIPANTS_SECTION_TAG,
SEALED_SECTION_TAG, decode_description, decode_sealed_description, encode_description,
encode_sealed_description,
};
use super::claim_frontier::ClaimFrontiers;
use super::conversation::{ConversationGenesis, ParticipantConversation};
use super::edge::ClosureState;
use super::membership::RetiredIdentity;
use super::storage::{
ConversationDescribeError, ConversationStateRestoreError, ParticipantConversationState,
ParticipantLifecycleRestore, ParticipantLifecycleView, StorageRestoreError,
describe_conversation,
};
pub type DigestParticipantView<'a> = ParticipantLifecycleView<
'a,
CredentialDigest,
CredentialDigest,
CredentialDigest,
CredentialDigest,
>;
pub type DigestConversationState = ParticipantConversationState<
CredentialDigest,
CredentialDigest,
CredentialDigest,
CredentialDigest,
>;
pub type DigestRetiredIdentity =
RetiredIdentity<CredentialDigest, CredentialDigest, CredentialDigest>;
type DigestParticipantRestore = ParticipantLifecycleRestore<
CredentialDigest,
CredentialDigest,
CredentialDigest,
CredentialDigest,
>;
const CAPSULE_MAGIC: [u8; 4] = *b"LPCK";
const CAPSULE_CODEC_MAJOR: u16 = 1;
const CAPSULE_CODEC_MINOR: u16 = 0;
const CAPSULE_CODEC_MINOR_SEALED: u16 = 1;
const CAPSULE_HEADER_LEN: usize = 28;
const SECTION_HEADER_LEN: u32 = 6;
const CONVERSATION_SECTION_TAG: u16 = 1;
const CONVERSATION_SECTION_LEN: u32 = 9;
const GENESIS_VALIDATED_FLAG: u8 = 0b1;
const CONVERSATION_FRAME_LEN: u32 = SECTION_HEADER_LEN + CONVERSATION_SECTION_LEN;
const SECTION_TAGS: [u16; 5] = [
CONVERSATION_SECTION_TAG,
PARTICIPANTS_SECTION_TAG,
FRONTIERS_SECTION_TAG,
CLOSURE_SECTION_TAG,
SEALED_SECTION_TAG,
];
#[derive(Debug)]
pub struct LifecycleView<'a> {
pub conversation: &'a ParticipantConversation,
pub lifecycle: Option<ConversationLifecycleView<'a>>,
pub sealed: Option<SealedLifecycleView<'a>>,
pub covered_log_sequence: u64,
}
#[derive(Debug)]
pub struct SealedLifecycleView<'a> {
pub participants: Vec<DigestParticipantView<'a>>,
pub ledger_high_watermark: u64,
}
#[derive(Debug)]
pub struct ConversationLifecycleView<'a> {
pub participants: Vec<DigestParticipantView<'a>>,
pub frontiers: &'a ClaimFrontiers,
pub retained_record_limit: u64,
pub closure: &'a ClosureState,
}
#[derive(Debug, PartialEq, Eq)]
pub struct LifecycleCapsule {
conversation_id: ConversationId,
covered_log_sequence: u64,
conversation: ConversationSection,
lifecycle: Option<LifecycleSections>,
sealed: Option<SealedSections>,
body_len: u32,
}
#[derive(Debug, PartialEq, Eq)]
struct SealedSections {
participants: Vec<DigestParticipantRestore>,
ledger_high_watermark: u64,
encoded: Vec<u8>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct ConversationSection {
next_event_ordinal: u64,
genesis_validated: bool,
}
impl ConversationSection {
const fn is_consistent(self) -> bool {
self.genesis_validated == (self.next_event_ordinal > 0)
}
}
#[derive(Debug, PartialEq, Eq)]
struct LifecycleSections {
description: DigestDescription<CredentialDigest>,
encoded: Vec<u8>,
}
impl LifecycleCapsule {
pub fn capture(view: &LifecycleView<'_>) -> Result<Self, CapsuleCaptureError> {
let conversation_id = view.conversation.conversation_id();
if view.lifecycle.is_some() && view.sealed.is_some() {
return Err(CapsuleCaptureError::SealedBesideLifecycle);
}
let lifecycle = view
.lifecycle
.as_ref()
.map(|lifecycle| capture_lifecycle(conversation_id, lifecycle))
.transpose()?;
let sealed = view.sealed.as_ref().map(capture_sealed).transpose()?;
let lifecycle_len = lifecycle
.as_ref()
.map_or(0, |sections| sections.encoded.len())
.checked_add(sealed.as_ref().map_or(0, |sections| sections.encoded.len()))
.ok_or(CapsuleCaptureError::LengthOverflow)?;
let body_len = u32::try_from(lifecycle_len)
.map_err(|_| CapsuleCaptureError::LengthOverflow)?
.checked_add(CONVERSATION_FRAME_LEN)
.ok_or(CapsuleCaptureError::LengthOverflow)?;
Ok(Self {
conversation_id,
covered_log_sequence: view.covered_log_sequence,
conversation: ConversationSection {
next_event_ordinal: view.conversation.next_event_ordinal(),
genesis_validated: view.conversation.genesis_validated(),
},
lifecycle,
sealed,
body_len,
})
}
#[must_use]
pub const fn is_sealed(&self) -> bool {
self.sealed.is_some()
}
#[must_use]
pub const fn conversation_id(&self) -> ConversationId {
self.conversation_id
}
#[must_use]
pub const fn covered_log_sequence(&self) -> u64 {
self.covered_log_sequence
}
#[must_use]
pub fn encode_canonical(&self) -> Vec<u8> {
let mut encoded = Vec::with_capacity(CAPSULE_HEADER_LEN + self.body_len as usize);
let minor = if self.sealed.is_some() {
CAPSULE_CODEC_MINOR_SEALED
} else {
CAPSULE_CODEC_MINOR
};
encoded.extend_from_slice(&CAPSULE_MAGIC);
encoded.extend_from_slice(&CAPSULE_CODEC_MAJOR.to_be_bytes());
encoded.extend_from_slice(&minor.to_be_bytes());
encoded.extend_from_slice(&self.conversation_id.to_be_bytes());
encoded.extend_from_slice(&self.covered_log_sequence.to_be_bytes());
encoded.extend_from_slice(&self.body_len.to_be_bytes());
encoded.extend_from_slice(&CONVERSATION_SECTION_TAG.to_be_bytes());
encoded.extend_from_slice(&CONVERSATION_SECTION_LEN.to_be_bytes());
encoded.extend_from_slice(&self.conversation.next_event_ordinal.to_be_bytes());
encoded.push(if self.conversation.genesis_validated {
GENESIS_VALIDATED_FLAG
} else {
0
});
if let Some(sections) = &self.lifecycle {
encoded.extend_from_slice(§ions.encoded);
}
if let Some(sections) = &self.sealed {
encoded.extend_from_slice(§ions.encoded);
}
encoded
}
pub fn decode_canonical(input: &[u8]) -> Result<Self, CapsuleDecodeError> {
if input.len() < CAPSULE_HEADER_LEN {
return Err(CapsuleDecodeError::Truncated {
required: CAPSULE_HEADER_LEN,
available: input.len(),
});
}
if input.get(0..4) != Some(CAPSULE_MAGIC.as_slice()) {
return Err(CapsuleDecodeError::InvalidMagic);
}
let major = take_u16(input, 4)?;
let minor = take_u16(input, 6)?;
if major != CAPSULE_CODEC_MAJOR || minor > CAPSULE_CODEC_MINOR_SEALED {
return Err(CapsuleDecodeError::UnsupportedCodec { major, minor });
}
let conversation_id = take_u64(input, 8)?;
let covered_log_sequence = take_u64(input, 16)?;
let declared_body_len = take_u32(input, 24)?;
let actual_body_len = input.len() - CAPSULE_HEADER_LEN;
if usize::try_from(declared_body_len).map_err(|_| CapsuleDecodeError::LengthOverflow)?
!= actual_body_len
{
return Err(CapsuleDecodeError::NonCanonicalLength {
declared_body_len,
actual_body_len,
});
}
let mut at = CAPSULE_HEADER_LEN;
let conversation =
decode_conversation(take_section(input, &mut at, CONVERSATION_SECTION_TAG)?)?;
let (lifecycle, sealed) = if input.len() - at < SECTION_HEADER_LEN as usize {
(None, None)
} else {
let start = at;
let participants = take_section(input, &mut at, PARTICIPANTS_SECTION_TAG)?;
let next_tag = if input.len() - at < SECTION_HEADER_LEN as usize {
None
} else {
Some(take_u16(input, at)?)
};
if next_tag == Some(SEALED_SECTION_TAG) {
if minor != CAPSULE_CODEC_MINOR_SEALED {
return Err(CapsuleDecodeError::UnknownSection {
tag: SEALED_SECTION_TAG,
});
}
let sealed = take_section(input, &mut at, SEALED_SECTION_TAG)?;
let (participants, ledger_high_watermark) =
decode_sealed_description::<32, CredentialDigest>(participants, sealed)?;
let encoded = input
.get(start..at)
.ok_or(CapsuleDecodeError::LengthOverflow)?
.to_vec();
(
None,
Some(SealedSections {
participants,
ledger_high_watermark,
encoded,
}),
)
} else {
(
Some(decode_lifecycle(
input,
&mut at,
start,
participants,
conversation_id,
)?),
None,
)
}
};
if minor == CAPSULE_CODEC_MINOR_SEALED && sealed.is_none() {
return Err(CapsuleDecodeError::MissingSection {
tag: SEALED_SECTION_TAG,
});
}
if at != input.len() {
return Err(CapsuleDecodeError::TrailingBytes {
trailing: input.len() - at,
});
}
Ok(Self {
conversation_id,
covered_log_sequence,
conversation,
lifecycle,
sealed,
body_len: declared_body_len,
})
}
#[doc(hidden)]
pub fn restore_with_binding_sources(
self,
sources: &[(ParticipantId, &[u8])],
) -> Result<RestoredLifecycle, BindingOriginSourceError> {
let mut sources = sources.iter();
if let Some(lifecycle) = &self.lifecycle {
for participant in &lifecycle.description.restore.participants {
if let super::storage::ParticipantLifecycleRestore::Live {
identity,
binding_origin: Some(expected),
..
} = participant
{
let Some(&(participant_id, bytes)) = sources.next() else {
return Err(BindingOriginSourceError::Association {
participant_id: identity.participant_id,
});
};
let origin = super::checkpoint_codec::decode_binding_origin_source(bytes)
.map_err(|error| BindingOriginSourceError::Decode {
participant_id,
error,
})?;
if participant_id != identity.participant_id || origin != *expected {
return Err(BindingOriginSourceError::Association { participant_id });
}
}
}
}
if let Some(&(participant_id, _)) = sources.next() {
return Err(BindingOriginSourceError::Association { participant_id });
}
self.restore().map_err(BindingOriginSourceError::Restore)
}
pub fn restore(self) -> Result<RestoredLifecycle, CapsuleRestoreError> {
if !self.conversation.is_consistent() {
return Err(CapsuleRestoreError::InconsistentConversation);
}
let lifecycle = self
.lifecycle
.map(|sections| sections.description.into_state())
.transpose()
.map_err(CapsuleRestoreError::Lifecycle)?;
let sealed = self.sealed.map(restore_sealed).transpose()?;
Ok(RestoredLifecycle {
covered_log_sequence: self.covered_log_sequence,
restored_fenced_sources: BTreeSet::new(),
conversation: ParticipantConversation::from_checkpoint(
ConversationGenesis::new(self.conversation_id),
self.conversation.next_event_ordinal,
self.conversation.genesis_validated,
),
lifecycle,
sealed,
})
}
}
fn capture_lifecycle(
conversation_id: ConversationId,
view: &ConversationLifecycleView<'_>,
) -> Result<LifecycleSections, CapsuleCaptureError> {
if view.frontiers.conversation_id() != conversation_id {
return Err(CapsuleCaptureError::ForeignConversation);
}
for pair in view.participants.windows(2) {
if let [earlier, later] = pair
&& earlier.participant_id() >= later.participant_id()
{
return Err(CapsuleCaptureError::ParticipantOutOfOrder {
participant_id: later.participant_id(),
});
}
}
let description = describe_conversation(
&view.participants,
view.frontiers,
view.retained_record_limit,
*view.closure,
)
.map_err(CapsuleCaptureError::Describe)?;
let encoded = encode_description(&description).ok_or(CapsuleCaptureError::LengthOverflow)?;
Ok(LifecycleSections {
description,
encoded,
})
}
fn capture_sealed(view: &SealedLifecycleView<'_>) -> Result<SealedSections, CapsuleCaptureError> {
for pair in view.participants.windows(2) {
if let [earlier, later] = pair
&& earlier.participant_id() >= later.participant_id()
{
return Err(CapsuleCaptureError::ParticipantOutOfOrder {
participant_id: later.participant_id(),
});
}
}
if let Some(live) = view
.participants
.iter()
.find(|participant| matches!(participant, DigestParticipantView::Live { .. }))
{
return Err(CapsuleCaptureError::LiveParticipantSealed {
participant_id: live.participant_id(),
});
}
let participants: Vec<_> = view
.participants
.iter()
.map(ParticipantLifecycleView::storage_description)
.collect();
let encoded = encode_sealed_description(&participants, view.ledger_high_watermark)
.ok_or(CapsuleCaptureError::LengthOverflow)?;
Ok(SealedSections {
participants,
ledger_high_watermark: view.ledger_high_watermark,
encoded,
})
}
fn restore_sealed(sections: SealedSections) -> Result<SealedLifecycle, CapsuleRestoreError> {
let mut retired = Vec::with_capacity(sections.participants.len());
for participant in sections.participants {
match participant {
ParticipantLifecycleRestore::Retired(identity) => {
retired.push(identity.restore().map_err(CapsuleRestoreError::Sealed)?);
}
ParticipantLifecycleRestore::Live { identity, .. } => {
return Err(CapsuleRestoreError::SealedLiveParticipant {
participant_id: identity.participant_id,
});
}
}
}
Ok(SealedLifecycle {
retired,
ledger_high_watermark: sections.ledger_high_watermark,
})
}
fn decode_lifecycle(
input: &[u8],
at: &mut usize,
start: usize,
participants: &[u8],
conversation_id: ConversationId,
) -> Result<LifecycleSections, CapsuleDecodeError> {
let frontiers = take_section(input, at, FRONTIERS_SECTION_TAG)?;
let closure = take_section(input, at, CLOSURE_SECTION_TAG)?;
let description = decode_description::<32, CredentialDigest>(participants, frontiers, closure)?;
if description.restore.frontiers.conversation_id != conversation_id {
return Err(CapsuleDecodeError::ForeignConversation {
tag: FRONTIERS_SECTION_TAG,
});
}
let encoded = input
.get(start..*at)
.ok_or(CapsuleDecodeError::LengthOverflow)?
.to_vec();
Ok(LifecycleSections {
description,
encoded,
})
}
fn take_section<'a>(
input: &'a [u8],
at: &mut usize,
expected: u16,
) -> Result<&'a [u8], CapsuleDecodeError> {
if *at == input.len() {
return Err(CapsuleDecodeError::MissingSection { tag: expected });
}
let tag = take_u16(input, *at)?;
let len = take_u32(input, *at + 2)?;
let start = *at + SECTION_HEADER_LEN as usize;
let end = start
.checked_add(usize::try_from(len).map_err(|_| CapsuleDecodeError::LengthOverflow)?)
.ok_or(CapsuleDecodeError::LengthOverflow)?;
let body = input.get(start..end).ok_or(CapsuleDecodeError::Truncated {
required: end,
available: input.len(),
})?;
if tag != expected {
return Err(if SECTION_TAGS.contains(&tag) {
CapsuleDecodeError::SectionOutOfOrder { tag }
} else {
CapsuleDecodeError::UnknownSection { tag }
});
}
*at = end;
Ok(body)
}
#[derive(Debug, PartialEq, Eq)]
pub struct RestoredLifecycle {
covered_log_sequence: u64,
restored_fenced_sources: BTreeSet<ParticipantId>,
conversation: ParticipantConversation,
lifecycle: Option<DigestConversationState>,
sealed: Option<SealedLifecycle>,
}
#[derive(Debug, PartialEq, Eq)]
pub struct SealedLifecycle {
retired: Vec<DigestRetiredIdentity>,
ledger_high_watermark: u64,
}
impl SealedLifecycle {
#[must_use]
pub fn retired(&self) -> &[DigestRetiredIdentity] {
&self.retired
}
#[must_use]
pub const fn ledger_high_watermark(&self) -> u64 {
self.ledger_high_watermark
}
#[must_use]
pub fn into_parts(self) -> (Vec<DigestRetiredIdentity>, u64) {
(self.retired, self.ledger_high_watermark)
}
}
impl RestoredLifecycle {
#[doc(hidden)]
pub fn restore_checkpoint_observer_projection(
&self,
bytes: &[u8],
participant_id: ParticipantId,
) -> Result<super::ObserverProgressProjection, super::OwnerCheckpointRestoreError> {
let (conversation_id, progress, source) =
super::checkpoint_codec::decode_observer_source(bytes)?;
if conversation_id != self.conversation.conversation_id()
|| source.participant_id() != participant_id
{
return Err(super::OwnerCheckpointRestoreError::Field(
"observer_projection.capsule_association",
));
}
let high_watermark = match (&self.lifecycle, &self.sealed) {
(Some(lifecycle), _) => lifecycle.frontiers().sequence().ledger().high_watermark(),
(None, Some(sealed)) => sealed.ledger_high_watermark,
(None, None) => {
return Err(super::OwnerCheckpointRestoreError::Field(
"observer_projection.lifecycle",
));
}
};
if u128::from(progress) > u128::from(high_watermark) + 1 {
return Err(super::OwnerCheckpointRestoreError::Field(
"observer_projection.delivery_coverage",
));
}
let generation = match (&self.lifecycle, &self.sealed) {
(Some(lifecycle), _) => match lifecycle
.participants()
.iter()
.find(|participant| participant.view().participant_id() == participant_id)
{
Some(super::RestoredParticipantLifecycle::Live { member, .. }) => {
Some(member.generation())
}
Some(super::RestoredParticipantLifecycle::Retired(retired)) => {
Some(retired.retired_generation())
}
None if lifecycle
.frontiers()
.active_identities()
.participants()
.iter()
.all(|active| active.participant_index() != participant_id) =>
{
None
}
None => {
return Err(super::OwnerCheckpointRestoreError::Field(
"observer_projection.participant",
));
}
},
(None, Some(sealed)) => sealed
.retired
.iter()
.find(|retired| retired.participant_id() == participant_id)
.map(super::RetiredIdentity::retired_generation),
(None, None) => {
return Err(super::OwnerCheckpointRestoreError::Field(
"observer_projection.lifecycle",
));
}
};
if let Some(generation) = generation
&& source
.binding_epoch()
.is_some_and(|epoch| epoch.capability_generation.get() > generation.get())
{
return Err(super::OwnerCheckpointRestoreError::Field(
"observer_projection.binding_generation",
));
}
Ok(super::ObserverProgressProjection::sourced(
conversation_id,
progress,
source,
))
}
#[doc(hidden)]
pub fn restore_checkpoint_fenced_attach(
&mut self,
source: super::FencedAttachCommitRestore,
marker_record: &[u8],
) -> Result<super::FencedAttachCommit, super::OwnerCheckpointRestoreError> {
use super::OwnerCheckpointRestoreError as Error;
if self
.restored_fenced_sources
.contains(&source.participant_id)
{
return Err(Error::Field("fenced_source.duplicate"));
}
let lifecycle = self
.lifecycle
.as_ref()
.ok_or(Error::Field("fenced_source.lifecycle"))?;
let origin = lifecycle
.participants()
.iter()
.find_map(|participant| match participant {
super::RestoredParticipantLifecycle::Live {
member,
binding_origin: Some(origin),
..
} if member.participant_id() == source.participant_id => Some(*origin),
_ => None,
})
.ok_or(Error::Field("fenced_source.binding_origin"))?;
if origin.binding_epoch() != source.new_binding_epoch
|| origin.recovered_marker()
!= Some((source.marker_delivery_seq, source.prior_binding_epoch))
{
return Err(Error::Field("fenced_source.origin_association"));
}
let record = lifecycle
.frontiers()
.checkpoint_recovered_marker_record(origin)
.ok_or(Error::Field("fenced_source.historical_marker"))?;
if super::checkpoint_codec::encode_marker_source(&record) != marker_record {
return Err(Error::Field("fenced_source.marker_bytes"));
}
let restored = source
.restore_with_record(&record)
.map_err(Error::Provenance);
record.consume();
let restored = restored?;
self.restored_fenced_sources.insert(source.participant_id);
Ok(restored)
}
#[must_use]
pub const fn covered_log_sequence(&self) -> u64 {
self.covered_log_sequence
}
#[must_use]
pub fn into_parts(self) -> RestoredLifecycleParts {
RestoredLifecycleParts {
conversation: self.conversation,
lifecycle: self.lifecycle,
sealed: self.sealed,
}
}
#[must_use]
pub const fn sealed(&self) -> Option<&SealedLifecycle> {
self.sealed.as_ref()
}
}
#[derive(Debug)]
#[non_exhaustive]
pub struct RestoredLifecycleParts {
pub conversation: ParticipantConversation,
pub lifecycle: Option<DigestConversationState>,
pub sealed: Option<SealedLifecycle>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum CapsuleCaptureError {
ForeignConversation,
ParticipantOutOfOrder {
participant_id: ParticipantId,
},
Describe(ConversationDescribeError),
LengthOverflow,
SealedBesideLifecycle,
LiveParticipantSealed {
participant_id: ParticipantId,
},
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum CapsuleDecodeError {
Truncated {
required: usize,
available: usize,
},
InvalidMagic,
UnsupportedCodec {
major: u16,
minor: u16,
},
NonCanonicalLength {
declared_body_len: u32,
actual_body_len: usize,
},
MissingSection {
tag: u16,
},
SectionOutOfOrder {
tag: u16,
},
UnknownSection {
tag: u16,
},
NonCanonicalSectionLength {
tag: u16,
declared: u32,
},
NonCanonicalSection {
tag: u16,
},
ParticipantOutOfOrder {
participant_id: ParticipantId,
},
ForeignConversation {
tag: u16,
},
TrailingBytes {
trailing: usize,
},
LengthOverflow,
}
#[derive(Clone, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum CapsuleRestoreError {
InconsistentConversation,
Lifecycle(ConversationStateRestoreError),
Sealed(StorageRestoreError),
SealedLiveParticipant {
participant_id: ParticipantId,
},
}
fn decode_conversation(body: &[u8]) -> Result<ConversationSection, CapsuleDecodeError> {
if body.len() != CONVERSATION_SECTION_LEN as usize {
return Err(CapsuleDecodeError::NonCanonicalSectionLength {
tag: CONVERSATION_SECTION_TAG,
declared: u32::try_from(body.len()).map_err(|_| CapsuleDecodeError::LengthOverflow)?,
});
}
let next_event_ordinal = take_u64(body, 0)?;
let flags = body.get(8).copied().ok_or(CapsuleDecodeError::Truncated {
required: 9,
available: body.len(),
})?;
if flags & !GENESIS_VALIDATED_FLAG != 0 {
return Err(CapsuleDecodeError::NonCanonicalSection {
tag: CONVERSATION_SECTION_TAG,
});
}
let section = ConversationSection {
next_event_ordinal,
genesis_validated: flags & GENESIS_VALIDATED_FLAG != 0,
};
if !section.is_consistent() {
return Err(CapsuleDecodeError::NonCanonicalSection {
tag: CONVERSATION_SECTION_TAG,
});
}
Ok(section)
}
fn take<const N: usize>(input: &[u8], start: usize) -> Result<[u8; N], CapsuleDecodeError> {
let end = start
.checked_add(N)
.ok_or(CapsuleDecodeError::LengthOverflow)?;
input
.get(start..end)
.ok_or(CapsuleDecodeError::Truncated {
required: end,
available: input.len(),
})?
.try_into()
.map_err(|_| CapsuleDecodeError::LengthOverflow)
}
fn take_u16(input: &[u8], start: usize) -> Result<u16, CapsuleDecodeError> {
take(input, start).map(u16::from_be_bytes)
}
fn take_u32(input: &[u8], start: usize) -> Result<u32, CapsuleDecodeError> {
take(input, start).map(u32::from_be_bytes)
}
fn take_u64(input: &[u8], start: usize) -> Result<u64, CapsuleDecodeError> {
take(input, start).map(u64::from_be_bytes)
}
#[doc(hidden)]
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum BindingOriginSourceError {
Decode {
participant_id: ParticipantId,
error: CapsuleDecodeError,
},
Association {
participant_id: ParticipantId,
},
Restore(CapsuleRestoreError),
}
impl RestoredLifecycle {
pub fn restore_cursor_episode(
self,
bytes: &[u8],
observer_progress: u64,
) -> Result<RestoredLifecycleWithCursorEpisode, super::StorageRestoreError> {
let lifecycle = self
.lifecycle
.as_ref()
.ok_or(super::StorageRestoreError::CursorEpisode)?;
let ClosureState::Owed { debt, .. } = lifecycle.closure() else {
return Err(super::StorageRestoreError::CursorEpisode);
};
let episode = super::NonzeroDebtCursorEpisode::decode_checkpoint(
bytes,
lifecycle.frontiers(),
debt,
observer_progress,
)?;
Ok(RestoredLifecycleWithCursorEpisode {
lifecycle: self,
episode,
})
}
}
#[derive(Debug)]
pub struct RestoredLifecycleWithCursorEpisode {
lifecycle: RestoredLifecycle,
episode: super::NonzeroDebtCursorEpisode,
}
impl RestoredLifecycleWithCursorEpisode {
#[must_use]
pub fn into_parts(self) -> (RestoredLifecycleParts, super::NonzeroDebtCursorEpisode) {
(self.lifecycle.into_parts(), self.episode)
}
}
impl LifecycleCapsule {
pub fn capture_with_binding_sources(
view: &LifecycleView<'_>,
sources: &[(ParticipantId, &[u8])],
) -> Result<Self, BindingOriginCaptureError> {
let Some(lifecycle) = view.lifecycle.as_ref() else {
if let Some(&(participant_id, _)) = sources.first() {
return Err(BindingOriginCaptureError::Association { participant_id });
}
return Self::capture(view).map_err(BindingOriginCaptureError::Capture);
};
let mut remaining = sources.iter().peekable();
let mut origins = Vec::new();
for participant in &lifecycle.participants {
match participant {
DigestParticipantView::Retired(_) => origins.push(None),
DigestParticipantView::Live {
member,
binding,
binding_origin,
..
} => {
let participant_id = member.participant_id();
let origin_required = !matches!(binding, super::BindingState::Detached)
|| member.latest_terminal().is_some()
|| binding_origin.is_some();
let Some(&&(source_id, bytes)) = remaining.peek() else {
if origin_required {
return Err(BindingOriginCaptureError::Association { participant_id });
}
origins.push(None);
continue;
};
if source_id != participant_id {
if origin_required || source_id < participant_id {
return Err(BindingOriginCaptureError::Association { participant_id });
}
origins.push(None);
continue;
}
remaining.next();
let origin = super::checkpoint_codec::decode_binding_origin_source(bytes)
.map_err(|error| BindingOriginCaptureError::Decode {
participant_id,
error,
})?;
super::storage::validate_binding_origin(origin, member, **binding).map_err(
|error| BindingOriginCaptureError::Validation {
participant_id,
error,
detail: origin_disagreement(&**binding, &origin, member),
},
)?;
if binding_origin.is_some_and(|expected| *expected != origin) {
return Err(BindingOriginCaptureError::Association { participant_id });
}
origins.push(Some(origin));
}
}
}
if let Some(&(participant_id, _)) = remaining.next() {
return Err(BindingOriginCaptureError::Association { participant_id });
}
let participants = lifecycle
.participants
.iter()
.zip(&origins)
.map(|(participant, origin)| match participant {
DigestParticipantView::Live {
member,
binding,
detach_cell,
..
} => DigestParticipantView::Live {
member,
binding,
detach_cell,
binding_origin: origin.as_ref(),
},
DigestParticipantView::Retired(retired) => DigestParticipantView::Retired(retired),
})
.collect();
let capsule = Self::capture(&LifecycleView {
sealed: None,
conversation: view.conversation,
covered_log_sequence: view.covered_log_sequence,
lifecycle: Some(ConversationLifecycleView {
participants,
frontiers: lifecycle.frontiers,
retained_record_limit: lifecycle.retained_record_limit,
closure: lifecycle.closure,
}),
})
.map_err(BindingOriginCaptureError::Capture)?;
let canonical = capsule.encode_canonical();
capsule
.restore()
.map_err(BindingOriginCaptureError::Restore)?;
Self::decode_canonical(&canonical).map_err(BindingOriginCaptureError::Canonical)
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum BindingOriginCaptureError {
Restore(CapsuleRestoreError),
Canonical(CapsuleDecodeError),
Decode {
participant_id: ParticipantId,
error: CapsuleDecodeError,
},
Association {
participant_id: ParticipantId,
},
Validation {
participant_id: ParticipantId,
error: super::StorageRestoreError,
detail: String,
},
Capture(CapsuleCaptureError),
}
fn origin_disagreement(
binding: &impl core::fmt::Debug,
origin: &super::BindingOrigin,
member: &super::LiveMember<CredentialDigest>,
) -> String {
format!(
"binding {binding:?}; origin for participant {} conversation {} epoch {:?}; \
member conversation {} generation {:?}; latest terminal epoch {:?}",
origin.participant_id(),
origin.conversation_id(),
origin.binding_epoch(),
member.conversation_id(),
member.generation(),
member
.latest_terminal()
.map(super::CommittedBindingTerminal::binding_epoch),
)
}