liminal-server 0.20.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, WriteOutcomeUnknown,
};

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 fault the next routed batch that carries an extension
/// row: refuse it before anything lands, or apply it and answer an unknown
/// outcome, with the read-back that settles the outcome left working or made
/// to fail once.
#[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_batch: AtomicBool,
    answer_unknown_after_next_extension_batch: AtomicBool,
    fail_readback_of_unknown: AtomicBool,
    fail_next_readback: AtomicBool,
}

/// Whether a routed batch writes a row (not a head) of a Unit 2 extension
/// stream: the codec writes the stream's own name into every row key.
fn carries_extension_row(items: &[liminal::durability::RoutedBatchItem]) -> bool {
    items.iter().any(|item| {
        super::routed_stream_codec::row_stream(&item.key)
            .is_some_and(|stream| stream.starts_with(OUTBOX_STREAM_PREFIX))
    })
}

/// The outcome a store answers when it cannot say whether the write landed.
fn outcome_unknown() -> DurabilityError {
    haematite::ApiError::Storage(haematite::DatabaseError::OutcomeUnknown { pid: 1 }).into()
}

/// What the store answers the read that cannot settle an unknown outcome.
fn unreadable() -> DurabilityError {
    haematite::ApiError::Storage(haematite::DatabaseError::IoError(std::io::Error::other(
        "the test failed the read-back of an unknown outcome",
    )))
    .into()
}

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_batch: AtomicBool::new(false),
            answer_unknown_after_next_extension_batch: AtomicBool::new(false),
            fail_readback_of_unknown: AtomicBool::new(false),
            fail_next_readback: 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 apply_routed_batch(
        &self,
        route_key: &str,
        items: Vec<liminal::durability::RoutedBatchItem>,
    ) -> Result<(), liminal::durability::DurabilityError> {
        let extension = carries_extension_row(&items);
        if extension
            && self
                .refuse_next_extension_batch
                .swap(false, Ordering::SeqCst)
        {
            return Err(DurabilityError::ConfigError(
                "the test refused the batch carrying this extension row".to_owned(),
            ));
        }
        self.inner.apply_routed_batch(route_key, items).await?;
        // The batch is the one durable operation with its one flush inside;
        // a flush that fails after the rows are applied beneath is an unknown
        // outcome over rows that landed.
        if extension
            && self
                .answer_unknown_after_next_extension_batch
                .swap(false, Ordering::SeqCst)
        {
            // The first routed read after this answer is the settle's
            // read-back of the batch's keys; armed, it fails once.
            if self.fail_readback_of_unknown.swap(false, Ordering::SeqCst) {
                self.fail_next_readback.store(true, Ordering::SeqCst);
            }
            return Err(outcome_unknown());
        }
        Ok(())
    }

    async fn get_routed(
        &self,
        route_key: &str,
        key: &[u8],
    ) -> Result<Option<Vec<u8>>, liminal::durability::DurabilityError> {
        if self.fail_next_readback.swap(false, Ordering::SeqCst) {
            return Err(unreadable());
        }
        let value = self.inner.get_routed(route_key, key).await?;
        // A row read through its routed key is a row read: the base rows and
        // extension rows a rebuild reads are routed rows now, and a rebuild
        // that reads an operation log's first row starts from its start.
        // Heads count nothing.
        if let Ok(super::routed_stream_codec::StreamKey::Row { stream, sequence }) =
            super::routed_stream_codec::decode_key(key)
        {
            if sequence == 0 && is_operation_log(stream) {
                self.log_reads_from_start.fetch_add(1, Ordering::SeqCst);
            }
            if value.is_some() {
                self.count(stream, 1);
            }
        }
        Ok(value)
    }

    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> {
        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> {
        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 Left base row that a `Left` operation wrote, by sequence, or `None`.
fn left_sequence(handler: &ProductionParticipantHandler, conversation_id: u64) -> Option<u64> {
    base_rows(handler, conversation_id)
        .ok()?
        .iter()
        .rev()
        .find_map(|(sequence, operation)| {
            matches!(operation, StoredOperation::Left { .. }).then_some(*sequence)
        })
}

/// How many Produced extension rows carry `source_log_sequence`.
fn produced_rows_for(rows: &[(u64, OutboxRow)], source_log_sequence: u64) -> usize {
    rows.iter()
        .filter(|(_, row)| {
            matches!(
                row,
                OutboxRow::Produced(batch) if batch.source_log_sequence() == source_log_sequence
            )
        })
        .count()
}

/// The store refuses the batch that carries a Leave's base row and its
/// extension row: one durable operation, so neither row lands, the leave is
/// answered as an unknown outcome under its attempt token, the next request
/// on the conversation reloads it and finds no leave to repair, and the leave
/// re-presented with its token commits both rows, once each.
///
/// The crash point between a base row and its extension row, refused by name
/// and written by the next request, closed with the batch: there is no row to
/// owe. A connection fate that fails part way latches the service fatal
/// instead, by its own design: its durable Open is its retry.
#[test]
fn a_refused_batch_lands_no_row_and_the_re_presented_leave_commits_both()
-> 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,
    )?;
    let leave = || {
        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 extension_before = extension_rows(&handler, CONVERSATION)?;
    store
        .refuse_next_extension_batch
        .store(true, Ordering::SeqCst);
    let refused = dispatch_tracked(
        &handler,
        leaver.connection,
        &mut leaver.conversations,
        leave(),
    )?;
    assert!(
        matches!(
            refused,
            ServerValue::WriteOutcomeUnknown(WriteOutcomeUnknown::Leave { .. })
        ),
        "the refused batch was not answered as an unknown outcome under the leave's token: {refused:?}"
    );
    assert!(
        !store.refuse_next_extension_batch.load(Ordering::SeqCst),
        "the armed refusal was never struck"
    );
    assert_eq!(
        left_sequence(&handler, CONVERSATION),
        None,
        "a refused batch left its Left base row durable"
    );
    assert_eq!(
        extension_rows(&handler, CONVERSATION)?,
        extension_before,
        "a refused batch left an extension row durable"
    );

    let next = admit(&handler, CONVERSATION, &mut peer, 0x92)?;
    assert!(matches!(next, ServerValue::RecordCommitted(_)), "{next:?}");
    assert_eq!(
        left_sequence(&handler, CONVERSATION),
        None,
        "the next request wrote a leave nobody committed"
    );

    let committed = dispatch_tracked(
        &handler,
        leaver.connection,
        &mut leaver.conversations,
        leave(),
    )?;
    assert!(
        matches!(committed, ServerValue::LeaveCommitted(_)),
        "the leave re-presented under its token did not commit: {committed:?}"
    );
    let left = left_sequence(&handler, CONVERSATION)
        .ok_or("the re-presented leave's Left base row was not durable")?;
    assert_eq!(
        produced_rows_for(&extension_rows(&handler, CONVERSATION)?, left),
        1,
        "the re-presented leave's extension row is not in the stream exactly once"
    );
    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(())
}

/// Enrolls a leaver and a peer on `conversation_id`, arms the store's faults
/// for the leaver's batch, dispatches the leave, and answers what the leave
/// answered.
fn leave_under_fault(
    store: &ReplayCountingStore,
    handler: &ProductionParticipantHandler,
    conversation_id: u64,
    connection: u64,
    fail_readback: bool,
) -> Result<(ServerValue, Member), Box<dyn Error>> {
    let mut leaver = enroll_and_attach(
        handler,
        conversation_id,
        ConnectionIncarnation::new(connection, 1),
        0xD1,
    )?;
    let peer = enroll_and_attach(
        handler,
        conversation_id,
        ConnectionIncarnation::new(connection, 2),
        0xE1,
    )?;
    store
        .answer_unknown_after_next_extension_batch
        .store(true, Ordering::SeqCst);
    store
        .fail_readback_of_unknown
        .store(fail_readback, Ordering::SeqCst);
    let answered = dispatch_tracked(
        handler,
        leaver.connection,
        &mut leaver.conversations,
        ClientRequest::Leave(LeaveRequest {
            conversation_id,
            participant_id: leaver.participant_id,
            capability_generation: leaver.generation,
            attach_secret: leaver.attach_secret,
            leave_attempt_token: LeaveAttemptToken::new([0xD2; 16]),
        }),
    )?;
    assert!(
        !store
            .answer_unknown_after_next_extension_batch
            .load(Ordering::SeqCst),
        "the armed unknown outcome was never struck"
    );
    assert!(
        !store.fail_readback_of_unknown.load(Ordering::SeqCst)
            && !store.fail_next_readback.load(Ordering::SeqCst),
        "the armed read-back fault was never struck"
    );
    Ok((answered, peer))
}

/// After a leave whose rows landed under an unknown outcome: the Left base
/// row is durable, its extension row is in the stream exactly once, the next
/// request keeps it rather than writing a second copy, and a reopen keeps it
/// too.
fn assert_landed_rows_kept_once(
    store: &Arc<ReplayCountingStore>,
    handler: ProductionParticipantHandler,
    config: ParticipantConfig,
    conversation_id: u64,
    peer: &mut Member,
    label: &str,
) -> Result<(), Box<dyn Error>> {
    let left =
        left_sequence(&handler, conversation_id).ok_or("the Left base row was not durable")?;
    assert_eq!(
        produced_rows_for(&extension_rows(&handler, conversation_id)?, left),
        1,
        "{label}: the landed extension row is not in the stream exactly once"
    );
    let next = admit(&handler, conversation_id, peer, 0xE2)?;
    assert!(matches!(next, ServerValue::RecordCommitted(_)), "{next:?}");
    let after = extension_rows(&handler, conversation_id)?;
    assert_eq!(
        produced_rows_for(&after, left),
        1,
        "{label}: 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
        )),
        "{label}: the next request's own extension row is missing"
    );
    let reopened = assert_complete_and_unrepaired(store, handler, config, conversation_id, label)?;
    assert_eq!(
        produced_rows_for(&extension_rows(&reopened, conversation_id)?, left),
        1,
        "{label}: a reopen wrote a second copy of the landed row"
    );
    Ok(())
}

/// The batch carrying a Leave's rows lands and the store answers an unknown
/// outcome: the commit reads its own keys back, finds every row standing as
/// written, and answers the leave as committed. The row is in the stream
/// exactly once and the next request keeps it rather than writing a second
/// copy.
///
/// Before the batch this was the flush that failed after an append that
/// landed, refused as unconfirmed; the batch settles it by read-back instead.
#[test]
fn an_unknown_outcome_whose_rows_landed_is_settled_by_read_back_and_kept_once()
-> 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 (answered, mut peer) = leave_under_fault(&store, &handler, CONVERSATION, 76, false)?;
    assert!(
        matches!(answered, ServerValue::LeaveCommitted(_)),
        "the read-back did not settle the landed rows as committed: {answered:?}"
    );
    assert_landed_rows_kept_once(
        &store,
        handler,
        config,
        CONVERSATION,
        &mut peer,
        "a leave settled by read-back",
    )
}

/// The batch lands, the store answers an unknown outcome, and the read-back
/// that would settle it fails: the leave is answered as an unknown outcome
/// under its attempt token, with the rows standing. The next request reloads
/// the conversation and keeps the landed row exactly once, and the leave
/// re-presented under its token is answered as the committed leave it is.
#[test]
fn an_unknown_outcome_that_cannot_be_read_back_is_answered_unknown_and_kept_once()
-> Result<(), Box<dyn Error>> {
    const CONVERSATION: u64 = 0x7E7;
    let store = ReplayCountingStore::new()?;
    let config = test_participant_config();
    let handler =
        ProductionParticipantHandler::new(Arc::clone(&store) as Arc<dyn DurableStore>, config)?;
    let (answered, mut peer) = leave_under_fault(&store, &handler, CONVERSATION, 77, true)?;
    assert!(
        matches!(
            answered,
            ServerValue::WriteOutcomeUnknown(WriteOutcomeUnknown::Leave { .. })
        ),
        "an unreadable outcome was not answered as unknown under the leave's token: {answered:?}"
    );
    assert_landed_rows_kept_once(
        &store,
        handler,
        config,
        CONVERSATION,
        &mut peer,
        "a leave whose outcome could not be read back",
    )
}