liminal-server 0.21.1

//! SRV-005: the shared (optionally clustered) channel supervisor used by the
//! server's connection services.
//!
//! Extracted from `services.rs` to keep that file within the 500-line limit. A
//! single [`ChannelCluster`] holds the one [`ChannelSupervisor`] every configured
//! channel runs on, plus the [`ClusterResolver`] shared with that supervisor's
//! scheduler when clustering is configured. Building it is the only place the
//! choice between a clustered and a plain supervisor is made.

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

use liminal::channel::{ChannelRestartPolicy, ChannelSupervisor, ClusterObserver};
use liminal::envelope::Envelope;

use crate::ServerError;
use crate::cluster::discovery::{ClusterResolver, as_resolver};
use crate::config::types::ClusterConfig;

/// The shared channel supervisor backing every configured channel (SRV-005).
///
/// Plus the cluster resolver when clustering is configured. All channels run on
/// this ONE supervisor's scheduler so they share the clustered distribution
/// transport; the resolver is the SAME instance handed to that scheduler, kept
/// here so the cluster can dial seeds and learn peer names on it.
#[derive(Clone, Debug)]
pub struct ChannelCluster {
    supervisor: ChannelSupervisor,
    resolver: Option<Arc<ClusterResolver>>,
    observer: Arc<ServerObserver>,
}

/// The one observer the channel supervisor reports to (SRV-012 R6): every
/// subscribe and every departure, explicit or a dropped subscription's
/// trapped exit, reaches the console hub once (attach and detach) and,
/// once the cluster has started, `ClusterSync`. Departures arrive on the
/// channel actor's thread, so nothing here waits on a channel.
#[derive(Debug, Default)]
pub struct ServerObserver {
    cluster: OnceLock<Arc<dyn ClusterObserver>>,
    console: OnceLock<Arc<crate::console::ConsoleHub>>,
    /// Hears every subscribe and departure after the console has, console
    /// subjects included. Test-only: the order a test waits on.
    #[cfg(test)]
    witness: OnceLock<Arc<dyn ClusterObserver>>,
    /// The channel whose departures are held back from the console, and the
    /// subscriber pids held so far. Test-only: a late detach, forced without
    /// blocking the channel actor's thread.
    #[cfg(test)]
    held_departures: std::sync::Mutex<Option<HeldDepartures>>,
}

/// A held channel and the subscriber pids whose departures it holds back.
/// Test-only.
#[cfg(test)]
type HeldDepartures = (String, Vec<u64>);

impl ServerObserver {
    /// Forwards to the cluster's observer from now on. A second cluster is
    /// refused: the first keeps the channel.
    pub fn set_cluster(&self, cluster: Arc<dyn ClusterObserver>) -> Result<(), ServerError> {
        self.cluster
            .set(cluster)
            .map_err(|_| ServerError::ClusterJoin {
                message: "the channel observer already forwards to a cluster".to_owned(),
            })
    }

    /// Tells `witness` of every subscribe and departure from now on, after
    /// the console has heard it. Test-only.
    #[cfg(test)]
    pub(crate) fn set_witness(&self, witness: Arc<dyn ClusterObserver>) -> Result<(), ServerError> {
        self.witness
            .set(witness)
            .map_err(|_| ServerError::ClusterJoin {
                message: "the channel observer already has a witness".to_owned(),
            })
    }

    /// Holds back every departure from `channel` before the console hears
    /// it, until [`Self::release_departures`] hands them on. Test-only.
    #[cfg(test)]
    pub(crate) fn hold_departures(&self, channel: &str) -> std::io::Result<()> {
        *self.held()? = Some((channel.to_owned(), Vec::new()));
        Ok(())
    }

    /// Hands every held departure on, as the channel reported it. Test-only.
    #[cfg(test)]
    pub(crate) fn release_departures(&self) -> std::io::Result<()> {
        let held = self.held()?.take();
        if let Some((channel, pids)) = held {
            for pid in pids {
                self.on_unsubscribe(&channel, pid);
            }
        }
        Ok(())
    }

    /// The held departures. A poisoned lock is a test hook's own failure,
    /// not a configuration or admission error, so it is reported as one.
    #[cfg(test)]
    fn held(&self) -> std::io::Result<std::sync::MutexGuard<'_, Option<HeldDepartures>>> {
        self.held_departures.lock().map_err(|error| {
            std::io::Error::other(format!("the held-departures lock was poisoned: {error}"))
        })
    }

    /// Keeps a departure from a held channel back, answering whether it did.
    /// A poisoned lock holds nothing: a callback cannot answer an error.
    #[cfg(test)]
    fn held_back(&self, channel: &str, subscriber_pid: u64) -> bool {
        let Ok(mut held) = self.held_departures.lock() else {
            return false;
        };
        match held.as_mut() {
            Some((held_channel, pids)) if held_channel == channel => {
                pids.push(subscriber_pid);
                true
            }
            _ => false,
        }
    }

    /// Attaches and detaches the console hub's taps from now on.
    pub fn set_console(&self, console: Arc<crate::console::ConsoleHub>) -> Result<(), ServerError> {
        self.console
            .set(console)
            .map_err(|_| ServerError::ConfigValidation {
                message: "the channel observer already feeds a console hub".to_owned(),
            })
    }
}

impl ServerObserver {
    /// A cluster peer joined or left: one topology change (SRV-012 R7).
    pub fn peer(&self, name: &str, present: bool) {
        if let Some(console) = self.console.get()
            && let Err(error) = console.emit(crate::console::TopologyChange::Peer {
                name: name.to_owned(),
                present,
            })
        {
            tracing::warn!(%error, peer = name, "console topology change was not applied");
        }
    }
}

impl ClusterObserver for ServerObserver {
    fn on_subscribe(&self, channel: &str, subscriber_pid: u64) {
        if let Some(console) = self.console.get()
            && let Err(error) = console.attach(channel)
        {
            tracing::warn!(%error, channel, "console tap did not attach");
        }
        #[cfg(test)]
        if let Some(witness) = self.witness.get() {
            witness.on_subscribe(channel, subscriber_pid);
        }
        // SRV-012 R6: a console subject is this server's own; it is never
        // advertised to a peer.
        if crate::console::tappable(channel)
            && let Some(cluster) = self.cluster.get()
        {
            cluster.on_subscribe(channel, subscriber_pid);
        }
    }

    fn on_unsubscribe(&self, channel: &str, subscriber_pid: u64) {
        #[cfg(test)]
        if self.held_back(channel, subscriber_pid) {
            return;
        }
        if let Some(console) = self.console.get() {
            if let Err(error) = console.detach(channel) {
                tracing::warn!(%error, channel, "console tap did not detach");
            }
            if let Err(error) = console.tap().forget_receiver(subscriber_pid) {
                console.observe_failed();
                tracing::warn!(%error, channel, "TAP_RECEIVER_FAILED");
            }
        }
        #[cfg(test)]
        if let Some(witness) = self.witness.get() {
            witness.on_unsubscribe(channel, subscriber_pid);
        }
        // SRV-012 R6: a console subject is this server's own; it is never
        // advertised to a peer.
        if crate::console::tappable(channel)
            && let Some(cluster) = self.cluster.get()
        {
            cluster.on_unsubscribe(channel, subscriber_pid);
        }
    }

    fn on_publish(&self, channel: &str, envelope: &Envelope) {
        // SRV-012 R4: console subjects are this server's own and never fan
        // out to peers.
        if !crate::console::tappable(channel) {
            return;
        }
        if let Some(cluster) = self.cluster.get() {
            cluster.on_publish(channel, envelope);
        }
    }
}

impl ChannelCluster {
    /// The observer the supervisor reports to.
    #[must_use]
    pub const fn observer(&self) -> &Arc<ServerObserver> {
        &self.observer
    }

    /// The shared channel supervisor.
    #[must_use]
    pub const fn supervisor(&self) -> &ChannelSupervisor {
        &self.supervisor
    }

    /// The cluster resolver, present only when clustering is configured.
    #[must_use]
    pub const fn resolver(&self) -> Option<&Arc<ClusterResolver>> {
        self.resolver.as_ref()
    }
}

/// The channel restart policy used for server-hosted channels. Matches the
/// library default (one-for-one with a bounded budget) and is passed explicitly
/// because the distribution-enabled constructor takes a policy.
fn server_channel_policy() -> ChannelRestartPolicy {
    ChannelRestartPolicy::default()
}

/// Builds the shared channel supervisor (SRV-005).
///
/// When `cluster` is `Some`, the supervisor's scheduler is distribution-enabled
/// with the configured node identity and cookie, and a [`ClusterResolver`] is
/// created and shared between the scheduler and the returned [`ChannelCluster`].
/// When `None`, an ordinary non-clustered supervisor is built.
///
/// # Errors
/// Returns [`ServerError`] when the underlying scheduler cannot start.
pub fn build_channel_cluster(
    cluster: Option<&ClusterConfig>,
) -> Result<ChannelCluster, ServerError> {
    let Some(cluster) = cluster else {
        let supervisor =
            ChannelSupervisor::with_policy(server_channel_policy()).map_err(|error| {
                ServerError::ConfigValidation {
                    message: format!("failed to start channel supervisor: {error}"),
                }
            })?;
        return Ok(with_observer(supervisor, None));
    };
    let resolver = Arc::new(ClusterResolver::new());
    let supervisor = ChannelSupervisor::with_distribution(
        cluster.node_name.clone(),
        node_creation(),
        cluster.cookie.clone(),
        as_resolver(Arc::clone(&resolver)),
        server_channel_policy(),
    )
    .map_err(|error| ServerError::ClusterJoin {
        message: format!("failed to start clustered channel supervisor: {error}"),
    })?;
    Ok(with_observer(supervisor, Some(resolver)))
}

fn with_observer(
    supervisor: ChannelSupervisor,
    resolver: Option<Arc<ClusterResolver>>,
) -> ChannelCluster {
    let observer = Arc::new(ServerObserver::default());
    supervisor.install_observer(Arc::clone(&observer) as Arc<dyn ClusterObserver>);
    ChannelCluster {
        supervisor,
        resolver,
        observer,
    }
}

/// A non-zero node-incarnation value for beamr distribution. Derived from the
/// process start time so two incarnations of the same node name on one host get
/// distinct creations; never zero (beamr reserves 0 for "unknown creation").
///
/// The distinct-incarnations property holds only while the clock reads after the
/// epoch. A clock fault — a host set before 1970 — folds every incarnation to the
/// same constant value, and beamr then cannot tell a restarted node from its own
/// predecessor. That case is not detected here and not reported; the non-zero
/// guarantee still holds, which is what keeps distribution functioning at all.
fn node_creation() -> u32 {
    // A pre-epoch clock yields `Err` and is folded to 0 (hence to the constant 1
    // below) rather than refused: distribution needs SOME non-zero creation to
    // start, and there is no better value to invent. This is the exact case in
    // which two incarnations become indistinguishable.
    let since_epoch = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_or(0, |duration| duration.as_secs());
    // Fold the 64-bit seconds into 31 bits (the low word, masked positive) so the
    // creation is a stable-per-process, non-zero value; the exact bits do not
    // matter, only that incarnations differ and none is zero.
    let folded = u32::try_from(since_epoch & u64::from(u32::MAX)).unwrap_or(u32::MAX);
    (folded & 0x7fff_ffff).max(1)
}