liminal-server 0.25.4

//! Re-execution of canonical transition inputs through `liminal-protocol`.
//!
//! Every header this module produces is the RESULT of applying the protocol's
//! own transitions to a resume seed and the rows that follow it. Nothing here
//! reads a header out of an event, because no event carries one.

use std::collections::BTreeMap;

use liminal_protocol::{
    lifecycle::{
        ConnectionIncarnationAllocationDecision, ConnectionIncarnationAllocator,
        ConnectionOrdinalExhaustion, DurableIncarnationReferences,
        ServerIncarnationStartupDecision, allocate_connection_incarnation,
        prepare_server_incarnation_startup,
    },
    wire::ConnectionIncarnation,
};

use super::ConnectionFateClass;
use super::codec::IncarnationEvent;
use super::error::IncarnationStreamError;
use crate::server::participant::dispatch::ConnectionFateWorkItem;

/// One replay-validated durable connection-fate work item.
#[derive(Clone, Debug, PartialEq, Eq)]
pub(in crate::server) struct ConnectionFateIntent {
    pub(in crate::server) open_sequence: u64,
    pub(in crate::server) allocation_sequence: u64,
    pub(in crate::server) startup_sequence: u64,
    pub(in crate::server) server_incarnation: u64,
    pub(in crate::server) declared_reference_bound: usize,
    pub(in crate::server) connection_incarnation: ConnectionIncarnation,
    pub(in crate::server) class: ConnectionFateClass,
    pub(in crate::server) declared_conversation_bound: usize,
    pub(in crate::server) conversations: Vec<u64>,
}

impl ConnectionFateIntent {
    /// Projects the exact handler input while retaining replay-generation audits here.
    #[must_use]
    pub(in crate::server) fn work_item(&self) -> ConnectionFateWorkItem {
        ConnectionFateWorkItem {
            open_sequence: self.open_sequence,
            connection_incarnation: self.connection_incarnation,
            class: self.class,
            tracked_conversations: self.conversations.clone(),
        }
    }
}

#[derive(Clone, Debug)]
pub(super) struct AllocationGeneration {
    pub(super) startup_sequence: u64,
    pub(super) server_incarnation: u64,
    pub(super) declared_reference_bound: Option<usize>,
    pub(super) active_allocations: BTreeMap<ConnectionIncarnation, u64>,
}

impl AllocationGeneration {
    pub(super) const fn new(startup_sequence: u64, server_incarnation: u64) -> Self {
        Self {
            startup_sequence,
            server_incarnation,
            declared_reference_bound: None,
            active_allocations: BTreeMap::new(),
        }
    }

    pub(super) const fn establish_bound(
        &mut self,
        stored_sequence: u64,
        declared_reference_bound: usize,
    ) -> Result<(), IncarnationStreamError> {
        if let Some(expected) = self.declared_reference_bound
            && expected != declared_reference_bound
        {
            return Err(IncarnationStreamError::GenerationReferenceBoundConflict {
                startup_sequence: self.startup_sequence,
                expected,
                actual: declared_reference_bound,
                stored_sequence,
            });
        }
        self.declared_reference_bound = Some(declared_reference_bound);
        Ok(())
    }
}

#[derive(Clone, Debug, Default)]
pub(super) struct UnmatchedConnectionFates {
    generation_startup_sequence: Option<u64>,
    pub(super) by_sequence: BTreeMap<u64, ConnectionFateIntent>,
}

impl UnmatchedConnectionFates {
    pub(super) fn open(
        &mut self,
        open_sequence: u64,
        generation: &AllocationGeneration,
        connection_incarnation: ConnectionIncarnation,
        class: ConnectionFateClass,
        declared_conversation_bound: usize,
        conversations: Vec<u64>,
    ) -> Result<ConnectionFateIntent, IncarnationStreamError> {
        let Some(&allocation_sequence) = generation.active_allocations.get(&connection_incarnation)
        else {
            return Err(IncarnationStreamError::UnknownConnectionAllocation {
                open_sequence,
                connection_incarnation,
            });
        };
        if let Some(expected_startup_sequence) = self.generation_startup_sequence
            && expected_startup_sequence != generation.startup_sequence
        {
            return Err(IncarnationStreamError::CrossGenerationOpen {
                open_sequence,
                expected_startup_sequence,
                actual_startup_sequence: generation.startup_sequence,
            });
        }
        if let Some(existing) = self
            .by_sequence
            .values()
            .find(|intent| intent.connection_incarnation == connection_incarnation)
        {
            return Err(IncarnationStreamError::DuplicateConnectionOpen {
                open_sequence,
                existing_open_sequence: existing.open_sequence,
            });
        }
        let declared_reference_bound = generation.declared_reference_bound.ok_or(
            IncarnationStreamError::UnknownConnectionAllocation {
                open_sequence,
                connection_incarnation,
            },
        )?;
        let active_count = self
            .by_sequence
            .len()
            .checked_add(1)
            .ok_or(IncarnationStreamError::EventLengthOverflow)?;
        if active_count > declared_reference_bound {
            return Err(
                IncarnationStreamError::HistoricalConnectionFateBoundExceeded {
                    startup_sequence: generation.startup_sequence,
                    bound: declared_reference_bound,
                    actual: active_count,
                    open_sequence,
                },
            );
        }
        self.generation_startup_sequence = Some(generation.startup_sequence);
        let intent = ConnectionFateIntent {
            open_sequence,
            allocation_sequence,
            startup_sequence: generation.startup_sequence,
            server_incarnation: generation.server_incarnation,
            declared_reference_bound,
            connection_incarnation,
            class,
            declared_conversation_bound,
            conversations,
        };
        self.by_sequence.insert(open_sequence, intent.clone());
        Ok(intent)
    }

    pub(super) fn complete(
        &mut self,
        complete_sequence: u64,
        open_sequence: u64,
    ) -> Result<(), IncarnationStreamError> {
        if self.by_sequence.remove(&open_sequence).is_none() {
            return Err(IncarnationStreamError::CompleteForAbsentOpen {
                complete_sequence,
                open_sequence,
            });
        }
        if self.by_sequence.is_empty() {
            self.generation_startup_sequence = None;
        }
        Ok(())
    }
}

#[derive(Debug)]
pub(super) struct ReplayAccumulator {
    pub(super) allocator: ConnectionIncarnationAllocator,
    pub(super) generation: Option<AllocationGeneration>,
    pub(super) unmatched: UnmatchedConnectionFates,
    pub(super) has_started: bool,
}

impl ReplayAccumulator {
    pub(super) fn apply(
        self,
        event: IncarnationEvent,
        stored_sequence: u64,
    ) -> Result<Self, IncarnationStreamError> {
        match event {
            IncarnationEvent::Startup => self.apply_startup(stored_sequence),
            IncarnationEvent::Allocate {
                declared_reference_bound,
                referenced_incarnations,
            } => self.apply_allocate(
                stored_sequence,
                declared_reference_bound,
                &referenced_incarnations,
            ),
            IncarnationEvent::OpenConnectionFate {
                connection_incarnation,
                class,
                declared_conversation_bound,
                conversations,
            } => self.apply_open(
                stored_sequence,
                connection_incarnation,
                class,
                declared_conversation_bound,
                conversations,
            ),
            IncarnationEvent::CompleteConnectionFate {
                open_event_sequence,
            } => self.apply_complete(stored_sequence, open_event_sequence),
        }
    }

    fn apply_startup(mut self, stored_sequence: u64) -> Result<Self, IncarnationStreamError> {
        self.allocator = match prepare_server_incarnation_startup(self.allocator) {
            ServerIncarnationStartupDecision::Fsync(intent) => intent.complete_after_fsync(),
            ServerIncarnationStartupDecision::Exhausted(_) => {
                return Err(IncarnationStreamError::StartupAfterServerExhaustion {
                    stored_sequence,
                });
            }
        };
        self.generation = Some(AllocationGeneration::new(
            stored_sequence,
            self.allocator.as_restore().server_incarnation,
        ));
        self.has_started = true;
        Ok(self)
    }

    fn apply_allocate(
        mut self,
        stored_sequence: u64,
        declared_reference_bound: usize,
        referenced_incarnations: &[ConnectionIncarnation],
    ) -> Result<Self, IncarnationStreamError> {
        if !self.has_started {
            return Err(IncarnationStreamError::AllocateBeforeStartup { stored_sequence });
        }
        let Some(generation) = self.generation.as_mut() else {
            return Err(IncarnationStreamError::AllocateBeforeStartup { stored_sequence });
        };
        generation.establish_bound(stored_sequence, declared_reference_bound)?;
        let references = DurableIncarnationReferences::try_new(
            referenced_incarnations,
            declared_reference_bound,
        )?;
        let (allocator, produced) =
            match allocate_connection_incarnation(self.allocator, references) {
                ConnectionIncarnationAllocationDecision::Allocated(allocation) => {
                    let produced = allocation.connection_incarnation();
                    (allocation.into_resulting(), Some(produced))
                }
                ConnectionIncarnationAllocationDecision::Exhausted(exhaustion) => {
                    if matches!(
                        &exhaustion,
                        ConnectionOrdinalExhaustion::AlreadyExhausted(_)
                    ) {
                        return Err(IncarnationStreamError::AllocateAfterOrdinalExhaustion {
                            stored_sequence,
                        });
                    }
                    (exhaustion.into_resulting(), None)
                }
            };
        self.allocator = allocator;
        retain_active_allocations(
            generation,
            referenced_incarnations,
            produced.map(|connection| (connection, stored_sequence)),
        );
        Ok(self)
    }

    fn apply_open(
        mut self,
        stored_sequence: u64,
        connection_incarnation: ConnectionIncarnation,
        class: ConnectionFateClass,
        declared_conversation_bound: usize,
        conversations: Vec<u64>,
    ) -> Result<Self, IncarnationStreamError> {
        let Some(generation) = self.generation.as_ref() else {
            return Err(IncarnationStreamError::UnknownConnectionAllocation {
                open_sequence: stored_sequence,
                connection_incarnation,
            });
        };
        drop(self.unmatched.open(
            stored_sequence,
            generation,
            connection_incarnation,
            class,
            declared_conversation_bound,
            conversations,
        )?);
        Ok(self)
    }

    fn apply_complete(
        mut self,
        stored_sequence: u64,
        open_event_sequence: u64,
    ) -> Result<Self, IncarnationStreamError> {
        self.unmatched
            .complete(stored_sequence, open_event_sequence)?;
        Ok(self)
    }
}

#[derive(Debug)]
pub(super) struct ReplayedIncarnationState {
    pub(super) allocator: ConnectionIncarnationAllocator,
    pub(super) next_sequence: u64,
    pub(super) generation: Option<AllocationGeneration>,
    pub(super) unmatched: UnmatchedConnectionFates,
    pub(super) has_started: bool,
}

pub(super) fn retain_active_allocations(
    generation: &mut AllocationGeneration,
    referenced_incarnations: &[ConnectionIncarnation],
    produced: Option<(ConnectionIncarnation, u64)>,
) {
    let prior = core::mem::take(&mut generation.active_allocations);
    generation.active_allocations = referenced_incarnations
        .iter()
        .filter_map(|incarnation| {
            prior
                .get(incarnation)
                .map(|sequence| (*incarnation, *sequence))
        })
        .collect();
    if let Some((incarnation, sequence)) = produced {
        generation.active_allocations.insert(incarnation, sequence);
    }
}