liminal-server 0.25.4

//! The complete revoked baseline and the membership authority share one cursor.

use crate::auth_membership::{AuthorityState, MembershipAuthority};
use crate::auth_revocation::{
    DecisionReply, GrantDecisionSource, GrantId, GrantLogId, Revision, RevocationRefusal,
};
use lys_pass::membership::GrantLog;
use serde::Serialize;
use std::collections::BTreeSet;
use std::sync::{Arc, Mutex};
use tokio::sync::watch;

/// Diagnostics contain counts and stable local error names, never response words.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct StreamDiagnostics {
    /// Readiness of both authorities at the applied fence.
    pub readiness: &'static str,
    /// Last successfully applied grant receipt revision.
    pub cursor: Option<u64>,
    /// Last successfully applied placement position.
    pub placed: Option<u64>,
    /// Stable cause; response bodies and remote refusal text are excluded.
    pub last_error: Option<&'static str>,
    /// Uncovered receipt positions, when a gap is known.
    pub missing: Option<u64>,
    /// Settled membership generation.
    pub generation: u64,
    /// Reads actually issued, including the sole waiting read at idle.
    pub requests: u64,
    /// Transport-ended resumptions.
    pub reopens: u64,
    /// True after the reader loop ended; the shutdown guard still joins its worker.
    pub stopped: bool,
}

/// A failed snapshot names its failure and carries no invented cursor or counts.
#[derive(Clone, Debug, Serialize)]
#[serde(untagged)]
pub enum StreamHealth {
    /// Counts read under the authority's trusted hold.
    Known(StreamDiagnostics),
    /// The hold cannot be trusted; admission remains closed.
    Unavailable {
        /// The owner cannot establish its state.
        readiness: &'static str,
        /// No cursor is claimed without the hold.
        cursor: Option<u64>,
        /// No placement cursor is claimed without the hold.
        placed: Option<u64>,
        /// The stable failure, without any untrusted words.
        last_error: &'static str,
    },
}

impl StreamHealth {
    /// Whether this snapshot establishes the complete readiness fence.
    #[must_use]
    pub fn is_ready(&self) -> bool {
        matches!(self, Self::Known(state) if state.readiness == "ready")
    }
}

pub(super) struct Inner {
    pub diagnostics: StreamDiagnostics,
    pub revoked: BTreeSet<GrantId>,
}

/// Locally readable authority; a poisoned lock closes admission by name.
pub struct StreamState {
    pub(super) reader_claim: std::sync::atomic::AtomicBool,
    pub(super) log: GrantLog,
    pub(super) inner: Mutex<Inner>,
    pub(super) changed: watch::Sender<Result<StreamDiagnostics, super::StreamError>>,
    pub(super) stop: watch::Sender<bool>,
}

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

impl StreamState {
    /// Installs a closed authority before any reader or protected connection exists.
    #[must_use]
    pub fn for_log(log: GrantLog) -> Arc<Self> {
        let diagnostics = StreamDiagnostics {
            readiness: "not_ready",
            cursor: None,
            placed: None,
            last_error: None,
            missing: None,
            generation: 0,
            requests: 0,
            reopens: 0,
            stopped: false,
        };
        Arc::new(Self {
            reader_claim: std::sync::atomic::AtomicBool::new(false),
            log,
            inner: Mutex::new(Inner {
                diagnostics: diagnostics.clone(),
                revoked: BTreeSet::new(),
            }),
            changed: watch::channel(Ok(diagnostics)).0,
            stop: watch::channel(false).0,
        })
    }

    /// Counts-only snapshot; nothing is answered from a poisoned hold.
    ///
    /// # Errors
    /// Names an untrusted authority hold rather than fabricating counters.
    pub fn snapshot(&self) -> Result<StreamDiagnostics, super::StreamError> {
        self.inner
            .lock()
            .map(|inner| inner.diagnostics.clone())
            .map_err(|_| super::StreamError::named("grant_stream_state_poisoned"))
    }

    /// Health reports a failure explicitly while preserving the liveness route.
    #[must_use]
    pub fn health(&self) -> StreamHealth {
        match self.snapshot() {
            Ok(status) => StreamHealth::Known(status),
            Err(error) => StreamHealth::Unavailable {
                readiness: "terminal",
                cursor: None,
                placed: None,
                last_error: error.name(),
            },
        }
    }

    /// Owned progress signal for consumers and tests; no timer or request is created.
    #[must_use]
    pub fn subscribe(&self) -> watch::Receiver<Result<StreamDiagnostics, super::StreamError>> {
        self.changed.subscribe()
    }

    /// Cancellation also interrupts in-flight membership and binding requests.
    #[must_use]
    pub fn cancellation(&self) -> watch::Receiver<bool> {
        self.stop.subscribe()
    }

    pub(super) fn update(
        &self,
        edit: impl FnOnce(&mut Inner) -> Result<(), super::StreamError>,
    ) -> Result<(), super::StreamError> {
        let result = (|| {
            let mut inner = self
                .inner
                .lock()
                .map_err(|_| super::StreamError::named("grant_stream_state_poisoned"))?;
            edit(&mut inner)?;
            Ok(inner.diagnostics.clone())
        })();
        drop(self.changed.send_replace(result.clone()));
        result.map(|_| ())
    }

    pub(super) fn unavailable(&self, name: &'static str) -> Result<(), super::StreamError> {
        self.update(|inner| {
            inner.diagnostics.readiness = "unavailable";
            inner.diagnostics.last_error = Some(name);
            Ok(())
        })
    }
}

impl MembershipAuthority for StreamState {
    fn state(&self) -> AuthorityState {
        let status = match self.snapshot() {
            Ok(status) => status,
            Err(error) => {
                return AuthorityState::Unavailable {
                    cause: error.name().into(),
                };
            }
        };
        match status.readiness {
            "ready" => {
                status
                    .cursor
                    .map_or(AuthorityState::NotReady, |revision| AuthorityState::Ready {
                        log: self.log.clone(),
                        revision,
                        generation: status.generation,
                    })
            }
            "not_ready" => AuthorityState::NotReady,
            "gap" => match (status.cursor, status.missing) {
                (Some(after), Some(missing)) => AuthorityState::Gap { after, missing },
                _ => AuthorityState::Unavailable {
                    cause: "revocation_stream_gap".into(),
                },
            },
            "reset" => AuthorityState::Reset,
            _ => AuthorityState::Unavailable {
                cause: status
                    .last_error
                    .unwrap_or("revocation_stream_unavailable")
                    .into(),
            },
        }
    }
}

impl GrantDecisionSource for StreamState {
    fn decide(
        &self,
        log: &GrantLogId,
        grants: &[GrantId],
        at_least: Revision,
    ) -> Result<DecisionReply, RevocationRefusal> {
        let inner = self
            .inner
            .lock()
            .map_err(|_| RevocationRefusal::StreamUnavailable)?;
        if log.identity() != self.log.identity || log.epoch() != self.log.epoch {
            return Err(RevocationRefusal::LogReset);
        }
        if inner.diagnostics.readiness != "ready" {
            return Err(RevocationRefusal::StreamUnavailable);
        }
        let revision = inner
            .diagnostics
            .cursor
            .ok_or(RevocationRefusal::RevisionUnavailable)?;
        if revision < at_least.get() {
            return Err(RevocationRefusal::RevisionUnavailable);
        }
        // The authenticated baseline lists every visible revoked ancestor. Absence
        // means not revoked only after contiguous replay and its Ready barrier.
        let answer = DecisionReply {
            log: log.clone(),
            revision: Revision::new(revision),
            decisions: grants
                .iter()
                .map(|grant| (grant.clone(), inner.revoked.contains(grant)))
                .collect(),
        };
        drop(inner);
        Ok(answer)
    }
}