liminal-server 0.24.9

//! 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;

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

/// `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";

/// 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 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(),
        );
        outcome.publish(started)
    }

    #[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(),
            )));
        }
    }
}