liminal-server 0.24.5

//! Participant production-log v2 migration contract.

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

use liminal::durability::bridge::block_on;
use liminal::durability::{
    DurabilityError, DurableStore, RoutedBatchItem, ScanCursor, ScanPage, StoredEntry,
    open_ephemeral,
};
use liminal_protocol::wire::{
    ClientRequest, ConnectionIncarnation, EnrollmentRequest, EnrollmentToken, Generation,
    LeaveAttemptToken, LeaveEnvelope, LeaveRequest, RecordAdmission, RecordAdmissionAttemptToken,
    RecordAdmissionEnvelope, ServerValue, WriteOutcomeUnknown,
};

use super::ProductionParticipantHandler;
use super::log::{
    DecodedStoredOperation, OperationLog, OperationLogError, OperationSchemaPhase, READ_BATCH_SIZE,
    SCHEMA_VERSION, STREAM_PREFIX, StoredOperation, legacy_rows,
};
use super::outbox_log::OUTBOX_STREAM_PREFIX;
use super::restore_checkpoint_locator::checkpoint_conversation_route;
use super::routed_stream_codec::row_stream;
use super::tests::{dispatch, open_disk_store_for_tests, test_participant_config};
use super::tests_w3_restore_fixture::stream_payloads;

const CONVERSATION: u64 = 0xF0C4;

/// A store that faults the next routed batch carrying a row of an operation
/// log. Every production commit is that one batch (`OperationBatch::commit`
/// to `apply_routed_batch`), with its one flush inside the store, so a fault
/// on `append` or `flush` is never issued; the batch is where it strikes. It
/// refuses the batch before anything lands, or applies it and answers an
/// unknown outcome, with the read-back that settles the outcome left working
/// or made to fail once.
#[derive(Debug)]
struct FaultStore {
    inner: Arc<dyn DurableStore>,
    refuse_next_batch: AtomicBool,
    answer_unknown_after_next_batch: AtomicBool,
    fail_readback_of_unknown: AtomicBool,
    fail_next_readback: AtomicBool,
}

/// The fault the next operation batch meets.
#[derive(Clone, Copy, Debug)]
enum Fault {
    /// Refused before anything lands.
    Refused,
    /// Applied, answered unknown, and read back as landed.
    UnknownLanded,
    /// Applied, answered unknown, and the read-back fails.
    UnknownUnreadable,
}

impl FaultStore {
    fn new() -> Result<Arc<Self>, Box<dyn Error>> {
        Ok(Arc::new(Self {
            inner: Arc::new(open_ephemeral(1)?),
            refuse_next_batch: AtomicBool::new(false),
            answer_unknown_after_next_batch: AtomicBool::new(false),
            fail_readback_of_unknown: AtomicBool::new(false),
            fail_next_readback: AtomicBool::new(false),
        }))
    }

    fn arm(&self, fault: Fault) {
        match fault {
            Fault::Refused => self.refuse_next_batch.store(true, Ordering::SeqCst),
            Fault::UnknownLanded => self
                .answer_unknown_after_next_batch
                .store(true, Ordering::SeqCst),
            Fault::UnknownUnreadable => {
                self.answer_unknown_after_next_batch
                    .store(true, Ordering::SeqCst);
                self.fail_readback_of_unknown.store(true, Ordering::SeqCst);
            }
        }
    }

    /// Whether every armed fault was struck: a fault left armed measured
    /// nothing.
    fn all_struck(&self) -> bool {
        [
            &self.refuse_next_batch,
            &self.answer_unknown_after_next_batch,
            &self.fail_readback_of_unknown,
            &self.fail_next_readback,
        ]
        .iter()
        .all(|armed| !armed.load(Ordering::SeqCst))
    }
}

/// Whether a routed batch writes a row (not a head) of an operation log: the
/// codec writes the stream's own name into every row key.
fn carries_operation_row(items: &[RoutedBatchItem]) -> bool {
    items
        .iter()
        .any(|item| row_stream(&item.key).is_some_and(|stream| stream.starts_with(STREAM_PREFIX)))
}

#[async_trait::async_trait]
impl DurableStore for FaultStore {
    async fn apply_routed_batch(
        &self,
        route_key: &str,
        items: Vec<RoutedBatchItem>,
    ) -> Result<(), DurabilityError> {
        let operation = carries_operation_row(&items);
        if operation && self.refuse_next_batch.swap(false, Ordering::SeqCst) {
            return Err(DurabilityError::ConfigError(
                "the test refused the batch carrying this operation row".to_owned(),
            ));
        }
        self.inner.apply_routed_batch(route_key, items).await?;
        if operation
            && self
                .answer_unknown_after_next_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(DurabilityError::OutcomeUnknown { pid: 1 });
        }
        Ok(())
    }

    async fn get_routed(
        &self,
        route_key: &str,
        key: &[u8],
    ) -> Result<Option<Vec<u8>>, DurabilityError> {
        if self.fail_next_readback.swap(false, Ordering::SeqCst) {
            return Err(DurabilityError::ConfigError(
                "the test failed the read-back of an unknown outcome".to_owned(),
            ));
        }
        self.inner.get_routed(route_key, key).await
    }

    async fn append(
        &self,
        stream_key: &str,
        payload: Vec<u8>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.inner.append(stream_key, payload, expected_seq).await
    }

    async fn append_all(
        &self,
        stream_key: &str,
        payloads: Vec<Vec<u8>>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.inner
            .append_all(stream_key, payloads, expected_seq)
            .await
    }

    async fn read_from(
        &self,
        stream_key: &str,
        offset: u64,
        limit: usize,
    ) -> Result<Vec<StoredEntry>, DurabilityError> {
        self.inner.read_from(stream_key, offset, limit).await
    }

    async fn read_at(
        &self,
        stream_key: &str,
        sequence: u64,
    ) -> Result<Option<StoredEntry>, DurabilityError> {
        self.inner.read_at(stream_key, sequence).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 store() -> Result<Arc<dyn DurableStore>, Box<dyn Error>> {
    Ok(Arc::new(open_ephemeral(1)?))
}

fn operation_stream() -> String {
    format!("{STREAM_PREFIX}{CONVERSATION}")
}

fn extension_stream() -> String {
    format!("{OUTBOX_STREAM_PREFIX}{CONVERSATION}")
}

fn append_literal(
    store: &Arc<dyn DurableStore>,
    conversation_id: u64,
    sequence: u64,
    bytes: &[u8],
) -> Result<(), Box<dyn Error>> {
    let stream_key = format!("{STREAM_PREFIX}{conversation_id}");
    let assigned = block_on(store.append(&stream_key, bytes.to_vec(), sequence))??;
    assert_eq!(assigned, sequence);
    block_on(store.flush())??;
    Ok(())
}

#[test]
fn empty_stream_and_all_v2_stream_restore() -> Result<(), Box<dyn Error>> {
    let store = store()?;
    let log = OperationLog::new(Arc::clone(&store), CONVERSATION);
    assert!(
        block_on(log.read_bounded_page(0, OperationSchemaPhase::V2Prefix, READ_BATCH_SIZE))??
            .rows
            .is_empty()
    );

    // An append is one routed batch on the conversation's route, so its row
    // is read where the commit put it: the routed page, which refuses a
    // stream with no routed head rather than answering it empty.
    block_on(log.append(&StoredOperation::Genesis { event: Vec::new() }, 0))??;
    let route = checkpoint_conversation_route(CONVERSATION);
    let page = block_on(log.read_routed_page(&route, 0, OperationSchemaPhase::V2Prefix))??;
    assert_eq!(page.rows.len(), 1);
    assert_eq!(page.rows[0].sequence, 0);
    assert!(matches!(
        page.rows[0].operation,
        DecodedStoredOperation::V3(StoredOperation::Genesis { .. })
    ));

    let raw = stream_payloads(&store, &operation_stream())?;
    let payload = raw.first().ok_or("the appended row is not in its stream")?;
    let json: serde_json::Value = serde_json::from_slice(payload)?;
    assert_eq!(json["schema_version"], SCHEMA_VERSION);
    Ok(())
}

#[test]
fn literal_v1_row_is_rejected_as_schema_version_one() -> Result<(), Box<dyn Error>> {
    let store = store()?;
    append_literal(
        &store,
        CONVERSATION,
        0,
        br#"{"schema_version":1,"operation":{"operation":"genesis","event":[]}}"#,
    )?;
    let log = OperationLog::new(store, CONVERSATION);
    let result =
        block_on(log.read_bounded_page(0, OperationSchemaPhase::V2Prefix, READ_BATCH_SIZE))?;
    assert!(matches!(result, Err(OperationLogError::SchemaVersion(1))));
    Ok(())
}

#[test]
fn mixed_v2_then_v1_page_is_rejected_before_returning_any_rows() -> Result<(), Box<dyn Error>> {
    // A legacy flat stream mixed in version is refused: a v1 row can only
    // follow a flat row, never a routed one, so the first row is flat too.
    let store = store()?;
    let log = OperationLog::new(Arc::clone(&store), CONVERSATION);
    block_on(log.append_flat(&StoredOperation::Genesis { event: Vec::new() }, 0))??;
    append_literal(
        &store,
        CONVERSATION,
        1,
        br#"{"schema_version":1,"operation":{"operation":"genesis","event":[]}}"#,
    )?;

    let result =
        block_on(log.read_bounded_page(0, OperationSchemaPhase::V2Prefix, READ_BATCH_SIZE))?;
    assert!(matches!(result, Err(OperationLogError::SchemaVersion(1))));
    Ok(())
}

#[test]
fn mixed_stream_fails_startup_without_publishing_an_authority() -> Result<(), Box<dyn Error>> {
    let home = tempfile::tempdir()?;
    let data_dir = home.path().join("durability");
    let conversation_id = 0xF0C5;
    {
        // A legacy flat stream mixed in version is refused: the enrollment
        // writes the flat rows a store from before pruning holds, so the v1
        // row below follows them in the same flat stream.
        let legacy = legacy_rows::arm();
        let store = open_disk_store_for_tests(&data_dir)?;
        let handler =
            ProductionParticipantHandler::new(Arc::clone(&store), test_participant_config())?;
        let enrolled = dispatch(
            &handler,
            ConnectionIncarnation::new(44, 1),
            ClientRequest::Enrollment(EnrollmentRequest {
                conversation_id,
                enrollment_token: EnrollmentToken::new([44; 16]),
            }),
        )?;
        assert!(matches!(enrolled, ServerValue::EnrollBound(_)));
        drop(handler);
        drop(legacy);
        append_literal(
            &store,
            conversation_id,
            2,
            br#"{"schema_version":1,"operation":{"operation":"genesis","event":[]}}"#,
        )?;
    }

    let store = open_disk_store_for_tests(&data_dir)?;
    // CONTAINMENT: the node starts. The two properties this test has always
    // protected are unchanged and are now asserted DIRECTLY rather than
    // inferred from a dead constructor: the mixed stream publishes no
    // authority, and the migration cause survives by name.
    let handler = ProductionParticipantHandler::new_restored(store, test_participant_config())
        .map_err(|error| {
            format!("CONTAINMENT: one unloadable conversation took the whole node down: {error}")
        })?;

    // DUR-020 (e5c6093d) and f8a21104: the first touch runs the
    // conversation's restore unit, and an unloadable conversation answers it
    // with its typed refusal carrying the cause, so no authority can be
    // reached through it at all. An `Ok` cell must still hold none.
    match handler.cell(conversation_id) {
        Err(crate::server::participant::ParticipantSemanticError::ConversationUnloadable {
            conversation_id: refused,
            reason,
        }) => {
            assert_eq!(
                refused, conversation_id,
                "the first touch refused another conversation"
            );
            assert!(
                reason.contains("schema version 1"),
                "the first touch refused the mixed stream, but its cause was lost: {reason}"
            );
        }
        Err(error) => return Err(format!("unexpected refusal: {error:?}").into()),
        Ok(cell) => {
            let owner = cell
                .lock()
                .map_err(|_| "conversation owner lock is poisoned")?;
            assert!(
                owner.is_none(),
                "the mixed v2/v1 stream published an authority anyway"
            );
            drop(owner);
        }
    }

    let unloadable = handler.unloadable_conversations();
    let reason = unloadable
        .get(&conversation_id)
        .ok_or("the mixed v2/v1 conversation was not recorded as unloadable")?;
    assert!(
        reason.contains("schema version 1"),
        "the migration failure was contained but its cause was lost: {reason}"
    );
    Ok(())
}

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

fn enrolled_fault_handler(
    fault: &Arc<FaultStore>,
    incarnation: ConnectionIncarnation,
) -> Result<ProductionParticipantHandler, Box<dyn Error>> {
    let store: Arc<dyn DurableStore> = fault.clone();
    let handler = ProductionParticipantHandler::new(store, test_participant_config())?;
    let value = dispatch(
        &handler,
        incarnation,
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id: CONVERSATION,
            enrollment_token: EnrollmentToken::new([0xF4; 16]),
        }),
    )?;
    assert!(matches!(value, ServerValue::EnrollBound(_)));
    Ok(handler)
}

fn enrolled_fault_leave_handler(
    fault: &Arc<FaultStore>,
    incarnation: ConnectionIncarnation,
) -> Result<(ProductionParticipantHandler, LeaveRequest), Box<dyn Error>> {
    let store: Arc<dyn DurableStore> = fault.clone();
    let handler = ProductionParticipantHandler::new(store, test_participant_config())?;
    let value = dispatch(
        &handler,
        incarnation,
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id: CONVERSATION,
            enrollment_token: EnrollmentToken::new([0xF5; 16]),
        }),
    )?;
    let ServerValue::EnrollBound(receipt) = value else {
        return Err(format!("fault-test enrollment did not bind: {value:?}").into());
    };
    let request = LeaveRequest {
        conversation_id: CONVERSATION,
        participant_id: receipt.participant_id(),
        capability_generation: Generation::ONE,
        attach_secret: receipt.attach_secret(),
        leave_attempt_token: LeaveAttemptToken::new([0xF6; 16]),
    };
    Ok((handler, request))
}

/// Whether no in-memory poststate is published for the test conversation.
fn owner_unpublished(handler: &ProductionParticipantHandler) -> Result<bool, Box<dyn Error>> {
    let cell = handler.cell(CONVERSATION)?;
    let unpublished = cell
        .lock()
        .map_err(|_| "test conversation owner lock poisoned")?
        .is_none();
    Ok(unpublished)
}

/// The payloads of one stream, in sequence order.
type Payloads = Vec<Vec<u8>>;

/// The operation stream and the extension stream of the test conversation,
/// each read where its commit put it: the flat prefix, then the routed rows.
fn both_streams(fault: &FaultStore) -> Result<(Payloads, Payloads), Box<dyn Error>> {
    Ok((
        stream_payloads(&fault.inner, &operation_stream())?,
        stream_payloads(&fault.inner, &extension_stream())?,
    ))
}

/// The operation stream after one batch landed: exactly one row past
/// `before`, a complete current-schema row of `operation`.
fn one_row_landed(
    fault: &FaultStore,
    before: usize,
    operation: &str,
) -> Result<Vec<Vec<u8>>, Box<dyn Error>> {
    let rows = stream_payloads(&fault.inner, &operation_stream())?;
    assert_eq!(
        rows.len(),
        before + 1,
        "the batch did not leave exactly one {operation} row in the operation stream"
    );
    let last = rows.last().ok_or("the operation stream is empty")?;
    let stored: serde_json::Value = serde_json::from_slice(last)?;
    assert_eq!(stored["schema_version"], SCHEMA_VERSION);
    assert_eq!(stored["operation"]["operation"], operation);
    Ok(rows)
}

/// Whether `value` is the write-outcome-unknown answer to `record(token)`,
/// carrying the request's envelope and its attempt token.
fn record_unknown(value: &ServerValue, token: u8) -> bool {
    let expected = RecordAdmissionEnvelope {
        conversation_id: CONVERSATION,
        participant_id: 0,
        capability_generation: Generation::ONE,
        record_admission_attempt_token: RecordAdmissionAttemptToken::new([token; 16]),
    };
    matches!(
        value,
        ServerValue::WriteOutcomeUnknown(WriteOutcomeUnknown::RecordAdmission { request })
            if *request == expected
    )
}

/// Whether `value` is the write-outcome-unknown answer to `leave`, carrying
/// the request's envelope and its attempt token.
fn leave_unknown(value: &ServerValue, leave: &LeaveRequest) -> bool {
    let expected = LeaveEnvelope {
        conversation_id: leave.conversation_id,
        participant_id: leave.participant_id,
        capability_generation: leave.capability_generation,
        leave_attempt_token: leave.leave_attempt_token,
    };
    matches!(
        value,
        ServerValue::WriteOutcomeUnknown(WriteOutcomeUnknown::Leave { request })
            if *request == expected
    )
}

/// Was `record_append_failure_publishes_no_response_or_poststate`.
#[test]
fn a_refused_record_batch_lands_no_row_and_the_re_presented_record_commits()
-> Result<(), Box<dyn Error>> {
    let fault = FaultStore::new()?;
    let incarnation = ConnectionIncarnation::new(45, 1);
    let handler = enrolled_fault_handler(&fault, incarnation)?;
    let before = both_streams(&fault)?;
    fault.arm(Fault::Refused);

    let refused = dispatch(&handler, incarnation, record(0xA1))?;
    assert!(
        record_unknown(&refused, 0xA1),
        "the refused batch was not answered as an unknown outcome under the record's token: \
         {refused:?}"
    );
    assert!(fault.all_struck(), "the armed refusal was never struck");
    assert!(
        owner_unpublished(&handler)?,
        "a refused batch published in-memory poststate"
    );
    assert_eq!(
        both_streams(&fault)?,
        before,
        "a refused batch left a row of the operation or extension stream durable"
    );

    let committed = dispatch(&handler, incarnation, record(0xA1))?;
    assert!(
        matches!(committed, ServerValue::RecordCommitted(_)),
        "the record re-presented under its token did not commit: {committed:?}"
    );
    one_row_landed(&fault, before.0.len(), "record_admission")?;
    Ok(())
}

/// Was `record_flush_failure_replays_only_a_complete_row`.
#[test]
fn a_record_batch_answered_unknown_whose_rows_landed_commits_and_replays_once()
-> Result<(), Box<dyn Error>> {
    let fault = FaultStore::new()?;
    let incarnation = ConnectionIncarnation::new(46, 1);
    let handler = enrolled_fault_handler(&fault, incarnation)?;
    let before = both_streams(&fault)?.0.len();
    fault.arm(Fault::UnknownLanded);

    let settled = dispatch(&handler, incarnation, record(0xB1))?;
    let ServerValue::RecordCommitted(settled) = settled else {
        return Err(format!("the read-back did not settle the landed row: {settled:?}").into());
    };
    assert!(
        fault.all_struck(),
        "the armed unknown outcome was never struck"
    );
    let landed = one_row_landed(&fault, before, "record_admission")?;

    drop(handler);
    let reopened = ProductionParticipantHandler::new(
        Arc::clone(&fault) as Arc<dyn DurableStore>,
        test_participant_config(),
    )?;
    let replayed = dispatch(&reopened, incarnation, record(0xB1))?;
    let ServerValue::RecordCommitted(replayed) = replayed else {
        return Err(format!("cold replay did not answer the landed record: {replayed:?}").into());
    };
    assert_eq!(replayed.delivery_seq(), settled.delivery_seq());
    assert_eq!(
        stream_payloads(&fault.inner, &operation_stream())?,
        landed,
        "cold replay wrote a second copy of a row that had landed"
    );
    let committed = dispatch(&reopened, incarnation, record(0xB2))?;
    let ServerValue::RecordCommitted(committed) = committed else {
        return Err("post-fault cold replay did not admit the next record".into());
    };
    assert!(committed.delivery_seq() > settled.delivery_seq());
    Ok(())
}

/// The unreadable shape of the record batch; it had no flat-era pin.
#[test]
fn a_record_batch_whose_unknown_outcome_cannot_be_read_back_is_answered_unknown_and_kept_once()
-> Result<(), Box<dyn Error>> {
    let fault = FaultStore::new()?;
    let incarnation = ConnectionIncarnation::new(49, 1);
    let handler = enrolled_fault_handler(&fault, incarnation)?;
    let before = both_streams(&fault)?.0.len();
    fault.arm(Fault::UnknownUnreadable);

    let answered = dispatch(&handler, incarnation, record(0xC1))?;
    assert!(
        record_unknown(&answered, 0xC1),
        "an unreadable outcome was not answered as unknown under the record's token: \
         {answered:?}"
    );
    assert!(
        fault.all_struck(),
        "the armed read-back fault was never struck"
    );
    assert!(
        owner_unpublished(&handler)?,
        "an unknown outcome published in-memory poststate"
    );
    let landed = one_row_landed(&fault, before, "record_admission")?;

    let replayed = dispatch(&handler, incarnation, record(0xC1))?;
    assert!(
        matches!(replayed, ServerValue::RecordCommitted(_)),
        "the record re-presented under its token was not answered committed: {replayed:?}"
    );
    assert_eq!(
        stream_payloads(&fault.inner, &operation_stream())?,
        landed,
        "the re-presented record wrote a second copy of a row that had landed"
    );
    Ok(())
}

/// Was `leave_append_failure_publishes_no_response_or_tombstone`.
#[test]
fn a_refused_leave_batch_lands_no_row_and_the_re_presented_leave_commits()
-> Result<(), Box<dyn Error>> {
    let fault = FaultStore::new()?;
    let incarnation = ConnectionIncarnation::new(47, 1);
    let (handler, request) = enrolled_fault_leave_handler(&fault, incarnation)?;
    let before = both_streams(&fault)?;
    fault.arm(Fault::Refused);

    let refused = dispatch(&handler, incarnation, ClientRequest::Leave(request.clone()))?;
    assert!(
        leave_unknown(&refused, &request),
        "the refused batch was not answered as an unknown outcome under the leave's token: \
         {refused:?}"
    );
    assert!(fault.all_struck(), "the armed refusal was never struck");
    assert!(
        owner_unpublished(&handler)?,
        "a refused batch published in-memory tombstone state"
    );
    assert_eq!(
        both_streams(&fault)?,
        before,
        "a refused batch left a Left row or its extension row durable"
    );

    let committed = dispatch(&handler, incarnation, ClientRequest::Leave(request))?;
    assert!(
        matches!(committed, ServerValue::LeaveCommitted(_)),
        "the leave re-presented under its token did not commit: {committed:?}"
    );
    one_row_landed(&fault, before.0.len(), "left")?;
    Ok(())
}

/// Was `leave_flush_failure_replays_only_a_complete_tombstone`.
#[test]
fn a_leave_batch_answered_unknown_whose_rows_landed_commits_and_replays_once()
-> Result<(), Box<dyn Error>> {
    let fault = FaultStore::new()?;
    let incarnation = ConnectionIncarnation::new(48, 1);
    let (handler, request) = enrolled_fault_leave_handler(&fault, incarnation)?;
    let before = both_streams(&fault)?.0.len();
    fault.arm(Fault::UnknownLanded);

    let settled = dispatch(&handler, incarnation, ClientRequest::Leave(request.clone()))?;
    assert!(
        matches!(settled, ServerValue::LeaveCommitted(_)),
        "the read-back did not settle the landed Left row as committed: {settled:?}"
    );
    assert!(
        fault.all_struck(),
        "the armed unknown outcome was never struck"
    );
    let landed = one_row_landed(&fault, before, "left")?;

    drop(handler);
    let reopened = ProductionParticipantHandler::new(
        Arc::clone(&fault) as Arc<dyn DurableStore>,
        test_participant_config(),
    )?;
    let replayed = dispatch(&reopened, incarnation, ClientRequest::Leave(request))?;
    assert!(
        matches!(replayed, ServerValue::LeaveCommitted(_)),
        "cold replay did not answer the landed leave: {replayed:?}"
    );
    assert_eq!(
        stream_payloads(&fault.inner, &operation_stream())?,
        landed,
        "cold replay wrote a second copy of a Left row that had landed"
    );
    Ok(())
}

/// The unreadable shape of the leave batch; it had no flat-era pin.
#[test]
fn a_leave_batch_whose_unknown_outcome_cannot_be_read_back_is_answered_unknown_and_kept_once()
-> Result<(), Box<dyn Error>> {
    let fault = FaultStore::new()?;
    let incarnation = ConnectionIncarnation::new(50, 1);
    let (handler, request) = enrolled_fault_leave_handler(&fault, incarnation)?;
    let before = both_streams(&fault)?.0.len();
    fault.arm(Fault::UnknownUnreadable);

    let answered = dispatch(&handler, incarnation, ClientRequest::Leave(request.clone()))?;
    assert!(
        leave_unknown(&answered, &request),
        "an unreadable outcome was not answered as unknown under the leave's token: \
         {answered:?}"
    );
    assert!(
        fault.all_struck(),
        "the armed read-back fault was never struck"
    );
    assert!(
        owner_unpublished(&handler)?,
        "an unknown outcome published in-memory tombstone state"
    );
    let landed = one_row_landed(&fault, before, "left")?;

    let replayed = dispatch(&handler, incarnation, ClientRequest::Leave(request))?;
    assert!(
        matches!(replayed, ServerValue::LeaveCommitted(_)),
        "the leave re-presented under its token was not answered committed: {replayed:?}"
    );
    assert_eq!(
        stream_payloads(&fault.inner, &operation_stream())?,
        landed,
        "the re-presented leave wrote a second copy of a Left row that had landed"
    );
    Ok(())
}