liminal-server 0.25.4

//! SRV-042 R2: a stop that begins while the incarnation start runs ends the
//! start at its next write, so the stop never waits out the rest of it; a
//! write already begun completes, the abandoned start refuses every later
//! admission with its phase and logs no error, and the next start over the
//! same store allocates strictly above every incarnation published before.
//!
//! Each test installs a process-wide subscriber, since the start runs on its
//! own thread; nextest runs each test in its own process.

use std::fmt::Write as _;
use std::net::{TcpListener, TcpStream};
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::mpsc::{Receiver, Sender, channel};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use liminal::durability::{
    DurabilityError, DurableStore, RoutedBatchItem, ScanCursor, ScanPage, StoredEntry,
    open_ephemeral,
};
use liminal_protocol::wire::ConnectionIncarnation;
use tracing_subscriber::prelude::*;

use super::ConnectionSupervisor;
use super::participant_only_services::supervisor_over;
use crate::ServerError;

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

/// How long a test waits on a signal before it reports the wait as the defect.
const WITHIN: Duration = Duration::from_secs(5);

/// The phase R2 names for a start the stop ended.
const ABANDONED: &str = "incarnation start abandoned at shutdown";

/// A store that, once armed, parks the first read (before it runs) or the
/// first write (after it has landed) until released, saying when it has; it
/// counts every write it begins.
#[derive(Debug)]
struct Gated {
    inner: Arc<dyn DurableStore>,
    read_armed: AtomicBool,
    write_armed: AtomicBool,
    entered: Mutex<Sender<()>>,
    release: Mutex<Receiver<()>>,
    writes: AtomicUsize,
}

impl Gated {
    fn new(inner: Arc<dyn DurableStore>) -> (Arc<Self>, Receiver<()>, Sender<()>) {
        let (entered_send, entered) = channel();
        let (release, released) = channel();
        let store = Arc::new(Self {
            inner,
            read_armed: AtomicBool::new(false),
            write_armed: AtomicBool::new(false),
            entered: Mutex::new(entered_send),
            release: Mutex::new(released),
            writes: AtomicUsize::new(0),
        });
        (store, entered, release)
    }

    fn writes(&self) -> usize {
        self.writes.load(Ordering::SeqCst)
    }

    /// Parks this thread once if `armed` is set, until the test releases it.
    /// The park is synchronous: liminal's `block_on` polls a store future
    /// once and needs it ready, so the store blocks its thread instead.
    fn park(&self, armed: &AtomicBool) -> Result<(), DurabilityError> {
        if !armed.swap(false, Ordering::SeqCst) {
            return Ok(());
        }
        let failed = |error: String| DurabilityError::ConfigError(error);
        self.entered
            .lock()
            .map_err(|error| failed(error.to_string()))?
            .send(())
            .map_err(|error| failed(error.to_string()))?;
        self.release
            .lock()
            .map_err(|error| failed(error.to_string()))?
            .recv()
            .map_err(|error| failed(error.to_string()))
    }

    fn before_read(&self) -> Result<(), DurabilityError> {
        self.park(&self.read_armed)
    }

    fn wrote<T>(&self, result: Result<T, DurabilityError>) -> Result<T, DurabilityError> {
        let value = result?;
        self.park(&self.write_armed)?;
        Ok(value)
    }

    fn begin_write(&self) {
        self.writes.fetch_add(1, Ordering::SeqCst);
    }
}

#[async_trait::async_trait]
impl DurableStore for Gated {
    async fn apply_routed_batch(
        &self,
        route_key: &str,
        items: Vec<RoutedBatchItem>,
    ) -> Result<(), DurabilityError> {
        self.begin_write();
        self.wrote(self.inner.apply_routed_batch(route_key, items).await)
    }

    async fn get_routed(
        &self,
        route_key: &str,
        key: &[u8],
    ) -> Result<Option<Vec<u8>>, DurabilityError> {
        self.before_read()?;
        self.inner.get_routed(route_key, key).await
    }

    async fn append(
        &self,
        stream_key: &str,
        payload: Vec<u8>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.begin_write();
        self.wrote(self.inner.append(stream_key, payload, expected_seq).await)
    }

    async fn append_all(
        &self,
        stream_key: &str,
        payloads: Vec<Vec<u8>>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.begin_write();
        self.wrote(
            self.inner
                .append_all(stream_key, payloads, expected_seq)
                .await,
        )
    }

    async fn read_from(
        &self,
        stream_key: &str,
        offset: u64,
        limit: usize,
    ) -> Result<Vec<StoredEntry>, DurabilityError> {
        self.before_read()?;
        self.inner.read_from(stream_key, offset, limit).await
    }

    async fn cas(&self, key: &str, old_value: u64, new_value: u64) -> Result<(), DurabilityError> {
        self.begin_write();
        self.wrote(self.inner.cas(key, old_value, new_value).await)
    }

    async fn read_stream_next_seq(&self, stream_key: &str) -> Result<Option<u64>, DurabilityError> {
        self.before_read()?;
        self.inner.read_stream_next_seq(stream_key).await
    }

    async fn read_value(&self, key: &str) -> Result<Option<u64>, DurabilityError> {
        self.before_read()?;
        self.inner.read_value(key).await
    }

    async fn scan_page(
        &self,
        prefix: &str,
        cursor: Option<ScanCursor>,
        limit: usize,
    ) -> Result<ScanPage, DurabilityError> {
        self.before_read()?;
        self.inner.scan_page(prefix, cursor, limit).await
    }

    async fn flush(&self) -> Result<(), DurabilityError> {
        self.begin_write();
        self.wrote(self.inner.flush().await)
    }
}

/// Every ERROR event, from any thread, as `target: fields`.
#[derive(Default)]
struct Errors(Arc<Mutex<Vec<String>>>);

struct Line(String);

impl tracing::field::Visit for Line {
    fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
        // Writing to a String cannot fail.
        if write!(self.0, " {}={value:?}", field.name()).is_err() {
            self.0.push_str(" <unwritten>");
        }
    }
}

impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for Errors {
    fn on_event(&self, event: &tracing::Event<'_>, _: tracing_subscriber::layer::Context<'_, S>) {
        if *event.metadata().level() != tracing::Level::ERROR {
            return;
        }
        let mut line = Line(format!("{}:", event.metadata().target()));
        event.record(&mut line);
        if let Ok(mut errors) = self.0.lock() {
            errors.push(line.0);
        }
    }
}

fn capture_errors() -> Result<Arc<Mutex<Vec<String>>>, Box<dyn std::error::Error>> {
    let errors = Errors::default();
    let lines = Arc::clone(&errors.0);
    tracing::subscriber::set_global_default(tracing_subscriber::registry().with(errors))?;
    Ok(lines)
}

/// Admits one TCP connection on `supervisor`, answering its incarnation.
fn admit_one(
    supervisor: &ConnectionSupervisor,
) -> Result<ConnectionIncarnation, Box<dyn std::error::Error>> {
    let listener = TcpListener::bind("127.0.0.1:0")?;
    let _client = TcpStream::connect(listener.local_addr()?)?;
    let (server, _) = listener.accept()?;
    let handle = supervisor.spawn_connection(server)?;
    Ok(handle
        .connection_incarnation()
        .ok_or("an admitted connection carries its incarnation")?)
}

/// One whole start and stop over `store`, answering the incarnation it
/// allocated.
fn boot_once(
    store: Arc<dyn DurableStore>,
) -> Result<ConnectionIncarnation, Box<dyn std::error::Error>> {
    let supervisor = supervisor_over(store)?;
    let allocated = admit_one(&supervisor)?;
    supervisor.shutdown();
    supervisor.finish_store_work()?;
    Ok(allocated)
}

fn is_abandoned(error: &ServerError) -> bool {
    matches!(error, ServerError::ParticipantIncarnation { phase, .. } if *phase == ABANDONED)
}

/// Boots once, then starts again over the gated store with `armed` set,
/// stops while the start is parked, releases it, and checks R2's outcome.
fn a_stop_while_parked(arm: impl FnOnce(&Gated)) -> TestResult {
    let errors = capture_errors()?;
    let inner: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let first = boot_once(Arc::clone(&inner))?;

    let (gated, entered, release) = Gated::new(Arc::clone(&inner));
    let store: Arc<dyn DurableStore> = gated.clone();
    let supervisor = supervisor_over(store)?;
    arm(&gated);
    supervisor.listening()?;
    entered.recv_timeout(WITHIN)?;
    supervisor.shutdown();
    let writes_at_shutdown = gated.writes();
    release.send(())?;
    let finished = supervisor.finish_store_work();
    assert_eq!(
        gated.writes(),
        writes_at_shutdown,
        "the start began a write after the shutdown began"
    );
    assert!(
        finished.is_ok(),
        "an abandoned start is a clean stop: {finished:?}"
    );
    let failure = supervisor.incarnation_start_failure();
    assert!(
        failure.as_ref().is_some_and(is_abandoned),
        "the start answers {ABANDONED:?}: {failure:?}"
    );
    let later = admit_one(&supervisor);
    assert!(
        later
            .as_ref()
            .is_err_and(|error| error.downcast_ref().is_some_and(is_abandoned)),
        "a later admission is refused with {ABANDONED:?}: {later:?}"
    );
    drop(supervisor);

    let next = boot_once(inner)?;
    assert!(
        next.server_incarnation > first.server_incarnation,
        "the next start allocates above every published incarnation: {next:?} after {first:?}"
    );
    // Over the whole test: the abandoned start, and the clean boots whose
    // stops can meet a connection before its readiness registers (SRV-043).
    let logged = errors.lock().map_err(|error| error.to_string())?.clone();
    assert!(
        logged.is_empty(),
        "an abandoned start, and a stop over a fresh connection, log no error: {logged:?}"
    );
    Ok(())
}

#[test]
fn a_shutdown_during_the_replay_ends_the_start_before_its_successor_epoch() -> TestResult {
    a_stop_while_parked(|gated| gated.read_armed.store(true, Ordering::SeqCst))
}

#[test]
fn a_shutdown_after_the_successor_append_landed_ends_the_start_and_the_next_start_boots_above_it()
-> TestResult {
    a_stop_while_parked(|gated| gated.write_armed.store(true, Ordering::SeqCst))
}