liminal-server 0.25.4

use std::sync::{Arc, Mutex};
use std::thread::JoinHandle;

use super::super::reclaim_placement::ReclaimRelease;
use super::{ConnectionIncarnationAuthority, 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`].
    ///
    /// SRV-042 R1: from here until the start publishes, an accepted
    /// connection is held (see [`Self::admit_when_started`]); the start thread
    /// runs what was held once it has, before a failed start stops the server,
    /// so a held client reads the start's refusal. The thread holds the
    /// supervisor weakly: it never keeps a stopped server alive.
    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();
        let held = Arc::clone(&self.inner.held);
        let supervisor = Arc::downgrade(&self.inner);
        held.starting();
        std::thread::Builder::new()
            .name("liminal-incarnation-start".to_owned())
            .spawn(move || {
                let started = authority.ensure_started();
                let admissions = held.started();
                if !admissions.is_empty()
                    && let Some(inner) = supervisor.upgrade()
                {
                    let supervisor = Self { inner };
                    for admit in admissions {
                        admit(&supervisor);
                    }
                }
                // SRV-048 R2: the anchor swap is settled here, after the
                // held connections are admitted; a refused swap stops the
                // server as a failed start does.
                let settled = started.and_then(|()| authority.settle_anchor());
                if let Err(error) = settled
                    && !ConnectionIncarnationAuthority::is_abandoned(&error)
                {
                    tracing::error!(%error, "the incarnation start or its anchor swap failed behind the bound listener; the server stops");
                    if let Some(fatal) = fatal {
                        fatal.initiate();
                    }
                }
            })
            .map(Some)
            .map_err(|error| {
                self.inner.held.stop();
                refusal(format!("the incarnation start could not begin: {error}"))
            })
    }

    /// Runs `admit` for a connection just accepted: at once, or, while the
    /// incarnation start runs, on the start thread once it has published, so
    /// the accept thread returns to accepting and a stop never waits on the
    /// start (SRV-042 R1). After shutdown it is dropped, closing its socket.
    pub(in crate::server) fn admit_when_started(&self, admit: impl FnOnce(&Self) + Send + 'static) {
        if let Some(admit) = self.inner.held.hold(Box::new(admit)) {
            admit(self);
        }
    }

    /// 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()))?;
        }
        self.held.stop();
        if let Some(authority) = &self.incarnations {
            authority.abandon();
        }
        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| {
        // The start thread can hold the last owner while it admits what it
        // held; dropping it there ends here, on the thread itself, which is
        // finishing: it is not joined from itself.
        if worker.thread().id() == std::thread::current().id() {
            return Ok(());
        }
        crate::server::boot_steps::stop_timed("incarnation_join", || 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(),
    }
}