liminal-server 0.21.2

//! The wire fixture keeps shutdown on its owning test thread.

use super::{Client, TestResult, config_with};
use crate::config::types::ServerConfig;
use crate::server::connection::LiminalConnectionServices;
use crate::server::self_streams::owner::PublisherRuntime;
use crate::server::self_streams::sink::SelfStreams;
use crate::server::shutdown::{ShutdownHandle, run_shutdown_sequence};
use crate::server::signals::SignalCounts;
use crate::server::{ConnectionSupervisor, ServerListener};
use liminal::durability::StoreCountsHandle;
use std::ops::Deref;
use std::sync::Arc;
use std::time::Duration;
use tempfile::TempDir;

pub(super) type Server = Fixture<Running>;

pub(super) struct Fixture<State> {
    state: Option<State>,
    shutdown: fn(State) -> TestResult,
}

impl<State> Fixture<State> {
    fn new(state: State, shutdown: fn(State) -> TestResult) -> Self {
        Self {
            state: Some(state),
            shutdown,
        }
    }

    pub(super) fn stop(mut self) -> TestResult {
        let state = self.state.take().ok_or("server fixture already stopped")?;
        (self.shutdown)(state)
    }
}

impl<State> Deref for Fixture<State> {
    type Target = State;

    fn deref(&self) -> &Self::Target {
        match self.state.as_ref() {
            Some(state) => state,
            None => panic!("server fixture already stopped"),
        }
    }
}

impl<State> Drop for Fixture<State> {
    fn drop(&mut self) {
        let Some(state) = self.state.take() else {
            return;
        };
        if let Err(error) = (self.shutdown)(state) {
            eprintln!("console wire fixture shutdown failed: {error}");
            // Preserve a test failure already unwinding while still naming teardown failure.
            if !std::thread::panicking() {
                panic!("console wire fixture shutdown failed: {error}");
            }
        }
    }
}

/// A server started as the runtime starts one, over an on-disk store.
pub(super) struct Running {
    pub(super) services: Arc<LiminalConnectionServices>,
    pub(super) listener: ServerListener,
    pub(super) publisher: PublisherRuntime,
    pub(super) counts: StoreCountsHandle,
    store: TempDir,
}

impl Fixture<Running> {
    pub(super) fn start(extra_channels: usize) -> TestResult<Self> {
        Self::start_with(extra_channels, &[])
    }

    /// [`Self::start`] with `named` configured as ephemeral channels too.
    pub(super) fn start_with(extra_channels: usize, named: &[&str]) -> TestResult<Self> {
        Self::start_verifying(extra_channels, named, None)
    }

    /// [`Self::start_with`], admitting connections by `pass` when given.
    pub(super) fn start_verifying(
        extra_channels: usize,
        named: &[&str],
        pass: Option<crate::auth_pass::PassVerifier>,
    ) -> TestResult<Self> {
        Self::start_configuring(extra_channels, named, pass, |_| Ok(()))
    }

    pub(super) fn start_configuring(
        extra_channels: usize,
        named: &[&str],
        pass: Option<crate::auth_pass::PassVerifier>,
        configure: impl FnOnce(&mut ServerConfig) -> TestResult,
    ) -> TestResult<Self> {
        let store = TempDir::new()?;
        let mut config = config_with(store.path(), extra_channels, named)?;
        configure(&mut config)?;
        let services = Arc::new(LiminalConnectionServices::from_config(&config)?);
        // The server's cost rows come on its owner's 10 s schedule, and no
        // console pin measures them: they are held out of every window.
        services.signals().wake().hold();
        let counts = services.store_counts();
        let shutdown = ShutdownHandle::new();
        let supervisor = ConnectionSupervisor::with_fatal_shutdown(
            services.clone(),
            None,
            pass,
            config.limits,
            shutdown.clone(),
        )?;
        let publisher = crate::server::runtime::start_self_stream_publisher(
            &services,
            &supervisor,
            &config,
            &shutdown,
        )?
        .ok_or("no self-stream publisher: the console streams need one")?;
        let listener = ServerListener::bind(&config, supervisor)?;
        Ok(Self::new(
            Running {
                services,
                listener,
                publisher,
                counts,
                store,
            },
            Running::stop,
        ))
    }

    pub(super) fn signals(&self) -> SignalCounts {
        self.services.signals().counts()
    }

    /// Waits until every connection opened or closed so far has had its
    /// `liminal.events` row written, and every row that writing queued in
    /// turn (the row of the store flush its write made), so a window opened
    /// next holds none of those rows.
    ///
    /// The connections' rows are counted when queued; the owner, released
    /// if a test held it, then drains the queue until nothing more is
    /// queued.
    pub(super) fn settled(&self) -> TestResult {
        let signals = self.signals();
        let events = signals.connections_opened + signals.connections_closed;
        if self
            .services
            .signals()
            .await_events(events, Duration::from_secs(30))
        {
            let streams = self.streams()?;
            streams.release_drainer()?;
            Ok(streams.drain_all()?)
        } else {
            Err(format!("{events} connection events not queued in 30 s").into())
        }
    }

    pub(super) fn connect(&self) -> TestResult<Client> {
        Client::connect(self.listener.local_addr())
    }

    /// The self-stream publisher's streams, whose drainer a test holds.
    pub(super) fn streams(&self) -> TestResult<Arc<SelfStreams>> {
        Ok(Arc::clone(
            self.services
                .signals()
                .streams()
                .ok_or("no self-stream publisher is installed")?,
        ))
    }
}

impl Running {
    /// Stops the server the way the runtime does, then removes its store.
    ///
    /// The shutdown sequence runs first, then the self-stream publisher; a
    /// failed removal fails the test instead of leaving residue.
    fn stop(self) -> TestResult {
        let Self {
            services,
            mut listener,
            mut publisher,
            counts,
            store,
        } = self;
        let supervisor = listener.supervisor();
        let stopped = run_shutdown_sequence(&mut listener, None, &supervisor, Duration::ZERO);
        if stopped.is_err() {
            supervisor.shutdown();
        }
        let finished = publisher.finish();
        services.channel_cluster().supervisor().shutdown();
        // Everything holding the store goes before its directory does, so no
        // late write lands in a directory being removed.
        drop((supervisor, listener, publisher, counts, services));
        let closed = store.close();
        let failures: Vec<_> = [
            stopped
                .err()
                .map(|error| format!("server shutdown: {error}")),
            finished
                .err()
                .map(|error| format!("publisher finish: {error}")),
            closed.err().map(|error| format!("store removal: {error}")),
        ]
        .into_iter()
        .flatten()
        .collect();
        if failures.is_empty() {
            Ok(())
        } else {
            Err(failures.join("; ").into())
        }
    }
}

#[test]
fn a_panicking_test_runs_fixture_shutdown_before_releasing_state() -> TestResult {
    use std::sync::Mutex;
    type Events = Arc<Mutex<Vec<(&'static str, std::thread::ThreadId)>>>;
    struct Probe {
        events: Events,
        fails: bool,
    }
    impl Probe {
        fn shutdown(self) -> TestResult {
            self.events
                .lock()
                .map_err(|_| "events poisoned")?
                .push(("shutdown", std::thread::current().id()));
            if self.fails {
                Err("forced shutdown failure".into())
            } else {
                Ok(())
            }
        }
    }
    impl Drop for Probe {
        fn drop(&mut self) {
            match self.events.lock() {
                Ok(mut events) => events.push(("release", std::thread::current().id())),
                Err(error) => panic!("fixture release probe failed: {error}"),
            }
        }
    }
    for fails in [false, true] {
        let events = Arc::new(Mutex::new(vec![]));
        let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let fixture = Fixture::new(
                Probe {
                    events: Arc::clone(&events),
                    fails,
                },
                Probe::shutdown,
            );
            assert!(fixture.state.is_some());
            panic!("forced fixture assertion failure");
        }));
        let panic = caught.err().ok_or("fixture assertion did not panic")?;
        assert_eq!(
            panic.downcast_ref::<&str>(),
            Some(&"forced fixture assertion failure")
        );
        let events = events.lock().map_err(|_| "events poisoned")?;
        let thread = std::thread::current().id();
        assert_eq!(*events, vec![("shutdown", thread), ("release", thread)]);
    }
    let events = Arc::new(Mutex::new(vec![]));
    Fixture::new(
        Probe {
            events: Arc::clone(&events),
            fails: false,
        },
        Probe::shutdown,
    )
    .stop()?;
    let events = events.lock().map_err(|_| "events poisoned")?;
    let thread = std::thread::current().id();
    assert_eq!(*events, vec![("shutdown", thread), ("release", thread)]);
    Ok(())
}