liminal-server 0.19.1

//! Every teardown of a connection that tracks a participant conversation
//! completes its `ConnectionLost` fate before the record goes.
//!
//! A teardown that skips the fate leaves the participant bound to a dead
//! connection: the churn convergence e2e's clean retry was answered `Bound`
//! for a binding whose connection had died at its readiness probe (box 12,
//! Vincent's 7 Oct 10:07 cause).

use std::error::Error;
use std::net::{TcpListener, TcpStream};
use std::sync::{Arc, Mutex, OnceLock, Weak};
use std::time::{Duration, Instant};

use liminal::durability::{DurableStore, open_ephemeral};
use liminal_protocol::wire::{ClientRequest, ServerValue};

use super::ConnectionProcess;
use crate::ServerError;
use crate::config::types::LimitsConfig;
use crate::server::connection::ConnectionSupervisor;
use crate::server::connection::services::{ConnectionServices, LiminalConnectionServices};
use crate::server::connection::supervisor::ConnectionRuntime;
use crate::server::participant::{
    ConnectionFateClass, ConnectionFateWorkItem, InstalledParticipantService,
    ParticipantConnectionContext, ParticipantConnectionConversations, ParticipantSemanticError,
    ParticipantSemanticHandler, ParticipantServiceFatal,
};
use crate::server::shutdown::ShutdownHandle;

const CONVERSATION: u64 = 0x_C0_17;

/// Records every connection fate the runtime hands it, and, once a runtime is
/// observed, how many connections had departed when each fate ran.
#[derive(Debug, Default)]
struct RecordingFates {
    fates: Mutex<Vec<ConnectionFateWorkItem>>,
    runtime: OnceLock<Weak<ConnectionRuntime>>,
    departures_at_fate: Mutex<Vec<u64>>,
}

impl RecordingFates {
    fn recorded(&self) -> Result<Vec<ConnectionFateWorkItem>, Box<dyn Error>> {
        Ok(self
            .fates
            .lock()
            .map_err(|_| "the recorded fates' lock was poisoned")?
            .clone())
    }

    fn departures_at_fate(&self) -> Result<Vec<u64>, Box<dyn Error>> {
        Ok(self
            .departures_at_fate
            .lock()
            .map_err(|_| "the departure snapshots' lock was poisoned")?
            .clone())
    }
}

impl ParticipantSemanticHandler for RecordingFates {
    fn service_fatal(&self) -> Result<Option<ParticipantServiceFatal>, ParticipantSemanticError> {
        Ok(None)
    }

    fn latch_connection_fate_intent_incomplete(
        &self,
        open_sequence: u64,
        conversation_id: u64,
    ) -> Result<ParticipantServiceFatal, ParticipantSemanticError> {
        Ok(ParticipantServiceFatal::ConnectionFateIntentIncomplete {
            open_sequence,
            conversation_id,
        })
    }

    fn handle_connection_fate(
        &self,
        work_item: ConnectionFateWorkItem,
    ) -> Result<(), ParticipantSemanticError> {
        if let Some(runtime) = self.runtime.get().and_then(Weak::upgrade) {
            let departures =
                runtime
                    .departures()
                    .map_err(|error| ParticipantSemanticError::Internal {
                        message: error.to_string(),
                    })?;
            self.departures_at_fate
                .lock()
                .map_err(|_| ParticipantSemanticError::Internal {
                    message: "the departure snapshots' lock was poisoned".to_owned(),
                })?
                .push(departures);
        }
        self.fates
            .lock()
            .map_err(|_| ParticipantSemanticError::Internal {
                message: "the recorded fates' lock was poisoned".to_owned(),
            })?
            .push(work_item);
        Ok(())
    }

    fn publication_conversation_limit(&self) -> u64 {
        1
    }

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

fn tcp_pair() -> Result<(TcpStream, TcpStream), Box<dyn Error>> {
    let listener = TcpListener::bind("127.0.0.1:0")?;
    let client = TcpStream::connect(listener.local_addr()?)?;
    let (server, _) = listener.accept()?;
    Ok((client, server))
}

/// A supervisor whose participant service records fates, and one live
/// connection on it whose durable incarnation the process under test takes.
struct Fixture {
    supervisor: ConnectionSupervisor,
    fates: Arc<RecordingFates>,
    pid: u64,
    _client: TcpStream,
    process: ConnectionProcess,
}

impl Fixture {
    fn new() -> Result<Self, Box<dyn Error>> {
        let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
        let fates = Arc::new(RecordingFates::default());
        let service = InstalledParticipantService::new(
            Arc::clone(&fates) as Arc<dyn ParticipantSemanticHandler>,
            store,
            u64::MAX,
        )
        .map_err(|error| format!("participant service configuration failed: {error:?}"))?;
        let services: Arc<dyn ConnectionServices> =
            Arc::new(LiminalConnectionServices::empty()?.with_participant_service(service));
        let supervisor = ConnectionSupervisor::with_fatal_shutdown(
            services,
            None,
            None,
            LimitsConfig::default(),
            ShutdownHandle::new(),
        )?;
        let (client, server) = tcp_pair()?;
        let connection = supervisor.spawn_connection(server)?;
        let pid = connection.pid();
        let incarnation = connection
            .connection_incarnation()
            .ok_or("participant connection lacks a durable incarnation")?;
        let holder = Arc::new(Mutex::new(None));
        let mut process = ConnectionProcess::from_holder(
            supervisor.runtime_for_test(),
            None,
            &holder,
            Some(incarnation),
        );
        process.state.participant_conversations.track(CONVERSATION);
        Ok(Self {
            supervisor,
            fates,
            pid,
            _client: client,
            process,
        })
    }

    /// The one fate the teardown completed, failing unless there is exactly one.
    fn the_fate(&self) -> Result<ConnectionFateWorkItem, Box<dyn Error>> {
        let recorded = self.fates.recorded()?;
        let [fate] = recorded.as_slice() else {
            return Err(format!("expected one completed fate, recorded {recorded:?}").into());
        };
        Ok(fate.clone())
    }
}

impl Drop for Fixture {
    fn drop(&mut self) {
        self.supervisor.shutdown();
    }
}

/// A readiness-probe failure (`fail_slice`, "connection readiness contract
/// failed") completes the `ConnectionLost` fate of the conversations it tracks.
#[test]
fn a_readiness_probe_failure_completes_the_connection_lost_fate() -> Result<(), Box<dyn Error>> {
    let mut fixture = Fixture::new()?;

    let _outcome = fixture.process.fail_slice(
        u64::MAX,
        &ServerError::ListenerAccept {
            message: "connection readiness probe failed: injected reset".to_owned(),
        },
    );

    let fate = fixture.the_fate()?;
    assert_eq!(fate.class, ConnectionFateClass::ConnectionLost);
    assert_eq!(fate.tracked_conversations, vec![CONVERSATION]);
    Ok(())
}

/// A participant publication fault ("participant publication failed; tearing
/// down the connection") completes the `ConnectionLost` fate of the
/// conversations it tracks.
#[test]
fn a_publication_teardown_completes_the_connection_lost_fate() -> Result<(), Box<dyn Error>> {
    let mut fixture = Fixture::new()?;
    fixture.process.fail_next_pump = true;

    let outcome = fixture.process.service_participant_pushes(u64::MAX);

    assert!(outcome.is_some(), "the publication fault did not tear down");
    let fate = fixture.the_fate()?;
    assert_eq!(fate.class, ConnectionFateClass::ConnectionLost);
    assert_eq!(fate.tracked_conversations, vec![CONVERSATION]);
    Ok(())
}

/// The public departure signal is raised only after the departed connection's
/// fate has completed: when the fate runs, the connection has not yet
/// departed, and a waiter parked on [`ConnectionSupervisor::await_departures`]
/// wakes to a fate already recorded.
#[test]
fn a_departure_is_raised_only_after_its_fate_completes() -> Result<(), Box<dyn Error>> {
    let mut fixture = Fixture::new()?;
    fixture
        .fates
        .runtime
        .set(Arc::downgrade(&fixture.supervisor.runtime_for_test()))
        .map_err(|_| "the fate recorder already observes a runtime")?;
    let before = fixture.supervisor.departures()?;
    let deadline = Instant::now() + Duration::from_secs(30);
    let waiter = {
        let supervisor = fixture.supervisor.clone();
        let fates = Arc::clone(&fixture.fates);
        std::thread::spawn(move || {
            let departed = supervisor
                .await_departures(before, 1, deadline)
                .map_err(|error| error.to_string())?;
            fates
                .recorded()
                .map(|recorded| (departed, recorded.len()))
                .map_err(|error| error.to_string())
        })
    };

    let pid = fixture.pid;
    let _outcome = fixture.process.fail_slice(
        pid,
        &ServerError::ListenerAccept {
            message: "connection readiness probe failed: injected reset".to_owned(),
        },
    );

    assert_eq!(
        fixture.fates.departures_at_fate()?,
        vec![before],
        "the connection had already departed when its fate ran"
    );
    let (departed, fates_at_wake) = waiter
        .join()
        .map_err(|_| "the departure waiter panicked")??;
    assert!(departed, "no departure was raised before the deadline");
    assert_eq!(
        fates_at_wake, 1,
        "the waiter woke before the fate was recorded"
    );
    assert_eq!(fixture.supervisor.departures()?, before.wrapping_add(1));
    Ok(())
}