liminal_protocol/lifecycle/
checkpoint.rs

1//! Lifecycle checkpoint capsule: every protocol-owned value of one
2//! conversation's lifecycle at a log position, in canonical bytes.
3//!
4//! A storage binding lends its live values to [`LifecycleCapsule::capture`]
5//! through [`LifecycleView`], which holds references only, so a binding can
6//! lend what it has but never fabricate a value. The capsule's bytes are this
7//! module's own stable v1 `LPCK` envelope; the binding stores them beside its
8//! own state and never reads their fields. [`LifecycleCapsule::restore`] is
9//! the only way from a capsule back to live values, and it validates the
10//! capsule as a whole first.
11//!
12//! The capsule cannot be built by a struct literal:
13//!
14//! ```compile_fail
15//! use liminal_protocol::lifecycle::LifecycleCapsule;
16//!
17//! let _ = LifecycleCapsule { conversation_id: 1, covered_log_sequence: 0 };
18//! ```
19
20use alloc::vec::Vec;
21
22use crate::wire::{ConversationId, ParticipantId};
23
24use super::checkpoint_codec::{
25    CLOSURE_SECTION_TAG, DigestDescription, FRONTIERS_SECTION_TAG, PARTICIPANTS_SECTION_TAG,
26    decode_description, encode_description,
27};
28use super::claim_frontier::ClaimFrontiers;
29use super::conversation::{ConversationGenesis, ParticipantConversation};
30use super::edge::ClosureState;
31use super::storage::{
32    ConversationDescribeError, ConversationStateRestoreError, ParticipantConversationState,
33    ParticipantLifecycleView, describe_conversation,
34};
35
36/// A participant view over the 32-byte digests every server stores.
37pub type DigestParticipantView<'a> =
38    ParticipantLifecycleView<'a, [u8; 32], [u8; 32], [u8; 32], [u8; 32]>;
39
40/// Whole-conversation state over the 32-byte digests every server stores.
41pub type DigestConversationState =
42    ParticipantConversationState<[u8; 32], [u8; 32], [u8; 32], [u8; 32]>;
43
44const CAPSULE_MAGIC: [u8; 4] = *b"LPCK";
45const CAPSULE_CODEC_MAJOR: u16 = 1;
46const CAPSULE_CODEC_MINOR: u16 = 0;
47/// Magic, major, minor, conversation id, covered log sequence, body length.
48const CAPSULE_HEADER_LEN: usize = 28;
49/// Section tag and section body length.
50const SECTION_HEADER_LEN: u32 = 6;
51const CONVERSATION_SECTION_TAG: u16 = 1;
52/// Next event ordinal and flags.
53const CONVERSATION_SECTION_LEN: u32 = 9;
54const GENESIS_VALIDATED_FLAG: u8 = 0b1;
55/// The conversation section with its header, which every capsule carries.
56const CONVERSATION_FRAME_LEN: u32 = SECTION_HEADER_LEN + CONVERSATION_SECTION_LEN;
57/// Every assigned section tag. The conversation section is always present;
58/// the participants, frontiers and closure sections follow it in this order,
59/// all three or none.
60const SECTION_TAGS: [u16; 4] = [
61    CONVERSATION_SECTION_TAG,
62    PARTICIPANTS_SECTION_TAG,
63    FRONTIERS_SECTION_TAG,
64    CLOSURE_SECTION_TAG,
65];
66/// Width of every generic digest the capsule carries: the enrollment and
67/// Leave fingerprints and both request verifiers.
68const DIGEST_LEN: usize = 32;
69
70/// What a storage binding lends to [`LifecycleCapsule::capture`]: references
71/// to the protocol values it holds, and the log position they cover.
72#[derive(Debug)]
73pub struct LifecycleView<'a> {
74    /// The conversation's event-sourced shell.
75    pub conversation: &'a ParticipantConversation,
76    /// The participants, frontiers and closure, once the conversation has
77    /// any; `None` before its first enrollment.
78    pub lifecycle: Option<ConversationLifecycleView<'a>>,
79    /// The last log sequence whose effects these values include.
80    pub covered_log_sequence: u64,
81}
82
83/// The lifecycle values of one enrolled conversation, lent by reference.
84#[derive(Debug)]
85pub struct ConversationLifecycleView<'a> {
86    /// Every live and retired participant, strictly ascending by participant
87    /// id.
88    pub participants: Vec<DigestParticipantView<'a>>,
89    /// Both coupled claim frontiers.
90    pub frontiers: &'a ClaimFrontiers,
91    /// The signed cap on retained record facts the frontier owner keeps.
92    pub retained_record_limit: u64,
93    /// The current closure debt and edge.
94    pub closure: &'a ClosureState,
95}
96
97/// Every protocol-owned value of one conversation's lifecycle at a log
98/// position. Built only by [`LifecycleCapsule::capture`] or
99/// [`LifecycleCapsule::decode_canonical`].
100#[derive(Debug, PartialEq, Eq)]
101pub struct LifecycleCapsule {
102    conversation_id: ConversationId,
103    covered_log_sequence: u64,
104    conversation: ConversationSection,
105    lifecycle: Option<LifecycleSections>,
106    /// Exact body length after the fixed header, proven to fit at capture or
107    /// read from the header at decode.
108    body_len: u32,
109}
110
111/// The conversation shell's state: its genesis is the conversation id.
112#[derive(Clone, Copy, Debug, PartialEq, Eq)]
113struct ConversationSection {
114    next_event_ordinal: u64,
115    genesis_validated: bool,
116}
117
118impl ConversationSection {
119    /// Genesis validation is the first event and happens once, so a shell
120    /// past ordinal 0 has validated its genesis and one at ordinal 0 has not.
121    const fn is_consistent(self) -> bool {
122        self.genesis_validated == (self.next_event_ordinal > 0)
123    }
124}
125
126/// The described participants, frontiers and closure, beside their exact
127/// canonical section bytes (headers included).
128#[derive(Debug, PartialEq, Eq)]
129struct LifecycleSections {
130    description: DigestDescription<DIGEST_LEN>,
131    encoded: Vec<u8>,
132}
133
134impl LifecycleCapsule {
135    /// Captures the lent values. Nothing is consumed; generic digests are
136    /// copied out of the borrowed values.
137    ///
138    /// # Errors
139    ///
140    /// Returns [`CapsuleCaptureError`] when the frontiers belong to another
141    /// conversation, the participants are not strictly ascending, the live
142    /// closure cannot be described, or a count or length does not fit `u32`.
143    pub fn capture(view: &LifecycleView<'_>) -> Result<Self, CapsuleCaptureError> {
144        let conversation_id = view.conversation.conversation_id();
145        let lifecycle = view
146            .lifecycle
147            .as_ref()
148            .map(|lifecycle| capture_lifecycle(conversation_id, lifecycle))
149            .transpose()?;
150        let lifecycle_len = lifecycle
151            .as_ref()
152            .map_or(0, |sections| sections.encoded.len());
153        let body_len = u32::try_from(lifecycle_len)
154            .map_err(|_| CapsuleCaptureError::LengthOverflow)?
155            .checked_add(CONVERSATION_FRAME_LEN)
156            .ok_or(CapsuleCaptureError::LengthOverflow)?;
157        Ok(Self {
158            conversation_id,
159            covered_log_sequence: view.covered_log_sequence,
160            conversation: ConversationSection {
161                next_event_ordinal: view.conversation.next_event_ordinal(),
162                genesis_validated: view.conversation.genesis_validated(),
163            },
164            lifecycle,
165            body_len,
166        })
167    }
168
169    /// The conversation these values belong to.
170    #[must_use]
171    pub const fn conversation_id(&self) -> ConversationId {
172        self.conversation_id
173    }
174
175    /// The last log sequence whose effects the capsule includes.
176    #[must_use]
177    pub const fn covered_log_sequence(&self) -> u64 {
178        self.covered_log_sequence
179    }
180
181    /// Encodes the stable v1 capsule envelope in network byte order.
182    #[must_use]
183    pub fn encode_canonical(&self) -> Vec<u8> {
184        let mut encoded = Vec::with_capacity(CAPSULE_HEADER_LEN + self.body_len as usize);
185        encoded.extend_from_slice(&CAPSULE_MAGIC);
186        encoded.extend_from_slice(&CAPSULE_CODEC_MAJOR.to_be_bytes());
187        encoded.extend_from_slice(&CAPSULE_CODEC_MINOR.to_be_bytes());
188        encoded.extend_from_slice(&self.conversation_id.to_be_bytes());
189        encoded.extend_from_slice(&self.covered_log_sequence.to_be_bytes());
190        encoded.extend_from_slice(&self.body_len.to_be_bytes());
191        encoded.extend_from_slice(&CONVERSATION_SECTION_TAG.to_be_bytes());
192        encoded.extend_from_slice(&CONVERSATION_SECTION_LEN.to_be_bytes());
193        encoded.extend_from_slice(&self.conversation.next_event_ordinal.to_be_bytes());
194        encoded.push(if self.conversation.genesis_validated {
195            GENESIS_VALIDATED_FLAG
196        } else {
197            0
198        });
199        if let Some(sections) = &self.lifecycle {
200            encoded.extend_from_slice(&sections.encoded);
201        }
202        encoded
203    }
204
205    /// Decodes one exact canonical v1 capsule.
206    ///
207    /// # Errors
208    ///
209    /// Returns [`CapsuleDecodeError`] for truncation, an invalid magic prefix,
210    /// an unsupported codec version, a declared length that differs from the
211    /// supplied bytes, a section out of order, missing, unknown or of the wrong
212    /// length, a section whose fields break their invariants, participants out
213    /// of order, or frontiers that belong to another conversation.
214    pub fn decode_canonical(input: &[u8]) -> Result<Self, CapsuleDecodeError> {
215        if input.len() < CAPSULE_HEADER_LEN {
216            return Err(CapsuleDecodeError::Truncated {
217                required: CAPSULE_HEADER_LEN,
218                available: input.len(),
219            });
220        }
221        if input.get(0..4) != Some(CAPSULE_MAGIC.as_slice()) {
222            return Err(CapsuleDecodeError::InvalidMagic);
223        }
224        let major = take_u16(input, 4)?;
225        let minor = take_u16(input, 6)?;
226        if (major, minor) != (CAPSULE_CODEC_MAJOR, CAPSULE_CODEC_MINOR) {
227            return Err(CapsuleDecodeError::UnsupportedCodec { major, minor });
228        }
229        let conversation_id = take_u64(input, 8)?;
230        let covered_log_sequence = take_u64(input, 16)?;
231        let declared_body_len = take_u32(input, 24)?;
232        let actual_body_len = input.len() - CAPSULE_HEADER_LEN;
233        if usize::try_from(declared_body_len).map_err(|_| CapsuleDecodeError::LengthOverflow)?
234            != actual_body_len
235        {
236            return Err(CapsuleDecodeError::NonCanonicalLength {
237                declared_body_len,
238                actual_body_len,
239            });
240        }
241        let mut at = CAPSULE_HEADER_LEN;
242        let conversation =
243            decode_conversation(take_section(input, &mut at, CONVERSATION_SECTION_TAG)?)?;
244        let lifecycle = if input.len() - at < SECTION_HEADER_LEN as usize {
245            None
246        } else {
247            Some(decode_lifecycle(input, &mut at, conversation_id)?)
248        };
249        if at != input.len() {
250            return Err(CapsuleDecodeError::TrailingBytes {
251                trailing: input.len() - at,
252            });
253        }
254        Ok(Self {
255            conversation_id,
256            covered_log_sequence,
257            conversation,
258            lifecycle,
259            body_len: declared_body_len,
260        })
261    }
262
263    /// Validates the capsule as a whole and hands back the live values.
264    ///
265    /// The lifecycle goes through the one joint participant-conversation
266    /// validation, through its crate-private entry that also checks and
267    /// installs the fenced-marker issuance latch the capsule carried: a latch
268    /// naming a take the restored state could not have granted is refused.
269    ///
270    /// # Errors
271    ///
272    /// Returns [`CapsuleRestoreError`] when the capsule's sections are
273    /// inconsistent with each other or with their own invariants.
274    pub fn restore(self) -> Result<RestoredLifecycle, CapsuleRestoreError> {
275        if !self.conversation.is_consistent() {
276            return Err(CapsuleRestoreError::InconsistentConversation);
277        }
278        let lifecycle = self
279            .lifecycle
280            .map(|sections| sections.description.into_state())
281            .transpose()
282            .map_err(CapsuleRestoreError::Lifecycle)?;
283        Ok(RestoredLifecycle {
284            covered_log_sequence: self.covered_log_sequence,
285            conversation: ParticipantConversation::from_checkpoint(
286                ConversationGenesis::new(self.conversation_id),
287                self.conversation.next_event_ordinal,
288                self.conversation.genesis_validated,
289            ),
290            lifecycle,
291        })
292    }
293}
294
295/// Describes and encodes one lent lifecycle, refusing a view the capsule
296/// could not restore as the same conversation.
297fn capture_lifecycle(
298    conversation_id: ConversationId,
299    view: &ConversationLifecycleView<'_>,
300) -> Result<LifecycleSections, CapsuleCaptureError> {
301    if view.frontiers.conversation_id() != conversation_id {
302        return Err(CapsuleCaptureError::ForeignConversation);
303    }
304    for pair in view.participants.windows(2) {
305        if let [earlier, later] = pair
306            && earlier.participant_id() >= later.participant_id()
307        {
308            return Err(CapsuleCaptureError::ParticipantOutOfOrder {
309                participant_id: later.participant_id(),
310            });
311        }
312    }
313    let description = describe_conversation(
314        &view.participants,
315        view.frontiers,
316        view.retained_record_limit,
317        *view.closure,
318    )
319    .map_err(CapsuleCaptureError::Describe)?;
320    let encoded = encode_description(&description).ok_or(CapsuleCaptureError::LengthOverflow)?;
321    Ok(LifecycleSections {
322        description,
323        encoded,
324    })
325}
326
327/// Decodes the participants, frontiers and closure sections, which follow the
328/// conversation section all together.
329fn decode_lifecycle(
330    input: &[u8],
331    at: &mut usize,
332    conversation_id: ConversationId,
333) -> Result<LifecycleSections, CapsuleDecodeError> {
334    let start = *at;
335    let participants = take_section(input, at, PARTICIPANTS_SECTION_TAG)?;
336    let frontiers = take_section(input, at, FRONTIERS_SECTION_TAG)?;
337    let closure = take_section(input, at, CLOSURE_SECTION_TAG)?;
338    let description = decode_description::<DIGEST_LEN>(participants, frontiers, closure)?;
339    if description.restore.frontiers.conversation_id != conversation_id {
340        return Err(CapsuleDecodeError::ForeignConversation {
341            tag: FRONTIERS_SECTION_TAG,
342        });
343    }
344    let encoded = input
345        .get(start..*at)
346        .ok_or(CapsuleDecodeError::LengthOverflow)?
347        .to_vec();
348    Ok(LifecycleSections {
349        description,
350        encoded,
351    })
352}
353
354/// Takes the next section, which must carry `expected`, and moves past it.
355fn take_section<'a>(
356    input: &'a [u8],
357    at: &mut usize,
358    expected: u16,
359) -> Result<&'a [u8], CapsuleDecodeError> {
360    if *at == input.len() {
361        return Err(CapsuleDecodeError::MissingSection { tag: expected });
362    }
363    let tag = take_u16(input, *at)?;
364    let len = take_u32(input, *at + 2)?;
365    let start = *at + SECTION_HEADER_LEN as usize;
366    let end = start
367        .checked_add(usize::try_from(len).map_err(|_| CapsuleDecodeError::LengthOverflow)?)
368        .ok_or(CapsuleDecodeError::LengthOverflow)?;
369    let body = input.get(start..end).ok_or(CapsuleDecodeError::Truncated {
370        required: end,
371        available: input.len(),
372    })?;
373    if tag != expected {
374        return Err(if SECTION_TAGS.contains(&tag) {
375            CapsuleDecodeError::SectionOutOfOrder { tag }
376        } else {
377            CapsuleDecodeError::UnknownSection { tag }
378        });
379    }
380    *at = end;
381    Ok(body)
382}
383
384/// Live values restored from one validated capsule.
385#[derive(Debug, PartialEq, Eq)]
386pub struct RestoredLifecycle {
387    covered_log_sequence: u64,
388    conversation: ParticipantConversation,
389    lifecycle: Option<DigestConversationState>,
390}
391
392impl RestoredLifecycle {
393    /// The last log sequence whose effects the restored values include; a
394    /// binding replays its log from the row after it.
395    #[must_use]
396    pub const fn covered_log_sequence(&self) -> u64 {
397        self.covered_log_sequence
398    }
399
400    /// Moves the restored values out, for the binding to install.
401    #[must_use]
402    pub fn into_parts(self) -> RestoredLifecycleParts {
403        RestoredLifecycleParts {
404            conversation: self.conversation,
405            lifecycle: self.lifecycle,
406        }
407    }
408}
409
410/// The restored values, each for the binding to install where it keeps it.
411/// It cannot be built outside the protocol: a binding receives one only from
412/// [`RestoredLifecycle::into_parts`].
413#[derive(Debug)]
414#[non_exhaustive]
415pub struct RestoredLifecycleParts {
416    /// The conversation's event-sourced shell.
417    pub conversation: ParticipantConversation,
418    /// Every participant, both frontiers and the closure, validated together,
419    /// with the fenced-marker issuance latch the capsule carried; `None` for a
420    /// conversation captured before its first enrollment.
421    pub lifecycle: Option<DigestConversationState>,
422}
423
424/// Live values that cannot be captured into a capsule.
425#[derive(Clone, Copy, Debug, PartialEq, Eq)]
426#[non_exhaustive]
427pub enum CapsuleCaptureError {
428    /// The lent frontiers belong to another conversation than the shell.
429    ForeignConversation,
430    /// The lent participants are not strictly ascending by participant id.
431    ParticipantOutOfOrder {
432        /// The first participant id not above its predecessor.
433        participant_id: ParticipantId,
434    },
435    /// The live conversation cannot be described back into restore input.
436    Describe(ConversationDescribeError),
437    /// A count or length does not fit its `u32` field.
438    LengthOverflow,
439}
440
441/// Stable canonical capsule-decode failure.
442#[derive(Clone, Copy, Debug, PartialEq, Eq)]
443#[non_exhaustive]
444pub enum CapsuleDecodeError {
445    /// Fewer bytes than the fixed envelope or selected field requires.
446    Truncated {
447        /// Minimum bytes required at the detected boundary.
448        required: usize,
449        /// Bytes supplied by the caller.
450        available: usize,
451    },
452    /// The capsule does not begin with the assigned `LPCK` magic prefix.
453    InvalidMagic,
454    /// The capsule codec version is not supported by this crate.
455    UnsupportedCodec {
456        /// Presented codec major version.
457        major: u16,
458        /// Presented codec minor version.
459        minor: u16,
460    },
461    /// The declared body length differs from the bytes supplied after the
462    /// fixed header.
463    NonCanonicalLength {
464        /// Length declared in the envelope.
465        declared_body_len: u32,
466        /// Bytes supplied after the fixed header.
467        actual_body_len: usize,
468    },
469    /// The body ended before a section the canonical order requires.
470    MissingSection {
471        /// Tag of the first absent section.
472        tag: u16,
473    },
474    /// A known section appeared where the canonical order has another.
475    SectionOutOfOrder {
476        /// Tag of the misplaced section.
477        tag: u16,
478    },
479    /// A section tag is unassigned.
480    UnknownSection {
481        /// Unrecognized section tag.
482        tag: u16,
483    },
484    /// A section's declared length is not its canonical length.
485    NonCanonicalSectionLength {
486        /// Tag of the section.
487        tag: u16,
488        /// Length the section declared.
489        declared: u32,
490    },
491    /// A complete section violates a field invariant.
492    NonCanonicalSection {
493        /// Tag of the section whose invariants failed.
494        tag: u16,
495    },
496    /// The participants section is not strictly ascending by participant id.
497    ParticipantOutOfOrder {
498        /// The first participant id not above its predecessor.
499        participant_id: ParticipantId,
500    },
501    /// A section names another conversation than the capsule header.
502    ForeignConversation {
503        /// Tag of the section naming the other conversation.
504        tag: u16,
505    },
506    /// Bytes remain after the last canonical section.
507    TrailingBytes {
508        /// Bytes after the last section.
509        trailing: usize,
510    },
511    /// A platform length conversion or offset addition overflowed.
512    LengthOverflow,
513}
514
515/// A capsule whose sections do not form one valid lifecycle.
516#[derive(Clone, Debug, PartialEq, Eq)]
517#[non_exhaustive]
518pub enum CapsuleRestoreError {
519    /// The shell's ordinal and genesis validation disagree.
520    InconsistentConversation,
521    /// The participants, frontiers and closure failed the joint
522    /// participant-conversation validation.
523    Lifecycle(ConversationStateRestoreError),
524}
525
526fn decode_conversation(body: &[u8]) -> Result<ConversationSection, CapsuleDecodeError> {
527    if body.len() != CONVERSATION_SECTION_LEN as usize {
528        return Err(CapsuleDecodeError::NonCanonicalSectionLength {
529            tag: CONVERSATION_SECTION_TAG,
530            declared: u32::try_from(body.len()).map_err(|_| CapsuleDecodeError::LengthOverflow)?,
531        });
532    }
533    let next_event_ordinal = take_u64(body, 0)?;
534    let flags = body.get(8).copied().ok_or(CapsuleDecodeError::Truncated {
535        required: 9,
536        available: body.len(),
537    })?;
538    if flags & !GENESIS_VALIDATED_FLAG != 0 {
539        return Err(CapsuleDecodeError::NonCanonicalSection {
540            tag: CONVERSATION_SECTION_TAG,
541        });
542    }
543    let section = ConversationSection {
544        next_event_ordinal,
545        genesis_validated: flags & GENESIS_VALIDATED_FLAG != 0,
546    };
547    if !section.is_consistent() {
548        return Err(CapsuleDecodeError::NonCanonicalSection {
549            tag: CONVERSATION_SECTION_TAG,
550        });
551    }
552    Ok(section)
553}
554
555fn take<const N: usize>(input: &[u8], start: usize) -> Result<[u8; N], CapsuleDecodeError> {
556    let end = start
557        .checked_add(N)
558        .ok_or(CapsuleDecodeError::LengthOverflow)?;
559    input
560        .get(start..end)
561        .ok_or(CapsuleDecodeError::Truncated {
562            required: end,
563            available: input.len(),
564        })?
565        .try_into()
566        .map_err(|_| CapsuleDecodeError::LengthOverflow)
567}
568
569fn take_u16(input: &[u8], start: usize) -> Result<u16, CapsuleDecodeError> {
570    take(input, start).map(u16::from_be_bytes)
571}
572
573fn take_u32(input: &[u8], start: usize) -> Result<u32, CapsuleDecodeError> {
574    take(input, start).map(u32::from_be_bytes)
575}
576
577fn take_u64(input: &[u8], start: usize) -> Result<u64, CapsuleDecodeError> {
578    take(input, start).map(u64::from_be_bytes)
579}
liminal_protocol/lifecycle/checkpoint.rs

liminal_protocol/lifecycle/
checkpoint.rs
