liminal-server 0.20.1

//! Validate-first source and presentation checks for composed fenced terminals.

use std::collections::BTreeMap;

use super::log::{
    DecodedStoredOperation, FencedAttachProofRefusal, OperationLog, OperationLogError,
    StoredAttachModeV3, StoredComposedTerminal, StoredComposedTerminalCause,
    StoredComposedTerminalKind, StoredDetachedCause, StoredDiedCause, StoredFinalizerPresentation,
    StoredOperationV3, StoredRecoveredFate, StoredRecoveredPresentation, StoredTerminalDisposition,
};
use super::restore_checkpoint::CheckpointError;
use super::state::StateError;
use super::state::checkpoint_body::{CheckpointBodyDecoder, CheckpointBodyEncoder, invalid};

/// Config-bounded replay state for open recovered finalizer reservations.
pub(super) struct ComposedTerminalValidation {
    recovered_reservations: BTreeMap<u64, u64>,
    maximum_recovered_reservations: usize,
}

/// Separates authoritative source failures from semantic audit refusals.
#[derive(Debug)]
pub(super) enum AuditSourceError<E> {
    Semantic(StateError),
    Source(E),
}
impl<E> From<StateError> for AuditSourceError<E> {
    fn from(error: StateError) -> Self {
        Self::Semantic(error)
    }
}

/// Move-owned pending/recovered source pairs, with configuration supplied at restore.
#[derive(Debug)]
pub(super) struct ComposedTerminalRestoreParts {
    recovered_reservations: BTreeMap<u64, u64>,
}

impl ComposedTerminalRestoreParts {
    /// Takes raw checkpoint pairs for validation at the covered base head.
    pub(super) const fn from_reservations(recovered_reservations: BTreeMap<u64, u64>) -> Self {
        Self {
            recovered_reservations,
        }
    }
}

impl ComposedTerminalValidation {
    /// Captures exact reservation authority by borrowing its active pairs.
    pub(super) fn encode_checkpoint(&self) -> Result<Vec<u8>, CheckpointError> {
        encode_reservations(&self.recovered_reservations)
    }

    /// Restores canonical pairs through the existing covered-head validator.
    pub(super) fn decode_checkpoint(
        bytes: &[u8],
        identity_slots: u64,
        covered_log_sequence: u64,
    ) -> Result<Self, CheckpointError> {
        let mut decoder = CheckpointBodyDecoder::new(bytes);
        let count = decoder.count("audit reservation count", 16)?;
        let mut reservations = BTreeMap::new();
        let mut previous = None;
        for _ in 0..count {
            let pending = decoder.u64("audit pending source")?;
            if previous.is_some_and(|value| value >= pending) {
                return Err(invalid("audit pending source order"));
            }
            previous = Some(pending);
            let recovered = decoder.u64("audit recovered source")?;
            reservations.insert(pending, recovered);
        }
        decoder.finish()?;
        Self::from_restore_parts(
            ComposedTerminalRestoreParts::from_reservations(reservations),
            identity_slots,
            covered_log_sequence,
        )
        .map_err(|source| CheckpointError::LiveState {
            source: Box::new(source),
        })
    }

    /// Validates saved pairs against the covered head and current capacity.
    pub(super) fn from_restore_parts(
        parts: ComposedTerminalRestoreParts,
        identity_slots: u64,
        covered_log_sequence: u64,
    ) -> Result<Self, StateError> {
        let mut owner = Self::for_identity_slots(identity_slots)?;
        if parts.recovered_reservations.len() > owner.maximum_recovered_reservations {
            return Err(composed_error(
                covered_log_sequence,
                FencedAttachProofRefusal::ComposedRecoveredReservationCapacity,
            ));
        }
        let mut recovered_sources = std::collections::BTreeSet::new();
        for (&pending, &recovered) in &parts.recovered_reservations {
            if pending >= recovered
                || recovered > covered_log_sequence
                || !recovered_sources.insert(recovered)
            {
                return Err(composed_error(
                    recovered,
                    FencedAttachProofRefusal::ComposedRecoveredReservationMismatch,
                ));
            }
        }
        owner.recovered_reservations = parts.recovered_reservations;
        Ok(owner)
    }

    pub(super) const fn new(maximum_recovered_reservations: usize) -> Self {
        Self {
            recovered_reservations: BTreeMap::new(),
            maximum_recovered_reservations,
        }
    }

    /// Sizes the reservation bound from the configured identity-slot count,
    /// refusing a count the address space cannot index rather than
    /// truncating it.
    pub(super) fn for_identity_slots(identity_slots: u64) -> Result<Self, StateError> {
        let maximum_recovered_reservations =
            usize::try_from(identity_slots).map_err(|_| StateError::AllocationExhausted {
                domain: "active recovered reservation count",
            })?;
        Ok(Self::new(maximum_recovered_reservations))
    }

    #[cfg(test)]
    pub(super) fn active_reservation_count(&self) -> usize {
        self.recovered_reservations.len()
    }

    /// Validates one decoded operation without mutating conversation authority.
    pub(super) async fn validate(
        &mut self,
        log: &OperationLog,
        sequence: u64,
        operation: &DecodedStoredOperation,
    ) -> Result<(), StateError> {
        match self
            .validate_from(&|source| log.read_at(source), sequence, operation)
            .await
        {
            Ok(()) => Ok(()),
            Err(AuditSourceError::Semantic(error)) => Err(error),
            Err(AuditSourceError::Source(error)) => Err(error.into()),
        }
    }

    pub(super) async fn validate_from<R, F, E>(
        &mut self,
        read_source: &R,
        sequence: u64,
        operation: &DecodedStoredOperation,
    ) -> Result<(), AuditSourceError<E>>
    where
        R: Fn(u64) -> F + Sync,
        F: std::future::Future<Output = Result<Option<super::log::DecodedOperation>, E>> + Send,
    {
        let DecodedStoredOperation::V3(operation) = operation else {
            return Ok(());
        };
        if let StoredOperationV3::Recovered { row, .. } = operation {
            return self
                .validate_recovered_reservation(sequence, row)
                .map_err(AuditSourceError::Semantic);
        }
        let StoredOperationV3::Attached { request, mode, .. } = operation else {
            return Ok(());
        };
        let StoredAttachModeV3::Fenced {
            prior_binding_epoch,
            composed_terminal: Some(terminal),
            ..
        } = mode.as_ref()
        else {
            return Ok(());
        };
        self.validate_pending_source(
            read_source,
            sequence,
            request.participant_id,
            *prior_binding_epoch,
            terminal,
        )
        .await?;
        match terminal.presentation {
            StoredFinalizerPresentation::PresentEnclosing => {
                if self
                    .recovered_reservations
                    .contains_key(&terminal.pending_source_sequence)
                {
                    return Err(composed_error(
                        sequence,
                        FencedAttachProofRefusal::ComposedRecoveredReservationMismatch,
                    )
                    .into());
                }
            }
            StoredFinalizerPresentation::ConsumeRecoveredReservation {
                recovered_source_sequence,
            } => {
                let reserved = self
                    .recovered_reservations
                    .remove(&terminal.pending_source_sequence);
                if reserved != Some(recovered_source_sequence) {
                    return Err(composed_error(
                        sequence,
                        FencedAttachProofRefusal::ComposedRecoveredReservationMismatch,
                    )
                    .into());
                }
                let recovered = read_source(recovered_source_sequence)
                    .await
                    .map_err(AuditSourceError::Source)?
                    .ok_or_else(|| {
                        composed_error(
                            sequence,
                            FencedAttachProofRefusal::ComposedRecoveredReservationMismatch,
                        )
                    })?;
                let DecodedStoredOperation::V3(StoredOperationV3::Recovered { row, .. }) =
                    recovered.operation
                else {
                    return Err(composed_error(
                        sequence,
                        FencedAttachProofRefusal::ComposedRecoveredReservationMismatch,
                    )
                    .into());
                };
                if row.presentation
                    != StoredRecoveredPresentation::RecoveredOwnsAndReservesFinalizer
                    || row.died_source_sequence != terminal.pending_source_sequence
                    || row.participant_id != request.participant_id
                {
                    return Err(composed_error(
                        sequence,
                        FencedAttachProofRefusal::ComposedRecoveredReservationMismatch,
                    )
                    .into());
                }
            }
        }
        Ok(())
    }

    fn validate_recovered_reservation(
        &mut self,
        sequence: u64,
        row: &StoredRecoveredFate,
    ) -> Result<(), StateError> {
        if row.presentation != StoredRecoveredPresentation::RecoveredOwnsAndReservesFinalizer {
            return Ok(());
        }
        if row.died_source_sequence >= sequence
            || self
                .recovered_reservations
                .contains_key(&row.died_source_sequence)
        {
            return Err(composed_error(
                sequence,
                FencedAttachProofRefusal::ComposedRecoveredReservationMismatch,
            ));
        }
        if self.recovered_reservations.len() >= self.maximum_recovered_reservations {
            return Err(composed_error(
                sequence,
                FencedAttachProofRefusal::ComposedRecoveredReservationCapacity,
            ));
        }
        self.recovered_reservations
            .insert(row.died_source_sequence, sequence);
        Ok(())
    }

    async fn validate_pending_source<R, F, E>(
        &self,
        read_source: &R,
        sequence: u64,
        participant_id: u64,
        prior_binding_epoch: super::log::StoredBindingEpoch,
        terminal: &StoredComposedTerminal,
    ) -> Result<(), AuditSourceError<E>>
    where
        R: Fn(u64) -> F + Sync,
        F: std::future::Future<Output = Result<Option<super::log::DecodedOperation>, E>> + Send,
    {
        let source = read_source(terminal.pending_source_sequence)
            .await
            .map_err(AuditSourceError::Source)?
            .ok_or_else(|| {
                composed_error(
                    sequence,
                    FencedAttachProofRefusal::ComposedPendingSourceMismatch,
                )
            })?;
        let matches = match source.operation {
            DecodedStoredOperation::V3(StoredOperationV3::Died { row }) => {
                terminal.kind == StoredComposedTerminalKind::Died
                    && row.participant_id == participant_id
                    && row.binding_epoch == prior_binding_epoch
                    && row.terminal_order == terminal.transaction_order
                    && row.disposition == StoredTerminalDisposition::Pending
                    && died_cause_matches(row.cause, terminal.cause)
            }
            DecodedStoredOperation::V3(StoredOperationV3::Detached { row }) => {
                terminal.kind == StoredComposedTerminalKind::Detached
                    && row.participant_id == participant_id
                    && row.binding_epoch == prior_binding_epoch
                    && row.terminal_order == terminal.transaction_order
                    && row.disposition == StoredTerminalDisposition::Pending
                    && detached_cause_matches(row.cause, terminal.cause)
            }
            DecodedStoredOperation::V2(_) | DecodedStoredOperation::V3(_) => false,
        };
        if matches {
            Ok(())
        } else {
            Err(composed_error(
                sequence,
                FencedAttachProofRefusal::ComposedPendingSourceMismatch,
            )
            .into())
        }
    }
}

const fn died_cause_matches(cause: StoredDiedCause, audit: StoredComposedTerminalCause) -> bool {
    match (cause, audit) {
        (StoredDiedCause::ConnectionLost, StoredComposedTerminalCause::ConnectionLost)
        | (StoredDiedCause::ProcessKilled, StoredComposedTerminalCause::ProcessKilled)
        | (StoredDiedCause::ProtocolError, StoredComposedTerminalCause::ProtocolError) => true,
        (
            StoredDiedCause::UncleanServerRestart {
                prior_server_incarnation: left,
            },
            StoredComposedTerminalCause::UncleanServerRestart {
                prior_server_incarnation: right,
            },
        ) => left == right,
        _ => false,
    }
}

const fn detached_cause_matches(
    cause: StoredDetachedCause,
    audit: StoredComposedTerminalCause,
) -> bool {
    matches!(
        (cause, audit),
        (
            StoredDetachedCause::CleanDeregister,
            StoredComposedTerminalCause::CleanDeregister
        ) | (
            StoredDetachedCause::ServerShutdown,
            StoredComposedTerminalCause::ServerShutdown
        )
    )
}

const fn composed_error(sequence: u64, reason: FencedAttachProofRefusal) -> StateError {
    StateError::Log(OperationLogError::FencedAttachProof { sequence, reason })
}

fn encode_reservations(reservations: &BTreeMap<u64, u64>) -> Result<Vec<u8>, CheckpointError> {
    let mut encoder = CheckpointBodyEncoder::new();
    encoder.count("audit reservation count", reservations.len())?;
    for (&pending, &recovered) in reservations {
        encoder.u64(pending);
        encoder.u64(recovered);
    }
    Ok(encoder.finish())
}