liminal-server 0.19.1

//! LIMINAL-FATE-1: a real pending terminal must not turn a second connection's
//! clean close or killed bound client into process-wide broker shutdown.
//!
//! Three participants reach the five-row retention cap through enrollment,
//! attach and publish. Killing the first leaves a durable Pending Died row;
//! closing the second then crosses the real socket/supervisor/WAL seam. The
//! survivor must still publish the drain, and a new client must still bind.
//! No injected semantic refusal and no synthetic pending marker is used.

use std::error::Error;
use std::sync::Arc;
use std::time::{Duration, Instant};

use liminal::durability::{DurableStore, bridge::block_on};
use liminal_protocol::wire::{
    AttachAttemptToken, ClientRequest, CredentialAttachRequest, DetachAttemptToken, DetachRequest,
    EnrollBound, EnrollmentRequest, EnrollmentToken, Generation, RecordAdmission,
    RecordAdmissionAttemptToken, ServerValue,
};

use crate::server::connection::ConnectionHandle;
use crate::server::shutdown::ShutdownHandle;

use super::e2e_tests::{SocketFixture, SocketPeer};
use super::log::{
    DecodedStoredOperation, OperationLog, StoredOperation, StoredTerminalDisposition,
};
use super::tests::test_participant_config;

const CONVERSATION: u64 = 619;
/// Scheduling watchdog only: fail a hung socket/fate fold rather than hanging
/// the test runner. State predicates, not elapsed time, establish completion.
const FATE_WATCHDOG: Duration = Duration::from_secs(10);
/// Avoid busy-spinning while the independently scheduled connection exits.
const POLL_INTERVAL: Duration = Duration::from_millis(10);

fn bound(value: ServerValue) -> Result<EnrollBound, Box<dyn Error>> {
    match value {
        ServerValue::EnrollBound(receipt) => Ok(receipt),
        other => Err(format!("fixture enrollment did not bind: {other:?}").into()),
    }
}

fn enroll(token: u8) -> ClientRequest {
    ClientRequest::Enrollment(EnrollmentRequest {
        conversation_id: CONVERSATION,
        enrollment_token: EnrollmentToken::new([token; 16]),
    })
}

fn publish(participant_id: u64, token: u8) -> ClientRequest {
    ClientRequest::RecordAdmission(RecordAdmission {
        conversation_id: CONVERSATION,
        participant_id,
        capability_generation: Generation::ONE,
        record_admission_attempt_token: RecordAdmissionAttemptToken::new([token; 16]),
        payload: vec![token],
    })
}

fn wait_for_exit(connection: &ConnectionHandle) -> Result<(), Box<dyn Error>> {
    let deadline = Instant::now() + FATE_WATCHDOG;
    while connection.is_live() {
        if Instant::now() >= deadline {
            return Err(format!(
                "connection {} did not finish its fate fold",
                connection.pid()
            )
            .into());
        }
        std::thread::sleep(POLL_INTERVAL);
    }
    Ok(())
}

fn terminal_is_pending(
    store: &Arc<dyn DurableStore>,
    participant_id: u64,
) -> Result<bool, Box<dyn Error>> {
    let log = OperationLog::new(Arc::clone(store), CONVERSATION);
    let mut sequence = 0;
    while let Some(entry) = block_on(log.read_at(sequence))?? {
        if let DecodedStoredOperation::V3(StoredOperation::Died { row }) = entry.operation {
            if row.participant_id == participant_id
                && row.disposition == StoredTerminalDisposition::Pending
                && row.drained.is_none()
            {
                return Ok(true);
            }
        }
        sequence += 1;
    }
    Ok(false)
}

fn pending_lane(
    server: &mut SocketFixture,
) -> Result<(SocketPeer, EnrollBound, u64), Box<dyn Error>> {
    let mut first = server.spawn_peer()?;
    let first_bound = bound(first.request(enroll(0x91))?)?;
    let attached = first.request(ClientRequest::CredentialAttach(CredentialAttachRequest {
        conversation_id: CONVERSATION,
        participant_id: first_bound.participant_id(),
        capability_generation: Generation::ONE,
        attach_secret: first_bound.attach_secret(),
        attach_attempt_token: AttachAttemptToken::new([0x92; 16]),
        accept_marker_delivery_seq: None,
    }))?;
    let ServerValue::AttachBound(attached) = attached else {
        return Err(format!("fixture attach did not bind: {attached:?}").into());
    };
    let mut second = server.spawn_peer()?;
    let second_bound = bound(second.request(enroll(0x93))?)?;
    let survivor = bound(server.request(enroll(0x94))?)?;
    let committed = first.request(ClientRequest::RecordAdmission(RecordAdmission {
        conversation_id: CONVERSATION,
        participant_id: first_bound.participant_id(),
        capability_generation: attached.origin_binding_epoch().capability_generation,
        record_admission_attempt_token: RecordAdmissionAttemptToken::new([0x95; 16]),
        payload: vec![0x95],
    }))?;
    assert!(
        matches!(committed, ServerValue::RecordCommitted(_)),
        "{committed:?}"
    );
    let first_connection = first.connection_handle()?;
    first.shutdown_transport()?;
    drop(first);
    wait_for_exit(&first_connection)?;
    assert!(
        terminal_is_pending(&server.durable_store(), first_bound.participant_id())?,
        "fixture did not durably occupy the candidate lane"
    );
    Ok((second, second_bound, survivor.participant_id()))
}

#[derive(Clone, Copy, Debug)]
enum CloseTrigger {
    CleanDisconnect,
    RefusedDetachThenEof,
    #[cfg(unix)]
    KilledClient,
}

fn keeps_broker_alive(trigger: CloseTrigger) -> Result<(), Box<dyn Error>> {
    let home = tempfile::tempdir()?;
    let mut config = test_participant_config();
    config.max_retained_record_rows = 5;
    let shutdown = ShutdownHandle::new();
    let mut server =
        SocketFixture::start_with_fatal_shutdown(home.path(), config, shutdown.clone())?;
    let (mut second, second_bound, survivor) = pending_lane(&mut server)?;
    let connection = second.connection_handle()?;
    match trigger {
        CloseTrigger::CleanDisconnect => second.disconnect()?,
        CloseTrigger::RefusedDetachThenEof => {
            let detached = second.request(ClientRequest::Detach(DetachRequest {
                conversation_id: CONVERSATION,
                participant_id: second_bound.participant_id(),
                capability_generation: Generation::ONE,
                detach_attempt_token: DetachAttemptToken::new([0x96; 16]),
            }));
            eprintln!("LIMINAL-FATE-1 DETACH response: {detached:?}");
            if detached.is_ok() {
                second.shutdown_transport()?;
            }
            // A failed-closed response must still reach the broker-liveness
            // assertion below; EOF alone does not prove process-wide shutdown.
        }
        #[cfg(unix)]
        CloseTrigger::KilledClient => second.kill_client_process()?,
    }
    wait_for_exit(&connection)?;
    let fresh = server.spawn_peer();
    let fresh_error = fresh.as_ref().err().map(ToString::to_string);
    assert!(
        !shutdown.is_initiated(),
        "LIMINAL-FATE-1 {trigger:?}: broker shutdown requested after pending fate; fresh admission: {fresh_error:?}"
    );
    let wal = crate::server::participant::incarnation_stream::IncarnationStream::new(
        server.durable_store(),
        crate::config::types::LimitsConfig::default().max_connections,
    );
    let parked = block_on(wal.unmatched_intents_for_test())??;
    if matches!(trigger, CloseTrigger::RefusedDetachThenEof) {
        // This additional request-level closure path did not leave an unmatched
        // Open in the measured fixture. It proves liveness, not parked replay.
        assert!(
            parked.is_empty(),
            "DETACH closure left unexpected WAL work: {parked:?}"
        );
    } else {
        assert_eq!(
            parked.len(),
            1,
            "exactly the second connection's Open must remain parked"
        );
        assert_eq!(
            Some(parked[0].connection_incarnation),
            connection.connection_incarnation()
        );
        assert_eq!(parked[0].conversations, vec![CONVERSATION]);
        eprintln!(
            "LIMINAL-FATE-1 {trigger:?}: parked durable Open {}",
            parked[0].open_sequence
        );
    }
    let mut fresh = fresh?;
    let admitted = fresh.request(ClientRequest::Enrollment(EnrollmentRequest {
        conversation_id: CONVERSATION + 1,
        enrollment_token: EnrollmentToken::new([0x97; 16]),
    }))?;
    bound(admitted)?;
    let drained = server.request(publish(survivor, 0x98))?;
    assert!(
        matches!(drained, ServerValue::RecordCommitted(_)),
        "drain: {drained:?}"
    );
    assert!(
        !shutdown.is_initiated(),
        "drain escalated to broker shutdown"
    );
    let outstanding = block_on(wal.unmatched_intents_for_test())??;
    assert!(
        outstanding.is_empty(),
        "publish failed to complete parked Opens {parked:?}: {outstanding:?}"
    );
    // A second response must not replay or append Complete for that Open again.
    let repeated = server.request(publish(survivor, 0x99))?;
    assert!(
        matches!(repeated, ServerValue::RecordCommitted(_)),
        "repeat: {repeated:?}"
    );
    assert!(block_on(wal.unmatched_intents_for_test())??.is_empty());
    drop(fresh);
    server.stop();
    Ok(())
}

#[test]
fn clean_disconnect_with_pending_fate_keeps_broker_alive() -> Result<(), Box<dyn Error>> {
    keeps_broker_alive(CloseTrigger::CleanDisconnect)
}

#[test]
fn refused_detach_then_eof_with_pending_fate_keeps_broker_alive() -> Result<(), Box<dyn Error>> {
    keeps_broker_alive(CloseTrigger::RefusedDetachThenEof)
}

#[test]
#[cfg(unix)]
fn killed_bound_client_with_pending_fate_keeps_broker_alive() -> Result<(), Box<dyn Error>> {
    keeps_broker_alive(CloseTrigger::KilledClient)
}