liminal-server 0.21.1

Standalone server for the liminal messaging bus
Documentation
//! Fresh restoration of the exact active fate-occurrence table.

use std::collections::BTreeMap;

use super::super::super::restore_checkpoint::{CheckpointError, CheckpointPositions};
use super::super::super::restore_checkpoint_history_key::{CheckpointHistoryKey, HistoryEntryKey};
use super::super::super::state::checkpoint_body::{CheckpointBodyDecoder, invalid};
use super::super::{FateOccurrenceKey, FateOccurrenceRouter, FateOccurrenceState};

impl FateOccurrenceRouter {
    pub(in crate::server::participant::production) fn decode_checkpoint(
        bytes: &[u8],
        positions: &CheckpointPositions,
    ) -> Result<Self, CheckpointError> {
        let mut decoder = CheckpointBodyDecoder::new(bytes);
        let count = decoder.count("fate occurrences", 8)?;
        let mut occurrences = BTreeMap::new();
        let mut previous = None;
        for _ in 0..count {
            let key_bytes = decoder.bytes("fate occurrence key")?;
            let history_key =
                CheckpointHistoryKey::decode_canonical(key_bytes, positions.conversation_id)
                    .map_err(|source| CheckpointError::LiveState {
                        source: Box::new(source),
                    })?;
            if history_key.encode_canonical() != key_bytes {
                return Err(invalid("canonical fate occurrence key"));
            }
            let HistoryEntryKey::FateOccurrence {
                participant_id,
                binding_epoch,
            } = history_key.entry()
            else {
                return Err(invalid("fate occurrence key kind"));
            };
            if participant_id >= positions.next_participant {
                return Err(invalid("fate occurrence participant"));
            }
            let key = FateOccurrenceKey {
                conversation_id: positions.conversation_id,
                participant_id,
                binding_epoch,
            };
            if previous.is_some_and(|last| last >= key) {
                return Err(invalid("fate occurrence order"));
            }
            previous = Some(key);
            let value_bytes = decoder.bytes("fate occurrence value")?;
            let value = FateOccurrenceState::decode_history_value(
                value_bytes,
                key,
                positions.covered_log_sequence,
            )
            .map_err(|source| CheckpointError::LiveState {
                source: Box::new(source),
            })?;
            if value
                .encode_history_value(key, positions.covered_log_sequence)
                .map_err(|source| CheckpointError::LiveState {
                    source: Box::new(source),
                })?
                != value_bytes
            {
                return Err(invalid("canonical fate occurrence value"));
            }
            occurrences.insert(key, value);
        }
        decoder.finish()?;
        Ok(Self { occurrences })
    }
}