liminal-server 0.24.5

#![cfg(test)]

//! A write the store applied but answered `OutcomeUnknown` for.
//!
//! haematite answers a write whose caller stopped waiting after the shard had
//! started it with `OutcomeUnknown`: it may have been applied, so the position
//! must be read again. These pins drive one write of each class through
//! [`UnknownOutcomeStore`], which applies the write and then answers that, and
//! assert what the design table
//! (`docs/design/durability/OUTCOME-UNKNOWN-WRITE-SITES-2026-10-07.md`) says
//! must hold: the live owner equals what the store holds, the next commit
//! succeeds, and the client is answered the committed value, never a failure
//! for a row that is durable.

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

use liminal::durability::{
    DurabilityError, DurableStore, RoutedBatchItem, ScanCursor, ScanPage, StoredEntry,
    bridge::block_on,
};
use liminal_protocol::wire::{
    AttachAttemptToken, ClientRequest, ConnectionIncarnation, CredentialAttachRequest,
    DetachAttemptToken, DetachRequest, EnrollBound, EnrollmentRequest, EnrollmentToken, Generation,
    LeaveAttemptToken, LeaveRequest, RecordAdmission, RecordAdmissionAttemptToken, ServerValue,
};

use super::ProductionParticipantHandler;
use super::log::{OperationLog, STREAM_PREFIX};
use super::observer::OBSERVER_STREAM_KEY;
use super::outbox_log::OUTBOX_STREAM_PREFIX;
use super::registry::REGISTRY_STREAM_KEY;
use super::restore_checkpoint_locator::checkpoint_conversation_route;
use super::tests::{decode_server_value, dispatch, dispatch_outcome, test_participant_config};
use super::tests_memory_store::MemoryStore;
use crate::server::participant::{ParticipantConnectionConversations, ParticipantDispatch};

/// Which kind of write the store answers `OutcomeUnknown` for.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum Write {
    Append,
    AppendIo,
    AppendAbsentIo,
    AppendAll,
    AppendAllIo,
    AppendAllAbsentIo,
    Cas,
    FlushUnknown,
    FlushIo,
    ReadValueIo,
    ReadValueAfterBaseIo,
    ReadValueAfterRegistryIo,
    LifecycleBatchIo,
}

/// Applies every write, and answers the first armed write of one kind to keys
/// under one prefix with `OutcomeUnknown` once it has been applied. Armed
/// unsettled, the first read under the prefix after that answer fails too, so
/// the write cannot be settled by reading it back.
#[derive(Debug)]
pub(super) struct UnknownOutcomeStore {
    inner: Arc<dyn DurableStore>,
    write: Write,
    prefix: String,
    armed: AtomicBool,
    blind_on_strike: AtomicBool,
    blind: AtomicBool,
    fenced: AtomicBool,
    phase_write_completed: AtomicBool,
    lifecycle_base_rows: AtomicU64,
    lifecycle_submissions: AtomicU64,
    cas_submissions: AtomicU64,
    pub(super) flushes: AtomicU64,
}

impl UnknownOutcomeStore {
    pub(super) fn new(inner: &Arc<dyn DurableStore>, write: Write, prefix: &str) -> Self {
        Self {
            inner: Arc::clone(inner),
            write,
            prefix: prefix.to_owned(),
            armed: AtomicBool::new(false),
            blind_on_strike: AtomicBool::new(false),
            blind: AtomicBool::new(false),
            fenced: AtomicBool::new(false),
            phase_write_completed: AtomicBool::new(false),
            lifecycle_base_rows: AtomicU64::new(0),
            lifecycle_submissions: AtomicU64::new(0),
            cas_submissions: AtomicU64::new(0),
            flushes: AtomicU64::new(0),
        }
    }

    pub(super) fn arm(&self) {
        self.phase_write_completed.store(false, Ordering::SeqCst);
        self.armed.store(true, Ordering::SeqCst);
    }

    pub(super) fn recover(&self) {
        self.fenced.store(false, Ordering::SeqCst);
    }

    fn readable(&self) -> Result<(), DurabilityError> {
        if self.fenced.load(Ordering::SeqCst) {
            Err(outcome_unknown())
        } else {
            Ok(())
        }
    }

    /// Arms the write, and the read after it to fail.
    pub(super) fn arm_unsettled(&self) {
        self.blind_on_strike.store(true, Ordering::SeqCst);
        self.arm();
    }

    /// Whether this write is the armed one, disarming it.
    fn strikes(&self, write: Write, key: &str) -> bool {
        let struck = write == self.write
            && key.starts_with(&self.prefix)
            && self.armed.swap(false, Ordering::SeqCst);
        if struck && self.blind_on_strike.swap(false, Ordering::SeqCst) {
            self.blind.store(true, Ordering::SeqCst);
        }
        struck
    }

    /// Whether this read is the one after the answer that fails, clearing it.
    fn blinded(&self, key: &str) -> bool {
        key.starts_with(&self.prefix) && self.blind.swap(false, Ordering::SeqCst)
    }

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

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

    /// Whether the armed write has been answered.
    pub(super) fn struck(&self) -> bool {
        !self.armed.load(Ordering::SeqCst)
    }
}

/// What the store answers a started write whose caller stopped waiting, as
/// it maps haematite's error.
fn outcome_unknown() -> DurabilityError {
    haematite::ApiError::Storage(haematite::DatabaseError::OutcomeUnknown { pid: 1 }).into()
}

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

/// The key a batch outside the measured prefix strikes under: no stream key
/// starts with it, so no armed write of the prefix fires.
const OUTSIDE_PREFIX: &str = "outside the measured prefix";

/// Whether a routed key names a stream under `needle`: the codec writes the
/// stream's own bytes into every head and row key.
fn contains(key: &[u8], needle: &str) -> bool {
    !needle.is_empty()
        && key
            .windows(needle.len())
            .any(|window| window == needle.as_bytes())
}

/// Whether any item of the batch is keyed under `needle`.
fn touches(items: &[RoutedBatchItem], needle: &str) -> bool {
    items.iter().any(|item| contains(&item.key, needle))
}

/// How many row items (not heads) the batch carries under `needle`: a row
/// key ends in the row marker and its eight sequence bytes.
fn rows_under(items: &[RoutedBatchItem], needle: &str) -> u64 {
    let rows = items
        .iter()
        .filter(|item| {
            contains(&item.key, needle)
                && item
                    .key
                    .len()
                    .checked_sub(9)
                    .and_then(|marker| item.key.get(marker))
                    == Some(&1)
        })
        .count();
    u64::try_from(rows).unwrap_or(u64::MAX)
}

#[async_trait::async_trait]
impl DurableStore for UnknownOutcomeStore {
    /// The one routed batch a dispatch commits stands where its flat appends
    /// stood: it strikes under the prefix when any row it carries belongs to
    /// a stream under it, and it is applied beneath before the answer lies.
    async fn apply_routed_batch(
        &self,
        route_key: &str,
        items: Vec<RoutedBatchItem>,
    ) -> Result<(), DurabilityError> {
        self.readable()?;
        // An empty prefix measures every batch; a named one measures the
        // batches carrying a row under it. A batch outside it strikes
        // nothing: its key names no stream under the prefix.
        let under = self.prefix.is_empty() || touches(&items, &self.prefix);
        let key = if under {
            self.prefix.as_str()
        } else {
            OUTSIDE_PREFIX
        };
        let base_rows = rows_under(&items, STREAM_PREFIX);
        let registry_rows = touches(&items, REGISTRY_STREAM_KEY);
        if self.write == Write::LifecycleBatchIo && under {
            self.lifecycle_submissions.fetch_add(1, Ordering::SeqCst);
            if self.phase_write_completed.load(Ordering::SeqCst)
                && self.strikes(Write::LifecycleBatchIo, key)
            {
                return Err(unreadable());
            }
        }
        if self.strikes(Write::AppendAbsentIo, key) || self.strikes(Write::AppendAllAbsentIo, key) {
            self.fenced.store(true, Ordering::SeqCst);
            return Err(unreadable());
        }
        self.inner.apply_routed_batch(route_key, items).await?;
        // The batch is the flush now: one durable operation with its one full
        // flush inside the store, and nothing flushes after it. A flush fault
        // therefore stands here, after the rows are applied beneath, so the
        // answer lies while the rows are durable, exactly as the flat flush
        // fault did behind a landed append.
        if under {
            self.flushes.fetch_add(1, Ordering::SeqCst);
        }
        if self.strikes(Write::FlushUnknown, key) {
            return Err(outcome_unknown());
        }
        if self.strikes(Write::FlushIo, key) {
            return Err(unreadable());
        }
        if self.write == Write::LifecycleBatchIo {
            self.lifecycle_base_rows
                .fetch_add(base_rows, Ordering::SeqCst);
        }
        let reaches_phase = match self.write {
            Write::ReadValueAfterBaseIo => base_rows > 0,
            Write::ReadValueAfterRegistryIo => registry_rows,
            _ => false,
        };
        if reaches_phase && self.armed.load(Ordering::SeqCst) {
            self.phase_write_completed.store(true, Ordering::SeqCst);
        }
        if self.strikes(Write::Append, key) || self.strikes(Write::AppendAll, key) {
            return Err(outcome_unknown());
        }
        if self.strikes(Write::AppendIo, key) || self.strikes(Write::AppendAllIo, key) {
            return Err(unreadable());
        }
        Ok(())
    }

    /// The first routed read under the prefix after a strike is the settle's
    /// read-back of the batch's keys; armed unsettled, it fails once, so the
    /// write stays unknown. The reads after it (the retry's restore) see what
    /// the store holds.
    async fn get_routed(
        &self,
        route_key: &str,
        key: &[u8],
    ) -> Result<Option<Vec<u8>>, DurabilityError> {
        self.readable()?;
        if contains(key, &self.prefix) && self.blind.swap(false, Ordering::SeqCst) {
            return Err(unreadable());
        }
        self.inner.get_routed(route_key, key).await
    }

    async fn scan_routed(
        &self,
        route_key: &str,
        from: &[u8],
        to: &[u8],
    ) -> Result<Vec<(Vec<u8>, Vec<u8>)>, DurabilityError> {
        self.readable()?;
        self.inner.scan_routed(route_key, from, to).await
    }

    async fn remove_routed_range(
        &self,
        route_key: &str,
        from: &[u8],
        to: &[u8],
    ) -> Result<u64, DurabilityError> {
        self.readable()?;
        self.inner.remove_routed_range(route_key, from, to).await
    }

    async fn append(
        &self,
        stream_key: &str,
        payload: Vec<u8>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.readable()?;
        if self.strikes(Write::AppendAbsentIo, stream_key) {
            self.fenced.store(true, Ordering::SeqCst);
            return Err(unreadable());
        }
        let applied = self.inner.append(stream_key, payload, expected_seq).await?;
        if self.write == Write::LifecycleBatchIo && stream_key.starts_with(STREAM_PREFIX) {
            self.lifecycle_base_rows.fetch_add(1, Ordering::SeqCst);
        }
        let reaches_phase = match self.write {
            Write::ReadValueAfterBaseIo => stream_key.starts_with(STREAM_PREFIX),
            Write::ReadValueAfterRegistryIo => stream_key == REGISTRY_STREAM_KEY,
            _ => false,
        };
        if reaches_phase && self.armed.load(Ordering::SeqCst) {
            self.phase_write_completed.store(true, Ordering::SeqCst);
        }
        if self.strikes(Write::Append, stream_key) {
            return Err(outcome_unknown());
        }
        if self.strikes(Write::AppendIo, stream_key) {
            return Err(unreadable());
        }
        Ok(applied)
    }

    async fn append_all(
        &self,
        stream_key: &str,
        payloads: Vec<Vec<u8>>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.readable()?;
        if self.write == Write::LifecycleBatchIo && stream_key.starts_with(&self.prefix) {
            self.lifecycle_submissions.fetch_add(1, Ordering::SeqCst);
            if self.phase_write_completed.load(Ordering::SeqCst)
                && self.strikes(Write::LifecycleBatchIo, stream_key)
            {
                return Err(unreadable());
            }
        }
        if self.strikes(Write::AppendAllAbsentIo, stream_key) {
            self.fenced.store(true, Ordering::SeqCst);
            return Err(unreadable());
        }
        let applied = self
            .inner
            .append_all(stream_key, payloads, expected_seq)
            .await?;
        if self.strikes(Write::AppendAll, stream_key) {
            return Err(outcome_unknown());
        }
        if self.strikes(Write::AppendAllIo, stream_key) {
            return Err(unreadable());
        }
        Ok(applied)
    }

    async fn read_from(
        &self,
        stream_key: &str,
        offset: u64,
        limit: usize,
    ) -> Result<Vec<StoredEntry>, DurabilityError> {
        self.readable()?;
        if self.blinded(stream_key) {
            return Err(unreadable());
        }
        self.inner.read_from(stream_key, offset, limit).await
    }

    async fn read_at(
        &self,
        stream_key: &str,
        sequence: u64,
    ) -> Result<Option<StoredEntry>, DurabilityError> {
        self.readable()?;
        if self.blinded(stream_key) {
            return Err(unreadable());
        }
        self.inner.read_at(stream_key, sequence).await
    }

    async fn cas(&self, key: &str, old_value: u64, new_value: u64) -> Result<(), DurabilityError> {
        self.readable()?;
        self.cas_submissions.fetch_add(1, Ordering::SeqCst);
        self.inner.cas(key, old_value, new_value).await?;
        if self.strikes(Write::Cas, key) {
            return Err(outcome_unknown());
        }
        Ok(())
    }

    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.readable()?;
        if self.strikes(Write::ReadValueIo, key)
            || (self.phase_write_completed.load(Ordering::SeqCst)
                && (self.strikes(Write::ReadValueAfterBaseIo, key)
                    || self.strikes(Write::ReadValueAfterRegistryIo, key)))
        {
            return Err(unreadable());
        }
        self.inner.read_value(key).await
    }

    async fn scan_page(
        &self,
        prefix: &str,
        cursor: Option<ScanCursor>,
        limit: usize,
    ) -> Result<ScanPage, DurabilityError> {
        self.readable()?;
        self.inner.scan_page(prefix, cursor, limit).await
    }

    async fn flush(&self) -> Result<(), DurabilityError> {
        self.readable()?;
        self.flushes.fetch_add(1, Ordering::SeqCst);
        if self.strikes(Write::FlushUnknown, &self.prefix) {
            return Err(outcome_unknown());
        }
        if self.strikes(Write::FlushIo, &self.prefix) {
            return Err(unreadable());
        }
        self.inner.flush().await?;
        if self.write == Write::LifecycleBatchIo
            && self.lifecycle_base_rows.load(Ordering::SeqCst) == 2
            && !self.phase_write_completed.swap(true, Ordering::SeqCst)
        {
            self.armed.store(true, Ordering::SeqCst);
        }
        Ok(())
    }
}

const FIRST: u64 = 0x0E_01;
const SECOND: u64 = 0x0E_02;
const SENDER: ConnectionIncarnation = ConnectionIncarnation::new(0x0E, 1);
const RECIPIENT: ConnectionIncarnation = ConnectionIncarnation::new(0x0E, 2);
const LATER: ConnectionIncarnation = ConnectionIncarnation::new(0x0E, 3);

/// The memory store beneath, the fault store over it, and a handler on the
/// fault store.
pub(super) type Faulty = (
    Arc<dyn DurableStore>,
    Arc<UnknownOutcomeStore>,
    ProductionParticipantHandler,
);

/// A handler over a memory store answering `write` to keys under `prefix`
/// with `OutcomeUnknown` once armed.
pub(super) fn faulty(write: Write, prefix: &str) -> Result<Faulty, Box<dyn Error>> {
    let memory: Arc<dyn DurableStore> = Arc::new(MemoryStore::default());
    let fault = Arc::new(UnknownOutcomeStore::new(&memory, write, prefix));
    let store: Arc<dyn DurableStore> = fault.clone();
    let handler = ProductionParticipantHandler::new(store, test_participant_config())?;
    Ok((memory, fault, handler))
}

fn enrollment(conversation_id: u64, token: u8) -> ClientRequest {
    ClientRequest::Enrollment(EnrollmentRequest {
        conversation_id,
        enrollment_token: EnrollmentToken::new([token; 16]),
    })
}

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

pub(super) fn record(conversation_id: u64, sender: u64, index: u8) -> ClientRequest {
    ClientRequest::RecordAdmission(RecordAdmission {
        conversation_id,
        participant_id: sender,
        capability_generation: Generation::ONE,
        record_admission_attempt_token: RecordAdmissionAttemptToken::new([index; 16]),
        payload: vec![index],
    })
}

/// The committed operation rows the store holds for `conversation_id`,
/// wherever a commit put them: flat below the native head, routed at and
/// after it. A conversation whose only operation was refused before its
/// batch committed has no routed head and holds no row; the routed read
/// alone refused that stream by name instead of counting none.
pub(super) fn stored_rows(
    memory: &Arc<dyn DurableStore>,
    conversation_id: u64,
) -> Result<u64, Box<dyn Error>> {
    let log = OperationLog::new(Arc::clone(memory), conversation_id);
    let route = checkpoint_conversation_route(conversation_id);
    let mut rows = 0_u64;
    while block_on(log.read_committed_at(&route, rows))??.is_some() {
        rows += 1;
    }
    Ok(rows)
}

/// The live owner's next log sequence for `conversation_id`.
fn live_rows(
    handler: &ProductionParticipantHandler,
    conversation_id: u64,
) -> Result<Option<u64>, Box<dyn Error>> {
    let cell = handler.cell(conversation_id)?;
    let owner = cell.lock().map_err(|_| "the owner lock was poisoned")?;
    let rows = owner.as_ref().map(|authority| authority.next_log_sequence);
    drop(owner);
    Ok(rows)
}

/// A sender and a recipient enrolled in `conversation_id`; answers the sender.
pub(super) fn two_members(
    handler: &ProductionParticipantHandler,
    conversation_id: u64,
) -> Result<u64, Box<dyn Error>> {
    let sender = enrolled(handler, SENDER, conversation_id, 0xE1)?;
    enrolled(handler, RECIPIENT, conversation_id, 0xE2)?;
    Ok(sender)
}

/// Class 1: a record whose operation row landed under an unknown outcome is
/// answered its committed value, the owner matches the store, and the next
/// record commits.
#[test]
fn a_record_whose_row_landed_under_an_unknown_outcome_is_answered_committed()
-> Result<(), Box<dyn Error>> {
    let (memory, fault, handler) = faulty(Write::Append, &format!("{STREAM_PREFIX}{FIRST}"))?;
    let sender = two_members(&handler, FIRST)?;
    fault.arm();
    let answered = dispatch(&handler, SENDER, record(FIRST, sender, 0x01));
    assert!(fault.struck(), "the record wrote no operation row");
    let next = dispatch(&handler, SENDER, record(FIRST, sender, 0x02));
    let stored = stored_rows(&memory, FIRST)?;
    let live = live_rows(&handler, FIRST)?;
    let answered = answered.map_err(|error| error.to_string());
    assert!(
        matches!(answered, Ok(ServerValue::RecordCommitted(_))),
        "a record whose row the store holds was answered {answered:?}, not its committed value"
    );
    assert!(
        matches!(next, Ok(ServerValue::RecordCommitted(_))),
        "the next record did not commit"
    );
    assert_eq!(
        live,
        Some(stored),
        "the live owner disagrees with the store"
    );
    Ok(())
}

/// Class 2: a record whose extension row landed under an unknown outcome is
/// answered its committed value, and the next record commits.
#[test]
fn a_record_whose_extension_row_landed_under_an_unknown_outcome_is_answered_committed()
-> Result<(), Box<dyn Error>> {
    let (memory, fault, handler) =
        faulty(Write::Append, &format!("{OUTBOX_STREAM_PREFIX}{FIRST}"))?;
    let sender = two_members(&handler, FIRST)?;
    fault.arm();
    let answered = dispatch(&handler, SENDER, record(FIRST, sender, 0x01));
    assert!(fault.struck(), "the record wrote no extension row");
    let next = dispatch(&handler, SENDER, record(FIRST, sender, 0x02));
    let stored = stored_rows(&memory, FIRST)?;
    let live = live_rows(&handler, FIRST)?;
    let answered = answered.map_err(|error| error.to_string());
    assert!(
        matches!(answered, Ok(ServerValue::RecordCommitted(_))),
        "a record whose extension row the store holds was answered {answered:?}, not its \
         committed value"
    );
    assert!(
        matches!(next, Ok(ServerValue::RecordCommitted(_))),
        "the next record did not commit"
    );
    assert_eq!(
        live,
        Some(stored),
        "the live owner disagrees with the store"
    );
    Ok(())
}

/// Class 3: a registry row that landed under an unknown outcome leaves the
/// next conversation creatable, and the first conversation's enrollment is
/// answered as the store settles it.
#[test]
fn a_registry_row_landed_under_an_unknown_outcome_leaves_the_next_conversation_creatable()
-> Result<(), Box<dyn Error>> {
    let (_memory, fault, handler) = faulty(Write::Append, REGISTRY_STREAM_KEY)?;
    fault.arm();
    let first = dispatch(&handler, SENDER, enrollment(FIRST, 0xE1));
    assert!(fault.struck(), "the first enrollment wrote no registry row");
    let second = dispatch(&handler, LATER, enrollment(SECOND, 0xE3));
    let first = first.map_err(|error| error.to_string());
    let second = second.map_err(|error| error.to_string());
    assert!(
        matches!(second, Ok(ServerValue::EnrollBound(_))),
        "after a registry row landed under an unknown outcome, the next conversation's creation \
         was answered {second:?}"
    );
    assert!(
        matches!(first, Ok(ServerValue::EnrollBound(_))),
        "the enrollment whose registry row the store holds was answered {first:?}"
    );
    Ok(())
}

/// Class 4: an observer row that landed under an unknown outcome leaves the
/// observer owner in step with its log: the enrollment that wrote it is
/// answered as the store settles it, and the next conversation is tracked.
#[test]
fn an_observer_row_landed_under_an_unknown_outcome_leaves_the_next_conversation_tracked()
-> Result<(), Box<dyn Error>> {
    let (_memory, fault, handler) = faulty(Write::AppendAll, OBSERVER_STREAM_KEY)?;
    fault.arm();
    let first = dispatch(&handler, SENDER, enrollment(FIRST, 0xE1));
    assert!(fault.struck(), "the first enrollment wrote no observer row");
    let second = dispatch(&handler, LATER, enrollment(SECOND, 0xE3));
    let first = first.map_err(|error| error.to_string());
    let second = second.map_err(|error| error.to_string());
    assert!(
        matches!(second, Ok(ServerValue::EnrollBound(_))),
        "after an observer row landed under an unknown outcome, the next conversation's \
         enrollment was answered {second:?}"
    );
    assert!(
        matches!(first, Ok(ServerValue::EnrollBound(_))),
        "the enrollment whose observer row the store holds was answered {first:?}"
    );
    Ok(())
}

/// What the client is told for one dispatch: the frame's discriminant, and
/// whether the connection closes after it.
fn told(outcome: &ParticipantDispatch) -> Result<String, Box<dyn Error>> {
    let (frame, closes) = match outcome {
        ParticipantDispatch::Respond(frame) => (frame, false),
        ParticipantDispatch::RespondThenClose(frame) => (frame, true),
        ParticipantDispatch::Fatal(_) => return Ok("no frame, then close".to_owned()),
        ParticipantDispatch::NotParticipant => return Ok("not a participant frame".to_owned()),
    };
    let mut bytes = vec![0; liminal::protocol::encoded_len(frame)?];
    let written = liminal::protocol::encode(frame, &mut bytes)?;
    bytes.truncate(written);
    let discriminant: [u8; 2] = bytes
        .get(14..16)
        .and_then(|discriminant| discriminant.try_into().ok())
        .ok_or("the response frame has no participant discriminant")?;
    let discriminant = u16::from_be_bytes(discriminant);
    Ok(if closes {
        format!("frame {discriminant:#06x}, then close")
    } else {
        format!("frame {discriminant:#06x}")
    })
}

/// Class 1 at the wire, when the store cannot say: a record whose row landed
/// under an unknown outcome that its read-back cannot settle is answered by
/// the named outcome-unknown frame (0x0128) on a connection that stays open,
/// never a bare close; its retry with the same token is answered its commit,
/// and the row is stored once.
#[test]
fn an_unsettled_record_is_answered_outcome_unknown_and_its_retry_its_commit()
-> Result<(), Box<dyn Error>> {
    let (memory, fault, handler) = faulty(Write::Append, &format!("{STREAM_PREFIX}{FIRST}"))?;
    let sender = two_members(&handler, FIRST)?;
    let before = stored_rows(&memory, FIRST)?;
    fault.arm_unsettled();
    let mut conversations = ParticipantConnectionConversations::default();
    let answered = dispatch_outcome(
        &handler,
        SENDER,
        &mut conversations,
        record(FIRST, sender, 0x01),
    )?;
    assert!(fault.struck(), "the record wrote no operation row");
    let answered = told(&answered)?;
    let retried = dispatch_outcome(
        &handler,
        SENDER,
        &mut conversations,
        record(FIRST, sender, 0x01),
    )?;
    let retried = match retried {
        ParticipantDispatch::Respond(frame) => match decode_server_value(&frame)? {
            ServerValue::RecordCommitted(committed) => {
                format!("committed at {:?}", committed.delivery_seq())
            }
            other => format!("{other:?}"),
        },
        other => told(&other)?,
    };
    let stored = stored_rows(&memory, FIRST)? - before;
    assert_eq!(
        (answered.as_str(), stored),
        ("frame 0x0128", 1),
        "a record the store could not settle was not answered by the outcome-unknown frame, or was stored other than once"
    );
    assert!(
        retried.starts_with("committed at"),
        "the same-token retry of an unsettled record was answered {retried}, not its commit"
    );
    Ok(())
}

/// Answers `request` under an unsettled outcome, then retries it with the
/// same token on the same connection: what the client is told first, the
/// retry's answer, and the operation rows the two wrote, counted on the store
/// beneath the fault.
fn unsettled_then_retried(
    handler: &ProductionParticipantHandler,
    (memory, fault): (&Arc<dyn DurableStore>, &UnknownOutcomeStore),
    connection: ConnectionIncarnation,
    request: &ClientRequest,
) -> Result<(String, String, u64), Box<dyn Error>> {
    let before = stored_rows(memory, FIRST)?;
    let mut conversations = ParticipantConnectionConversations::default();
    fault.arm_unsettled();
    let first = dispatch_outcome(handler, connection, &mut conversations, request.clone())?;
    if !fault.struck() {
        return Err("the request wrote no operation row".into());
    }
    let first = told(&first)?;
    let retried = dispatch_outcome(handler, connection, &mut conversations, request.clone())?;
    let retried = match retried {
        ParticipantDispatch::Respond(frame) => {
            format!("{:?}", decode_server_value(&frame)?.discriminant())
        }
        other => told(&other)?,
    };
    Ok((first, retried, stored_rows(memory, FIRST)? - before))
}

/// A bound member of `FIRST` on `connection`.
fn bound(
    handler: &ProductionParticipantHandler,
    connection: ConnectionIncarnation,
    token: u8,
) -> Result<EnrollBound, Box<dyn Error>> {
    let value = dispatch(handler, connection, enrollment(FIRST, token))?;
    let ServerValue::EnrollBound(bound) = value else {
        return Err(format!("enrollment {token:#x} did not bind: {value:?}").into());
    };
    Ok(bound)
}

fn detach_request(member: &EnrollBound, token: u8) -> ClientRequest {
    ClientRequest::Detach(DetachRequest {
        conversation_id: FIRST,
        participant_id: member.participant_id(),
        capability_generation: member.capability_generation(),
        detach_attempt_token: DetachAttemptToken::new([token; 16]),
    })
}

/// Each token-carrying operation whose operation row landed under an outcome
/// its read-back cannot settle is told the outcome-unknown frame on an open
/// connection, and the same request with the same token, on that connection,
/// is answered the operation's original commit: a binding by its receipt's
/// replay (`Bound`), a detach or leave by its commit. Either way the
/// operation's row is stored once: the retry writes none.
#[test]
fn an_unsettled_enrollment_is_answered_outcome_unknown_and_its_retry_its_binding()
-> Result<(), Box<dyn Error>> {
    let (memory, fault, handler) = faulty(Write::Append, &format!("{STREAM_PREFIX}{FIRST}"))?;
    bound(&handler, SENDER, 0xE1)?;
    let answered = unsettled_then_retried(
        &handler,
        (&memory, &fault),
        RECIPIENT,
        &enrollment(FIRST, 0xE2),
    )?;
    assert_eq!(
        answered,
        ("frame 0x0128".to_owned(), "Bound".to_owned(), 1),
        "an unsettled enrollment was not told outcome unknown, then bound on its retry, with its \
         operation stored once"
    );
    Ok(())
}

#[test]
fn an_unsettled_credential_attach_is_answered_outcome_unknown_and_its_retry_its_binding()
-> Result<(), Box<dyn Error>> {
    let (memory, fault, handler) = faulty(Write::Append, &format!("{STREAM_PREFIX}{FIRST}"))?;
    let member = bound(&handler, SENDER, 0xE1)?;
    let detached = dispatch(&handler, SENDER, detach_request(&member, 0xD1))?;
    if !matches!(detached, ServerValue::DetachCommitted(_)) {
        return Err(format!("the member did not detach: {detached:?}").into());
    }
    let attach = ClientRequest::CredentialAttach(CredentialAttachRequest {
        conversation_id: FIRST,
        participant_id: member.participant_id(),
        capability_generation: member.capability_generation(),
        attach_secret: member.attach_secret(),
        attach_attempt_token: AttachAttemptToken::new([0xA1; 16]),
        accept_marker_delivery_seq: None,
    });
    let answered = unsettled_then_retried(&handler, (&memory, &fault), LATER, &attach)?;
    assert_eq!(
        answered,
        ("frame 0x0128".to_owned(), "Bound".to_owned(), 1),
        "an unsettled credential attach was not told outcome unknown, then bound on its retry, with its \
         operation stored once"
    );
    Ok(())
}

#[test]
fn an_unsettled_detach_is_answered_outcome_unknown_and_its_retry_its_commit()
-> Result<(), Box<dyn Error>> {
    let (memory, fault, handler) = faulty(Write::Append, &format!("{STREAM_PREFIX}{FIRST}"))?;
    let member = bound(&handler, SENDER, 0xE1)?;
    let answered = unsettled_then_retried(
        &handler,
        (&memory, &fault),
        SENDER,
        &detach_request(&member, 0xD1),
    )?;
    assert_eq!(
        answered,
        ("frame 0x0128".to_owned(), "DetachCommitted".to_owned(), 1),
        "an unsettled detach was not told outcome unknown, then committed on its retry, with its \
         operation stored once"
    );
    Ok(())
}

#[test]
fn an_unsettled_leave_is_answered_outcome_unknown_and_its_retry_its_commit()
-> Result<(), Box<dyn Error>> {
    let (memory, fault, handler) = faulty(Write::Append, &format!("{STREAM_PREFIX}{FIRST}"))?;
    let member = bound(&handler, SENDER, 0xE1)?;
    let leave = ClientRequest::Leave(LeaveRequest {
        conversation_id: FIRST,
        participant_id: member.participant_id(),
        capability_generation: member.capability_generation(),
        attach_secret: member.attach_secret(),
        leave_attempt_token: LeaveAttemptToken::new([0x1E; 16]),
    });
    let answered = unsettled_then_retried(&handler, (&memory, &fault), SENDER, &leave)?;
    assert_eq!(
        answered,
        ("frame 0x0128".to_owned(), "LeaveCommitted".to_owned(), 1),
        "an unsettled leave was not told outcome unknown, then committed on its retry, with its \
         operation stored once"
    );
    Ok(())
}

#[test]
fn a_fenced_cas_never_reaches_its_inner_store_and_recovery_retains_errors()
-> Result<(), Box<dyn Error>> {
    let memory: Arc<dyn DurableStore> = Arc::new(MemoryStore::default());
    let cursor = "fence:cursor";
    let source = "fence:source";
    block_on(memory.cas(cursor, 0, 7))??;
    let fault = UnknownOutcomeStore::new(&memory, Write::AppendAbsentIo, source);
    fault.arm();
    let failure = block_on(fault.append(source, vec![1], 0))?;
    assert!(matches!(
        failure,
        Err(DurabilityError::StoreError(haematite::ApiError::Storage(
            haematite::DatabaseError::IoError(_)
        )))
    ));
    assert!(fault.struck(), "the fence injection was not reached");
    assert!(
        fault.fenced.load(Ordering::SeqCst),
        "the fake was not fenced"
    );
    assert!(block_on(memory.read_from(source, 0, 1))??.is_empty());
    assert!(matches!(
        block_on(fault.read_value(cursor))?,
        Err(DurabilityError::OutcomeUnknown { pid: 1 })
    ));
    let fenced_cas = block_on(fault.cas(cursor, 7, 8))?;
    assert!(fault.fenced.load(Ordering::SeqCst));
    let stored = block_on(memory.read_value(cursor))??;
    assert_eq!(
        stored,
        Some(7),
        "a fenced CAS reached the delegated mutation"
    );
    assert_eq!(fault.cas_submissions.load(Ordering::SeqCst), 0);
    assert!(matches!(
        fenced_cas,
        Err(DurabilityError::OutcomeUnknown { pid: 1 })
    ));

    fault.recover();
    assert!(!fault.fenced.load(Ordering::SeqCst));
    let original_error = block_on(fault.cas(cursor, 0, 9))?;
    assert!(matches!(
        original_error,
        Err(DurabilityError::CursorRegression {
            stored: 7,
            attempted: 0
        })
    ));
    assert_eq!(fault.cas_submissions.load(Ordering::SeqCst), 1);
    assert_eq!(block_on(memory.read_value(cursor))??, Some(7));
    block_on(fault.cas(cursor, 7, 8))??;
    assert_eq!(fault.cas_submissions.load(Ordering::SeqCst), 2);
    assert_eq!(block_on(memory.read_value(cursor))??, Some(8));
    Ok(())
}