liminal-protocol 0.10.0

//! Lifecycle checkpoint capsule: every protocol-owned value of one
//! conversation's lifecycle at a log position, in canonical bytes.
//!
//! A storage binding lends its live values to [`LifecycleCapsule::capture`]
//! through [`LifecycleView`], which holds references only, so a binding can
//! lend what it has but never fabricate a value. The capsule's bytes are this
//! module's own stable v1 `LPCK` envelope; the binding stores them beside its
//! own state and never reads their fields. [`LifecycleCapsule::restore`] is
//! the only way from a capsule back to live values, and it validates the
//! capsule as a whole first.
//!
//! The capsule cannot be built by a struct literal:
//!
//! ```compile_fail
//! use liminal_protocol::lifecycle::LifecycleCapsule;
//!
//! let _ = LifecycleCapsule { conversation_id: 1, covered_log_sequence: 0 };
//! ```

use alloc::vec::Vec;

use crate::wire::{ConversationId, ParticipantId};

use super::checkpoint_codec::{
    CLOSURE_SECTION_TAG, DigestDescription, FRONTIERS_SECTION_TAG, PARTICIPANTS_SECTION_TAG,
    decode_description, encode_description,
};
use super::claim_frontier::ClaimFrontiers;
use super::conversation::{ConversationGenesis, ParticipantConversation};
use super::edge::ClosureState;
use super::storage::{
    ConversationDescribeError, ConversationStateRestoreError, ParticipantConversationState,
    ParticipantLifecycleView, describe_conversation,
};

/// A participant view over the 32-byte digests every server stores.
pub type DigestParticipantView<'a> =
    ParticipantLifecycleView<'a, [u8; 32], [u8; 32], [u8; 32], [u8; 32]>;

/// Whole-conversation state over the 32-byte digests every server stores.
pub type DigestConversationState =
    ParticipantConversationState<[u8; 32], [u8; 32], [u8; 32], [u8; 32]>;

const CAPSULE_MAGIC: [u8; 4] = *b"LPCK";
const CAPSULE_CODEC_MAJOR: u16 = 1;
const CAPSULE_CODEC_MINOR: u16 = 0;
/// Magic, major, minor, conversation id, covered log sequence, body length.
const CAPSULE_HEADER_LEN: usize = 28;
/// Section tag and section body length.
const SECTION_HEADER_LEN: u32 = 6;
const CONVERSATION_SECTION_TAG: u16 = 1;
/// Next event ordinal and flags.
const CONVERSATION_SECTION_LEN: u32 = 9;
const GENESIS_VALIDATED_FLAG: u8 = 0b1;
/// The conversation section with its header, which every capsule carries.
const CONVERSATION_FRAME_LEN: u32 = SECTION_HEADER_LEN + CONVERSATION_SECTION_LEN;
/// Every assigned section tag. The conversation section is always present;
/// the participants, frontiers and closure sections follow it in this order,
/// all three or none.
const SECTION_TAGS: [u16; 4] = [
    CONVERSATION_SECTION_TAG,
    PARTICIPANTS_SECTION_TAG,
    FRONTIERS_SECTION_TAG,
    CLOSURE_SECTION_TAG,
];
/// Width of every generic digest the capsule carries: the enrollment and
/// Leave fingerprints and both request verifiers.
const DIGEST_LEN: usize = 32;

/// What a storage binding lends to [`LifecycleCapsule::capture`]: references
/// to the protocol values it holds, and the log position they cover.
#[derive(Debug)]
pub struct LifecycleView<'a> {
    /// The conversation's event-sourced shell.
    pub conversation: &'a ParticipantConversation,
    /// The participants, frontiers and closure, once the conversation has
    /// any; `None` before its first enrollment.
    pub lifecycle: Option<ConversationLifecycleView<'a>>,
    /// The last log sequence whose effects these values include.
    pub covered_log_sequence: u64,
}

/// The lifecycle values of one enrolled conversation, lent by reference.
#[derive(Debug)]
pub struct ConversationLifecycleView<'a> {
    /// Every live and retired participant, strictly ascending by participant
    /// id.
    pub participants: Vec<DigestParticipantView<'a>>,
    /// Both coupled claim frontiers.
    pub frontiers: &'a ClaimFrontiers,
    /// The signed cap on retained record facts the frontier owner keeps.
    pub retained_record_limit: u64,
    /// The current closure debt and edge.
    pub closure: &'a ClosureState,
}

/// Every protocol-owned value of one conversation's lifecycle at a log
/// position. Built only by [`LifecycleCapsule::capture`] or
/// [`LifecycleCapsule::decode_canonical`].
#[derive(Debug, PartialEq, Eq)]
pub struct LifecycleCapsule {
    conversation_id: ConversationId,
    covered_log_sequence: u64,
    conversation: ConversationSection,
    lifecycle: Option<LifecycleSections>,
    /// Exact body length after the fixed header, proven to fit at capture or
    /// read from the header at decode.
    body_len: u32,
}

/// The conversation shell's state: its genesis is the conversation id.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct ConversationSection {
    next_event_ordinal: u64,
    genesis_validated: bool,
}

impl ConversationSection {
    /// Genesis validation is the first event and happens once, so a shell
    /// past ordinal 0 has validated its genesis and one at ordinal 0 has not.
    const fn is_consistent(self) -> bool {
        self.genesis_validated == (self.next_event_ordinal > 0)
    }
}

/// The described participants, frontiers and closure, beside their exact
/// canonical section bytes (headers included).
#[derive(Debug, PartialEq, Eq)]
struct LifecycleSections {
    description: DigestDescription<DIGEST_LEN>,
    encoded: Vec<u8>,
}

impl LifecycleCapsule {
    /// Captures the lent values. Nothing is consumed; generic digests are
    /// copied out of the borrowed values.
    ///
    /// # Errors
    ///
    /// Returns [`CapsuleCaptureError`] when the frontiers belong to another
    /// conversation, the participants are not strictly ascending, the live
    /// closure cannot be described, or a count or length does not fit `u32`.
    pub fn capture(view: &LifecycleView<'_>) -> Result<Self, CapsuleCaptureError> {
        let conversation_id = view.conversation.conversation_id();
        let lifecycle = view
            .lifecycle
            .as_ref()
            .map(|lifecycle| capture_lifecycle(conversation_id, lifecycle))
            .transpose()?;
        let lifecycle_len = lifecycle
            .as_ref()
            .map_or(0, |sections| sections.encoded.len());
        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,
            body_len,
        })
    }

    /// The conversation these values belong to.
    #[must_use]
    pub const fn conversation_id(&self) -> ConversationId {
        self.conversation_id
    }

    /// The last log sequence whose effects the capsule includes.
    #[must_use]
    pub const fn covered_log_sequence(&self) -> u64 {
        self.covered_log_sequence
    }

    /// Encodes the stable v1 capsule envelope in network byte order.
    #[must_use]
    pub fn encode_canonical(&self) -> Vec<u8> {
        let mut encoded = Vec::with_capacity(CAPSULE_HEADER_LEN + self.body_len as usize);
        encoded.extend_from_slice(&CAPSULE_MAGIC);
        encoded.extend_from_slice(&CAPSULE_CODEC_MAJOR.to_be_bytes());
        encoded.extend_from_slice(&CAPSULE_CODEC_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(&sections.encoded);
        }
        encoded
    }

    /// Decodes one exact canonical v1 capsule.
    ///
    /// # Errors
    ///
    /// Returns [`CapsuleDecodeError`] for truncation, an invalid magic prefix,
    /// an unsupported codec version, a declared length that differs from the
    /// supplied bytes, a section out of order, missing, unknown or of the wrong
    /// length, a section whose fields break their invariants, participants out
    /// of order, or frontiers that belong to another conversation.
    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, minor) != (CAPSULE_CODEC_MAJOR, CAPSULE_CODEC_MINOR) {
            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 = if input.len() - at < SECTION_HEADER_LEN as usize {
            None
        } else {
            Some(decode_lifecycle(input, &mut at, conversation_id)?)
        };
        if at != input.len() {
            return Err(CapsuleDecodeError::TrailingBytes {
                trailing: input.len() - at,
            });
        }
        Ok(Self {
            conversation_id,
            covered_log_sequence,
            conversation,
            lifecycle,
            body_len: declared_body_len,
        })
    }

    /// Validates the capsule as a whole and hands back the live values.
    ///
    /// The lifecycle goes through the one joint participant-conversation
    /// validation, through its crate-private entry that also checks and
    /// installs the fenced-marker issuance latch the capsule carried: a latch
    /// naming a take the restored state could not have granted is refused.
    ///
    /// # Errors
    ///
    /// Returns [`CapsuleRestoreError`] when the capsule's sections are
    /// inconsistent with each other or with their own invariants.
    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)?;
        Ok(RestoredLifecycle {
            covered_log_sequence: self.covered_log_sequence,
            conversation: ParticipantConversation::from_checkpoint(
                ConversationGenesis::new(self.conversation_id),
                self.conversation.next_event_ordinal,
                self.conversation.genesis_validated,
            ),
            lifecycle,
        })
    }
}

/// Describes and encodes one lent lifecycle, refusing a view the capsule
/// could not restore as the same conversation.
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,
    })
}

/// Decodes the participants, frontiers and closure sections, which follow the
/// conversation section all together.
fn decode_lifecycle(
    input: &[u8],
    at: &mut usize,
    conversation_id: ConversationId,
) -> Result<LifecycleSections, CapsuleDecodeError> {
    let start = *at;
    let participants = take_section(input, at, PARTICIPANTS_SECTION_TAG)?;
    let frontiers = take_section(input, at, FRONTIERS_SECTION_TAG)?;
    let closure = take_section(input, at, CLOSURE_SECTION_TAG)?;
    let description = decode_description::<DIGEST_LEN>(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,
    })
}

/// Takes the next section, which must carry `expected`, and moves past it.
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)
}

/// Live values restored from one validated capsule.
#[derive(Debug, PartialEq, Eq)]
pub struct RestoredLifecycle {
    covered_log_sequence: u64,
    conversation: ParticipantConversation,
    lifecycle: Option<DigestConversationState>,
}

impl RestoredLifecycle {
    /// The last log sequence whose effects the restored values include; a
    /// binding replays its log from the row after it.
    #[must_use]
    pub const fn covered_log_sequence(&self) -> u64 {
        self.covered_log_sequence
    }

    /// Moves the restored values out, for the binding to install.
    #[must_use]
    pub fn into_parts(self) -> RestoredLifecycleParts {
        RestoredLifecycleParts {
            conversation: self.conversation,
            lifecycle: self.lifecycle,
        }
    }
}

/// The restored values, each for the binding to install where it keeps it.
/// It cannot be built outside the protocol: a binding receives one only from
/// [`RestoredLifecycle::into_parts`].
#[derive(Debug)]
#[non_exhaustive]
pub struct RestoredLifecycleParts {
    /// The conversation's event-sourced shell.
    pub conversation: ParticipantConversation,
    /// Every participant, both frontiers and the closure, validated together,
    /// with the fenced-marker issuance latch the capsule carried; `None` for a
    /// conversation captured before its first enrollment.
    pub lifecycle: Option<DigestConversationState>,
}

/// Live values that cannot be captured into a capsule.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum CapsuleCaptureError {
    /// The lent frontiers belong to another conversation than the shell.
    ForeignConversation,
    /// The lent participants are not strictly ascending by participant id.
    ParticipantOutOfOrder {
        /// The first participant id not above its predecessor.
        participant_id: ParticipantId,
    },
    /// The live conversation cannot be described back into restore input.
    Describe(ConversationDescribeError),
    /// A count or length does not fit its `u32` field.
    LengthOverflow,
}

/// Stable canonical capsule-decode failure.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum CapsuleDecodeError {
    /// Fewer bytes than the fixed envelope or selected field requires.
    Truncated {
        /// Minimum bytes required at the detected boundary.
        required: usize,
        /// Bytes supplied by the caller.
        available: usize,
    },
    /// The capsule does not begin with the assigned `LPCK` magic prefix.
    InvalidMagic,
    /// The capsule codec version is not supported by this crate.
    UnsupportedCodec {
        /// Presented codec major version.
        major: u16,
        /// Presented codec minor version.
        minor: u16,
    },
    /// The declared body length differs from the bytes supplied after the
    /// fixed header.
    NonCanonicalLength {
        /// Length declared in the envelope.
        declared_body_len: u32,
        /// Bytes supplied after the fixed header.
        actual_body_len: usize,
    },
    /// The body ended before a section the canonical order requires.
    MissingSection {
        /// Tag of the first absent section.
        tag: u16,
    },
    /// A known section appeared where the canonical order has another.
    SectionOutOfOrder {
        /// Tag of the misplaced section.
        tag: u16,
    },
    /// A section tag is unassigned.
    UnknownSection {
        /// Unrecognized section tag.
        tag: u16,
    },
    /// A section's declared length is not its canonical length.
    NonCanonicalSectionLength {
        /// Tag of the section.
        tag: u16,
        /// Length the section declared.
        declared: u32,
    },
    /// A complete section violates a field invariant.
    NonCanonicalSection {
        /// Tag of the section whose invariants failed.
        tag: u16,
    },
    /// The participants section is not strictly ascending by participant id.
    ParticipantOutOfOrder {
        /// The first participant id not above its predecessor.
        participant_id: ParticipantId,
    },
    /// A section names another conversation than the capsule header.
    ForeignConversation {
        /// Tag of the section naming the other conversation.
        tag: u16,
    },
    /// Bytes remain after the last canonical section.
    TrailingBytes {
        /// Bytes after the last section.
        trailing: usize,
    },
    /// A platform length conversion or offset addition overflowed.
    LengthOverflow,
}

/// A capsule whose sections do not form one valid lifecycle.
#[derive(Clone, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum CapsuleRestoreError {
    /// The shell's ordinal and genesis validation disagree.
    InconsistentConversation,
    /// The participants, frontiers and closure failed the joint
    /// participant-conversation validation.
    Lifecycle(ConversationStateRestoreError),
}

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)
}