liminal-server 0.25.4

//! SRV-041 R2: the incarnation start runs once, behind the bind; an admission
//! that arrives while it runs waits and is admitted or refused by its outcome.

use std::sync::mpsc::{Receiver, Sender, TryRecvError, channel};
use std::sync::{Arc, Mutex};

use liminal::durability::{CountingStore, DurableStore, bridge::block_on, open_ephemeral};
use liminal_protocol::wire::{ClientRequest, ConnectionIncarnation, ServerValue};
use tracing_subscriber::prelude::*;

use super::incarnation::{ConnectionIncarnationAuthority, IncarnationStartInputs};
use crate::ServerError;
use crate::health::AdmissionReadiness;
use crate::server::participant::incarnation_stream::{
    ConnectionFateClass, IncarnationStream, encode_allocate_event_fixture,
    encode_open_connection_fate_event_fixture, encode_startup_event_fixture,
};
use crate::server::participant::{
    ConnectionFateWorkItem, ParticipantConnectionContext, ParticipantConnectionConversations,
    ParticipantSemanticError, ParticipantSemanticHandler,
};

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

fn internal(error: impl std::fmt::Display) -> ParticipantSemanticError {
    ParticipantSemanticError::Internal {
        message: error.to_string(),
    }
}

/// A participant whose restart repair says it has begun, then parks until the
/// test answers whether it succeeds.
#[derive(Debug)]
struct GatedRepair {
    entered: Mutex<Sender<()>>,
    release: Mutex<Receiver<bool>>,
}

impl ParticipantSemanticHandler for GatedRepair {
    fn handle(
        &self,
        _context: ParticipantConnectionContext,
        _conversations: &mut ParticipantConnectionConversations,
        _request: ClientRequest,
    ) -> Result<ServerValue, ParticipantSemanticError> {
        Err(ParticipantSemanticError::Unavailable)
    }

    fn handle_connection_fate(
        &self,
        _work_item: ConnectionFateWorkItem,
    ) -> Result<(), ParticipantSemanticError> {
        Ok(())
    }

    fn repair_unclean_server_restart(
        &self,
        _current_server_incarnation: u64,
    ) -> Result<(), ParticipantSemanticError> {
        self.entered
            .lock()
            .map_err(internal)?
            .send(())
            .map_err(internal)?;
        let succeeds = self
            .release
            .lock()
            .map_err(internal)?
            .recv()
            .map_err(internal)?;
        if succeeds {
            Ok(())
        } else {
            Err(internal("injected restart repair refusal"))
        }
    }
}

/// An authority built and not started, its start parked in the restart repair
/// once begun, with an admission already parked behind it.
struct Held {
    authority: Arc<ConnectionIncarnationAuthority>,
    admission: AdmissionReadiness,
    release: Sender<bool>,
    starter: std::thread::JoinHandle<Result<(), ServerError>>,
    admitted: Receiver<Result<ConnectionIncarnation, ServerError>>,
    admitting: std::thread::JoinHandle<()>,
}

fn held() -> Result<Held, Box<dyn std::error::Error>> {
    let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let (entered_send, entered) = channel();
    let (release, released) = channel();
    let admission = AdmissionReadiness::available();
    let authority = Arc::new(ConnectionIncarnationAuthority::starting(
        IncarnationStartInputs {
            store,
            maximum_references: 4,
            maximum_conversations: 3,
            handler: Arc::new(GatedRepair {
                entered: Mutex::new(entered_send),
                release: Mutex::new(released),
            }),
        },
        admission.clone(),
    )?);
    assert!(!admission.is_available(), "ready is false before the start");
    let starting = Arc::clone(&authority);
    let starter = std::thread::spawn(move || starting.ensure_started());
    entered.recv()?;
    let (waiting_send, waiting) = channel();
    authority.on_wait_for_test(waiting_send);
    let (admitted_send, admitted) = channel();
    let allocating = Arc::clone(&authority);
    let admitting = std::thread::spawn(move || {
        let _sent = admitted_send.send(allocating.allocate(&[]));
    });
    waiting.recv()?;
    assert!(
        matches!(admitted.try_recv(), Err(TryRecvError::Empty)),
        "an admission during the start waits; it is never refused for starting"
    );
    assert!(!admission.is_available(), "ready is false while starting");
    Ok(Held {
        authority,
        admission,
        release,
        starter,
        admitted,
        admitting,
    })
}

#[test]
fn an_admission_during_the_start_waits_and_is_admitted_when_it_finishes() -> TestResult {
    let held = held()?;
    held.release.send(true)?;
    assert_eq!(held.admitted.recv()??, ConnectionIncarnation::new(1, 0));
    held.starter.join().map_err(|_| "the start panicked")??;
    held.admitting
        .join()
        .map_err(|_| "the admission panicked")?;
    assert!(held.admission.is_available());
    assert!(held.authority.start_failure().is_none());
    Ok(())
}

fn repair_refused(result: &Result<impl std::fmt::Debug, ServerError>) -> bool {
    matches!(
        result,
        Err(ServerError::ParticipantIncarnation {
            phase: "server restart repair",
            ..
        })
    )
}

#[test]
fn a_failed_start_refuses_the_waiting_admission_with_its_error() -> TestResult {
    let held = held()?;
    held.release.send(false)?;
    let refused = held.admitted.recv()?;
    assert!(repair_refused(&refused), "{refused:?}");
    let started = held.starter.join().map_err(|_| "the start panicked")?;
    assert!(repair_refused(&started), "{started:?}");
    held.admitting
        .join()
        .map_err(|_| "the admission panicked")?;
    let later = held.authority.allocate(&[]);
    assert!(repair_refused(&later), "a later admission: {later:?}");
    let failure = held.authority.start_failure().ok_or("no start failure")?;
    assert!(repair_refused(&Err::<(), _>(failure)));
    assert!(!held.admission.is_available());
    Ok(())
}

/// A participant that absorbs every fate and repair at once.
#[derive(Debug)]
struct Absorbing;

impl ParticipantSemanticHandler for Absorbing {
    fn handle(
        &self,
        _context: ParticipantConnectionContext,
        _conversations: &mut ParticipantConnectionConversations,
        _request: ClientRequest,
    ) -> Result<ServerValue, ParticipantSemanticError> {
        Err(ParticipantSemanticError::Unavailable)
    }

    fn handle_connection_fate(
        &self,
        _work_item: ConnectionFateWorkItem,
    ) -> Result<(), ParticipantSemanticError> {
        Ok(())
    }
}

/// The boot step names logged on `liminal::boot`.
#[derive(Default)]
struct Names(Mutex<Vec<String>>);
struct Capture(Arc<Names>);

#[derive(Default)]
struct Name(Option<String>);

impl tracing::field::Visit for Name {
    fn record_debug(&mut self, _: &tracing::field::Field, _: &dyn std::fmt::Debug) {}
    fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
        if field.name() == "name" {
            self.0 = Some(value.to_owned());
        }
    }
}

impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for Capture {
    fn on_event(&self, event: &tracing::Event<'_>, _: tracing_subscriber::layer::Context<'_, S>) {
        if event.metadata().target() != "liminal::boot" {
            return;
        }
        let mut name = Name::default();
        event.record(&mut name);
        if let (Some(name), Ok(mut names)) = (name.0, self.0.0.lock()) {
            names.push(name);
        }
    }
}

/// SRV-041 R1, pinned with R2: a start over an unclean predecessor names its
/// recovery, and the start runs only when it is asked for, never at build.
#[test]
fn a_start_after_an_unclean_stop_names_its_recovery() -> TestResult {
    let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let payloads = [
        encode_startup_event_fixture()?,
        encode_allocate_event_fixture(4, &[])?,
        encode_open_connection_fate_event_fixture(
            ConnectionIncarnation::new(1, 0),
            ConnectionFateClass::ConnectionLost,
            3,
            &[5, 8],
        )?,
    ];
    for (sequence, payload) in payloads.into_iter().enumerate() {
        let sequence = u64::try_from(sequence)?;
        block_on(store.append(IncarnationStream::stream_key(), payload, sequence))??;
    }
    block_on(store.flush())??;
    let names = Arc::new(Names::default());
    let subscriber = tracing_subscriber::registry().with(Capture(Arc::clone(&names)));
    let (built, started) = tracing::subscriber::with_default(subscriber, || {
        let authority = ConnectionIncarnationAuthority::starting(
            IncarnationStartInputs {
                store,
                maximum_references: 4,
                maximum_conversations: 3,
                handler: Arc::new(Absorbing),
            },
            AdmissionReadiness::available(),
        )?;
        let built = names.0.lock().map(|logged| logged.clone()).ok();
        authority.ensure_started()?;
        Ok::<_, ServerError>((built, authority.allocate(&[])?))
    })?;
    assert_eq!(
        built.ok_or("captured names poisoned")?,
        Vec::<String>::new()
    );
    assert_eq!(started, ConnectionIncarnation::new(2, 0));
    let logged = names
        .0
        .lock()
        .map_err(|_| "captured names poisoned")?
        .clone();
    for name in [
        "incarnation_startup",
        "incarnation_recovery",
        "restart_repair",
    ] {
        assert_eq!(
            logged.iter().filter(|step| *step == name).count(),
            1,
            "{name} once: {logged:?}"
        );
    }
    Ok(())
}

/// SRV-048 R2: the start publishes on its flushed Startup row; the anchor
/// swap is not part of it.
#[test]
fn the_start_publishes_before_the_anchor_swap() -> TestResult {
    let counting = Arc::new(CountingStore::new(open_ephemeral(1)?));
    let store: Arc<dyn DurableStore> = counting.clone();
    block_on(store.append(
        &IncarnationStream::epoch_stream_key(1),
        encode_startup_event_fixture()?,
        0,
    ))??;
    block_on(store.flush())??;
    block_on(store.cas(IncarnationStream::epoch_anchor_key(), 0, 1))??;
    let admission = AdmissionReadiness::available();
    let authority = ConnectionIncarnationAuthority::starting(
        IncarnationStartInputs {
            store: Arc::clone(&store),
            maximum_references: 4,
            maximum_conversations: 3,
            handler: Arc::new(Absorbing),
        },
        admission.clone(),
    )?;
    let before = counting.snapshot().cas_calls;
    authority.ensure_started()?;
    assert!(admission.is_available(), "the start published");
    assert_eq!(authority.allocate(&[])?, ConnectionIncarnation::new(2, 0));
    assert_eq!(
        counting.snapshot().cas_calls - before,
        0,
        "the start swaps no anchor"
    );
    assert_eq!(
        block_on(store.read_value(IncarnationStream::epoch_anchor_key()))??,
        Some(1),
        "the anchor still names the predecessor once the start has published"
    );
    Ok(())
}