liminal-server 0.25.4

//! The incarnation start behind the bind (SRV-041 R2): the supervisor builds
//! an authority that has read nothing, and the first caller that needs the
//! stream (the background start `listening()` begins, or an admission that
//! arrives first) runs the replay, recovery and transition flush once while
//! every other caller waits on its outcome. No caller is refused for starting.

use std::sync::Arc;

use liminal::durability::{DurableStore, bridge::block_on};

use super::{ConnectionIncarnationAuthority, ConnectionIncarnationAuthorityState as State};
use crate::ServerError;
use crate::health::AdmissionReadiness;
use crate::server::participant::ParticipantSemanticHandler;
use crate::server::participant::incarnation_stream::AnchorSwap;

/// `ServerError::ParticipantIncarnation::phase` for an operation that reached
/// the authority before its start, which only the teardown paths can do.
pub(in crate::server::connection) const NOT_STARTED_PHASE: &str = "incarnation start pending";

/// `ServerError::ParticipantIncarnation::phase` for an anchor swap the store
/// refused after the start published (SRV-048 R2).
pub(in crate::server::connection) const ANCHOR_SETTLE_PHASE: &str = "epoch anchor settle";

fn anchor_refusal(message: impl Into<String>) -> ServerError {
    ServerError::ParticipantIncarnation {
        phase: ANCHOR_SETTLE_PHASE,
        message: message.into(),
    }
}

/// What the start reads with, held from the supervisor's build to the start.
#[derive(Debug)]
pub(in crate::server::connection) struct IncarnationStartInputs {
    pub(in crate::server::connection) store: Arc<dyn DurableStore>,
    pub(in crate::server::connection) maximum_references: usize,
    pub(in crate::server::connection) maximum_conversations: u64,
    pub(in crate::server::connection) handler: Arc<dyn ParticipantSemanticHandler>,
}

/// A failed start, kept so every later admission is refused with the same
/// typed error the start returned.
#[derive(Debug)]
pub(in crate::server::connection) enum StartFailure {
    Incarnation {
        phase: &'static str,
        message: String,
    },
    ServerIncarnationExhausted,
    Other(String),
}

impl StartFailure {
    fn of(error: &ServerError) -> Self {
        match error {
            ServerError::ParticipantIncarnation { phase, message } => Self::Incarnation {
                phase,
                message: message.clone(),
            },
            ServerError::ServerIncarnationExhausted => Self::ServerIncarnationExhausted,
            other => Self::Other(other.to_string()),
        }
    }

    pub(in crate::server::connection) fn error(&self) -> ServerError {
        match self {
            Self::Incarnation { phase, message } => ServerError::ParticipantIncarnation {
                phase,
                message: message.clone(),
            },
            Self::ServerIncarnationExhausted => ServerError::ServerIncarnationExhausted,
            Self::Other(message) => ServerError::ParticipantIncarnation {
                phase: "server startup",
                message: message.clone(),
            },
        }
    }
}

impl ConnectionIncarnationAuthority {
    /// An authority that has read nothing, reporting admission unavailable
    /// until its start finishes.
    ///
    /// # Errors
    ///
    /// The signed conversation bound does not fit this platform, found before
    /// bind as today.
    pub(in crate::server::connection) fn starting(
        inputs: IncarnationStartInputs,
        admission: AdmissionReadiness,
    ) -> Result<Self, ServerError> {
        let maximum_conversations = super::conversation_bound(inputs.maximum_conversations)?;
        admission.set_available(false);
        Ok(Self::with_state(
            State::NotStarted(Box::new(inputs)),
            maximum_conversations,
            admission,
        ))
    }

    /// Runs the start when no one has begun it, or waits for the one begun;
    /// answers once the stream is ready, or the start's error.
    ///
    /// # Errors
    ///
    /// The start failed: its typed error, the same for every caller.
    pub(in crate::server::connection) fn ensure_started(&self) -> Result<(), ServerError> {
        match self.claim()? {
            Claim::Run(inputs) => self.start(&inputs),
            Claim::Done(result) => result,
        }
    }

    /// Claims the start when no one has begun it, or waits out the one begun
    /// and answers its outcome.
    fn claim(&self) -> Result<Claim, ServerError> {
        let mut state = self
            .started
            .wait_while(
                Self::lock(&self.state, "incarnation start lock")?,
                |state| {
                    let starting = matches!(state, State::Starting);
                    #[cfg(test)]
                    if starting {
                        self.note_waiting();
                    }
                    starting
                },
            )
            .map_err(|error| ServerError::ParticipantIncarnation {
                phase: "incarnation start wait",
                message: error.to_string(),
            })?;
        let claim = match std::mem::replace(&mut *state, State::Starting) {
            State::NotStarted(inputs) => Claim::Run(inputs),
            State::StartFailed(failure) => {
                let error = failure.error();
                *state = State::StartFailed(failure);
                Claim::Done(Err(error))
            }
            started => {
                *state = started;
                Claim::Done(Ok(()))
            }
        };
        drop(state);
        Ok(claim)
    }

    /// The server is stopping: a start in progress ends at its next step
    /// boundary, and one not yet begun ends before it reads (SRV-042 R2).
    pub(in crate::server::connection) fn abandon(&self) {
        self.stopping
            .store(true, std::sync::atomic::Ordering::SeqCst);
    }

    /// Whether `error` is a start the server's stop ended, not a failure.
    pub(in crate::server::connection) fn is_abandoned(error: &ServerError) -> bool {
        matches!(
            error,
            ServerError::ParticipantIncarnation { phase, .. } if *phase == super::stream_start::ABANDONED_PHASE
        )
    }

    /// The start's error, once a start has failed; nothing otherwise.
    pub(in crate::server::connection) fn start_failure(&self) -> Option<ServerError> {
        match &*self.state.lock().ok()? {
            State::StartFailed(failure) => Some(failure.error()),
            _ => None,
        }
    }

    /// Runs the start outside the lock, then publishes its outcome and wakes
    /// every caller parked behind it. A start that unwinds is published as a
    /// failure by its guard, so no caller is left waiting.
    fn start(&self, inputs: &IncarnationStartInputs) -> Result<(), ServerError> {
        let mut outcome = Outcome {
            authority: self,
            published: false,
        };
        let started = Self::start_stream(
            Arc::clone(&inputs.store),
            inputs.maximum_references,
            inputs.handler.as_ref(),
            &self.stopping,
        )
        .and_then(|mut stream| {
            self.park_anchor_swap(stream.take_anchor_swap())?;
            Ok(stream)
        });
        outcome.publish(started)
    }

    /// Parks a started stream's anchor swap for the start thread to settle
    /// once admission is open (SRV-048 R2).
    pub(in crate::server::connection) fn park_anchor_swap(
        &self,
        swap: Option<AnchorSwap>,
    ) -> Result<(), ServerError> {
        *self
            .anchor_swap
            .lock()
            .map_err(|_| anchor_refusal("the anchor swap slot is poisoned"))? = swap;
        Ok(())
    }

    /// Settles the parked anchor swap; nothing when none is parked. Run by
    /// the start thread after the held connections are admitted, whether or
    /// not the server is stopping: the shutdown joins that thread before the
    /// store's own work finishes (SRV-048 R2).
    ///
    /// # Errors
    ///
    /// Returns [`ANCHOR_SETTLE_PHASE`] naming the swap's from and to when the
    /// store refuses it.
    pub(in crate::server::connection) fn settle_anchor(&self) -> Result<(), ServerError> {
        let parked = self
            .anchor_swap
            .lock()
            .map_err(|_| anchor_refusal("the anchor swap slot is poisoned"))?
            .take();
        let Some(swap) = parked else {
            return Ok(());
        };
        let (from, to) = (swap.from(), swap.to());
        block_on(swap.settle())
            .map_err(|error| anchor_refusal(format!("the swap from {from} to {to}: {error}")))?
            .map_err(|error| anchor_refusal(format!("the swap from {from} to {to}: {error}")))
    }

    #[cfg(test)]
    fn note_waiting(&self) {
        if let Some(waiting) = self.on_wait.lock().ok().and_then(|slot| slot.clone()) {
            let _sent = waiting.send(());
        }
    }

    /// Tells `waiting` each time a caller parks behind a start in progress.
    #[cfg(test)]
    pub(in crate::server::connection) fn on_wait_for_test(
        &self,
        waiting: std::sync::mpsc::Sender<()>,
    ) {
        if let Ok(mut slot) = self.on_wait.lock() {
            *slot = Some(waiting);
        }
    }
}

/// What a caller of [`ConnectionIncarnationAuthority::ensure_started`] does.
enum Claim {
    /// Run the start with these inputs; no one else will.
    Run(Box<IncarnationStartInputs>),
    /// The start has finished with this outcome.
    Done(Result<(), ServerError>),
}

/// Publishes a start's outcome exactly once, as a failure when the start
/// unwound before publishing it.
struct Outcome<'authority> {
    authority: &'authority ConnectionIncarnationAuthority,
    published: bool,
}

impl Outcome<'_> {
    fn publish(
        &mut self,
        started: Result<super::StartedIncarnationStream, ServerError>,
    ) -> Result<(), ServerError> {
        let (next, result) = match started {
            Ok(stream) => (State::Ready(stream), Ok(())),
            Err(error) => (State::StartFailed(StartFailure::of(&error)), Err(error)),
        };
        let ready = matches!(next, State::Ready(_));
        self.set(next);
        if ready {
            self.authority.admission.set_available(true);
        }
        result
    }

    fn set(&mut self, next: State) {
        self.published = true;
        match self.authority.state.lock() {
            Ok(mut state) => *state = next,
            Err(poisoned) => *poisoned.into_inner() = next,
        }
        self.authority.started.notify_all();
    }
}

impl Drop for Outcome<'_> {
    fn drop(&mut self) {
        if !self.published {
            tracing::error!(
                "the incarnation start unwound before it finished; admission is refused"
            );
            self.set(State::StartFailed(StartFailure::Other(
                "the incarnation start unwound before it finished".to_owned(),
            )));
        }
    }
}