liminal-server 0.25.4

use std::error::Error;
use std::sync::Arc;
use std::time::Duration;

use liminal::durability::bridge::block_on;

use crate::config::{LimitsConfig, ServerConfig, ServicesConfig};
use crate::server::connection::services::{LiminalConnectionServices, open_persistent_store};
use crate::server::connection::{ConnectionSupervisor, LoopbackDuplex};
use crate::server::shutdown::ShutdownHandle;

#[test]
fn shutdown_releases_production_store_holders_before_reopen() -> Result<(), Box<dyn Error>> {
    let root = tempfile::tempdir()?;
    let config = ServerConfig {
        listen_address: "127.0.0.1:0".parse()?,
        extra_listen_addresses: Vec::new(),
        health_listen_address: "127.0.0.1:0".parse()?,
        drain_timeout_ms: 0,
        channels: Vec::new(),
        routing_rules: Vec::new(),
        persistence_path: Some(root.path().to_path_buf()),
        cluster: None,
        auth: None,
        services: ServicesConfig::default(),
        limits: LimitsConfig::default(),
        participant: Some(crate::server::participant::test_participant_config()),
        websocket: None,
    };
    let services = Arc::new(LiminalConnectionServices::from_config(&config)?);
    let shutdown = ShutdownHandle::new();
    let supervisor = ConnectionSupervisor::with_fatal_shutdown(
        services.clone(),
        None,
        None,
        config.limits,
        shutdown.clone(),
    )?;
    let mut publisher = crate::server::runtime::start_self_stream_publisher(
        &services,
        &supervisor,
        &config,
        &shutdown,
    )?
    .ok_or("self-stream publisher was not installed")?;
    let (client, server) = LoopbackDuplex::bounded(256 * 1024);
    let connection = supervisor.spawn_loopback_connection(server)?;
    let pid = connection.pid();
    let (reached, release, done) = supervisor.install_reclaim_barrier(pid);
    supervisor
        .scheduler()
        .terminate_process(pid, beamr::process::ExitReason::Error);
    reached.wait();
    supervisor.inner.runtime.finish(pid);
    drop(connection);
    supervisor.listening()?;
    let store = services.durable_store();
    block_on(store.cas("shutdown-reopen", 0, 41))??;
    drop(store);

    let (phase, observed) = std::sync::mpsc::channel();
    *supervisor
        .inner
        .reclaim_stop_observer
        .lock()
        .map_err(|_| "stop observer poisoned")? = Some(phase.clone());
    let reopen_path = root.path().to_path_buf();
    let counts = services.store_counts();
    let worker = std::thread::spawn(move || {
        let stopped = crate::server::runtime::runtime_links::shutdown(
            &supervisor,
            Duration::ZERO,
            || Ok(()),
            (
                &mut crate::server::runtime::runtime_links::ConfiguredLinks::default(),
                || publisher.finish(),
            ),
        );
        let after_flush = counts.snapshot();
        drop(client);
        drop(publisher);
        drop(supervisor);
        drop(services);
        assert_eq!(
            counts.snapshot().since(&after_flush),
            liminal::durability::StoreCounts::default()
        );
        phase
            .send(false)
            .map_err(|error| crate::ServerError::ListenerAccept {
                message: error.to_string(),
            })?;
        stopped?;
        open_persistent_store(&reopen_path)
    });
    let joining = observed.recv()?;
    let reopened = if joining {
        release.wait();
        done.wait();
        worker.join().map_err(|_| "shutdown worker panicked")?
    } else {
        let result = worker.join().map_err(|_| "shutdown worker panicked")?;
        release.wait();
        done.wait();
        result
    }?;
    assert_eq!(block_on(reopened.read_value("shutdown-reopen"))??, Some(41));
    Ok(())
}

#[test]
fn shutdown_joins_the_spine_release_owner_before_reopen() -> Result<(), Box<dyn Error>> {
    let root = tempfile::tempdir()?;
    let config = ServerConfig {
        listen_address: "127.0.0.1:0".parse()?,
        extra_listen_addresses: Vec::new(),
        health_listen_address: "127.0.0.1:0".parse()?,
        drain_timeout_ms: 0,
        channels: Vec::new(),
        routing_rules: Vec::new(),
        persistence_path: Some(root.path().to_path_buf()),
        cluster: None,
        auth: None,
        services: ServicesConfig::default(),
        limits: LimitsConfig::default(),
        participant: None,
        websocket: None,
    };
    let services = Arc::new(LiminalConnectionServices::from_config(&config)?);
    let supervisor = ConnectionSupervisor::with_services(services.clone())?;
    let spine = services
        .durable_store()
        .reclaim_spine()
        .ok_or("missing spine")?;
    let (entered, entry) = std::sync::mpsc::channel();
    let (release, released) = std::sync::mpsc::channel();
    let owner = crate::server::connection::reclaim_placement::ReclaimRelease::start(move || {
        let retained_spine = spine;
        assert!(entered.send(()).is_ok());
        let result = released.recv();
        drop(retained_spine);
        assert!(result.is_ok());
    })?;
    supervisor.retain_reclaim_release(owner)?;
    entry.recv()?;
    let (joining, observed) = std::sync::mpsc::channel();
    *supervisor
        .inner
        .reclaim_stop_observer
        .lock()
        .map_err(|_| "stop observer poisoned")? = Some(joining);
    let reopen_path = root.path().to_path_buf();
    let worker = std::thread::spawn(move || {
        let scheduler = Arc::downgrade(&supervisor.scheduler());
        let runtime = supervisor.runtime_weak();
        crate::server::runtime::runtime_links::shutdown(
            &supervisor,
            Duration::ZERO,
            || Ok(()),
            (
                &mut crate::server::runtime::runtime_links::ConfiguredLinks::default(),
                || Ok(()),
            ),
        )?;
        drop(supervisor);
        drop(services);
        assert!(scheduler.upgrade().is_none());
        assert!(runtime.upgrade().is_none());
        open_persistent_store(&reopen_path)
    });
    assert!(observed.recv()?);
    release.send(())?;
    let reopened = worker.join().map_err(|_| "shutdown worker panicked")??;
    assert!(block_on(reopened.read_value("missing"))??.is_none());
    Ok(())
}