liminal-server 0.20.0

use std::error::Error;
use std::io::{Read, Write};
use std::net::{SocketAddr, TcpListener, TcpStream};
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, Weak};
use std::time::Duration;

use liminal::durability::{
    DurabilityError, DurableStore, ScanCursor, ScanPage, StoredEntry, bridge::block_on,
    open_ephemeral,
};
use liminal::protocol::{
    Frame, ProtocolError, ProtocolVersion, decode as decode_generic, encode as encode_generic,
    encoded_len as generic_encoded_len,
};
use liminal_protocol::lifecycle::{
    BindingRequiredLookupResult, BindingState, BindingTerminalAdmitError,
    ParticipantBindingRequest, PresentedIdentity, lookup_binding_required,
};
use liminal_protocol::wire::{
    ClientRequest, Generation, PARTICIPANT_FRAME_TYPE, ParticipantFrame, RecordAdmission,
    RecordAdmissionAttemptToken, ServerValue, encode as encode_participant,
    encoded_len as participant_encoded_len,
};

use super::{ConnectionRuntime, ConnectionSupervisor, ParticipantSemanticHandler};
use crate::ServerError;
use crate::config::types::{LimitsConfig, ServerConfig, ServicesConfig};
use crate::server::connection::services::{ConnectionServices, LiminalConnectionServices};
use crate::server::listener::ServerListener;
use crate::server::participant::dispatch_impact::DispatchImpactAccumulator;
use crate::server::participant::incarnation_stream::{
    ConnectionFateClass, IncarnationStartup, IncarnationStream,
};
use crate::server::participant::{
    ConnectionFateWorkItem, InstalledParticipantService, ParticipantConnectionContext,
    ParticipantConnectionConversations, ParticipantSemanticError, ParticipantSemanticOutcome,
    ParticipantServiceFatal,
};
use crate::server::shutdown::{ShutdownHandle, run_shutdown_sequence};

const FIXTURE_CONVERSATION_LIMIT: u64 = 1;

#[derive(Debug)]
struct FateHandler {
    fatal: Mutex<Option<ParticipantServiceFatal>>,
    fail_handler: bool,
}

impl FateHandler {
    fn failing() -> Self {
        Self {
            fatal: Mutex::new(None),
            fail_handler: true,
        }
    }

    fn completing() -> Self {
        Self {
            fatal: Mutex::new(None),
            fail_handler: false,
        }
    }
}

impl ParticipantSemanticHandler for FateHandler {
    fn service_fatal(&self) -> Result<Option<ParticipantServiceFatal>, ParticipantSemanticError> {
        self.fatal.lock().map(|fatal| fatal.clone()).map_err(|_| {
            ParticipantSemanticError::Internal {
                message: "fatal-composition fixture latch is poisoned".to_owned(),
            }
        })
    }

    fn latch_connection_fate_intent_incomplete(
        &self,
        open_sequence: u64,
        conversation_id: u64,
    ) -> Result<ParticipantServiceFatal, ParticipantSemanticError> {
        let fatal = self
            .fatal
            .lock()
            .map_err(|_| ParticipantSemanticError::Internal {
                message: "fatal-composition fixture latch is poisoned".to_owned(),
            })?
            .get_or_insert_with(|| ParticipantServiceFatal::ConnectionFateIntentIncomplete {
                open_sequence,
                conversation_id,
            })
            .clone();
        Ok(fatal)
    }

    fn handle_connection_fate(
        &self,
        work_item: ConnectionFateWorkItem,
    ) -> Result<(), ParticipantSemanticError> {
        if self.fail_handler {
            return Err(ParticipantSemanticError::Internal {
                message: format!(
                    "injected failure while completing Open {}",
                    work_item.open_sequence
                ),
            });
        }
        Ok(())
    }

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

    fn handle(
        &self,
        context: ParticipantConnectionContext,
        conversations: &mut ParticipantConnectionConversations,
        request: ClientRequest,
    ) -> Result<ServerValue, ParticipantSemanticError> {
        let _ = (context, conversations, request);
        Err(ParticipantSemanticError::Unavailable)
    }
}

#[derive(Debug)]
struct FailNthFlush {
    inner: Arc<dyn DurableStore>,
    flush_count: AtomicUsize,
    fail_at: usize,
}

impl FailNthFlush {
    fn new(inner: Arc<dyn DurableStore>, fail_at: usize) -> Self {
        Self {
            inner,
            flush_count: AtomicUsize::new(0),
            fail_at,
        }
    }
}

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

    async fn get_routed(
        &self,
        route_key: &str,
        key: &[u8],
    ) -> Result<Option<Vec<u8>>, liminal::durability::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> {
        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> {
        let previous = self
            .flush_count
            .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |count| {
                count.checked_add(1)
            })
            .map_err(|_| {
                DurabilityError::ConfigError("fixture flush counter exhausted".to_owned())
            })?;
        let flush = previous.checked_add(1).ok_or_else(|| {
            DurabilityError::ConfigError("fixture flush counter exhausted".to_owned())
        })?;
        if flush == self.fail_at {
            return Err(DurabilityError::ConfigError(format!(
                "injected incarnation flush failure {flush}"
            )));
        }
        self.inner.flush().await
    }
}

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

/// Rows this process's own incarnation epoch holds.
///
/// Since INCARNATION-STREAM-BOUND each server incarnation's canonical inputs
/// live in their own stream, so counting rows means naming the generation. This
/// fixture builds exactly one supervisor, whose incarnation is therefore 1.
const FIXTURE_SERVER_INCARNATION: u64 = 1;

fn stream_len(store: &Arc<dyn DurableStore>) -> Result<usize, Box<dyn Error>> {
    Ok(block_on(store.read_from(
        &IncarnationStream::epoch_stream_key(FIXTURE_SERVER_INCARNATION),
        0,
        LimitsConfig::default().max_connections,
    ))??
    .len())
}

#[test]
fn post_open_fatal_stops_new_opens_and_admission_and_activates_normal_shutdown()
-> Result<(), Box<dyn Error>> {
    const CONVERSATION_ID: u64 = 31;

    let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let handler = Arc::new(FateHandler::failing());
    let participant_service =
        InstalledParticipantService::new(handler, Arc::clone(&store), u64::MAX)
            .map_err(|error| format!("participant service configuration failed: {error:?}"))?;
    let services: Arc<dyn ConnectionServices> =
        Arc::new(LiminalConnectionServices::empty()?.with_participant_service(participant_service));
    let shutdown = ShutdownHandle::new();
    let supervisor = ConnectionSupervisor::with_fatal_shutdown(
        services,
        None,
        None,
        LimitsConfig::default(),
        shutdown.clone(),
    )?;
    let (client, server) = tcp_pair()?;
    let connection = supervisor.spawn_connection(server)?;
    let incarnation = connection
        .connection_incarnation()
        .ok_or("participant connection lacks a durable incarnation")?;
    let entries_before_open = stream_len(&store)?;

    let failed = supervisor.inner.runtime.complete_connection_fate(
        Some(incarnation),
        ConnectionFateClass::ConnectionLost,
        &[CONVERSATION_ID],
    );
    let failure = match failed {
        Ok(()) => return Err("injected post-Open handler failure unexpectedly completed".into()),
        Err(error) => error,
    };
    let ServerError::ParticipantServiceFatal { fatal } = failure else {
        return Err(format!("post-Open failure returned the wrong server error: {failure}").into());
    };
    let ParticipantServiceFatal::ConnectionFateIntentIncomplete {
        open_sequence,
        conversation_id,
    } = &fatal;
    assert_eq!(*conversation_id, CONVERSATION_ID);
    assert!(shutdown.is_initiated());
    let observed_fatal = supervisor.participant_service_fatal()?;
    assert_eq!(observed_fatal.as_ref(), Some(&fatal));

    let entries_after_open = stream_len(&store)?;
    assert_eq!(
        entries_after_open,
        entries_before_open
            .checked_add(1)
            .ok_or("incarnation-stream fixture length overflow")?,
        "the failed fold must leave exactly its durable Open unmatched"
    );

    let stopped_open = supervisor.inner.runtime.complete_connection_fate(
        Some(incarnation),
        ConnectionFateClass::ServerShutdown,
        &[CONVERSATION_ID],
    );
    assert!(matches!(
        stopped_open,
        Err(ServerError::ParticipantServiceFatal { fatal: observed }) if observed == fatal
    ));
    assert_eq!(
        stream_len(&store)?,
        entries_after_open,
        "a latched service must not append another Open"
    );

    let (refused_client, refused_server) = tcp_pair()?;
    let refused = supervisor.spawn_connection(refused_server);
    assert!(matches!(
        refused,
        Err(ServerError::ParticipantServiceFatal { fatal: observed }) if observed == fatal
    ));
    assert_eq!(supervisor.active_connection_count(), 1);
    assert_eq!(stream_len(&store)?, entries_after_open);
    assert!(
        *open_sequence > 0,
        "the fatal must name the durable Open sequence"
    );

    drop(refused_client);
    drop(client);
    supervisor.shutdown();
    Ok(())
}

#[test]
fn complete_flush_failure_latches_fatal_and_activates_normal_shutdown() -> Result<(), Box<dyn Error>>
{
    const CONVERSATION_ID: u64 = 37;
    /// Flush ordinal of the connection-fate Complete's fsync.
    ///
    /// Startup costs two flushes since INCARNATION-STREAM-BOUND — this
    /// process's epoch Startup row, then the anchor that comes to name it
    /// (`participant/incarnation_stream/epoch.rs`) — and the connection's
    /// Allocate and Open follow, so the Complete is the fifth.
    const COMPLETE_FLUSH_NUMBER: usize = 5;

    let inner: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let store: Arc<dyn DurableStore> =
        Arc::new(FailNthFlush::new(Arc::clone(&inner), COMPLETE_FLUSH_NUMBER));
    let handler = Arc::new(FateHandler::completing());
    let participant_service =
        InstalledParticipantService::new(handler, Arc::clone(&store), u64::MAX)
            .map_err(|error| format!("participant service configuration failed: {error:?}"))?;
    let services: Arc<dyn ConnectionServices> =
        Arc::new(LiminalConnectionServices::empty()?.with_participant_service(participant_service));
    let shutdown = ShutdownHandle::new();
    let supervisor = ConnectionSupervisor::with_fatal_shutdown(
        services,
        None,
        None,
        LimitsConfig::default(),
        shutdown.clone(),
    )?;
    let (client, server) = tcp_pair()?;
    let connection = supervisor.spawn_connection(server)?;
    let incarnation = connection
        .connection_incarnation()
        .ok_or("participant connection lacks a durable incarnation")?;
    let entries_before_open = stream_len(&inner)?;

    let failed = supervisor.inner.runtime.complete_connection_fate(
        Some(incarnation),
        ConnectionFateClass::ServerShutdown,
        &[CONVERSATION_ID],
    );
    let failure = match failed {
        Ok(()) => return Err("injected Complete flush failure unexpectedly completed".into()),
        Err(error) => error,
    };
    let ServerError::ParticipantServiceFatal { fatal } = failure else {
        return Err(format!("Complete failure returned the wrong server error: {failure}").into());
    };
    let ParticipantServiceFatal::ConnectionFateIntentIncomplete {
        open_sequence,
        conversation_id,
    } = &fatal;
    assert_eq!(*conversation_id, CONVERSATION_ID);
    assert!(*open_sequence > 0);
    assert!(shutdown.is_initiated());
    let observed_fatal = supervisor.participant_service_fatal()?;
    assert_eq!(observed_fatal.as_ref(), Some(&fatal));

    let entries_after_failure = stream_len(&inner)?;
    assert_eq!(
        entries_after_failure,
        entries_before_open
            .checked_add(2)
            .ok_or("incarnation-stream fixture length overflow")?,
        "Open and the ambiguously flushed Complete must both reach the wrapped store"
    );

    let (refused_client, refused_server) = tcp_pair()?;
    let refused = supervisor.spawn_connection(refused_server);
    assert!(matches!(
        refused,
        Err(ServerError::ParticipantServiceFatal { fatal: observed }) if observed == fatal
    ));
    assert_eq!(stream_len(&inner)?, entries_after_failure);

    drop(refused_client);
    drop(client);
    supervisor.shutdown();
    Ok(())
}

/// Conversation the lane fixture's parked fates wait on.
const LANE_CONVERSATION: u64 = 41;
/// A conversation whose requests drain nothing.
const QUIET_CONVERSATION: u64 = 43;

/// One conversation's candidate lane, reduced to the three answers the
/// connection runtime acts on: occupied (`Precedence`), drained by a request
/// or by the stop drain (the tell), and a fold that fails for any other reason.
#[derive(Debug, Default)]
struct LaneHandler {
    fatal: Mutex<Option<ParticipantServiceFatal>>,
    occupied: AtomicBool,
    fail_fold: AtomicBool,
    refuse_stop_drain: AtomicBool,
    folds: AtomicUsize,
    /// The runtime whose parked-fate index every fold probes.
    probed: Mutex<Option<Weak<ConnectionRuntime>>>,
    /// Per fold: whether the parked-fate index was free while it ran.
    index_free_during_fold: Mutex<Vec<bool>>,
}

impl LaneHandler {
    fn occupied() -> Self {
        Self {
            occupied: AtomicBool::new(true),
            ..Self::default()
        }
    }

    fn internal(message: &str) -> ParticipantSemanticError {
        ParticipantSemanticError::Internal {
            message: message.to_owned(),
        }
    }

    fn probe_index(&self) -> Result<(), ParticipantSemanticError> {
        let probed = self
            .probed
            .lock()
            .map_err(|_| Self::internal("lane fixture probe slot is poisoned"))?
            .as_ref()
            .and_then(Weak::upgrade);
        if let Some(runtime) = probed {
            let free = runtime.parked_connection_fates.try_lock().is_ok();
            self.index_free_during_fold
                .lock()
                .map_err(|_| Self::internal("lane fixture probe record is poisoned"))?
                .push(free);
        }
        Ok(())
    }

    fn unknown_participant(
        request: ClientRequest,
    ) -> Result<ServerValue, ParticipantSemanticError> {
        let ClientRequest::RecordAdmission(record) = request else {
            return Err(Self::internal("lane fixture expected record admission"));
        };
        let lookup = lookup_binding_required::<[u8; 1], [u8; 1], [u8; 1]>(
            PresentedIdentity::Absent,
            &BindingState::Detached,
            None,
            &ParticipantBindingRequest::RecordAdmission(record),
        );
        let BindingRequiredLookupResult::ParticipantUnknown(value) = lookup else {
            return Err(Self::internal(
                "lane fixture lookup did not select ParticipantUnknown",
            ));
        };
        Ok(ServerValue::ParticipantUnknown(value))
    }
}

impl ParticipantSemanticHandler for LaneHandler {
    fn service_fatal(&self) -> Result<Option<ParticipantServiceFatal>, ParticipantSemanticError> {
        self.fatal
            .lock()
            .map(|fatal| fatal.clone())
            .map_err(|_| Self::internal("lane fixture latch is poisoned"))
    }

    fn latch_connection_fate_intent_incomplete(
        &self,
        open_sequence: u64,
        conversation_id: u64,
    ) -> Result<ParticipantServiceFatal, ParticipantSemanticError> {
        Ok(self
            .fatal
            .lock()
            .map_err(|_| Self::internal("lane fixture latch is poisoned"))?
            .get_or_insert_with(|| ParticipantServiceFatal::ConnectionFateIntentIncomplete {
                open_sequence,
                conversation_id,
            })
            .clone())
    }

    fn handle_connection_fate(
        &self,
        work_item: ConnectionFateWorkItem,
    ) -> Result<(), ParticipantSemanticError> {
        self.folds.fetch_add(1, Ordering::SeqCst);
        self.probe_index()?;
        if self.occupied.load(Ordering::SeqCst) {
            return Err(ParticipantSemanticError::BindingTerminalAdmissionRefused {
                error: BindingTerminalAdmitError::Precedence,
            });
        }
        if self.fail_fold.load(Ordering::SeqCst) {
            return Err(Self::internal(&format!(
                "injected fold failure for Open {}",
                work_item.open_sequence
            )));
        }
        Ok(())
    }

    fn drain_candidate_lane_with_impact(
        &self,
        conversation_id: u64,
    ) -> ParticipantSemanticOutcome<()> {
        if self.refuse_stop_drain.load(Ordering::SeqCst) {
            return ParticipantSemanticOutcome::unchanged(Err(Self::internal(
                "injected stop-drain refusal",
            )));
        }
        let mut impact = DispatchImpactAccumulator::new();
        if conversation_id == LANE_CONVERSATION && self.occupied.swap(false, Ordering::SeqCst) {
            impact.record_candidate_lane_drained();
        }
        ParticipantSemanticOutcome::new(Ok(()), impact.finish(conversation_id))
    }

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

    /// A record admission into [`LANE_CONVERSATION`] drains the lane, as a
    /// publish's `DrainFirst` does; one anywhere else drains nothing.
    fn handle_with_impact(
        &self,
        context: ParticipantConnectionContext,
        conversations: &mut ParticipantConnectionConversations,
        request: ClientRequest,
    ) -> ParticipantSemanticOutcome<ServerValue> {
        let _ = (context, conversations);
        let ClientRequest::RecordAdmission(record) = &request else {
            return ParticipantSemanticOutcome::unchanged(Err(Self::internal(
                "lane fixture expected record admission",
            )));
        };
        let conversation_id = record.conversation_id;
        let mut impact = DispatchImpactAccumulator::new();
        if conversation_id == LANE_CONVERSATION && self.occupied.swap(false, Ordering::SeqCst) {
            impact.record_candidate_lane_drained();
        }
        ParticipantSemanticOutcome::new(
            Self::unknown_participant(request),
            impact.finish(conversation_id),
        )
    }

    fn handle(
        &self,
        context: ParticipantConnectionContext,
        conversations: &mut ParticipantConnectionConversations,
        request: ClientRequest,
    ) -> Result<ServerValue, ParticipantSemanticError> {
        self.handle_with_impact(context, conversations, request)
            .into_parts()
            .0
    }
}

struct LaneFixture {
    handler: Arc<LaneHandler>,
    store: Arc<dyn DurableStore>,
    supervisor: ConnectionSupervisor,
    /// The connection whose fate parks; kept open so its own teardown adds nothing.
    _parked_client: TcpStream,
    open_sequence: u64,
}

impl LaneFixture {
    /// Builds a supervisor over the lane handler and parks one connection
    /// fate on [`LANE_CONVERSATION`]'s occupied lane.
    fn parked() -> Result<Self, Box<dyn Error>> {
        crate::metrics::init();
        let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
        let handler = Arc::new(LaneHandler::occupied());
        let participant_service = InstalledParticipantService::new(
            Arc::clone(&handler) as Arc<dyn ParticipantSemanticHandler>,
            Arc::clone(&store),
            u64::MAX,
        )
        .map_err(|error| format!("participant service configuration failed: {error:?}"))?;
        let services: Arc<dyn ConnectionServices> = Arc::new(
            LiminalConnectionServices::empty()?.with_participant_service(participant_service),
        );
        let supervisor = ConnectionSupervisor::with_fatal_shutdown(
            services,
            None,
            None,
            LimitsConfig::default(),
            ShutdownHandle::new(),
        )?;
        *handler
            .probed
            .lock()
            .map_err(|_| "lane fixture probe slot is poisoned")? =
            Some(Arc::downgrade(&supervisor.inner.runtime));
        let (parked_client, server) = tcp_pair()?;
        let connection = supervisor.spawn_connection(server)?;
        let incarnation = connection
            .connection_incarnation()
            .ok_or("participant connection lacks a durable incarnation")?;
        supervisor.inner.runtime.complete_connection_fate(
            Some(incarnation),
            ConnectionFateClass::ConnectionLost,
            &[LANE_CONVERSATION],
        )?;
        let parked = supervisor
            .inner
            .runtime
            .parked_connection_fate_sequences()?;
        let [open_sequence] = parked.as_slice() else {
            return Err(format!("expected exactly one parked fate, found {parked:?}").into());
        };
        Ok(Self {
            open_sequence: *open_sequence,
            handler,
            store,
            supervisor,
            _parked_client: parked_client,
        })
    }

    fn parked_sequences(&self) -> Result<Vec<u64>, Box<dyn Error>> {
        Ok(self
            .supervisor
            .inner
            .runtime
            .parked_connection_fate_sequences()?)
    }

    /// A second, unrelated participant connection past its `Connect`.
    fn participant_client(&self) -> Result<(TcpStream, Vec<u8>), Box<dyn Error>> {
        let (mut client, server) = tcp_pair()?;
        client.set_read_timeout(Some(Duration::from_secs(5)))?;
        client.set_write_timeout(Some(Duration::from_secs(5)))?;
        self.supervisor.spawn_connection(server)?;
        client.write_all(&encode_frame(&Frame::Connect {
            flags: 0,
            min_version: ProtocolVersion::new(1, 0),
            max_version: ProtocolVersion::new(1, 0),
            auth_token: Vec::new(),
        })?)?;
        let mut inbound = Vec::new();
        let ack = read_frame(&mut client, &mut inbound)?;
        if !matches!(ack, Frame::ConnectAck { .. }) {
            return Err(format!("participant client was not acknowledged: {ack:?}").into());
        }
        Ok((client, inbound))
    }

    /// Whether the next boot over this store would have an Open to reconcile.
    fn next_boot_requires_recovery(&self) -> Result<bool, Box<dyn Error>> {
        let startup = block_on(
            IncarnationStream::new(
                Arc::clone(&self.store),
                LimitsConfig::default().max_connections,
            )
            .startup(),
        )??;
        Ok(matches!(startup, IncarnationStartup::RecoveryRequired(_)))
    }
}

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

fn encode_frame(frame: &Frame) -> Result<Vec<u8>, Box<dyn Error>> {
    let mut bytes = vec![0; generic_encoded_len(frame)?];
    let written = encode_generic(frame, &mut bytes)?;
    bytes.truncate(written);
    Ok(bytes)
}

fn encode_request(request: ClientRequest) -> Result<Vec<u8>, String> {
    let frame = ParticipantFrame::ClientRequest(request);
    let mut bytes = vec![0; participant_encoded_len(&frame).map_err(|error| format!("{error:?}"))?];
    let written = encode_participant(&frame, &mut bytes).map_err(|error| format!("{error:?}"))?;
    bytes.truncate(written);
    Ok(bytes)
}

fn read_frame(socket: &mut TcpStream, buffer: &mut Vec<u8>) -> Result<Frame, Box<dyn Error>> {
    loop {
        match decode_generic(buffer) {
            Ok((frame, consumed)) => {
                buffer.drain(..consumed);
                return Ok(frame);
            }
            Err(
                ProtocolError::IncompleteHeader { .. } | ProtocolError::TruncatedPayload { .. },
            ) => {
                let mut chunk = [0_u8; 512];
                let read = socket.read(&mut chunk)?;
                if read == 0 {
                    return Err("connection closed before a complete frame arrived".into());
                }
                buffer.extend_from_slice(chunk.get(..read).unwrap_or(&[]));
            }
            Err(error) => return Err(Box::new(error)),
        }
    }
}

/// Sends one record admission and requires its participant reply.
fn request_reply(
    client: &mut TcpStream,
    inbound: &mut Vec<u8>,
    request: ClientRequest,
) -> Result<(), Box<dyn Error>> {
    client.write_all(&encode_request(request)?)?;
    let reply = read_frame(client, inbound)?;
    if !matches!(
        reply,
        Frame::Unknown {
            type_id: PARTICIPANT_FRAME_TYPE,
            ..
        }
    ) {
        return Err(format!("the request was not answered with its reply: {reply:?}").into());
    }
    Ok(())
}

/// P2: a parked fate is retried by the drain that clears its lane and by
/// nothing else. An unrelated reply leaves it untouched, the draining request
/// completes it with no further traffic, and the gauge follows the index.
#[test]
fn a_parked_fate_is_completed_by_the_drain_that_clears_its_lane() -> Result<(), Box<dyn Error>> {
    let fixture = LaneFixture::parked()?;
    assert_eq!(crate::metrics::connection_fates_parked_value(), Some(1));
    let folds_parked = fixture.handler.folds.load(Ordering::SeqCst);
    let (mut client, mut inbound) = fixture.participant_client()?;
    let rows_parked = stream_len(&fixture.store)?;

    request_reply(
        &mut client,
        &mut inbound,
        record_admission(QUIET_CONVERSATION, 0x51),
    )?;
    assert_eq!(
        fixture.handler.folds.load(Ordering::SeqCst),
        folds_parked,
        "a reply that drained nothing retried the parked fate"
    );
    assert_eq!(fixture.parked_sequences()?, vec![fixture.open_sequence]);

    request_reply(
        &mut client,
        &mut inbound,
        record_admission(LANE_CONVERSATION, 0x52),
    )?;
    assert_eq!(
        fixture.handler.folds.load(Ordering::SeqCst),
        folds_parked + 1,
        "the drain must retry the parked fate exactly once"
    );
    assert!(fixture.parked_sequences()?.is_empty());
    assert_eq!(crate::metrics::connection_fates_parked_value(), Some(0));
    assert_eq!(
        stream_len(&fixture.store)?,
        rows_parked + 1,
        "the retried fate must append exactly its Complete"
    );

    drop(client);
    fixture.supervisor.shutdown();
    Ok(())
}

/// P3: no fold and no durable Complete runs under the parked-fate index lock,
/// neither the first fold that parks nor the retry a drain drives.
#[test]
fn a_parked_fate_fold_holds_no_index_lock() -> Result<(), Box<dyn Error>> {
    let fixture = LaneFixture::parked()?;
    let (mut client, mut inbound) = fixture.participant_client()?;
    request_reply(
        &mut client,
        &mut inbound,
        record_admission(LANE_CONVERSATION, 0x53),
    )?;
    assert!(fixture.parked_sequences()?.is_empty());

    let observed = fixture
        .handler
        .index_free_during_fold
        .lock()
        .map_err(|_| "lane fixture probe record is poisoned")?
        .clone();
    assert_eq!(observed.len(), 2, "one parking fold and one retry fold");
    assert!(
        observed.iter().all(|free| *free),
        "a fold ran while the parked-fate index was locked: {observed:?}"
    );

    drop(client);
    fixture.supervisor.shutdown();
    Ok(())
}

/// P4: a failed retry of another connection's parked fate belongs to that
/// fate. The request whose drain triggered it keeps its reply, while the
/// fate's own failure takes the existing fatal policy, named by its Open.
#[test]
fn a_failed_retry_does_not_cost_the_draining_request_its_reply() -> Result<(), Box<dyn Error>> {
    let fixture = LaneFixture::parked()?;
    fixture.handler.fail_fold.store(true, Ordering::SeqCst);
    let (mut client, mut inbound) = fixture.participant_client()?;

    request_reply(
        &mut client,
        &mut inbound,
        record_admission(LANE_CONVERSATION, 0x54),
    )?;

    let fatal = fixture
        .supervisor
        .participant_service_fatal()?
        .ok_or("the failed retry did not take the fatal policy")?;
    let ParticipantServiceFatal::ConnectionFateIntentIncomplete { open_sequence, .. } = fatal;
    assert_eq!(open_sequence, fixture.open_sequence);

    drop(client);
    fixture.supervisor.shutdown();
    Ok(())
}

fn shutdown_listener(supervisor: &ConnectionSupervisor) -> Result<ServerListener, Box<dyn Error>> {
    let config = ServerConfig {
        listen_address: "127.0.0.1:0".parse()?,
        health_listen_address: "127.0.0.1:0".parse()?,
        drain_timeout_ms: 0,
        channels: Vec::new(),
        routing_rules: Vec::new(),
        persistence_path: None,
        cluster: None,
        auth: None,
        services: ServicesConfig::default(),
        limits: LimitsConfig::default(),
        participant: None,
        websocket: None,
    };
    Ok(ServerListener::bind(&config, supervisor.clone())?)
}

/// P1: a clean stop drains the lane its parked fate waits on and completes the
/// fate before it reports success, so the next boot has no Open to reconcile.
#[test]
fn a_clean_stop_settles_a_parked_fate_before_reporting_clean() -> Result<(), Box<dyn Error>> {
    let fixture = LaneFixture::parked()?;
    let mut listener = shutdown_listener(&fixture.supervisor)?;

    run_shutdown_sequence(&mut listener, None, &fixture.supervisor, Duration::ZERO)?;

    assert!(fixture.parked_sequences()?.is_empty());
    assert_eq!(crate::metrics::connection_fates_parked_value(), Some(0));
    assert!(
        !fixture.next_boot_requires_recovery()?,
        "a clean stop left the next boot an Open to reconcile"
    );
    Ok(())
}

/// P1, the other pole: a stop that cannot drain the lane does not report
/// clean. It names the Open it left, and that Open is still there for the next
/// boot to reconcile.
#[test]
fn a_stop_that_cannot_settle_a_parked_fate_names_it() -> Result<(), Box<dyn Error>> {
    let fixture = LaneFixture::parked()?;
    fixture
        .handler
        .refuse_stop_drain
        .store(true, Ordering::SeqCst);
    let mut listener = shutdown_listener(&fixture.supervisor)?;

    let log = crate::test_log::CapturedLog::default();
    let stopped = log.capture(|| {
        run_shutdown_sequence(&mut listener, None, &fixture.supervisor, Duration::ZERO)
    });

    let Err(ServerError::ShutdownUnsettledConnectionFates { open_sequences }) = stopped else {
        return Err(format!("an unsettled stop was not reported by name: {stopped:?}").into());
    };
    assert_eq!(open_sequences, vec![fixture.open_sequence]);
    let output = log.contents();
    assert!(
        output.contains(&format!("open_sequence={}", fixture.open_sequence))
            && output.contains("left a parked connection fate unsettled"),
        "the unsettled fate was not named in the stop's log: {output}"
    );
    assert!(fixture.next_boot_requires_recovery()?);
    Ok(())
}