liminal-server 0.25.4

//! 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;
use crate::server::signals::SignalCounts;
use crate::server::{ConnectionSupervisor, ServerListener};
use liminal::durability::StoreCountsHandle;
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 running(&self) -> TestResult<&State> {
        self.state
            .as_ref()
            .ok_or_else(|| "server fixture already stopped".into())
    }

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

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.
            assert!(
                std::thread::panicking(),
                "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) -> TestResult<SignalCounts> {
        Ok(self.running()?.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
            .running()?
            .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.running()?.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.running()?
                .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.
    ///
    /// Publisher finalization belongs to the coordinator before the final
    /// flush; 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 = stop_server(&supervisor, || listener.stop_accepting(), &mut publisher);
        // 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}")),
            closed.err().map(|error| format!("store removal: {error}")),
        ]
        .into_iter()
        .flatten()
        .collect();
        if failures.is_empty() {
            Ok(())
        } else {
            Err(failures.join("; ").into())
        }
    }
}

fn stop_server(
    supervisor: &ConnectionSupervisor,
    stop_accepting: impl FnOnce() -> Result<(), crate::ServerError>,
    publisher: &mut PublisherRuntime,
) -> Result<(), crate::ServerError> {
    crate::server::runtime::runtime_links::shutdown(
        supervisor,
        Duration::ZERO,
        stop_accepting,
        (
            &mut crate::server::runtime::runtime_links::ConfiguredLinks::default(),
            || publisher.finish(),
        ),
    )
}

#[path = "fixture/stop_tests.rs"]
mod stop_tests;

#[test]
fn a_fixture_without_running_state_returns_a_named_error() -> TestResult {
    let mut fixture = Fixture::new(7_u64, |_| Ok(()));
    let state = fixture.state.take().ok_or("fixture state missing")?;
    (fixture.shutdown)(state)?;
    let error = fixture
        .running()
        .err()
        .ok_or("stopped fixture returned state")?;
    assert_eq!(error.to_string(), "server fixture already stopped");
    Ok(())
}

#[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) {
            let released = {
                let locked = self.events.lock();
                match locked {
                    Ok(mut events) => {
                        events.push(("release", std::thread::current().id()));
                        drop(events);
                        Ok(())
                    }
                    Err(error) => {
                        let message = error.to_string();
                        drop(error);
                        Err(message)
                    }
                }
            };
            assert!(
                released.is_ok(),
                "fixture release probe failed: {released:?}"
            );
        }
    }
    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());
            std::panic::resume_unwind(Box::new("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(())
}