liminal-server 0.19.1

//! Terminal sources (Died, connection-close and explicit Detached, Left, and
//! the candidate-lane drains) write their own Unit 2 extension row in place,
//! right after their base row, and no commit rebuilds the conversation from
//! its log: not a connection drop, not a leave, and not the unclean-restart
//! repair a start runs over every conversation.

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

use liminal::durability::{
    DurabilityError, DurableStore, ScanCursor, ScanPage, StoredEntry, open_ephemeral,
};
use liminal_protocol::wire::{
    AttachAttemptToken, ClientRequest, ConnectionIncarnation, CredentialAttachRequest,
    EnrollmentRequest, EnrollmentToken, Generation, LeaveAttemptToken, LeaveRequest,
    RecordAdmission, RecordAdmissionAttemptToken, ServerValue,
};

use crate::config::types::ParticipantConfig;
use crate::server::participant::{
    ConnectionFateClass, ConnectionFateWorkItem, ParticipantConnectionConversations,
    ParticipantSemanticHandler,
};

use super::ProductionParticipantHandler;
use super::log::{GENERATION_STREAM_PREFIX, STREAM_PREFIX, StoredOperation};
use super::outbox_log::{OUTBOX_STREAM_PREFIX, OutboxRow};
use super::tests::{dispatch_tracked, test_participant_config};
use super::tests_p0_64_live_extension::{base_rows, extension_rows, owed_rows_match_written_rows};

// Leave, Died and connection-close Detached measured against history length.
#[path = "tests_lifecycle_history_cost.rs"]
mod lifecycle_history_cost;

/// A store that counts every row read from any conversation's operation log
/// (base rows) and Unit 2 extension stream, through every read the store
/// offers (`read_from`, `read_at`, `scan`), plus how many reads of an
/// operation log start at its first row, the start of every rebuild from the
/// log. It can also refuse the next extension append, or let the next one land
/// and fail the flush after it.
#[derive(Debug)]
struct ReplayCountingStore {
    inner: Arc<dyn DurableStore>,
    log_reads_from_start: AtomicU64,
    base_rows_read: AtomicU64,
    extension_rows_read: AtomicU64,
    all_rows_read: AtomicU64,
    refuse_next_extension_append: AtomicBool,
    fail_flush_after_next_extension_append: AtomicBool,
    flush_failure_armed: AtomicBool,
}

impl ReplayCountingStore {
    fn new() -> Result<Arc<Self>, Box<dyn Error>> {
        Ok(Arc::new(Self {
            inner: Arc::new(open_ephemeral(1)?),
            log_reads_from_start: AtomicU64::new(0),
            base_rows_read: AtomicU64::new(0),
            extension_rows_read: AtomicU64::new(0),
            all_rows_read: AtomicU64::new(0),
            refuse_next_extension_append: AtomicBool::new(false),
            fail_flush_after_next_extension_append: AtomicBool::new(false),
            flush_failure_armed: AtomicBool::new(false),
        }))
    }

    fn replays(&self) -> u64 {
        self.log_reads_from_start.load(Ordering::SeqCst)
    }

    /// Base rows and extension rows read since the last reset.
    fn rows_read(&self) -> (u64, u64) {
        (
            self.base_rows_read.load(Ordering::SeqCst),
            self.extension_rows_read.load(Ordering::SeqCst),
        )
    }

    /// Rows read from any stream at all since the last reset.
    fn all_rows_read(&self) -> u64 {
        self.all_rows_read.load(Ordering::SeqCst)
    }

    fn reset(&self) {
        self.log_reads_from_start.store(0, Ordering::SeqCst);
        self.base_rows_read.store(0, Ordering::SeqCst);
        self.extension_rows_read.store(0, Ordering::SeqCst);
        self.all_rows_read.store(0, Ordering::SeqCst);
    }

    fn count(&self, key: &str, rows: usize) {
        let rows = u64::try_from(rows).unwrap_or(u64::MAX);
        self.all_rows_read.fetch_add(rows, Ordering::SeqCst);
        if covers_operation_log(key) {
            self.base_rows_read.fetch_add(rows, Ordering::SeqCst);
        }
        if covers(key, OUTBOX_STREAM_PREFIX) {
            self.extension_rows_read.fetch_add(rows, Ordering::SeqCst);
        }
    }
}

fn is_operation_log(stream_key: &str) -> bool {
    stream_key.starts_with(STREAM_PREFIX) || stream_key.starts_with(GENERATION_STREAM_PREFIX)
}

/// Whether a stream key, or a scan prefix, reaches streams under `prefix`.
fn covers(key: &str, prefix: &str) -> bool {
    key.starts_with(prefix) || prefix.starts_with(key)
}

fn covers_operation_log(key: &str) -> bool {
    covers(key, STREAM_PREFIX) || covers(key, GENERATION_STREAM_PREFIX)
}

#[async_trait::async_trait]
impl DurableStore for ReplayCountingStore {
    async fn append(
        &self,
        stream_key: &str,
        payload: Vec<u8>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        let extension = stream_key.starts_with(OUTBOX_STREAM_PREFIX);
        if extension
            && self
                .refuse_next_extension_append
                .swap(false, Ordering::SeqCst)
        {
            return Err(DurabilityError::ConfigError(
                "the test refused this extension append".to_owned(),
            ));
        }
        let assigned = self.inner.append(stream_key, payload, expected_seq).await?;
        if extension
            && self
                .fail_flush_after_next_extension_append
                .swap(false, Ordering::SeqCst)
        {
            self.flush_failure_armed.store(true, Ordering::SeqCst);
        }
        Ok(assigned)
    }

    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> {
        if offset == 0 && is_operation_log(stream_key) {
            self.log_reads_from_start.fetch_add(1, Ordering::SeqCst);
        }
        let rows = self.inner.read_from(stream_key, offset, limit).await?;
        self.count(stream_key, rows.len());
        Ok(rows)
    }

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

    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> {
        let page = self.inner.scan_page(prefix, cursor, limit).await?;
        self.count(prefix, page.entries.len());
        Ok(page)
    }

    async fn flush(&self) -> Result<(), DurabilityError> {
        if self.flush_failure_armed.swap(false, Ordering::SeqCst) {
            return Err(DurabilityError::ConfigError(
                "the test failed the flush after an extension append that landed".to_owned(),
            ));
        }
        self.inner.flush().await
    }
}

struct Member {
    connection: ConnectionIncarnation,
    conversations: ParticipantConnectionConversations,
    participant_id: u64,
    generation: Generation,
    attach_secret: liminal_protocol::wire::AttachSecret,
}

fn enroll_and_attach(
    handler: &ProductionParticipantHandler,
    conversation_id: u64,
    connection: ConnectionIncarnation,
    token: u8,
) -> Result<Member, Box<dyn Error>> {
    let mut conversations = ParticipantConnectionConversations::default();
    let enrolled = dispatch_tracked(
        handler,
        connection,
        &mut conversations,
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id,
            enrollment_token: EnrollmentToken::new([token; 16]),
        }),
    )?;
    let ServerValue::EnrollBound(enrolled) = enrolled else {
        return Err(format!("enrollment {token:#x} did not bind: {enrolled:?}").into());
    };
    let attached = dispatch_tracked(
        handler,
        connection,
        &mut conversations,
        ClientRequest::CredentialAttach(CredentialAttachRequest {
            conversation_id,
            participant_id: enrolled.participant_id(),
            capability_generation: Generation::ONE,
            attach_secret: enrolled.attach_secret(),
            attach_attempt_token: AttachAttemptToken::new([token.wrapping_add(1); 16]),
            accept_marker_delivery_seq: None,
        }),
    )?;
    let ServerValue::AttachBound(attached) = attached else {
        return Err(format!("attach {token:#x} did not bind: {attached:?}").into());
    };
    Ok(Member {
        connection,
        conversations,
        participant_id: enrolled.participant_id(),
        generation: attached.origin_binding_epoch().capability_generation,
        attach_secret: attached.attach_secret(),
    })
}

fn admit(
    handler: &ProductionParticipantHandler,
    conversation_id: u64,
    member: &mut Member,
    token: u8,
) -> Result<ServerValue, Box<dyn Error>> {
    dispatch_tracked(
        handler,
        member.connection,
        &mut member.conversations,
        ClientRequest::RecordAdmission(RecordAdmission {
            conversation_id,
            participant_id: member.participant_id,
            capability_generation: member.generation,
            record_admission_attempt_token: RecordAdmissionAttemptToken::new([token; 16]),
            payload: vec![token],
        }),
    )
}

fn drop_connection(
    handler: &ProductionParticipantHandler,
    member: &Member,
    class: ConnectionFateClass,
    open_sequence: u64,
) -> Result<(), Box<dyn Error>> {
    handler
        .apply_connection_fate_with_impacts(&ConnectionFateWorkItem {
            open_sequence,
            connection_incarnation: member.connection,
            class,
            tracked_conversations: member.conversations.tracked_conversations(),
        })
        .into_result()?;
    Ok(())
}

/// Every owing base row has exactly its one extension row, and a cold load of
/// the same store neither writes nor changes any extension row.
fn assert_complete_and_unrepaired(
    store: &Arc<ReplayCountingStore>,
    handler: ProductionParticipantHandler,
    config: ParticipantConfig,
    conversation_id: u64,
    label: &str,
) -> Result<ProductionParticipantHandler, Box<dyn Error>> {
    let base = base_rows(&handler, conversation_id)?;
    let live_extension = extension_rows(&handler, conversation_id)?;
    owed_rows_match_written_rows(&base, &live_extension, label);
    drop(handler);
    let reopened =
        ProductionParticipantHandler::new(Arc::clone(store) as Arc<dyn DurableStore>, config)?;
    assert_eq!(
        extension_rows(&reopened, conversation_id)?,
        live_extension,
        "after {label} a cold load changed the extension stream, so the live commit left a row \
         for it to repair"
    );
    Ok(reopened)
}

fn base_kinds(
    handler: &ProductionParticipantHandler,
    conversation_id: u64,
) -> Result<Vec<&'static str>, Box<dyn Error>> {
    Ok(base_rows(handler, conversation_id)?
        .iter()
        .map(|(_, operation)| match operation {
            StoredOperation::Died { .. } => "died",
            StoredOperation::Detached { .. } => "detached",
            StoredOperation::Left { .. } => "left",
            StoredOperation::RecordAdmission { .. } => "record",
            StoredOperation::Ordinary { .. } => "ordinary",
            _ => "other",
        })
        .collect())
}

#[test]
fn connection_loss_and_close_commit_their_rows_in_place_with_no_replay()
-> Result<(), Box<dyn Error>> {
    const CONVERSATION: u64 = 0x7E1;
    let store = ReplayCountingStore::new()?;
    let config = test_participant_config();
    let handler =
        ProductionParticipantHandler::new(Arc::clone(&store) as Arc<dyn DurableStore>, config)?;
    let lost = enroll_and_attach(
        &handler,
        CONVERSATION,
        ConnectionIncarnation::new(71, 1),
        0x11,
    )?;
    let closed = enroll_and_attach(
        &handler,
        CONVERSATION,
        ConnectionIncarnation::new(71, 2),
        0x21,
    )?;
    let _peer = enroll_and_attach(
        &handler,
        CONVERSATION,
        ConnectionIncarnation::new(71, 3),
        0x31,
    )?;

    store.reset();
    drop_connection(&handler, &lost, ConnectionFateClass::ConnectionLost, 5)?;
    drop_connection(&handler, &closed, ConnectionFateClass::CleanDisconnect, 6)?;
    assert_eq!(
        store.replays(),
        0,
        "a connection drop rebuilt the conversation from its log"
    );
    assert_eq!(
        store.rows_read(),
        (0, 0),
        "a connection drop read base or extension rows"
    );
    let kinds = base_kinds(&handler, CONVERSATION)?;
    assert!(
        kinds.contains(&"died") && kinds.contains(&"detached"),
        "{kinds:?}"
    );
    assert_complete_and_unrepaired(&store, handler, config, CONVERSATION, "connection drops")?;
    Ok(())
}

#[test]
fn leave_commits_its_row_in_place_with_no_replay() -> Result<(), Box<dyn Error>> {
    const CONVERSATION: u64 = 0x7E2;
    let store = ReplayCountingStore::new()?;
    let config = test_participant_config();
    let handler =
        ProductionParticipantHandler::new(Arc::clone(&store) as Arc<dyn DurableStore>, config)?;
    let mut leaver = enroll_and_attach(
        &handler,
        CONVERSATION,
        ConnectionIncarnation::new(72, 1),
        0x41,
    )?;
    let _peer = enroll_and_attach(
        &handler,
        CONVERSATION,
        ConnectionIncarnation::new(72, 2),
        0x51,
    )?;

    store.reset();
    let left = dispatch_tracked(
        &handler,
        leaver.connection,
        &mut leaver.conversations,
        ClientRequest::Leave(LeaveRequest {
            conversation_id: CONVERSATION,
            participant_id: leaver.participant_id,
            capability_generation: leaver.generation,
            attach_secret: leaver.attach_secret,
            leave_attempt_token: LeaveAttemptToken::new([0x42; 16]),
        }),
    )?;
    assert!(matches!(left, ServerValue::LeaveCommitted(_)), "{left:?}");
    assert_eq!(
        store.replays(),
        0,
        "a leave rebuilt the conversation from its log"
    );
    assert_eq!(
        store.rows_read(),
        (0, 0),
        "a leave read base or extension rows"
    );
    assert_complete_and_unrepaired(&store, handler, config, CONVERSATION, "a leave")?;
    Ok(())
}

/// Retention at its cap with a peer that never acks, so a lost connection's
/// terminal pends; the peer's next record then drains that terminal and
/// commits its own record in ONE operation — two owing base rows — which is
/// exactly the shape whose rows must reach the extension stream in base order.
#[test]
fn a_drained_terminal_and_the_record_that_drained_it_complete_in_base_order()
-> Result<(), Box<dyn Error>> {
    const CONVERSATION: u64 = 0x7E3;
    let mut config = test_participant_config();
    config.max_retained_record_rows = 4;
    let store = ReplayCountingStore::new()?;
    let handler =
        ProductionParticipantHandler::new(Arc::clone(&store) as Arc<dyn DurableStore>, config)?;
    let mut victim = enroll_and_attach(
        &handler,
        CONVERSATION,
        ConnectionIncarnation::new(73, 1),
        0x61,
    )?;
    let mut peer_conversations = ParticipantConnectionConversations::default();
    let peer = dispatch_tracked(
        &handler,
        ConnectionIncarnation::new(73, 2),
        &mut peer_conversations,
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id: CONVERSATION,
            enrollment_token: EnrollmentToken::new([0x71; 16]),
        }),
    )?;
    let ServerValue::EnrollBound(peer) = peer else {
        return Err(format!("peer enrollment did not bind: {peer:?}").into());
    };
    let committed = admit(&handler, CONVERSATION, &mut victim, 0x62)?;
    assert!(
        matches!(committed, ServerValue::RecordCommitted(_)),
        "{committed:?}"
    );

    store.reset();
    drop_connection(&handler, &victim, ConnectionFateClass::ConnectionLost, 9)?;
    assert_eq!(
        store.replays(),
        0,
        "the drop rebuilt the conversation from its log"
    );
    let mut peer_member = Member {
        connection: ConnectionIncarnation::new(73, 2),
        conversations: peer_conversations,
        participant_id: peer.participant_id(),
        generation: Generation::ONE,
        attach_secret: peer.attach_secret(),
    };
    let before = base_rows(&handler, CONVERSATION)?.len();
    store.reset();
    let drained = admit(&handler, CONVERSATION, &mut peer_member, 0x72)?;
    assert!(
        matches!(drained, ServerValue::RecordCommitted(_)),
        "{drained:?}"
    );
    assert_eq!(
        store.replays(),
        0,
        "the drop or the draining record rebuilt the conversation from its log"
    );

    let base = base_rows(&handler, CONVERSATION)?;
    let appended: Vec<_> = base
        .iter()
        .skip(before)
        .map(|(sequence, operation)| (*sequence, operation.clone()))
        .collect();
    assert!(
        matches!(
            appended.as_slice(),
            [
                (_, StoredOperation::Died { .. }),
                ..,
                (_, StoredOperation::RecordAdmission { .. })
            ]
        ),
        "the peer's record did not drain the pending terminal first: {:?}",
        base_kinds(&handler, CONVERSATION)?
    );
    let extension = extension_rows(&handler, CONVERSATION)?;
    let sources: Vec<u64> = extension
        .iter()
        .filter_map(|(_, row)| match row {
            OutboxRow::Produced(batch) => Some(batch.source_log_sequence()),
            OutboxRow::AckAdvanced {
                source_log_sequence,
                ..
            } => Some(*source_log_sequence),
            OutboxRow::MarkerAckCommitted(_) => None,
        })
        .collect();
    let mut ascending = sources.clone();
    ascending.sort_unstable();
    assert_eq!(
        sources, ascending,
        "extension rows are out of base-row order"
    );
    assert_complete_and_unrepaired(
        &store,
        handler,
        config,
        CONVERSATION,
        "a drain and its record",
    )?;
    Ok(())
}

/// The crash point between a base row and its extension row: the commit is
/// refused by name, and the next request on the conversation writes the
/// missing row before anything of its own.
///
/// A Leave carries it here. A connection fate that fails part way latches the
/// service fatal instead, by its own design: its durable Open is its retry.
#[test]
fn an_unwritten_extension_row_is_refused_by_name_and_written_by_the_next_request()
-> Result<(), Box<dyn Error>> {
    const CONVERSATION: u64 = 0x7E4;
    let store = ReplayCountingStore::new()?;
    let config = test_participant_config();
    let handler =
        ProductionParticipantHandler::new(Arc::clone(&store) as Arc<dyn DurableStore>, config)?;
    let mut leaver = enroll_and_attach(
        &handler,
        CONVERSATION,
        ConnectionIncarnation::new(74, 1),
        0x81,
    )?;
    let mut peer = enroll_and_attach(
        &handler,
        CONVERSATION,
        ConnectionIncarnation::new(74, 2),
        0x91,
    )?;

    store
        .refuse_next_extension_append
        .store(true, Ordering::SeqCst);
    let refused = dispatch_tracked(
        &handler,
        leaver.connection,
        &mut leaver.conversations,
        ClientRequest::Leave(LeaveRequest {
            conversation_id: CONVERSATION,
            participant_id: leaver.participant_id,
            capability_generation: leaver.generation,
            attach_secret: leaver.attach_secret,
            leave_attempt_token: LeaveAttemptToken::new([0x82; 16]),
        }),
    );
    let text = match refused {
        Ok(value) => {
            return Err(
                format!("the leave answered although its row was refused: {value:?}").into(),
            );
        }
        Err(error) => error.to_string(),
    };
    assert!(
        text.contains("is not confirmed durable")
            && text.contains("the next request on it reloads"),
        "the refusal does not name the missing row and the act that writes it: {text}"
    );
    let base = base_rows(&handler, CONVERSATION)?;
    let left_sequence = base
        .iter()
        .rev()
        .find_map(|(sequence, operation)| {
            matches!(operation, StoredOperation::Left { .. }).then_some(*sequence)
        })
        .ok_or("the Left base row was not durable")?;
    let before = extension_rows(&handler, CONVERSATION)?;
    assert!(
        !before.iter().any(|(_, row)| matches!(
            row,
            OutboxRow::Produced(batch) if batch.source_log_sequence() == left_sequence
        )),
        "the refused extension row was written anyway"
    );

    let next = admit(&handler, CONVERSATION, &mut peer, 0x92)?;
    assert!(matches!(next, ServerValue::RecordCommitted(_)), "{next:?}");
    let after = extension_rows(&handler, CONVERSATION)?;
    let left_row = after
        .iter()
        .position(|(_, row)| {
            matches!(
                row,
                OutboxRow::Produced(batch) if batch.source_log_sequence() == left_sequence
            )
        })
        .ok_or("the next request did not write the missing extension row")?;
    let record_row = after
        .iter()
        .position(|(_, row)| {
            matches!(
                row,
                OutboxRow::Produced(batch) if batch.source_log_sequence() > left_sequence
            )
        })
        .ok_or("the next request's own extension row is missing")?;
    assert!(
        left_row < record_row,
        "the missing row was written after the next request's own row"
    );
    assert_complete_and_unrepaired(&store, handler, config, CONVERSATION, "the recovered leave")?;
    Ok(())
}

/// A start's unclean-restart repair walks every conversation and closes each
/// binding the previous server process left Bound. It reads no operation-log
/// row, however long the conversation's history: its work is the bindings it
/// closes, never the rows behind them.
#[test]
fn restart_repair_reads_no_log_rows_whatever_the_history_length() -> Result<(), Box<dyn Error>> {
    const CONVERSATION: u64 = 0x7E5;
    const PRIOR_SERVER: u64 = 75;
    let mut measured = Vec::new();
    for records in [2_u16, 96] {
        let store = ReplayCountingStore::new()?;
        let config = test_participant_config();
        let handler =
            ProductionParticipantHandler::new(Arc::clone(&store) as Arc<dyn DurableStore>, config)?;
        let mut writer = enroll_and_attach(
            &handler,
            CONVERSATION,
            ConnectionIncarnation::new(PRIOR_SERVER, 1),
            0xA1,
        )?;
        let _reader = enroll_and_attach(
            &handler,
            CONVERSATION,
            ConnectionIncarnation::new(PRIOR_SERVER, 2),
            0xB1,
        )?;
        for index in 0..records {
            let [high, low] = index.to_be_bytes();
            let mut token = [0xC0; 16];
            token[0] = high;
            token[1] = low;
            let committed = dispatch_tracked(
                &handler,
                writer.connection,
                &mut writer.conversations,
                ClientRequest::RecordAdmission(RecordAdmission {
                    conversation_id: CONVERSATION,
                    participant_id: writer.participant_id,
                    capability_generation: writer.generation,
                    record_admission_attempt_token: RecordAdmissionAttemptToken::new(token),
                    payload: vec![low],
                }),
            )?;
            assert!(
                matches!(committed, ServerValue::RecordCommitted(_)),
                "{committed:?}"
            );
        }
        let history = base_rows(&handler, CONVERSATION)?.len();
        drop(handler);

        let reopened =
            ProductionParticipantHandler::new(Arc::clone(&store) as Arc<dyn DurableStore>, config)?;
        store.reset();
        reopened.repair_unclean_server_restart(PRIOR_SERVER + 1)?;
        assert_eq!(
            (store.replays(), store.rows_read()),
            (0, (0, 0)),
            "the restart repair over {history} rows of history read base or extension rows \
             (rebuilds from row 0, (base rows, extension rows))"
        );
        let kinds = base_kinds(&reopened, CONVERSATION)?;
        assert_eq!(
            kinds.iter().filter(|kind| **kind == "died").count(),
            2,
            "the repair did not close both bindings the prior process left Bound: {kinds:?}"
        );
        measured.push((history, store.rows_read()));
        assert_complete_and_unrepaired(
            &store,
            reopened,
            config,
            CONVERSATION,
            "an unclean-restart repair",
        )?;
    }
    assert!(
        measured[0].0 < measured[1].0,
        "the two runs did not differ in history length: {measured:?}"
    );
    Ok(())
}

/// The extension append lands and the flush after it fails: the commit is
/// refused as unconfirmed, the row is already in the stream, and the next
/// request keeps it rather than writing a second copy.
#[test]
fn an_extension_row_whose_flush_failed_is_kept_once_by_the_next_request()
-> Result<(), Box<dyn Error>> {
    const CONVERSATION: u64 = 0x7E6;
    let store = ReplayCountingStore::new()?;
    let config = test_participant_config();
    let handler =
        ProductionParticipantHandler::new(Arc::clone(&store) as Arc<dyn DurableStore>, config)?;
    let mut leaver = enroll_and_attach(
        &handler,
        CONVERSATION,
        ConnectionIncarnation::new(76, 1),
        0xD1,
    )?;
    let mut peer = enroll_and_attach(
        &handler,
        CONVERSATION,
        ConnectionIncarnation::new(76, 2),
        0xE1,
    )?;

    store
        .fail_flush_after_next_extension_append
        .store(true, Ordering::SeqCst);
    let refused = dispatch_tracked(
        &handler,
        leaver.connection,
        &mut leaver.conversations,
        ClientRequest::Leave(LeaveRequest {
            conversation_id: CONVERSATION,
            participant_id: leaver.participant_id,
            capability_generation: leaver.generation,
            attach_secret: leaver.attach_secret,
            leave_attempt_token: LeaveAttemptToken::new([0xD2; 16]),
        }),
    );
    let text = match refused {
        Ok(value) => {
            return Err(
                format!("the leave answered although its row's flush failed: {value:?}").into(),
            );
        }
        Err(error) => error.to_string(),
    };
    assert!(
        text.contains("is not confirmed durable") && text.contains("flush after an extension"),
        "the refusal does not say the row is unconfirmed and keep its cause: {text}"
    );
    let left_sequence = base_rows(&handler, CONVERSATION)?
        .iter()
        .rev()
        .find_map(|(sequence, operation)| {
            matches!(operation, StoredOperation::Left { .. }).then_some(*sequence)
        })
        .ok_or("the Left base row was not durable")?;
    let rows_for_left = |rows: &[(u64, OutboxRow)]| {
        rows.iter()
            .filter(|(_, row)| {
                matches!(
                    row,
                    OutboxRow::Produced(batch) if batch.source_log_sequence() == left_sequence
                )
            })
            .count()
    };
    assert_eq!(
        rows_for_left(&extension_rows(&handler, CONVERSATION)?),
        1,
        "the landed extension row is not in the stream exactly once"
    );

    let next = admit(&handler, CONVERSATION, &mut peer, 0xE2)?;
    assert!(matches!(next, ServerValue::RecordCommitted(_)), "{next:?}");
    let after = extension_rows(&handler, CONVERSATION)?;
    assert_eq!(
        rows_for_left(&after),
        1,
        "the next request wrote a second copy of a row that had landed"
    );
    assert!(
        after.iter().any(|(_, row)| matches!(
            row,
            OutboxRow::Produced(batch) if batch.source_log_sequence() > left_sequence
        )),
        "the next request's own extension row is missing"
    );
    let reopened = assert_complete_and_unrepaired(
        &store,
        handler,
        config,
        CONVERSATION,
        "a leave whose flush failed",
    )?;
    assert_eq!(
        rows_for_left(&extension_rows(&reopened, CONVERSATION)?),
        1,
        "a reopen wrote a second copy of the landed row"
    );
    Ok(())
}