liminal-server 0.20.1

//! SRV-012 R8: one pin per console bookkeeping failure path. Each poisons the
//! named lock as a panic on a thread holding it would, runs the client
//! operation that drives the hook, and finds the operation answering as it
//! would without the console while the named counter moves from 0 to 1.

use std::error::Error;
use std::path::Path;
use std::sync::{Arc, Condvar, Mutex};
use std::time::Duration;

use liminal::Envelope;
use liminal::channel::ClusterObserver;
use liminal::conversation::EchoBehaviour;
use liminal::protocol::{CausalContext, MessageEnvelope, SchemaId};

use super::hub::HubLock;
use super::{ConsoleCountsSnapshot, TOPOLOGY_SUBJECT};
use crate::config::types::{ChannelDef, LimitsConfig, ServerConfig, ServicesConfig};
use crate::server::connection::services::{
    ConnectionServices, LiminalConnectionServices, roster_reads,
};

type TestResult = Result<(), Box<dyn Error>>;

const ORDERS_TAP: &str = "console.tap.orders";

fn channel(name: &str) -> ChannelDef {
    ChannelDef {
        name: name.to_owned(),
        schema_ref: None,
        durable: false,
        loaded_schema: None,
    }
}

/// Services with `orders` and the console's channels configured ephemeral,
/// so a subscribe to them needs no running console publisher.
fn services(persistence_path: &Path) -> Result<LiminalConnectionServices, Box<dyn Error>> {
    services_with(
        vec![
            channel("orders"),
            channel(TOPOLOGY_SUBJECT),
            channel(ORDERS_TAP),
        ],
        persistence_path,
    )
}

fn services_with(
    channels: Vec<ChannelDef>,
    persistence_path: &Path,
) -> Result<LiminalConnectionServices, Box<dyn Error>> {
    Ok(LiminalConnectionServices::from_config(&ServerConfig {
        listen_address: "127.0.0.1:0".parse()?,
        health_listen_address: "127.0.0.1:0".parse()?,
        drain_timeout_ms: 30_000,
        channels,
        routing_rules: Vec::new(),
        persistence_path: Some(persistence_path.to_path_buf()),
        cluster: None,
        auth: None,
        services: ServicesConfig::default(),
        limits: LimitsConfig::default(),
        participant: None,
        websocket: None,
    })?)
}

fn envelope() -> MessageEnvelope {
    MessageEnvelope::new(
        SchemaId::new([0_u8; SchemaId::WIRE_LEN]),
        CausalContext::independent(),
        br#"{"order":1}"#.to_vec(),
    )
}

fn counts(services: &LiminalConnectionServices) -> ConsoleCountsSnapshot {
    services.console().counts()
}

/// Counts departures. The library reports one on the channel actor's thread
/// after the console's detach has run (the server observer detaches first,
/// then forwards here), so waiting on it orders a test's read after the
/// detach.
#[derive(Debug, Default)]
struct Departures {
    seen: Mutex<u64>,
    changed: Condvar,
}

impl Departures {
    fn install(services: &LiminalConnectionServices) -> Result<Arc<Self>, Box<dyn Error>> {
        let departures = Arc::new(Self::default());
        services
            .channel_cluster()
            .observer()
            .set_witness(Arc::clone(&departures) as Arc<dyn ClusterObserver>)?;
        Ok(departures)
    }

    /// Waits for `target` departures; the minute is a guard, never a deadline.
    fn wait_for(&self, target: u64) -> TestResult {
        let (seen, guard) = self
            .changed
            .wait_timeout_while(
                self.seen.lock().map_err(|_| "departures poisoned")?,
                Duration::from_secs(60),
                |seen| *seen < target,
            )
            .map_err(|_| "departures poisoned")?;
        let count = *seen;
        drop(seen);
        if guard.timed_out() {
            return Err(format!("{count} departures, expected {target}").into());
        }
        Ok(())
    }
}

impl ClusterObserver for Departures {
    fn on_subscribe(&self, _channel: &str, _subscriber_pid: u64) {}

    fn on_unsubscribe(&self, _channel: &str, _subscriber_pid: u64) {
        if let Ok(mut seen) = self.seen.lock() {
            *seen = seen.saturating_add(1);
        }
        self.changed.notify_all();
    }

    fn on_publish(&self, _channel: &str, _envelope: &Envelope) {}
}

/// A responder registered while the topology is poisoned is registered;
/// the change it could not apply is one `topology_emit` failure.
#[test]
fn a_poisoned_topology_counts_one_emit_failure() -> TestResult {
    let home = tempfile::tempdir()?;
    let services = services(home.path())?;
    assert_eq!(counts(&services).topology_emit_failures, 0);
    assert!(services.console().poison(HubLock::Topology));
    services.register_responder("orders.reply", Arc::new(EchoBehaviour))?;
    assert!(
        services
            .responder_subjects()?
            .contains(&"orders.reply".to_owned())
    );
    assert_eq!(counts(&services).topology_emit_failures, 1);
    Ok(())
}

/// A publish the tap cannot observe is still delivered.
#[test]
fn a_poisoned_tap_counts_one_observe_failure() -> TestResult {
    let home = tempfile::tempdir()?;
    let services = services(home.path())?;
    let _taker = services.subscribe("orders", &[], None)?;
    let _tap = services.subscribe(ORDERS_TAP, &[], None)?;
    assert_eq!(counts(&services).tap_observe_failures, 0);
    assert!(services.console().tap().poison_subscribers());
    let outcome = services.publish(7, "orders", &envelope(), None)?;
    assert!(
        outcome.delivered,
        "the publish answers as it would without the console"
    );
    assert_eq!(counts(&services).tap_observe_failures, 1);
    Ok(())
}

/// A tap subscribe while the tap table is poisoned is admitted and served;
/// the tap it could not start is one attach failure.
#[test]
fn a_poisoned_tap_table_counts_one_attach_failure() -> TestResult {
    let home = tempfile::tempdir()?;
    let services = services(home.path())?;
    assert!(services.console().poison(HubLock::Taps));
    let subscription = services.subscribe(ORDERS_TAP, &[], None);
    assert!(
        subscription.is_ok(),
        "the subscribe answers as it would without the console"
    );
    assert_eq!(counts(&services).tap_attach_failures, 1);
    Ok(())
}

/// An unsubscribe while the tap table is poisoned unsubscribes; the tap it
/// could not stop is one detach failure.
#[test]
fn a_poisoned_tap_table_counts_one_detach_failure() -> TestResult {
    let home = tempfile::tempdir()?;
    let services = services(home.path())?;
    let departures = Departures::install(&services)?;
    let subscription = services.subscribe(ORDERS_TAP, &[], None)?;
    assert_eq!(counts(&services).tap_detach_failures, 0);
    assert!(services.console().poison(HubLock::Taps));
    services.unsubscribe(subscription)?;
    departures.wait_for(1)?;
    assert_eq!(counts(&services).tap_detach_failures, 1);
    Ok(())
}

/// A topology change due to a subscriber, with the notifier slot poisoned,
/// is applied; the wake that reached nobody is one `console_wake` failure.
#[test]
fn a_poisoned_notifier_counts_one_wake_failure() -> TestResult {
    let home = tempfile::tempdir()?;
    let services = services(home.path())?;
    let _topology = services.subscribe(TOPOLOGY_SUBJECT, &[], None)?;
    assert_eq!(counts(&services).console_wake_failures, 0);
    assert!(services.console().poison(HubLock::Notifier));
    services.register_responder("orders.reply", Arc::new(EchoBehaviour))?;
    assert_eq!(counts(&services).console_wake_failures, 1);
    Ok(())
}

/// The last unsubscribe from a tap whose subscriber list is poisoned
/// unsubscribes; the tap subscription that could not leave the list is one
/// `tap_unsubscribe` failure.
#[test]
fn a_poisoned_subscriber_list_counts_one_unsubscribe_failure() -> TestResult {
    let home = tempfile::tempdir()?;
    let services = services(home.path())?;
    let departures = Departures::install(&services)?;
    let subscription = services.subscribe(ORDERS_TAP, &[], None)?;
    assert_eq!(counts(&services).tap_unsubscribe_failures, 0);
    assert!(services.console().tap().poison_subscribers());
    services.unsubscribe(subscription)?;
    departures.wait_for(1)?;
    assert_eq!(counts(&services).tap_unsubscribe_failures, 1);
    Ok(())
}

/// SRV-012 R7: with twenty channels, one topology snapshot takes the roster
/// read lock once, and sees all twenty.
#[test]
fn the_snapshot_reads_the_roster_once() -> TestResult {
    let home = tempfile::tempdir()?;
    let services = services_with(
        (0..20)
            .map(|index| channel(&format!("fixture.{index:02}")))
            .collect(),
        home.path(),
    )?;
    let before = roster_reads();
    let snapshot = super::snapshot(&services)?;
    assert_eq!(roster_reads() - before, 1);
    assert_eq!(snapshot.channels.len(), 20);
    Ok(())
}