liminal-server 0.24.9

//! DUR-021 R4: the reclaim status and quiesce routes of the health endpoint.
//!
//! The endpoint reads each shard's reclaim status from the store on each
//! request, and offers one operator route that calls `quiesce_reclaim` once,
//! the only call site of it. Readiness never reads this: a server with a pass
//! running is ready.

use std::sync::{Arc, OnceLock, Weak};

use haematite::EventStore;
use haematite::db::{ReclaimStatus, ShardReclaimAnswer};

/// The spine, installed once the services exist.
///
/// Empty before, so the route answers that nothing is installed rather than
/// inventing a status. Held weakly: the health endpoint is torn down last,
/// and it never keeps the spine open past the services' own close.
#[derive(Debug, Clone, Default)]
pub struct SharedReclaimSpine {
    spine: Arc<OnceLock<Weak<EventStore>>>,
}

/// What the quiesce route answers: whether the store rested, in its own
/// words when it did not, and the status read after the call.
#[derive(Debug, Clone, serde::Serialize)]
pub struct QuiesceAnswer {
    pub quiesced: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub refused: Option<String>,
    /// Passes the call ran to their end, summed over the shards; zero is the
    /// store's word that no pass was needed.
    pub passes_run: u64,
    pub status: ReclaimStatus,
}

impl SharedReclaimSpine {
    /// Publishes the spine onto the routes, once: a server opens one spine,
    /// and a second install is refused by the answer (`false`), never taken.
    pub fn install(&self, spine: &Arc<EventStore>) -> bool {
        self.spine.set(Arc::downgrade(spine)).is_ok()
    }

    /// The spine while it is still open; `None` before install and after the
    /// services close it.
    fn spine(&self) -> Option<Arc<EventStore>> {
        self.spine.get().and_then(Weak::upgrade)
    }

    /// Each shard's status, read from the store now; `None` before install.
    #[must_use]
    pub fn status(&self) -> Option<ReclaimStatus> {
        self.spine().map(|spine| spine.database().reclaim_status())
    }

    /// Drives the pass to rest once and answers what the store said and
    /// its status after; `None` before install.
    #[must_use]
    pub fn quiesce(&self) -> Option<QuiesceAnswer> {
        let spine = self.spine()?;
        let database = spine.database();
        let before = passes_finished(&database.reclaim_status());
        let (quiesced, refused) = match database.quiesce_reclaim() {
            Ok(()) => (true, None),
            Err(error) => (false, Some(error.to_string())),
        };
        let status = database.reclaim_status();
        Some(QuiesceAnswer {
            quiesced,
            refused,
            passes_run: passes_finished(&status).saturating_sub(before),
            status,
        })
    }
}

/// Passes finished over every shard that answered.
fn passes_finished(status: &ReclaimStatus) -> u64 {
    status
        .shards
        .iter()
        .map(|answer| match answer {
            ShardReclaimAnswer::Answered { status, .. } => status.passes_finished,
            ShardReclaimAnswer::NotAnswering { .. } => 0,
        })
        .sum()
}