liminal-server 0.25.4

//! One owner of Lys's consumer-opened grant log, stopped and joined before store release.

#[path = "grant_stream_owner/binding.rs"]
mod binding;
#[path = "grant_stream_owner/config.rs"]
mod config;
#[path = "grant_stream_owner/configured.rs"]
mod configured;
#[path = "grant_stream_owner/engine.rs"]
mod engine;
#[path = "grant_stream_owner/http.rs"]
mod http;
#[path = "grant_stream_owner/state.rs"]
mod state;

pub use config::{GrantStreamConfig, GrantSubjectConfig};
pub use state::{StreamDiagnostics, StreamHealth, StreamState};

use crate::auth_lys::live::drafts::{DraftDecisionHook, PullReport};
use crate::config::types::LysConfig;
use lys_identity::grants::change_stream::{ChangesPage, ChangesRequest, LogName};
use std::sync::Arc;
use std::thread::JoinHandle;

/// Only stable local names cross diagnostics, errors, Debug and tracing.
#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
#[error("{name}")]
pub struct StreamError {
    name: &'static str,
    transport_ended: bool,
}

impl StreamError {
    pub(super) const fn named(name: &'static str) -> Self {
        Self {
            name,
            transport_ended: false,
        }
    }
    const fn transport() -> Self {
        Self {
            name: "revocation_stream_unavailable",
            transport_ended: true,
        }
    }
    /// The named failure, containing no transport error or remote response text.
    #[must_use]
    pub const fn name(self) -> &'static str {
        self.name
    }
}

/// The guard must remain on the server start stack until its shutdown finalizer joins it.
pub struct GrantStreamOwner {
    state: Arc<StreamState>,
    worker: Option<JoinHandle<Result<(), StreamError>>>,
    draft_hook: Option<Arc<dyn DraftDecisionHook>>,
}

impl std::fmt::Debug for GrantStreamOwner {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("GrantStreamOwner")
            .field("state", &self.state)
            .field("worker", &self.worker.is_some())
            .finish_non_exhaustive()
    }
}

impl GrantStreamOwner {
    /// Starts the configured owner on the standalone server's existing ask runtime.
    ///
    /// # Errors
    /// A missing runtime or any refusal from [`Self::start_configured`].
    pub fn start_for_server(
        config: &crate::config::types::ServerConfig,
        supervisor: &crate::server::ConnectionSupervisor,
        health: &crate::health::HealthServerHandle,
        runtime: Option<&tokio::runtime::Runtime>,
        draft_hook: Option<Arc<dyn DraftDecisionHook>>,
    ) -> Result<Option<Self>, crate::ServerError> {
        let Some(lys) = config.auth.as_ref().and_then(|auth| auth.lys.as_ref()) else {
            return Ok(None);
        };
        if lys.membership.is_none() && lys.grant_stream.is_none() {
            // SRV-039: with no owner, the root question is asked once at start.
            crate::auth_lys::root_presence_start::start(lys, health, runtime)?;
            return Ok(None);
        }
        let runtime = runtime.ok_or_else(|| config_error("grant_stream_runtime_missing"))?;
        Self::start_configured(
            lys,
            supervisor,
            health,
            runtime.handle().clone(),
            draft_hook,
        )
    }

    /// Installs both authorities before any connection and starts exactly one read.
    ///
    /// The supplied handle is the server's already owned Lys runtime; shutdown
    /// joins this guard before that runtime or the store can be released.
    ///
    /// # Errors
    /// Refuses invalid bounds, unreadable credentials or keys, a foreign membership
    /// log, a second owner or authority, and failure to spawn the owned worker.
    pub fn start_configured(
        lys: &LysConfig,
        supervisor: &crate::server::ConnectionSupervisor,
        health: &crate::health::HealthServerHandle,
        handle: tokio::runtime::Handle,
        draft_hook: Option<Arc<dyn DraftDecisionHook>>,
    ) -> Result<Option<Self>, crate::ServerError> {
        let Some(config) = lys.grant_stream.as_ref() else {
            if lys.membership.is_some() {
                return Err(config_error(
                    "auth.lys.membership requires auth.lys.grant_stream",
                ));
            }
            return Ok(None);
        };
        let configured::Installation {
            state,
            engine,
            http,
            root,
        } = configured::prepare(lys, config, supervisor, health, handle.clone())?;
        let worker_state = Arc::clone(&state);
        let worker = std::thread::Builder::new()
            .name("liminal-grants".into())
            .spawn(move || {
                let result = handle.block_on(read(engine, http, root));
                if let Err(error) = result {
                    worker_state.update(|inner| {
                        if !matches!(inner.diagnostics.readiness, "gap" | "reset") {
                            inner.diagnostics.readiness = "terminal";
                        }
                        inner.diagnostics.last_error = Some(error.name());
                        inner.diagnostics.stopped = true;
                        Ok(())
                    })?;
                } else {
                    worker_state.update(|inner| {
                        inner.diagnostics.readiness = "unavailable";
                        inner.diagnostics.last_error = Some("grant_stream_stopped");
                        inner.diagnostics.stopped = true;
                        Ok(())
                    })?;
                }
                result
            })
            .map_err(|_| config_error("grant_stream_worker_unavailable"))?;
        Ok(Some(Self {
            state,
            worker: Some(worker),
            draft_hook,
        }))
    }

    /// The health/progress authority installed at start.
    #[must_use]
    pub fn state(&self) -> Arc<StreamState> {
        Arc::clone(&self.state)
    }

    /// Hook for an explicitly accepted published draft decision, never a grant change.
    /// The current Lys frame contract has no such variant; unknown rows remain terminal.
    ///
    /// # Errors
    /// A missing hook or a named approved-pull failure, retained without draft words.
    pub async fn draft_decided(&self) -> Result<PullReport, StreamError> {
        let hook = self
            .draft_hook
            .as_ref()
            .ok_or(StreamError::named("grant_stream_draft_hook_missing"))?;
        let result = hook
            .draft_decided()
            .await
            .map_err(|_| StreamError::named("grant_stream_draft_pull_failed"));
        if let Err(error) = result.as_ref() {
            self.state.update(|inner| {
                inner.diagnostics.last_error = Some(error.name());
                Ok(())
            })?;
        }
        result
    }

    /// Cancels every pending HTTP ask and joins the worker, including on startup unwind.
    ///
    /// # Errors
    /// The terminal reader failure or an unexpectedly terminated worker.
    pub fn finish(&mut self) -> Result<(), StreamError> {
        self.draft_hook = None;
        self.state.stop.send_replace(true);
        match self.worker.take() {
            Some(worker) => worker
                .join()
                .map_err(|_| StreamError::named("grant_stream_worker_panicked"))?,
            None => Ok(()),
        }
    }
}

impl Drop for GrantStreamOwner {
    fn drop(&mut self) {
        if let Err(error) = self.finish() {
            tracing::error!(cause = error.name(), "grant stream shutdown failed");
        }
    }
}

fn config_error(message: &str) -> crate::ServerError {
    crate::ServerError::ConfigValidation {
        message: message.into(),
    }
}
fn as_config(error: StreamError) -> crate::ServerError {
    config_error(error.name())
}

/// The shutdown finalizer joins the reader before flushing or releasing the store.
///
/// # Errors
/// The owner's terminal failure or a failed worker join, by its stable name.
pub fn finish_configured(owner: &mut Option<GrantStreamOwner>) -> Result<(), crate::ServerError> {
    owner
        .as_mut()
        .map_or(Ok(()), |owner| owner.finish().map_err(as_config))
}

pub(crate) fn read_credential(
    path: &std::path::Path,
    bound: usize,
) -> Result<zeroize::Zeroizing<String>, crate::ServerError> {
    http::read_file(path, bound).map_err(as_config)
}

async fn read(
    mut engine: engine::Engine,
    http: http::Http,
    root: Arc<crate::auth_lys::root_presence::RootPresence>,
) -> Result<(), StreamError> {
    let result = read_pages(&mut engine, &http, &root).await;
    if let Err(error) = result.as_ref() {
        engine.terminal(error.name())?;
    }
    result
}

async fn read_pages(
    engine: &mut engine::Engine,
    http: &http::Http,
    root: &crate::auth_lys::root_presence::RootPresence,
) -> Result<(), StreamError> {
    let mut reopens = 0_usize;
    loop {
        if *http.stop.borrow() {
            engine.lost()?;
            return Ok(());
        }
        let before = engine.state.snapshot()?;
        let question = ChangesRequest {
            log: Some(LogName {
                identity: engine.state.log.identity.clone(),
                epoch: engine.state.log.epoch,
            }),
            after: before.cursor,
            limit: engine.config.page_positions,
            wait: true,
            placements: true,
            placed: before.placed,
        };
        engine.state.update(|inner| {
            inner.diagnostics.requests = inner
                .diagnostics
                .requests
                .checked_add(1)
                .ok_or(StreamError::named("grant_stream_counter_exhausted"))?;
            Ok(())
        })?;
        let page = match http
            .post::<_, ChangesPage>("grants/changes", &question)
            .await
        {
            Ok(page) => page,
            Err(error) if error.name() == "grant_stream_stopped" => {
                engine.lost()?;
                return Ok(());
            }
            Err(error) if error.transport_ended => {
                engine.lost()?;
                if reopens >= engine.config.max_transport_reopens {
                    return Err(error);
                }
                reopens += 1;
                engine.state.update(|inner| {
                    inner.diagnostics.reopens = inner
                        .diagnostics
                        .reopens
                        .checked_add(1)
                        .ok_or(StreamError::named("grant_stream_counter_exhausted"))?;
                    Ok(())
                })?;
                continue;
            }
            Err(error) => {
                engine.lost()?;
                return Err(error);
            }
        };
        if let Err(error) = engine.page(page) {
            if error.name() == "grant_stream_membership_failed" && *http.stop.borrow() {
                engine.lost()?;
                return Ok(());
            }
            return Err(error);
        }
        // SRV-038: asked on the ready fence's own event, never on a timer.
        if engine.take_root_owed() {
            root.ask_now().await;
        }
        let after = engine.state.snapshot()?;
        if (before.cursor, before.placed) == (after.cursor, after.placed) {
            return Err(StreamError::named("grant_stream_no_progress"));
        }
        reopens = 0;
    }
}

#[cfg(test)]
#[path = "grant_stream_owner/tests.rs"]
pub(crate) mod tests;