liminal-server 0.14.3

//! ONE PASS, NOT TWO: the refusal guarantee of the merged replay.
//!
//! `ConversationAuthority::replay` used to walk the whole log twice — a
//! schema-and-contiguity pass, then the apply pass re-reading every page.
//! The passes are merged; this file pins that nothing the two-pass shape
//! promised was given up in the merge:
//!
//! - a log whose schema audit fails at entry N is refused WHOLE, with the
//!   SAME typed error naming the SAME entry the standalone pass named;
//! - the node still boots, the refusal is attributed to the conversation by
//!   id and entry, and no owner is ever installed for it — nothing partial
//!   reaches the handler — while the neighbour on the same store is served;
//! - an apply failure EARLIER in the log does not outrank that audit
//!   failure, because the two-pass replay audited everything first;
//! - a log refused by its audit gets no durable write, even when the pass
//!   had already reached an extension repair before the refused entry;
//! - and the point of the merge: a cold replay reads each operation-log row
//!   EXACTLY ONCE, with a control proving the counter sees a second pass.

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

use liminal::durability::{DurabilityError, DurableStore, StoredEntry, bridge::block_on};
use liminal_protocol::wire::{
    ClientRequest, ConnectionIncarnation, EnrollmentRequest, EnrollmentToken, Generation,
    ParticipantAck, RecordAdmission, RecordAdmissionAttemptToken, ServerValue,
};

use crate::server::participant::ParticipantConnectionConversations;

use super::ProductionParticipantHandler;
use super::log::{
    FencedAttachProofRefusal, OperationLog, OperationLogError, READ_BATCH_SIZE, STREAM_PREFIX,
    StoredBindingEpoch, StoredOperation, StoredRecoveredFate, StoredRecoveredPresentation,
};
use super::ops_session_replay::validate_operation_schema;
use super::outbox::ConversationOutboxLimits;
use super::outbox_log::OutboxLog;
use super::outbox_replay::RestoreError;
use super::state::{ConversationAuthority, StateError};
use super::tests::{dispatch, dispatch_tracked, test_participant_config};
use super::tests_boot_containment::{
    HEALTHY_CONVERSATION, POISONED_CONVERSATION, seed_two_conversations,
};
use super::tests_w3_restore_fixture::{
    CONVERSATION as REPAIR_CONVERSATION, OutboxAppendFaultStore, append_payload, enrollment,
    extension_key, new_store, operation_key, seed_enrollment, stream_payloads,
};

const fn epoch(seed: u64) -> StoredBindingEpoch {
    StoredBindingEpoch {
        server_incarnation: seed,
        connection_ordinal: seed,
        capability_generation: seed,
    }
}

/// A `Recovered` reservation whose `died_source_sequence` is not behind it.
///
/// Decode and `validate_durable` both accept it; the composed-terminal audit
/// refuses it on sight, at exactly `at`, so it isolates the audit.
fn audit_refused_row(at: u64) -> StoredOperation {
    StoredOperation::Recovered {
        row: StoredRecoveredFate {
            participant_id: 0,
            last_dead_binding_epoch: epoch(1),
            died_source_sequence: at,
            fenced_attached_source_sequence: at,
            prior_binding_epoch: epoch(1),
            marker_delivery_seq: 1,
            resulting_floor: 1,
            presentation: StoredRecoveredPresentation::RecoveredOwnsAndReservesFinalizer,
        },
        event: at.to_be_bytes().to_vec(),
    }
}

/// Appends [`audit_refused_row`] at `conversation_id`'s current head and
/// returns that head.
fn append_audit_refused_row(
    store: &Arc<dyn DurableStore>,
    conversation_id: u64,
) -> Result<u64, Box<dyn Error>> {
    let head =
        u64::try_from(stream_payloads(store, &format!("{STREAM_PREFIX}{conversation_id}"))?.len())?;
    let log = OperationLog::new(Arc::clone(store), conversation_id);
    block_on(log.append(&audit_refused_row(head), head))??;
    Ok(head)
}

fn is_the_audit_refusal_at(error: &StateError, sequence: u64) -> bool {
    matches!(
        error,
        StateError::Log(OperationLogError::FencedAttachProof {
            sequence: refused,
            reason: FencedAttachProofRefusal::ComposedRecoveredReservationMismatch,
        }) if *refused == sequence
    )
}

/// The production single-pass replay, called directly.
fn one_pass(
    store: &Arc<dyn DurableStore>,
    conversation_id: u64,
) -> Result<Result<ConversationAuthority, RestoreError>, Box<dyn Error>> {
    let config = test_participant_config();
    let log = OperationLog::new(Arc::clone(store), conversation_id);
    let outbox_log = OutboxLog::new(Arc::clone(store), conversation_id);
    let limits =
        ConversationOutboxLimits::try_new(config.max_retained_record_rows, config.identity_slots)?;
    Ok(block_on(ConversationAuthority::replay(
        conversation_id,
        &log,
        &outbox_log,
        &config,
        limits,
    ))?)
}

/// The former standalone audit pass, kept as the two-pass reference.
fn two_pass_audit(
    store: &Arc<dyn DurableStore>,
    conversation_id: u64,
) -> Result<Result<(), StateError>, Box<dyn Error>> {
    let log = OperationLog::new(Arc::clone(store), conversation_id);
    Ok(block_on(validate_operation_schema(
        &log,
        test_participant_config().identity_slots,
    ))?)
}

/// The refusal a single-pass replay gave, or why it was not an audit refusal.
fn audit_refusal(
    store: &Arc<dyn DurableStore>,
    conversation_id: u64,
) -> Result<StateError, Box<dyn Error>> {
    match one_pass(store, conversation_id)? {
        Err(RestoreError::Semantic(refusal)) => Ok(refusal),
        Err(RestoreError::Extension(other)) => {
            Err(format!("the replay refused on its extension log instead: {other}").into())
        }
        Ok(_) => Err("the audit-refused log replayed".into()),
    }
}

fn enroll(
    handler: &ProductionParticipantHandler,
    conversation_id: u64,
    token: u8,
    connection: ConnectionIncarnation,
) -> Result<ServerValue, Box<dyn Error>> {
    dispatch_tracked(
        handler,
        connection,
        &mut ParticipantConnectionConversations::default(),
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id,
            enrollment_token: EnrollmentToken::new([token; 16]),
        }),
    )
}

/// A store whose second enrollment committed its operation row but lost its
/// Unit 2 extension row, so the next replay owes exactly one repair row.
struct LostProjection {
    store: Arc<dyn DurableStore>,
    /// The extension stream as the lost append left it.
    extension: Vec<Vec<u8>>,
}

/// Builds a [`LostProjection`] through the production seam.
fn lost_projection_store() -> Result<LostProjection, Box<dyn Error>> {
    let inner = new_store()?;
    let faults = Arc::new(OutboxAppendFaultStore::new(Arc::clone(&inner)));
    let store: Arc<dyn DurableStore> = faults.clone();
    let handler = seed_enrollment(&store)?;
    faults.set_fail(true);
    let refused = dispatch(
        &handler,
        ConnectionIncarnation::new(REPAIR_CONVERSATION, 2),
        enrollment(2),
    );
    faults.set_fail(false);
    if refused.is_ok() {
        return Err("the outbox append fault did not refuse the enrollment".into());
    }
    drop(handler);
    let extension = stream_payloads(&store, &extension_key())?;
    Ok(LostProjection { store, extension })
}

/// The one-pass replay refuses at entry N with the exact typed error, and the
/// exact text, the standalone two-pass audit gives for the same log.
#[test]
fn schema_failure_at_entry_n_is_refused_whole_with_the_two_pass_error() -> Result<(), Box<dyn Error>>
{
    let store = seed_two_conversations()?;
    let poisoned_at = append_audit_refused_row(&store, POISONED_CONVERSATION)?;
    assert!(poisoned_at > 0, "the seed wrote no rows to poison behind");

    let Err(two_pass) = two_pass_audit(&store, POISONED_CONVERSATION)? else {
        return Err("the standalone audit accepted the poisoned log".into());
    };
    assert!(
        is_the_audit_refusal_at(&two_pass, poisoned_at),
        "standalone audit refused elsewhere: {two_pass:?}"
    );

    let merged = audit_refusal(&store, POISONED_CONVERSATION)?;
    assert!(
        is_the_audit_refusal_at(&merged, poisoned_at),
        "one-pass replay refused elsewhere: {merged:?}"
    );
    assert_eq!(merged.to_string(), two_pass.to_string());
    Ok(())
}

/// Booting over the poisoned store installs no owner for the refused
/// conversation, attributes the refusal to it by id AND entry, and keeps
/// serving the neighbour.
#[test]
fn schema_failure_at_entry_n_registers_no_conversation_and_names_it() -> Result<(), Box<dyn Error>>
{
    let store = seed_two_conversations()?;
    let poisoned_at = append_audit_refused_row(&store, POISONED_CONVERSATION)?;

    let handler = ProductionParticipantHandler::new(Arc::clone(&store), test_participant_config())?;

    let unloadable = handler.unloadable_conversations();
    let reason = unloadable
        .get(&POISONED_CONVERSATION)
        .ok_or("the poisoned conversation was not recorded as unloadable")?;
    assert!(
        reason.contains(&format!("sequence {poisoned_at}")),
        "the refusal does not name entry {poisoned_at}: {reason}"
    );
    assert!(
        !unloadable.contains_key(&HEALTHY_CONVERSATION),
        "the healthy neighbour was refused: {unloadable:?}"
    );

    let cell = handler.cell(POISONED_CONVERSATION)?;
    let owner = cell
        .lock()
        .map_err(|_| "poisoned conversation owner lock is poisoned")?;
    let installed = owner.is_some();
    drop(owner);
    assert!(
        !installed,
        "a partial authority was installed for the refused conversation"
    );

    let refused = enroll(
        &handler,
        POISONED_CONVERSATION,
        0xD1,
        ConnectionIncarnation::new(801, 1),
    );
    let Err(refused) = refused else {
        return Err(format!("the refused conversation answered a request: {refused:?}").into());
    };
    let refused = refused.to_string();
    assert!(
        refused.contains(&POISONED_CONVERSATION.to_string()),
        "the request refusal does not name conversation {POISONED_CONVERSATION}: {refused}"
    );

    let served = enroll(
        &handler,
        HEALTHY_CONVERSATION,
        0xD2,
        ConnectionIncarnation::new(802, 1),
    )?;
    assert!(
        matches!(served, ServerValue::EnrollBound(_)),
        "the healthy neighbour stopped serving: {served:?}"
    );
    Ok(())
}

/// An entry M that audits clean but cannot be applied, followed by an entry
/// N the audit refuses: the refusal names N, as the audit-first replay did,
/// and matches the frozen two-pass reference byte for byte.
#[test]
fn an_earlier_apply_failure_does_not_outrank_a_later_audit_failure() -> Result<(), Box<dyn Error>> {
    let store = new_store()?;
    drop(seed_enrollment(&store)?);
    let base = stream_payloads(&store, &operation_key())?;
    let genesis = base
        .first()
        .cloned()
        .ok_or("seeded log has no genesis row")?;
    let apply_failure_at = u64::try_from(base.len())?;
    append_payload(&store, &operation_key(), genesis, apply_failure_at)?;

    // Control: a duplicated genesis audits clean and fails only when
    // applied, so M is an apply-phase failure and nothing else.
    assert!(
        matches!(two_pass_audit(&store, REPAIR_CONVERSATION)?, Ok(())),
        "the duplicated genesis row failed the audit, so it cannot stand for an apply failure"
    );
    let apply_only = match one_pass(&store, REPAIR_CONVERSATION)? {
        Ok(_) => return Err("a duplicated genesis row replayed".into()),
        Err(error) => error.to_string(),
    };

    let audit_failure_at = append_audit_refused_row(&store, REPAIR_CONVERSATION)?;
    assert!(audit_failure_at > apply_failure_at);
    let Err(two_pass) = two_pass_audit(&store, REPAIR_CONVERSATION)? else {
        return Err("the standalone audit accepted the poisoned log".into());
    };
    assert!(
        is_the_audit_refusal_at(&two_pass, audit_failure_at),
        "standalone audit refused elsewhere: {two_pass:?}"
    );
    let merged = audit_refusal(&store, REPAIR_CONVERSATION)?;
    assert!(
        is_the_audit_refusal_at(&merged, audit_failure_at),
        "the apply failure at {apply_failure_at} outranked the audit failure at \
         {audit_failure_at}: {merged:?}"
    );
    assert_eq!(merged.to_string(), two_pass.to_string());
    assert_ne!(
        merged.to_string(),
        apply_only,
        "the fixture cannot tell the two refusals apart"
    );

    let handler = ProductionParticipantHandler::new(Arc::clone(&store), test_participant_config())?;
    let log = OperationLog::new(Arc::clone(&store), REPAIR_CONVERSATION);
    let reference = handler
        .replay_aggregate_reference(REPAIR_CONVERSATION, &log)
        .err()
        .ok_or("the two-pass reference replayed the poisoned log")?;
    let production = handler
        .replay_and_repair(REPAIR_CONVERSATION, &log)
        .err()
        .ok_or("the production replay replayed the poisoned log")?;
    assert_eq!(production.to_string(), reference.to_string());
    Ok(())
}

/// The pass reaches an owed extension repair BEFORE the entry its audit
/// refuses: the repair stays staged and is never written, so the refused
/// conversation's store is byte-identical to what the audit-first replay
/// left.
#[test]
fn an_audit_refusal_after_a_staged_repair_writes_nothing() -> Result<(), Box<dyn Error>> {
    // Control: the same lost projection with nothing after it IS repaired,
    // so a repair really is owed in the arm below.
    let LostProjection {
        store: control,
        extension: control_before,
    } = lost_projection_store()?;
    assert_eq!(control_before.len(), 1);
    if let Err(refused) = one_pass(&control, REPAIR_CONVERSATION)? {
        return Err(format!("the unpoisoned control refused: {refused}").into());
    }
    assert_eq!(
        stream_payloads(&control, &extension_key())?.len(),
        2,
        "the control owed no repair, so the arm below proves nothing"
    );

    let LostProjection {
        store,
        extension: before,
    } = lost_projection_store()?;
    let poisoned_at = append_audit_refused_row(&store, REPAIR_CONVERSATION)?;
    let refusal = audit_refusal(&store, REPAIR_CONVERSATION)?;
    assert!(
        is_the_audit_refusal_at(&refusal, poisoned_at),
        "refused elsewhere: {refusal:?}"
    );
    assert_eq!(
        stream_payloads(&store, &extension_key())?,
        before,
        "a log refused by its audit had a repair row written"
    );

    let handler = ProductionParticipantHandler::new(Arc::clone(&store), test_participant_config())?;
    assert!(
        handler
            .unloadable_conversations()
            .contains_key(&REPAIR_CONVERSATION),
        "the boot did not refuse the poisoned conversation"
    );
    let log = OperationLog::new(Arc::clone(&store), REPAIR_CONVERSATION);
    let reference = handler
        .replay_aggregate_reference(REPAIR_CONVERSATION, &log)
        .err()
        .ok_or("the two-pass reference replayed the poisoned log")?;
    let production = handler
        .replay_and_repair(REPAIR_CONVERSATION, &log)
        .err()
        .ok_or("the production replay replayed the poisoned log")?;
    assert_eq!(production.to_string(), reference.to_string());
    assert_eq!(
        stream_payloads(&store, &extension_key())?,
        before,
        "a boot or a request over the refused log wrote a repair row"
    );
    Ok(())
}

/// The conversation the read-count pin writes; its own id, so no fixture
/// sharing a store can land rows in the stream it counts.
const READ_ONCE_CONVERSATION: u64 = 0x0E_1C_E0;

/// Counts the rows served from ONE conversation's operation-log stream, by
/// paged read and by point read alike.
#[derive(Debug)]
struct OperationRowCounter {
    inner: Arc<dyn DurableStore>,
    stream_key: String,
    rows: AtomicU64,
}

impl OperationRowCounter {
    fn new(inner: Arc<dyn DurableStore>, conversation_id: u64) -> Self {
        Self {
            inner,
            stream_key: format!("{STREAM_PREFIX}{conversation_id}"),
            rows: AtomicU64::new(0),
        }
    }

    fn take(&self) -> u64 {
        self.rows.swap(0, Ordering::SeqCst)
    }

    fn count(&self, stream_key: &str, rows: usize) -> Result<(), DurabilityError> {
        if stream_key == self.stream_key {
            let rows = u64::try_from(rows).map_err(|_| {
                DurabilityError::ConfigError(format!("{rows} rows exceed the row counter"))
            })?;
            self.rows.fetch_add(rows, Ordering::SeqCst);
        }
        Ok(())
    }
}

#[async_trait::async_trait]
impl DurableStore for OperationRowCounter {
    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 read_from(
        &self,
        stream_key: &str,
        offset: u64,
        limit: usize,
    ) -> Result<Vec<StoredEntry>, DurabilityError> {
        let entries = self.inner.read_from(stream_key, offset, limit).await?;
        self.count(stream_key, entries.len())?;
        Ok(entries)
    }

    async fn read_at(
        &self,
        stream_key: &str,
        sequence: u64,
    ) -> Result<Option<StoredEntry>, DurabilityError> {
        let entry = self.inner.read_at(stream_key, sequence).await?;
        self.count(stream_key, usize::from(entry.is_some()))?;
        Ok(entry)
    }

    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_value(&self, key: &str) -> Result<Option<u64>, DurabilityError> {
        self.inner.read_value(key).await
    }

    async fn scan(&self, prefix: &str) -> Result<Vec<StoredEntry>, DurabilityError> {
        self.inner.scan(prefix).await
    }

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

fn enrolled_participant(
    handler: &ProductionParticipantHandler,
    token: u8,
    connection: ConnectionIncarnation,
) -> Result<u64, Box<dyn Error>> {
    let value = enroll(handler, READ_ONCE_CONVERSATION, token, connection)?;
    let ServerValue::EnrollBound(receipt) = value else {
        return Err(format!("read-count enrollment {token:#x} did not bind: {value:?}").into());
    };
    Ok(receipt.participant_id())
}

fn acknowledge(
    handler: &ProductionParticipantHandler,
    connection: ConnectionIncarnation,
    participant_id: u64,
    through_seq: u64,
) -> Result<(), Box<dyn Error>> {
    let value = dispatch(
        handler,
        connection,
        ClientRequest::ParticipantAck(ParticipantAck {
            conversation_id: READ_ONCE_CONVERSATION,
            participant_id,
            capability_generation: Generation::ONE,
            through_seq,
        }),
    )?;
    if !matches!(value, ServerValue::AckCommitted(_)) {
        return Err(
            format!("read-count ack through {through_seq} did not commit: {value:?}").into(),
        );
    }
    Ok(())
}

/// Writes records and acks through the production seam until the operation
/// log spans more than two replay pages, so the pass crosses page boundaries.
fn seed_multi_page_log(store: &Arc<dyn DurableStore>) -> Result<u64, Box<dyn Error>> {
    let recipient_connection = ConnectionIncarnation::new(0xE1, 1);
    let sender_connection = ConnectionIncarnation::new(0xE1, 2);
    let handler = ProductionParticipantHandler::new(Arc::clone(store), test_participant_config())?;
    let recipient = enrolled_participant(&handler, 0xE1, recipient_connection)?;
    let sender = enrolled_participant(&handler, 0xE2, sender_connection)?;
    acknowledge(&handler, recipient_connection, recipient, 2)?;
    let target = u64::try_from(READ_BATCH_SIZE)?
        .checked_mul(2)
        .and_then(|rows| rows.checked_add(1))
        .ok_or("read-count target overflowed")?;
    let mut nonce = 0_u8;
    loop {
        let rows = u64::try_from(
            stream_payloads(store, &format!("{STREAM_PREFIX}{READ_ONCE_CONVERSATION}"))?.len(),
        )?;
        if rows > target {
            return Ok(rows);
        }
        let value = dispatch(
            &handler,
            sender_connection,
            ClientRequest::RecordAdmission(RecordAdmission {
                conversation_id: READ_ONCE_CONVERSATION,
                participant_id: sender,
                capability_generation: Generation::ONE,
                record_admission_attempt_token: RecordAdmissionAttemptToken::new([nonce; 16]),
                payload: vec![nonce],
            }),
        )?;
        let ServerValue::RecordCommitted(committed) = value else {
            return Err(format!("read-count record {nonce} did not commit: {value:?}").into());
        };
        acknowledge(
            &handler,
            recipient_connection,
            recipient,
            committed.delivery_seq(),
        )?;
        nonce = nonce.checked_add(1).ok_or("read-count nonce overflowed")?;
    }
}

/// The point of the merge: a cold replay of a log spanning several pages
/// reads each of its rows exactly once. The control runs the
/// removed audit pass in front of the same replay over the same counter and
/// sees exactly twice the rows, so the single count is a measurement, not a
/// counter that cannot see a second pass.
#[test]
fn a_cold_replay_reads_each_operation_row_exactly_once() -> Result<(), Box<dyn Error>> {
    let inner = new_store()?;
    let rows = seed_multi_page_log(&inner)?;
    let counter = Arc::new(OperationRowCounter::new(inner, READ_ONCE_CONVERSATION));
    let store: Arc<dyn DurableStore> = counter.clone();

    let replayed = match one_pass(&store, READ_ONCE_CONVERSATION)? {
        Ok(authority) => authority.next_log_sequence,
        Err(refused) => return Err(format!("the multi-page log refused: {refused}").into()),
    };
    assert_eq!(replayed, rows);
    let read_once = counter.take();
    assert_eq!(
        read_once, rows,
        "a cold replay of a {rows}-row log read {read_once} operation rows; every row must be \
         read exactly once"
    );

    if let Err(refused) = two_pass_audit(&store, READ_ONCE_CONVERSATION)? {
        return Err(format!("the control's audit refused: {refused}").into());
    }
    if let Err(refused) = one_pass(&store, READ_ONCE_CONVERSATION)? {
        return Err(format!("the control's replay refused: {refused}").into());
    }
    let read_twice = counter.take();
    assert_eq!(
        read_twice,
        rows.checked_mul(2).ok_or("control row count overflowed")?,
        "the counter did not see the control's second pass, so the single count above is not a \
         measurement"
    );
    Ok(())
}