liminal-server 0.25.4

//! Fault records remain readable independently of the state that was refused.

use super::{BridgeCounters, ServerError, failure, life, lock};
use std::collections::BTreeMap;
use std::net::{Shutdown, TcpStream};
use std::sync::atomic::AtomicU64;
#[cfg(feature = "test-support")]
use std::sync::atomic::Ordering;
use std::sync::{Arc, Condvar, Mutex, OnceLock, Weak};

#[derive(Debug, Default)]
pub(super) struct Observation {
    pub(super) life: OnceLock<crate::link::LinkLife>,
    previous_lifecycle: crate::link::LinkLifeCounters,
    #[cfg(feature = "test-support")]
    registry_signal: OnceLock<RegistrySignal>,
    #[cfg(feature = "test-support")]
    pub(super) stop_before_ack: std::sync::atomic::AtomicBool,
    #[cfg(feature = "test-support")]
    pub(super) prepared_crossing: Mutex<Option<super::prepared::PreparedCrossing>>,
    pub(super) counters: Mutex<BridgeCounters>,
    pub(super) settled: AtomicU64,
    pub(super) changed: Condvar,
    pub(super) wake: Arc<Wake>,
}

impl Observation {
    #[cfg(test)]
    pub(super) fn new(interrupt: TcpStream) -> Arc<Self> {
        Self::with_counts(Some(interrupt), BridgeCounters::default(), 0)
    }

    pub(super) fn resume(
        interrupt: TcpStream,
        counters: BridgeCounters,
    ) -> Result<Arc<Self>, ServerError> {
        Self::retaining(Some(interrupt), counters)
    }
    pub(super) fn retaining(
        interrupt: Option<TcpStream>,
        counters: BridgeCounters,
    ) -> Result<Arc<Self>, ServerError> {
        let settled = counters
            .crossed
            .checked_add(counters.refused)
            .and_then(|count| count.checked_add(counters.too_large))
            .ok_or_else(|| failure("configured link answer counter overflow"))?;
        Ok(Self::with_counts(interrupt, counters, settled))
    }
    fn with_counts(
        interrupt: Option<TcpStream>,
        counters: BridgeCounters,
        settled: u64,
    ) -> Arc<Self> {
        Arc::new_cyclic(|observation| Self {
            wake: Arc::new(Wake {
                observation: observation.clone(),
                interrupt,
                ..Wake::default()
            }),
            previous_lifecycle: counters.lifecycle,
            counters: Mutex::new(counters),
            settled: AtomicU64::new(settled),
            ..Self::default()
        })
    }
    pub(super) fn notify_changes(&self) {
        self.changed.notify_all();
        #[cfg(feature = "test-support")]
        if let Some((registry, changed)) = self.registry_signal.get()
            && let (Some(registry), Some(changed)) = (registry.upgrade(), changed.upgrade())
        {
            match lock(&registry, "registry notification") {
                Ok(guard) => {
                    changed.notify_all();
                    drop(guard);
                }
                Err(error) => self.wake.refuse(&error),
            }
        }
    }

    pub(super) fn snapshot(&self) -> BridgeCounters {
        let mut counters = match lock(&self.counters, "counters snapshot") {
            Ok(counters) => counters.clone(),
            Err(error) => {
                self.wake.refuse(&error);
                BridgeCounters::default()
            }
        };
        counters.error = self.wake.reason.get().cloned().or(counters.error);
        match super::counts::lifecycle(life::counts(self.life.get()), self.previous_lifecycle) {
            Ok(lifecycle) => counters.lifecycle = lifecycle,
            Err(error) => {
                self.wake.refuse(&error);
                counters.error = Some(error.to_string());
            }
        }
        counters
    }

    #[cfg(feature = "test-support")]
    pub(super) fn observe_answers(&self, count: u64, bound: std::time::Duration) -> bool {
        let result = lock(&self.counters, "answer observation").and_then(|counters| {
            self.changed
                .wait_timeout_while(counters, bound, |counters| {
                    self.settled.load(Ordering::Acquire) < count && counters.error.is_none()
                })
                .map_err(|_| failure("link_bridge_poisoned: answer observation"))
        });
        match result {
            Ok((counters, _)) => {
                let observed = self.settled.load(Ordering::Acquire) >= count
                    && counters.error.is_none()
                    && self.wake.check().is_ok();
                drop(counters);
                observed
            }
            Err(error) => {
                self.wake.refuse(&error);
                false
            }
        }
    }
}

type RegistryMap = Mutex<BTreeMap<String, Arc<Observation>>>;
#[cfg(feature = "test-support")]
type RegistrySignal = (Weak<RegistryMap>, Weak<Condvar>);

/// Live per-link observations shared with the health endpoint.
#[derive(Clone, Debug, Default)]
pub struct BridgeRegistry(
    pub(super) Arc<RegistryMap>,
    Arc<RegistryFault>,
    Arc<Condvar>,
);

#[derive(Debug, Default)]
struct RegistryFault {
    reason: OnceLock<String>,
    stops: OnceLock<Box<RegistryStop>>,
}

#[derive(Debug)]
struct RegistryStop {
    id: String,
    wake: Mutex<Weak<Wake>>,
    next: OnceLock<Box<Self>>,
}

impl BridgeRegistry {
    /// Bounds only an observation of the registry's signalled changes.
    #[cfg(feature = "test-support")]
    /// # Errors
    /// The registry lock was poisoned.
    pub fn observe(
        &self,
        bound: std::time::Duration,
        mut reached: impl FnMut(&BTreeMap<String, super::BridgeCounters>) -> bool,
    ) -> Result<bool, ServerError> {
        let (observations, _) = self
            .2
            .wait_timeout_while(
                lock(&self.0, "registry observation")?,
                bound,
                |observations| {
                    let snapshot = observations
                        .iter()
                        .map(|(id, observation)| (id.clone(), observation.snapshot()))
                        .collect();
                    !reached(&snapshot)
                },
            )
            .map_err(|_| failure("registry observation poisoned"))?;
        let snapshot = observations
            .iter()
            .map(|(id, observation)| (id.clone(), observation.snapshot()))
            .collect();
        drop(observations);
        Ok(reached(&snapshot))
    }

    /// Reads all configured bridge counters without scheduling or touching storage.
    #[must_use]
    pub fn snapshot(&self) -> BTreeMap<String, BridgeCounters> {
        match lock(&self.0, "health registry") {
            Ok(observations) => observations
                .iter()
                .map(|(id, observation)| (id.clone(), observation.snapshot()))
                .collect(),
            Err(error) => {
                self.refuse(&error);
                BTreeMap::new()
            }
        }
    }

    pub(crate) fn remember(&self, id: String, counters: BridgeCounters) -> Result<(), ServerError> {
        let observation = Observation::retaining(None, counters)?;
        self.install(id, &observation);
        observation.wake.check()
    }
    /// Records a refused dial and answers whether its reason differs from
    /// the last refusal's, so the retry owner tells a changed one by name
    /// (ACCESS-009: a dial that fails after Lys issues is told, not silent).
    pub(crate) fn dial_refused(&self, id: &str, reason: String) -> Result<bool, ServerError> {
        let mut counters = self.snapshot().remove(id).unwrap_or_default();
        let changed = counters.last_refusal.as_ref() != Some(&reason);
        counters.dial_refused = counters
            .dial_refused
            .checked_add(1)
            .ok_or_else(|| failure("configured dial refusal counter overflow"))?;
        counters.last_refusal = Some(reason);
        self.remember(id.to_owned(), counters)?;
        Ok(changed)
    }
    pub(crate) fn open_refused(&self, id: &str, reason: String) -> Result<bool, ServerError> {
        let mut counters = self.snapshot().remove(id).unwrap_or_default();
        let changed = counters.last_refusal.as_ref() != Some(&reason);
        counters.refused = counters
            .refused
            .checked_add(1)
            .ok_or_else(|| failure("configured open refusal counter overflow"))?;
        counters.last_refusal = Some(reason);
        self.remember(id.to_owned(), counters)?;
        Ok(changed)
    }
    #[cfg(feature = "test-support")]
    pub(crate) fn observe_answers(
        &self,
        ids: &[String],
        count: u64,
        bound: std::time::Duration,
    ) -> Result<bool, ServerError> {
        let reached = |observations: &BTreeMap<String, Arc<Observation>>| {
            ids.iter().all(|id| {
                observations.get(id).is_some_and(|observation| {
                    observation.settled.load(Ordering::Acquire) >= count
                        && observation.snapshot().error.is_none()
                })
            })
        };
        let (observations, _) = self
            .2
            .wait_timeout_while(
                lock(&self.0, "configured answer observation")?,
                bound,
                |observations| !reached(observations),
            )
            .map_err(|_| failure("configured answer observation poisoned"))?;
        let observed = reached(&observations);
        drop(observations);
        Ok(observed)
    }
    pub(super) fn install(&self, id: String, observation: &Arc<Observation>) {
        if let Err(error) = self.register_stop(&id, observation) {
            observation.wake.refuse(&error);
            return;
        }
        #[cfg(feature = "test-support")]
        if observation
            .registry_signal
            .set((Arc::downgrade(&self.0), Arc::downgrade(&self.2)))
            .is_err()
        {
            observation
                .wake
                .refuse(&failure("bridge observation registry already installed"));
            return;
        }
        let result = lock(&self.0, "health registry installation").map(|mut registry| {
            registry.insert(id, Arc::clone(observation));
            self.2.notify_all();
        });
        if let Err(error) = result {
            self.refuse(&error);
        }
        if let Some(reason) = self.1.reason.get() {
            observation.wake.refuse(&failure(reason.clone()));
        }
    }
    fn register_stop(&self, id: &str, observation: &Arc<Observation>) -> Result<(), ServerError> {
        // Replacement bridges update the same stop node instead of retaining an outage's history.
        let mut next = &self.1.stops;
        loop {
            if let Some(existing) = next.get() {
                if existing.id == id {
                    *lock(&existing.wake, "registry stop replacement")? =
                        Arc::downgrade(&observation.wake);
                    return Ok(());
                }
                next = &existing.next;
                continue;
            }
            let stop = Box::new(RegistryStop {
                id: id.to_owned(),
                wake: Mutex::new(Arc::downgrade(&observation.wake)),
                next: OnceLock::new(),
            });
            if next.set(stop).is_ok() {
                return Ok(());
            }
        }
    }

    fn refuse(&self, error: &ServerError) {
        if self.1.reason.set(error.to_string()).is_err() {
            return;
        }
        tracing::error!(%error, "link bridge registry refused poisoned state");
        let mut next = self.1.stops.get();
        while let Some(stop) = next {
            match lock(&stop.wake, "registry stop refusal") {
                Ok(wake) => {
                    if let Some(wake) = wake.upgrade() {
                        wake.refuse(error);
                    }
                }
                Err(error) => tracing::error!(%error, "registry stop refused poisoned state"),
            }
            next = stop.next.get();
        }
    }
}

#[derive(Debug, Default)]
pub(super) struct WakeState {
    ready: bool,
    stopping: bool,
}

#[derive(Debug, Default)]
pub(super) struct Wake {
    pub(super) state: Mutex<WakeState>,
    changed: Condvar,
    reason: OnceLock<String>,
    observation: Weak<Observation>,
    interrupt: Option<TcpStream>,
}

impl Wake {
    pub(super) fn notify(&self) {
        match lock(&self.state, "wake notification") {
            Ok(mut state) => state.ready = true,
            Err(error) => self.refuse(&error),
        }
        self.changed.notify_one();
    }

    pub(super) fn stopping(&self) -> bool {
        match lock(&self.state, "wake stop observation") {
            Ok(state) => state.stopping,
            Err(error) => {
                self.refuse(&error);
                true
            }
        }
    }

    pub(super) fn stop(&self) {
        match lock(&self.state, "wake stop") {
            Ok(mut state) => state.stopping = true,
            Err(error) => self.refuse(&error),
        }
        self.changed.notify_all();
    }

    pub(super) fn wait(&self) -> bool {
        let result = lock(&self.state, "wake wait").and_then(|state| {
            self.changed
                .wait_while(state, |state| !state.ready && !state.stopping)
                .map_err(|_| failure("link_bridge_poisoned: wake wait"))
        });
        match result {
            Ok(mut state) => {
                state.ready = false;
                !state.stopping
            }
            Err(error) => {
                self.refuse(&error);
                false
            }
        }
    }

    pub(super) fn check(&self) -> Result<(), ServerError> {
        self.reason
            .get()
            .map_or(Ok(()), |reason| Err(failure(reason.clone())))
    }

    pub(super) fn refuse(&self, error: &ServerError) {
        if self.reason.set(error.to_string()).is_err() {
            return;
        }
        tracing::error!(%error, "link bridge stopped as a fault");
        match self.state.lock() {
            Ok(mut state) => state.stopping = true,
            Err(_) => self.changed.notify_all(),
        }
        self.changed.notify_all();
        if let Some(interrupt) = &self.interrupt {
            match interrupt.shutdown(Shutdown::Both) {
                Ok(()) => {}
                Err(error) if error.kind() == std::io::ErrorKind::NotConnected => {
                    tracing::info!(%error, "faulted bridge carrier already closed");
                }
                Err(error) => tracing::error!(%error, "faulted bridge interruption failed"),
            }
        }
        if let Some(observation) = self.observation.upgrade() {
            match observation.counters.lock() {
                Ok(mut counters) => counters.error = Some(error.to_string()),
                Err(_) => observation.changed.notify_all(),
            }
            observation.changed.notify_all();
        }
    }
}