liminal-server 0.24.9

use std::error::Error;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};

use liminal::durability::{
    DurabilityError, DurableStore, ScanCursor, ScanPage, StoredEntry, bridge::block_on,
    open_ephemeral,
};
use liminal_protocol::wire::{ClientRequest, EnrollmentRequest, EnrollmentToken, ServerValue};

use crate::config::types::ParticipantConfig;
use crate::server::participant::incarnation_stream::{
    ConnectionFateClass, ConnectionFateIntent, IncarnationAllocation, IncarnationStartup,
    IncarnationStream,
};
use crate::server::participant::{
    ParticipantConnectionConversations, ParticipantSemanticError, ParticipantSemanticHandler,
    ParticipantServiceFatal,
};

use super::ProductionParticipantHandler;
use super::log::{
    DecodedStoredOperation, OperationLog, STREAM_PREFIX, StoredDiedCause, StoredOperation,
};
use super::routed_stream_codec::row_stream;
use super::tests::{dispatch_tracked, test_participant_config};

const CONVERSATIONS: [u64; 4] = [101, 103, 107, 109];
const FAILURE_CONVERSATION: u64 = CONVERSATIONS[2];

#[derive(Debug)]
pub(super) struct FailOneStreamAppend {
    inner: Arc<dyn DurableStore>,
    target: Mutex<Option<String>>,
    fired: AtomicBool,
}

impl FailOneStreamAppend {
    pub(super) fn new(inner: Arc<dyn DurableStore>) -> Self {
        Self {
            inner,
            target: Mutex::new(None),
            fired: AtomicBool::new(false),
        }
    }

    pub(super) fn arm(&self, stream_key: String) -> Result<(), Box<dyn Error>> {
        let mut target = self
            .target
            .lock()
            .map_err(|_| "connection-fate append gate lock was poisoned")?;
        *target = Some(stream_key);
        drop(target);
        self.fired.store(false, Ordering::SeqCst);
        Ok(())
    }

    pub(super) fn fired(&self) -> bool {
        self.fired.load(Ordering::SeqCst)
    }
}

#[async_trait::async_trait]
impl DurableStore for FailOneStreamAppend {
    /// A batch holding a row of the armed stream is refused whole, as the
    /// native append of that stream was: nothing of it is durable.
    async fn apply_routed_batch(
        &self,
        route_key: &str,
        items: Vec<liminal::durability::RoutedBatchItem>,
    ) -> Result<(), liminal::durability::DurabilityError> {
        let armed = {
            let mut target = self.target.lock().map_err(|_| {
                DurabilityError::ConfigError(
                    "connection-fate append gate lock was poisoned".to_owned(),
                )
            })?;
            let held = target.as_deref().is_some_and(|stream| {
                items
                    .iter()
                    .any(|item| row_stream(&item.key) == Some(stream))
            });
            if held { target.take() } else { None }
        };
        if let Some(stream) = armed {
            self.fired.store(true, Ordering::SeqCst);
            return Err(DurabilityError::ConfigError(format!(
                "injected post-middle append failure for {stream}"
            )));
        }
        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> {
        let must_fail = {
            let mut target = self.target.lock().map_err(|_| {
                DurabilityError::ConfigError(
                    "connection-fate append gate lock was poisoned".to_owned(),
                )
            })?;
            if target.as_deref() == Some(stream_key) {
                target.take();
                true
            } else {
                false
            }
        };
        if must_fail {
            self.fired.store(true, Ordering::SeqCst);
            return Err(DurabilityError::ConfigError(format!(
                "injected post-middle append failure for {stream_key}"
            )));
        }
        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> {
        let must_fail = {
            let mut target = self.target.lock().map_err(|_| {
                DurabilityError::ConfigError(
                    "connection-fate append gate lock was poisoned".to_owned(),
                )
            })?;
            if target.as_deref() == Some(stream_key) {
                target.take();
                true
            } else {
                false
            }
        };
        if must_fail {
            self.fired.store(true, Ordering::SeqCst);
            return Err(DurabilityError::ConfigError(format!(
                "injected post-middle append failure for {stream_key}"
            )));
        }
        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> {
        self.inner.flush().await
    }
}

fn operation_rows(
    store: Arc<dyn DurableStore>,
    conversation_id: u64,
) -> Result<Vec<StoredOperation>, Box<dyn Error>> {
    let log = OperationLog::new(store, conversation_id);
    let mut rows = Vec::new();
    let mut sequence = 0_u64;
    while let Some(decoded) = block_on(log.read_at(sequence))?? {
        let DecodedStoredOperation::V3(operation) = decoded.operation else {
            return Err(format!(
                "conversation {conversation_id} sequence {sequence} was not schema v3"
            )
            .into());
        };
        rows.push(operation);
        sequence = sequence
            .checked_add(1)
            .ok_or("operation-row fixture sequence overflow")?;
    }
    Ok(rows)
}

/// The incarnation this fixture's single interrupted server process owns.
///
/// Since INCARNATION-STREAM-BOUND each incarnation's canonical inputs live in
/// their own epoch stream, and the recovery Completes below land in the epoch
/// that carries their Opens.
const FIXTURE_SERVER_INCARNATION: u64 = 1;

fn incarnation_rows(store: &Arc<dyn DurableStore>) -> Result<Vec<StoredEntry>, Box<dyn Error>> {
    let limit = usize::try_from(test_participant_config().max_retained_record_rows)?;
    Ok(block_on(store.read_from(
        &IncarnationStream::epoch_stream_key(FIXTURE_SERVER_INCARNATION),
        0,
        limit,
    ))??)
}

struct InterruptedFold {
    store: Arc<dyn DurableStore>,
    config: ParticipantConfig,
    reference_bound: usize,
    intent: ConnectionFateIntent,
}

fn enroll_conversations(
    handler: &ProductionParticipantHandler,
    connection_incarnation: liminal_protocol::wire::ConnectionIncarnation,
) -> Result<Vec<u64>, Box<dyn Error>> {
    let mut tracked = ParticipantConnectionConversations::default();
    for (index, conversation_id) in CONVERSATIONS.iter().copied().enumerate() {
        let token_byte = u8::try_from(
            index
                .checked_add(1)
                .ok_or("enrollment-token fixture index overflow")?,
        )?;
        let enrolled = dispatch_tracked(
            handler,
            connection_incarnation,
            &mut tracked,
            ClientRequest::Enrollment(EnrollmentRequest {
                conversation_id,
                enrollment_token: EnrollmentToken::new([token_byte; 16]),
            }),
        )?;
        if !matches!(enrolled, ServerValue::EnrollBound(_)) {
            return Err(format!(
                "conversation {conversation_id} did not establish a real Bound slot: {enrolled:?}"
            )
            .into());
        }
    }
    let tracked_conversations = tracked.tracked_conversations();
    assert_eq!(tracked_conversations, CONVERSATIONS);
    Ok(tracked_conversations)
}

fn assert_interrupted_rows(
    store: &Arc<dyn DurableStore>,
    open_sequence: u64,
) -> Result<Vec<StoredOperation>, Box<dyn Error>> {
    let first_rows = operation_rows(Arc::clone(store), CONVERSATIONS[0])?;
    let middle_rows = operation_rows(Arc::clone(store), CONVERSATIONS[1])?;
    let failure_rows = operation_rows(Arc::clone(store), CONVERSATIONS[2])?;
    let tail_rows = operation_rows(Arc::clone(store), CONVERSATIONS[3])?;
    assert_eq!(first_rows.len(), middle_rows.len());
    assert_eq!(
        first_rows.len(),
        failure_rows
            .len()
            .checked_add(1)
            .ok_or("post-middle row-count fixture overflow")?,
        "the first and middle source rows must flush before the injected failure"
    );
    assert_eq!(failure_rows.len(), tail_rows.len());
    for rows in [&first_rows, &middle_rows] {
        let Some(StoredOperation::Died { row }) = rows.last() else {
            return Err("completed pre-failure conversation lacks its terminal Died row".into());
        };
        assert_eq!(row.cause, StoredDiedCause::ConnectionLost);
        assert_eq!(row.connection_intent_sequence, Some(open_sequence));
    }
    Ok(failure_rows)
}

fn interrupt_after_middle() -> Result<InterruptedFold, Box<dyn Error>> {
    let inner: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let gated = Arc::new(FailOneStreamAppend::new(Arc::clone(&inner)));
    let store: Arc<dyn DurableStore> = gated.clone();
    let config = test_participant_config();
    let reference_bound = usize::try_from(config.max_semantic_conversations_per_connection)?;
    let startup = block_on(IncarnationStream::new(Arc::clone(&store), reference_bound).startup())??;
    let IncarnationStartup::Started(mut started) = startup else {
        return Err("fresh incarnation stream did not start".into());
    };
    let allocation = block_on(started.allocate(&[]))??;
    let IncarnationAllocation::Allocated {
        connection_incarnation,
        ..
    } = allocation
    else {
        return Err("fresh incarnation allocation was exhausted".into());
    };
    let handler = Arc::new(ProductionParticipantHandler::new(
        Arc::clone(&store),
        config,
    )?);
    let tracked_conversations = enroll_conversations(handler.as_ref(), connection_incarnation)?;
    let intent = block_on(started.open_connection_fate(
        connection_incarnation,
        ConnectionFateClass::ConnectionLost,
        reference_bound,
        &tracked_conversations,
    ))??;
    gated.arm(format!("{STREAM_PREFIX}{FAILURE_CONVERSATION}"))?;

    let failed = handler.handle_connection_fate(intent.work_item());
    assert!(matches!(
        failed,
        Err(ParticipantSemanticError::Internal { .. })
    ));
    assert!(
        gated.fired(),
        "the deterministic post-middle gate did not fire"
    );
    let fatal = handler
        .service_fatal()?
        .ok_or("post-Open failure did not latch participant service fatal")?;
    assert_eq!(
        fatal,
        ParticipantServiceFatal::ConnectionFateIntentIncomplete {
            open_sequence: intent.open_sequence,
            conversation_id: FAILURE_CONVERSATION,
        }
    );
    let failure_rows = assert_interrupted_rows(&store, intent.open_sequence)?;
    let incarnation_before_refusal = incarnation_rows(&store)?;
    let refused = handler.handle_connection_fate(intent.work_item());
    assert!(matches!(
        refused,
        Err(ParticipantSemanticError::ServiceFatal(observed)) if observed == fatal
    ));
    assert_eq!(incarnation_rows(&store)?, incarnation_before_refusal);
    assert_eq!(
        operation_rows(Arc::clone(&store), FAILURE_CONVERSATION)?,
        failure_rows
    );

    let durable_before_drop: Vec<_> = CONVERSATIONS
        .iter()
        .copied()
        .map(|conversation_id| operation_rows(Arc::clone(&store), conversation_id))
        .collect::<Result<_, _>>()?;
    drop(handler);
    drop(started);
    let durable_after_drop: Vec<_> = CONVERSATIONS
        .iter()
        .copied()
        .map(|conversation_id| operation_rows(Arc::clone(&store), conversation_id))
        .collect::<Result<_, _>>()?;
    assert_eq!(durable_after_drop, durable_before_drop);
    Ok(InterruptedFold {
        store,
        config,
        reference_bound,
        intent,
    })
}

fn recover_interrupted_tail(interrupted: &InterruptedFold) -> Result<(), Box<dyn Error>> {
    let cold =
        ProductionParticipantHandler::new(Arc::clone(&interrupted.store), interrupted.config)?;
    let startup = block_on(
        IncarnationStream::new(Arc::clone(&interrupted.store), interrupted.reference_bound)
            .startup(),
    )??;
    let IncarnationStartup::RecoveryRequired(mut recovery) = startup else {
        return Err("unmatched post-Open failure did not require startup recovery".into());
    };
    let intents = recovery.intents();
    assert_eq!(
        intents.as_slice(),
        core::slice::from_ref(&interrupted.intent)
    );
    cold.handle_connection_fate(intents[0].work_item())?;
    for conversation_id in CONVERSATIONS {
        // DUR-024 (5b37bafa1): a conversation not yet restored holds its fate
        // and applies it first in its own restore unit, so the tail's Died
        // rows are written when each conversation restores.
        cold.cell(conversation_id)?;
        let rows = operation_rows(Arc::clone(&interrupted.store), conversation_id)?;
        let Some(StoredOperation::Died { row }) = rows.last() else {
            return Err(format!(
                "startup tail completion omitted conversation {conversation_id} terminal row; its last row is {:?}",
                rows.last()
            )
            .into());
        };
        assert_eq!(row.cause, StoredDiedCause::ConnectionLost);
        assert_eq!(
            row.connection_intent_sequence,
            Some(interrupted.intent.open_sequence)
        );
    }
    let before_complete = incarnation_rows(&interrupted.store)?;
    block_on(recovery.complete(interrupted.intent.open_sequence))??;
    let after_complete = incarnation_rows(&interrupted.store)?;
    assert_eq!(
        after_complete.len(),
        before_complete
            .len()
            .checked_add(1)
            .ok_or("incarnation-row fixture count overflow")?,
        "Complete must append only after every tail binding source is observable"
    );
    let resumed = block_on(recovery.finish_startup())??;
    assert!(matches!(resumed, IncarnationStartup::Started(_)));
    assert_eq!(cold.service_fatal()?, None);
    Ok(())
}

pub(super) fn run_post_middle_failure_recovery() -> Result<(), Box<dyn Error>> {
    let interrupted = interrupt_after_middle()?;
    recover_interrupted_tail(&interrupted)
}

#[test]
fn connection_fate_intent_failure_on_middle_conversation_completes_every_tail_binding()
-> Result<(), Box<dyn Error>> {
    run_post_middle_failure_recovery()
}

#[test]
fn post_open_middle_failure_latches_service_fatal_and_startup_completes_tail()
-> Result<(), Box<dyn Error>> {
    run_post_middle_failure_recovery()
}