liminal-server 0.23.1

Standalone server for the liminal messaging bus
Documentation
//! Canonical server-owned slot receipts and provenance beside validated lifecycle values.

use super::super::facts::Digest;
use super::super::restore_checkpoint::{CheckpointError, CheckpointPositions};
use super::checkpoint_body::{CheckpointBodyDecoder, CheckpointBodyEncoder, invalid};
use super::{AttachProvenanceRecord, AttachReceiptState, RetainedSource, Slot};
use liminal_protocol::lifecycle::{
    BindingOrigin, BindingState, CredentialAttachLiveReceipt, DetachCell, EnrollmentLiveReceipt,
    LiveMember,
};
use liminal_protocol::wire::{
    AttachAttemptToken, AttachSecret, DetachAttemptToken, Generation, ReceiptExpiryReason,
};
use std::collections::BTreeMap;

#[derive(Debug, thiserror::Error)]
#[error("slot receipt codec refused: {0:?}")]
struct ReceiptCodecError(liminal_protocol::wire::CodecError);

fn wire(error: liminal_protocol::wire::CodecError) -> CheckpointError {
    CheckpointError::LiveState {
        source: Box::new(ReceiptCodecError(error)),
    }
}

pub(super) struct SlotRestoreParts {
    pub(super) retained_source: RetainedSource,
    enrollment_receipt: EnrollmentLiveReceipt,
    enrollment_receipt_expires_at: u128,
    enrollment_provenance_expires_at: u128,
    enrollment_receipt_ended: Option<ReceiptExpiryReason>,
    enrollment_provenance_displaced: bool,
    attach: Option<AttachReceiptState>,
    attach_provenance: BTreeMap<[u8; 16], AttachProvenanceRecord>,
    attach_secret: AttachSecret,
    exact_detach_token: Option<DetachAttemptToken>,
    binding_fate: Option<Vec<u8>>,
}

impl Slot {
    pub(super) fn encode_checkpoint(&self) -> Result<Vec<u8>, CheckpointError> {
        if self.enrollment_receipt.committed() != &self.enrollment_outcome {
            return Err(invalid("enrollment receipt outcome association"));
        }
        let mut encoder = CheckpointBodyEncoder::new();
        encoder.u64(self.checkpoint_source().log_sequence);
        encoder.bytes(
            "binding origin source",
            &self.checkpoint_source().binding_origin,
        )?;
        encoder.bytes(
            "enrollment receipt",
            &self.enrollment_receipt.encode_committed_body(),
        )?;
        encoder.raw(&self.enrollment_receipt_expires_at.to_le_bytes());
        encoder.raw(&self.enrollment_provenance_expires_at.to_le_bytes());
        encoder.u8(reason(self.enrollment_receipt_ended));
        encoder.u8(u8::from(self.enrollment_provenance_displaced));
        encode_attach(&mut encoder, self.attach.as_ref())?;
        encoder.count("attach provenance", self.attach_provenance.len())?;
        for (token, record) in &self.attach_provenance {
            encoder.raw(token);
            encoder.u64(record.result_generation.get());
            encoder.u8(reason(Some(record.reason)));
            encoder.raw(&record.provenance_expires_at.to_le_bytes());
        }
        encoder.raw(self.attach_secret.as_bytes());
        encoder.u8(u8::from(self.exact_detach_token.is_some()));
        if let Some(token) = self.exact_detach_token.as_ref() {
            encoder.raw(token.as_bytes());
        }
        let fate = self
            .binding_fate
            .as_ref()
            .map(super::checkpoint_binding_fate::encode_pending)
            .transpose()?;
        encoder.bytes("slot sealed fate", fate.as_deref().unwrap_or(&[]))?;
        Ok(encoder.finish())
    }
}

impl SlotRestoreParts {
    pub(super) fn stage_fenced_source(
        &self,
        participant_id: u64,
        positions: &CheckpointPositions,
        lifecycle: &mut liminal_protocol::lifecycle::RestoredLifecycle,
    ) -> Result<Option<liminal_protocol::lifecycle::FencedAttachCommit>, CheckpointError> {
        self.binding_fate
            .as_deref()
            .map(|bytes| {
                super::checkpoint_binding_fate::stage_native_source(
                    bytes,
                    positions,
                    participant_id,
                    lifecycle,
                )
            })
            .transpose()
            .map(Option::flatten)
    }

    pub(super) fn decode_checkpoint(
        bytes: &[u8],
        participant_id: u64,
        positions: &CheckpointPositions,
    ) -> Result<Self, CheckpointError> {
        let mut decoder = CheckpointBodyDecoder::new(bytes);
        let log_sequence = decoder.u64("slot source sequence")?;
        if log_sequence == 0 || log_sequence > positions.covered_log_sequence {
            return Err(invalid("slot source coverage"));
        }
        let binding_origin = decoder.bytes("binding origin source")?.to_vec();
        let enrollment_receipt =
            EnrollmentLiveReceipt::decode_committed_body(decoder.bytes("enrollment receipt")?)
                .map_err(wire)?;
        let outcome = enrollment_receipt.committed();
        if outcome.conversation_id() != positions.conversation_id
            || outcome.participant_id() != participant_id
        {
            return Err(invalid("enrollment receipt identity"));
        }
        let enrollment_receipt_expires_at = wide(&mut decoder)?;
        let enrollment_provenance_expires_at = wide(&mut decoder)?;
        if enrollment_receipt_expires_at != outcome.receipt_expires_at()
            || enrollment_provenance_expires_at != outcome.provenance_expires_at()
        {
            return Err(invalid("enrollment receipt deadlines"));
        }
        let enrollment_receipt_ended = decode_reason(decoder.u8("enrollment receipt reason")?)?;
        let enrollment_provenance_displaced = boolean(&mut decoder)?;
        let attach = decode_attach(&mut decoder, participant_id, positions.conversation_id)?;
        let count = decoder.count("attach provenance", 41)?;
        let mut attach_provenance = BTreeMap::new();
        let mut previous = None;
        for _ in 0..count {
            let token = fixed::<16>(&mut decoder)?;
            if previous.is_some_and(|value| value >= token) {
                return Err(invalid("attach provenance order"));
            }
            previous = Some(token);
            let result_generation = Generation::new(decoder.u64("provenance generation")?)
                .ok_or_else(|| invalid("provenance generation"))?;
            let reason = decode_reason(decoder.u8("attach provenance reason")?)?
                .ok_or_else(|| invalid("attach provenance reason"))?;
            let provenance_expires_at = wide(&mut decoder)?;
            attach_provenance.insert(
                token,
                AttachProvenanceRecord {
                    result_generation,
                    reason,
                    provenance_expires_at,
                },
            );
        }
        let attach_secret = AttachSecret::new(fixed::<32>(&mut decoder)?);
        let exact_detach_token = if boolean(&mut decoder)? {
            Some(DetachAttemptToken::new(fixed::<16>(&mut decoder)?))
        } else {
            None
        };
        let fate = decoder.bytes("slot sealed fate")?;
        let binding_fate = (!fate.is_empty()).then(|| fate.to_vec());
        decoder.finish()?;
        Ok(Self {
            retained_source: RetainedSource {
                log_sequence,
                binding_origin,
            },
            enrollment_receipt,
            enrollment_receipt_expires_at,
            enrollment_provenance_expires_at,
            enrollment_receipt_ended,
            enrollment_provenance_displaced,
            attach,
            attach_provenance,
            attach_secret,
            exact_detach_token,
            binding_fate,
        })
    }

    pub(super) fn install(
        self,
        member: LiveMember<Digest>,
        binding: BindingState,
        cell: DetachCell<Digest>,
        origin: Option<&BindingOrigin>,
        positions: &CheckpointPositions,
        recovered: Option<liminal_protocol::lifecycle::FencedAttachCommit>,
    ) -> Result<Slot, CheckpointError> {
        if self.enrollment_receipt.committed().participant_id() != member.participant_id()
            || self.enrollment_receipt.committed().conversation_id() != member.conversation_id()
            || self.attach_secret != member.attach_secret()
        {
            return Err(invalid("slot member association"));
        }
        let binding_fate = self
            .binding_fate
            .as_deref()
            .map(|bytes| {
                super::checkpoint_binding_fate::decode_pending(
                    bytes,
                    origin,
                    member.cursor(),
                    positions,
                    super::checkpoint_binding_fate::FateRestoreIdentity::Slot {
                        participant_id: member.participant_id(),
                    },
                    recovered,
                )
            })
            .transpose()?;
        if binding_fate
            .as_ref()
            .is_some_and(|fate| fate.attached_source_sequence != self.retained_source.log_sequence)
        {
            return Err(invalid("slot fate binding-origin source association"));
        }
        let enrollment_outcome = self.enrollment_receipt.clone().into_committed();
        Ok(Slot {
            retained_source: self.retained_source,
            member,
            binding,
            binding_fate,
            cell,
            enrollment_receipt: self.enrollment_receipt,
            enrollment_outcome,
            enrollment_receipt_expires_at: self.enrollment_receipt_expires_at,
            enrollment_provenance_expires_at: self.enrollment_provenance_expires_at,
            enrollment_receipt_ended: self.enrollment_receipt_ended,
            enrollment_provenance_displaced: self.enrollment_provenance_displaced,
            attach: self.attach,
            attach_provenance: self.attach_provenance,
            attach_secret: self.attach_secret,
            exact_detach_token: self.exact_detach_token,
        })
    }
}

fn encode_attach(
    encoder: &mut CheckpointBodyEncoder,
    attach: Option<&AttachReceiptState>,
) -> Result<(), CheckpointError> {
    encoder.u8(u8::from(attach.is_some()));
    let Some(attach) = attach else { return Ok(()) };
    if attach.receipt.committed() != &attach.outcome {
        return Err(invalid("attach receipt outcome association"));
    }
    encoder.bytes("attach receipt", &attach.receipt.encode_committed_body())?;
    encoder.raw(attach.token.as_bytes());
    encoder.raw(&attach.verifier);
    encoder.u64(attach.result_generation.get());
    encoder.raw(&attach.receipt_expires_at.to_le_bytes());
    encoder.raw(&attach.provenance_expires_at.to_le_bytes());
    Ok(())
}

fn decode_attach(
    decoder: &mut CheckpointBodyDecoder<'_>,
    participant_id: u64,
    conversation_id: u64,
) -> Result<Option<AttachReceiptState>, CheckpointError> {
    if !boolean(decoder)? {
        return Ok(None);
    }
    let receipt =
        CredentialAttachLiveReceipt::decode_committed_body(decoder.bytes("attach receipt")?)
            .map_err(wire)?;
    let token = AttachAttemptToken::new(fixed::<16>(decoder)?);
    let verifier = fixed::<32>(decoder)?;
    let result_generation = Generation::new(decoder.u64("attach generation")?)
        .ok_or_else(|| invalid("attach generation"))?;
    let receipt_expires_at = wide(decoder)?;
    let provenance_expires_at = wide(decoder)?;
    let outcome = receipt.committed();
    if outcome.conversation_id() != conversation_id
        || outcome.participant_id() != participant_id
        || outcome.token() != token
        || outcome.capability_generation() != result_generation
        || outcome.receipt_expires_at() != receipt_expires_at
        || outcome.provenance_expires_at() != provenance_expires_at
    {
        return Err(invalid("attach receipt association"));
    }
    let outcome = receipt.clone().into_committed();
    Ok(Some(AttachReceiptState {
        token,
        receipt,
        outcome,
        verifier,
        result_generation,
        receipt_expires_at,
        provenance_expires_at,
    }))
}

const fn reason(reason: Option<ReceiptExpiryReason>) -> u8 {
    match reason {
        None => 0,
        Some(ReceiptExpiryReason::Deadline) => 1,
        Some(ReceiptExpiryReason::Superseded) => 2,
    }
}
fn decode_reason(tag: u8) -> Result<Option<ReceiptExpiryReason>, CheckpointError> {
    match tag {
        0 => Ok(None),
        1 => Ok(Some(ReceiptExpiryReason::Deadline)),
        2 => Ok(Some(ReceiptExpiryReason::Superseded)),
        _ => Err(invalid("receipt reason")),
    }
}
fn boolean(decoder: &mut CheckpointBodyDecoder<'_>) -> Result<bool, CheckpointError> {
    match decoder.u8("slot flag")? {
        0 => Ok(false),
        1 => Ok(true),
        _ => Err(invalid("slot flag")),
    }
}
fn fixed<const N: usize>(
    decoder: &mut CheckpointBodyDecoder<'_>,
) -> Result<[u8; N], CheckpointError> {
    decoder
        .take("slot fixed field", N)?
        .try_into()
        .map_err(|_| invalid("slot fixed field"))
}
fn wide(decoder: &mut CheckpointBodyDecoder<'_>) -> Result<u128, CheckpointError> {
    Ok(u128::from_le_bytes(fixed::<16>(decoder)?))
}

#[cfg(test)]
mod tests {
    use super::*;
    use liminal_protocol::wire::{
        BindingEpoch, ConnectionIncarnation, EnrollBound, EnrollmentToken,
    };
    #[test]
    fn borrowed_enrollment_receipt_matches_the_existing_move_writer() -> Result<(), CheckpointError>
    {
        let outcome = EnrollBound::new(
            7,
            EnrollmentToken::new([2; 16]),
            0,
            AttachSecret::new([3; 32]),
            BindingEpoch::new(ConnectionIncarnation::new(1, 1), Generation::ONE),
            5,
            8,
        )
        .ok_or_else(|| invalid("fixture enrollment"))?;
        let receipt = EnrollmentLiveReceipt::from_commit(outcome);
        let borrowed = receipt.encode_committed_body();
        let moved =
            EnrollmentLiveReceipt::from_commit(receipt.into_committed()).encode_committed_body();
        assert_eq!(borrowed, moved);
        let restored = EnrollmentLiveReceipt::decode_committed_body(&borrowed).map_err(wire)?;
        assert_eq!(restored.encode_committed_body(), borrowed);
        Ok(())
    }
}