liminal-server 0.25.4

//! Export subscriptions forward through one keyed carrier, settling each answer before the next send.
//!
//! Crossings are judged on the opened link resource at both ends, within its configured export prefixes.
//! A named import refusal records the subject and reason, advances its subscription past that message, and leaves the link open for other exports.

use crate::ServerError;
use crate::link_carrier::choice::{ChoiceLink as DialedLink, Interrupt as TcpStream};
use crate::server::connection::services::LiminalConnectionServices;
use subscriptions::{ExportSubscriptions, PendingDelivery};

mod counts;
mod crossing;
use crossing::cross;
mod fault;
mod life;
#[cfg(feature = "test-support")]
mod prepared;
mod start;
pub use fault::BridgeRegistry;
use fault::{Observation, Wake};
#[cfg(test)]
mod cancellation_tests;
#[cfg(test)]
mod fault_tests;
mod stored;
pub(crate) mod subscriptions;
use life::topology;
use std::collections::BTreeMap;
use std::net::Shutdown;
use std::sync::atomic::Ordering;
use std::sync::{Arc, Mutex};
use std::thread::JoinHandle;

/// The crossing-refusal code on the importing link's stream.
const LINK_CROSSING_REFUSED_CODE: u16 = 0x0203;

/// Observations for one link; positions belong to its local subscriptions.
#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize)]
pub struct BridgeCounters {
    /// Opens and ends observed by the keyed carrier.
    #[serde(flatten)]
    pub lifecycle: crate::link::LinkLifeCounters,
    /// Acknowledged crossings.
    pub crossed: u64,
    /// Configured dial attempts for which every carrier refused.
    pub dial_refused: u64,
    /// Crossings refused at either end and opens refused by the peer.
    pub refused: u64,
    /// Refused crossings counted for each subject, logged at its first refusal.
    pub refused_subjects: BTreeMap<String, u64>,
    /// Envelopes exceeding one Noise frame.
    pub too_large: u64,
    /// The startup durable position of each exported subject.
    pub resumed_from: BTreeMap<String, u64>,
    /// The last named refusal, when one was observed.
    pub last_refusal: Option<String>,
    /// A forwarding failure that stopped this link.
    pub error: Option<String>,
}

fn reached(prefix: &str, subject: &str) -> bool {
    subject == prefix
        || subject
            .strip_prefix(prefix)
            .is_some_and(|rest| rest.starts_with('.'))
}

fn consumer_id(link: &str, prefix: &str) -> String {
    let mut hash = blake3::Hasher::new();
    hash.update(link.as_bytes());
    hash.update(&[0]);
    hash.update(prefix.as_bytes());
    format!("link-bridge-{}", hash.finalize().to_hex())
}

fn failure(message: impl Into<String>) -> ServerError {
    ServerError::ListenerAccept {
        message: message.into(),
    }
}

/// Owns one prefix subscription per configured export and one carrier worker.
#[derive(Debug)]
pub struct LinkBridge {
    link_id: String,
    observation: Arc<Observation>,
    wake: Arc<Wake>,
    interrupt: TcpStream,
    worker: Option<JoinHandle<Result<(), ServerError>>>,
}

impl LinkBridge {
    /// Starts an already keyed link over bounded local subscriptions.
    ///
    /// # Errors
    /// Prefix validation, subscription recovery, or worker creation failed.
    pub fn start(
        link: impl Into<DialedLink>,
        services: Arc<LiminalConnectionServices>,
    ) -> Result<Self, ServerError> {
        let link = link.into();
        let peer = hex::encode(link.transport.peer().ed25519_public());
        Self::start_named(link, services, peer)
    }

    /// The current counters for this link.
    #[must_use]
    pub fn counters(&self) -> BridgeCounters {
        self.observation.snapshot()
    }

    /// Shares this link's live observations with a health registry.
    pub fn install_observations(&self, registry: &BridgeRegistry) {
        registry.install(self.link_id.clone(), &self.observation);
    }

    /// Arms a signalled test stop after the next ack, before checkpointing its subscription.
    #[cfg(feature = "test-support")]
    pub fn stop_before_next_acknowledge(&self) {
        self.observation
            .stop_before_ack
            .store(true, Ordering::Release);
    }

    /// Bounds only a test observation of settled crossing answers.
    #[cfg(feature = "test-support")]
    #[must_use]
    pub fn observe_answers(&self, count: u64, bound: std::time::Duration) -> bool {
        self.observation.observe_answers(count, bound)
    }

    /// Stops the subscription worker and interrupts an unanswered carrier read.
    ///
    /// # Errors
    /// The socket interruption or worker join failed, or forwarding had already failed.
    pub fn finish(mut self) -> Result<(), ServerError> {
        self.stop_worker()
    }

    fn stop_worker(&mut self) -> Result<(), ServerError> {
        // Mark the owned stop before the carrier can observe its closed lifecycle.
        self.wake.stop();
        let ended = self.observation.end_life(crate::link::LinkEnd::Shutdown);
        let interruption = self.interrupt.shutdown(Shutdown::Both);
        if let Some(worker) = self.worker.take() {
            worker
                .join()
                .map_err(|_| failure("link bridge worker panicked"))??;
        }
        let error = self.observation.snapshot().error;
        if let Some(error) = error {
            return Err(failure(error));
        }
        ended?;
        match interruption {
            Ok(()) => Ok(()),
            Err(error) if error.kind() == std::io::ErrorKind::NotConnected => {
                tracing::info!(link = %self.link_id,
                    "bridge carrier already closed at stop; subscription position retained");
                Ok(())
            }
            Err(error) => Err(failure(format!("link bridge interruption: {error}"))),
        }
    }
}

impl Drop for LinkBridge {
    fn drop(&mut self) {
        if self.worker.is_some() {
            if let Err(error) = self.stop_worker() {
                tracing::error!(link = %self.link_id, %error, "link bridge shutdown failed");
            }
        }
    }
}

fn run_worker(
    link: &mut DialedLink,
    feeds: &ExportSubscriptions,
    wake: &Wake,
    observation: &Observation,
    services: &LiminalConnectionServices,
    peer: &str,
) -> Result<(), ServerError> {
    topology(services, &link.link_id, peer, observation, None);
    let result = forward(link, feeds, wake, observation, services, peer);
    if let Err(error) = &result {
        wake.refuse(error);
    }
    observation.worker_ended(&result)?;
    topology(
        services,
        &link.link_id,
        peer,
        observation,
        Some(
            result
                .as_ref()
                .err()
                .map_or_else(|| "stopped".to_owned(), ToString::to_string),
        ),
    );
    observation.notify_changes();
    result
}

enum Crossing {
    Acknowledged,
    Refused(Option<String>),
    TooLarge,
    Stopped,
}

fn forward(
    link: &mut DialedLink,
    feeds: &ExportSubscriptions,
    wake: &Wake,
    observation: &Observation,
    services: &LiminalConnectionServices,
    peer: &str,
) -> Result<(), ServerError> {
    loop {
        wake.check()?;
        if wake.stopping() || observation.ended()? {
            wake.check()?;
            return Ok(());
        }
        if let Err(error) = link.carrier.refuse_unsolicited() {
            if wake.stopping() || observation.ended()? {
                wake.check()?;
                return Ok(());
            }
            return Err(failure(error.to_string()));
        }
        let Some(mut source) = feeds.next_ready()? else {
            if !wake.wait() {
                wake.check()?;
                return Ok(());
            }
            continue;
        };
        #[cfg(feature = "test-support")]
        let answer = match prepared::take(observation, &mut source)? {
            Some(sealed) => crossing::exchange(link, &source.subject, &sealed, wake)?,
            None => cross(link, &mut source, wake)?,
        };
        #[cfg(not(feature = "test-support"))]
        let answer = cross(link, &mut source, wake)?;
        if matches!(answer, Crossing::Stopped) {
            return Ok(());
        }
        let mut counters = lock(&observation.counters, "counters")?;
        #[cfg(feature = "test-support")]
        let stop_before_ack = matches!(answer, Crossing::Acknowledged)
            && observation.stop_before_ack.swap(false, Ordering::AcqRel);
        #[cfg(not(feature = "test-support"))]
        let stop_before_ack = false;
        match answer {
            Crossing::Acknowledged => {
                counters.crossed = counters
                    .crossed
                    .checked_add(1)
                    .ok_or_else(|| failure("bridge crossing counter overflow"))?;
            }
            Crossing::Refused(reason) => {
                counters.refused = counters
                    .refused
                    .checked_add(1)
                    .ok_or_else(|| failure("bridge refusal counter overflow"))?;
                let refused = counters
                    .refused_subjects
                    .entry(source.subject.clone())
                    .or_default();
                *refused = refused
                    .checked_add(1)
                    .ok_or_else(|| failure("bridge subject refusal counter overflow"))?;
                if *refused == 1 {
                    tracing::warn!(link = %link.link_id, subject = %source.subject, reason = ?reason,
                        "link crossing refused; subscription advances without retry");
                }
                counters.last_refusal = reason;
            }
            Crossing::TooLarge => {
                counters.too_large = counters
                    .too_large
                    .checked_add(1)
                    .ok_or_else(|| failure("bridge too_large counter overflow"))?;
                counters.last_refusal = Some("too_large".to_owned());
                tracing::warn!(link = %link.link_id, subject = %source.subject, reason = "too_large",
                    "link crossing refused locally");
            }
            Crossing::Stopped => return Ok(()),
        }
        drop(counters);
        if !stop_before_ack {
            source.acknowledge()?;
        }
        topology(services, &link.link_id, peer, observation, None);
        // Pair settlement with the observer's mutex so arming cannot miss the signal.
        let guard = lock(&observation.counters, "settlement counters")?;
        observation
            .settled
            .fetch_update(Ordering::Release, Ordering::Relaxed, |count| {
                count.checked_add(1)
            })
            .map_err(|_| failure("bridge settled answer counter overflow"))?;
        drop(guard);
        observation.notify_changes();
        if stop_before_ack {
            return Ok(());
        }
    }
}

fn lock<'a, Value>(
    mutex: &'a Mutex<Value>,
    state: &str,
) -> Result<std::sync::MutexGuard<'a, Value>, ServerError> {
    mutex
        .lock()
        .map_err(|_| failure(format!("link_bridge_poisoned: {state}")))
}