liminal_server/server/connection/

supervisor_lifecycle.rs

use std::sync::Mutex;
use std::thread::JoinHandle;

use super::super::reclaim_placement::ReclaimRelease;
use super::{ConnectionSupervisor, SupervisorInner};
use crate::ServerError;

#[derive(Debug, Default)]
pub(super) struct BackgroundStart {
    started: bool,
    stopped: bool,
    worker: Option<JoinHandle<()>>,
    /// The incarnation start behind the bind (SRV-041 R2).
    incarnation: Option<JoinHandle<()>>,
    failure: Option<String>,
    reclaim_releases: Vec<ReclaimRelease>,
}

impl ConnectionSupervisor {
    pub(crate) fn retain_reclaim_release(
        &self,
        release: ReclaimRelease,
    ) -> Result<(), ServerError> {
        let mut start = self
            .inner
            .background
            .lock()
            .map_err(|_| refusal("connection background start lock poisoned"))?;
        if start.stopped {
            return Err(refusal("spine release cannot be registered after shutdown"));
        }
        start.reclaim_releases.push(release);
        drop(start);
        Ok(())
    }

    pub(crate) fn listening(&self) -> Result<(), ServerError> {
        let mut start = self
            .inner
            .background
            .lock()
            .map_err(|_| refusal("connection background start lock poisoned"))?;
        if start.stopped {
            return Err(refusal(
                "connection background work cannot start after shutdown",
            ));
        }
        if !start.started {
            start.worker = self.inner.runtime.services_arc().listening();
            start.incarnation = self.start_incarnation()?;
            start.started = true;
        }
        drop(start);
        Ok(())
    }

    /// Begins the incarnation start on its own thread. A start that fails
    /// stops the server through the fatal shutdown; its typed error is
    /// answered by [`Self::incarnation_start_failure`].
    fn start_incarnation(&self) -> Result<Option<JoinHandle<()>>, ServerError> {
        let Some(authority) = self.inner.incarnations.clone() else {
            return Ok(None);
        };
        let fatal = self.inner.runtime.fatal_shutdown.clone();
        std::thread::Builder::new()
            .name("liminal-incarnation-start".to_owned())
            .spawn(move || {
                if let Err(error) = authority.ensure_started() {
                    tracing::error!(%error, "the incarnation start failed behind the bound listener; the server stops");
                    if let Some(fatal) = fatal {
                        fatal.initiate();
                    }
                }
            })
            .map(Some)
            .map_err(|error| refusal(format!("the incarnation start could not begin: {error}")))
    }

    /// The incarnation start's typed error once it has failed, which a server
    /// stopped by that failure returns; nothing while it has not.
    #[must_use]
    pub fn incarnation_start_failure(&self) -> Option<ServerError> {
        self.inner
            .incarnations
            .as_ref()
            .and_then(|authority| authority.start_failure())
    }

    pub(in crate::server) fn finish_store_work(&self) -> Result<(), ServerError> {
        self.inner.finish_store_work()
    }

    pub(in crate::server) fn release_store_holders(&self) -> Result<(), ServerError> {
        self.inner.runtime.services().release_store_holders()
    }
}

impl SupervisorInner {
    fn finish_store_work(&self) -> Result<(), ServerError> {
        #[cfg(test)]
        let observer = self
            .reclaim_stop_observer
            .lock()
            .map_err(|_| refusal("reclamation stop observer lock poisoned"))?
            .take();
        #[cfg(test)]
        if let Some(observer) = observer {
            observer
                .send(true)
                .map_err(|error| refusal(error.to_string()))?;
        }
        first_failure([
            self.reclaim.finish(&self.scheduler),
            join_background(&self.background),
            self.runtime.services().finish_background_store_work(),
        ])
    }
}

impl Drop for SupervisorInner {
    fn drop(&mut self) {
        self.scheduler.shutdown();
        if let Err(error) = self.finish_store_work() {
            tracing::error!(%error, "connection supervisor store teardown failed");
        }
    }
}

fn join_background(background: &Mutex<BackgroundStart>) -> Result<(), ServerError> {
    let mut start = background
        .lock()
        .map_err(|_| refusal("connection background join lock poisoned"))?;
    start.stopped = true;
    if let Some(failure) = &start.failure {
        return Err(refusal(failure.clone()));
    }
    let joined = start.worker.take().map_or(Ok(()), |worker| {
        worker
            .join()
            .map_err(|_| refusal("connection background store work panicked during shutdown"))
    });
    let started = start.incarnation.take().map_or(Ok(()), |worker| {
        worker
            .join()
            .map_err(|_| refusal("the incarnation start panicked during shutdown"))
    });
    let mut results = vec![joined, started];
    for release in &mut start.reclaim_releases {
        results.push(release.finish());
    }
    start.reclaim_releases.clear();
    let mut first = None;
    for result in results {
        if let Err(error) = result {
            tracing::error!(%error, "background store teardown failed");
            if first.is_none() {
                first = Some(error);
            }
        }
    }
    if let Some(error) = first {
        start.failure = Some(error.to_string());
        return Err(error);
    }
    drop(start);
    Ok(())
}

fn first_failure<const N: usize>(results: [Result<(), ServerError>; N]) -> Result<(), ServerError> {
    let mut first = None;
    for result in results {
        if let Err(error) = result {
            tracing::error!(%error, "connection store work teardown failed");
            if first.is_none() {
                first = Some(error);
            }
        }
    }
    first.map_or(Ok(()), Err)
}

fn refusal(message: impl Into<String>) -> ServerError {
    ServerError::ListenerAccept {
        message: message.into(),
    }
}