liminal-server 0.25.4

//! SRV-048 R2: the incarnation start publishes on its flushed Startup row, and
//! the start thread settles the epoch anchor's swap after the held
//! connections are admitted; a refused swap stops the server.

use std::sync::Arc;

use liminal::durability::{
    CountingStore, DurabilityError, DurableStore, RoutedBatchItem, ScanCursor, ScanPage,
    StoredEntry, bridge::block_on, open_ephemeral,
};

use super::ConnectionSupervisor;
use super::incarnation::{
    ANCHOR_SETTLE_PHASE, ConnectionIncarnationAuthority, IncarnationStartInputs,
};
use super::incarnation_tests::{config, services, store};
use crate::ServerError;
use crate::health::AdmissionReadiness;
use crate::server::listener::ServerListener;
use crate::server::participant::incarnation_stream::IncarnationStream;
use crate::server::participant::{
    ConnectionFateWorkItem, ParticipantConnectionContext, ParticipantConnectionConversations,
    ParticipantSemanticError, ParticipantSemanticHandler,
};
use liminal_protocol::wire::{ClientRequest, ServerValue};

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

fn anchor(store: &Arc<dyn DurableStore>) -> Result<Option<u64>, Box<dyn std::error::Error>> {
    Ok(block_on(
        store.read_value(IncarnationStream::epoch_anchor_key()),
    )??)
}

/// 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(())
    }
}

/// An authority's start publishes with no swap made; its settle makes the
/// one swap, and a second settle makes none.
#[test]
fn the_authority_settles_the_swap_its_start_parked() -> TestResult {
    let counting = Arc::new(CountingStore::new(open_ephemeral(1)?));
    let observed: Arc<dyn DurableStore> = counting.clone();
    let admission = AdmissionReadiness::available();
    let authority = ConnectionIncarnationAuthority::starting(
        IncarnationStartInputs {
            store: Arc::clone(&observed),
            maximum_references: 4,
            maximum_conversations: 3,
            handler: Arc::new(Absorbing),
        },
        admission.clone(),
    )?;
    authority.ensure_started()?;
    assert!(admission.is_available(), "the start published");
    assert_eq!(
        counting.snapshot().cas_calls,
        0,
        "the start swaps no anchor"
    );
    assert_eq!(anchor(&observed)?, None);

    authority.settle_anchor()?;
    assert_eq!(counting.snapshot().cas_calls, 1);
    assert_eq!(anchor(&observed)?, Some(1), "the settle names the start");
    authority.settle_anchor()?;
    assert_eq!(
        counting.snapshot().cas_calls,
        1,
        "a second settle swaps nothing"
    );
    Ok(())
}

/// The supervisor's start thread settles the swap: a bound start that the
/// shutdown has joined leaves the anchor naming its incarnation.
#[test]
fn a_bound_start_settles_the_anchor_on_the_start_thread() -> TestResult {
    let store = store()?;
    let config = config()?;
    let supervisor = ConnectionSupervisor::with_services(services(&config, Arc::clone(&store))?)?;
    let listener = ServerListener::bind(&config, supervisor.clone())?;
    assert_eq!(supervisor.current_server_incarnation()?, Some(1));
    listener.shutdown()?;
    supervisor.finish_store_work()?;
    assert_eq!(
        anchor(&store)?,
        Some(1),
        "the start thread, joined by the shutdown, settled the swap"
    );
    supervisor.shutdown();
    Ok(())
}

/// A store whose compare-and-swap of the epoch anchor refuses.
#[derive(Debug)]
struct RefusingAnchor {
    inner: Arc<dyn DurableStore>,
}

#[async_trait::async_trait]
impl DurableStore for RefusingAnchor {
    async fn apply_routed_batch(
        &self,
        route_key: &str,
        items: Vec<RoutedBatchItem>,
    ) -> Result<(), DurabilityError> {
        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.inner.get_routed(route_key, key).await
    }

    async fn append(
        &self,
        stream_key: &str,
        payload: Vec<u8>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        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.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.inner.read_from(stream_key, offset, limit).await
    }

    async fn cas(&self, key: &str, old_value: u64, new_value: u64) -> Result<(), DurabilityError> {
        if key == IncarnationStream::epoch_anchor_key() {
            return Err(DurabilityError::ConfigError(
                "injected anchor swap refusal".to_owned(),
            ));
        }
        self.inner.cas(key, old_value, new_value).await
    }

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

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

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

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

/// A refused settle names the swap's from and to under its own phase.
#[test]
fn a_refused_settle_names_its_from_and_to() -> TestResult {
    let refusing: Arc<dyn DurableStore> = Arc::new(RefusingAnchor { inner: store()? });
    let authority = ConnectionIncarnationAuthority::starting(
        IncarnationStartInputs {
            store: refusing,
            maximum_references: 4,
            maximum_conversations: 3,
            handler: Arc::new(Absorbing),
        },
        AdmissionReadiness::available(),
    )?;
    authority.ensure_started()?;
    match authority.settle_anchor() {
        Err(ServerError::ParticipantIncarnation { phase, message }) => {
            assert_eq!(phase, ANCHOR_SETTLE_PHASE);
            assert!(message.contains("from 0 to 1"), "{message}");
        }
        other => return Err(format!("the refused settle answered {other:?}").into()),
    }
    Ok(())
}

/// A swap the store refuses after the start published stops the server
/// through the fatal shutdown, the published epoch durable and the anchor
/// unmoved.
#[test]
fn a_refused_settle_stops_the_server() -> TestResult {
    let inner = store()?;
    let refusing: Arc<dyn DurableStore> = Arc::new(RefusingAnchor {
        inner: Arc::clone(&inner),
    });
    let config = config()?;
    let shutdown = crate::server::shutdown::ShutdownHandle::new();
    let supervisor = ConnectionSupervisor::with_fatal_shutdown(
        services(&config, refusing)?,
        None,
        None,
        config.limits,
        shutdown.clone(),
    )?;
    let listener = ServerListener::bind(&config, supervisor.clone())?;
    shutdown.wait();
    assert_eq!(
        supervisor.current_server_incarnation()?,
        Some(1),
        "the start published before its swap was refused"
    );
    assert_eq!(anchor(&inner)?, None, "the refused swap moved nothing");
    assert_eq!(
        block_on(inner.read_from(&IncarnationStream::epoch_stream_key(1), 0, 8))??.len(),
        1,
        "the published epoch's Startup row is durable"
    );
    listener.shutdown()?;
    supervisor.shutdown();
    Ok(())
}