liminal-server 0.24.9

//! Eleven-oracle W3 acceptance additions (items three through eleven).

use std::error::Error;
use std::sync::Arc;

use liminal::durability::DurableStore;
use liminal::durability::bridge::block_on;
use liminal_protocol::wire::{
    ClientRequest, ConnectionIncarnation, InvalidObserverEpoch, ObserverRecoveryHandshake,
    ObserverRefusal, ParticipantDelivery, ParticipantRecord, ServerValue,
};

use crate::server::participant::{ParticipantOfferedProgress, ParticipantSemanticHandler};

use super::ProductionParticipantHandler;
use super::log::{OperationLog, STREAM_PREFIX, legacy_rows};
use super::outbox::{ConversationOutbox, ConversationOutboxLimits};
use super::outbox_log::{
    OUTBOX_SCHEMA_VERSION, OUTBOX_STREAM_PREFIX, OutboxLog, OutboxLogError, OutboxRow,
    ProducedBatch, ProducedSourceKind, ProjectedRecord, RestorePageAccountingGuard,
    UNIT2_OUTBOX_RESTORE_BATCH_ROWS, encode_row,
};
use super::outbox_replay::{RestoreError, RestoreError::Extension};
use super::restore_checkpoint_locator::{CheckpointLocator, checkpoint_conversation_route};
use super::restore_checkpoint_rebuild::{BootCheckpointError, restore_at_boot};
use super::state::ConversationAuthority;
use super::tests::{dispatch, test_participant_config};
use super::tests_marker_ack::{commit_exact_marker_ack, record_exact_marker_offer};
use super::tests_marker_ack_fixture::{
    MarkerFixture, marker_fixture_config, marker_protocol_snapshot,
    prepare_marker_fixture_with_store,
};
use super::tests_memory_store::MemoryStore;
use super::tests_w3_restore_fixture::{
    CONVERSATION, CursorFaultArm, CutPageStore, OutboxAppendFaultStore, PhaseAwareStore,
    ShortPageStore, append_payload, duplicate_extension_to, enrollment, enrollment_outcome_unknown,
    extension_key, new_store, operation_key, seed_enrollment, stream_payloads,
};

/// What a load refused by its durable log says: the request seam's load is the
/// boot's checkpoint restore (`restore_at_boot`), whose refusal carries the
/// typed restore error after these words. The words a refused extension
/// stream once carried, "participant Unit 2 extension log failed:", belong to
/// the frozen aggregate reference alone now.
const CHECKPOINT_RESTORE_FAILED: &str = "participant checkpoint restore failed: ";

/// The extension decode refusal a restore over a malformed suffix answers,
/// as the load reports it.
fn malformed_suffix_refusal() -> String {
    format!("{CHECKPOINT_RESTORE_FAILED}Unit 2 extension row ended before row_kind")
}

/// The pages one pass over `rows` extension rows loads: full pages and one
/// short page; the confirming empty read decodes nothing and is not a page.
fn pages_for(rows: usize) -> usize {
    rows.div_ceil(UNIT2_OUTBOX_RESTORE_BATCH_ROWS)
}

fn more_than_two_pages() -> Result<usize, Box<dyn Error>> {
    UNIT2_OUTBOX_RESTORE_BATCH_ROWS
        .checked_mul(2)
        .and_then(|rows| rows.checked_add(1))
        .ok_or_else(|| "W3 fixture row count overflowed".into())
}

fn clear_owner(handler: &ProductionParticipantHandler) -> Result<(), Box<dyn Error>> {
    let cell = handler.cell(CONVERSATION)?;
    let mut owner = cell.lock().map_err(|_| "W3 owner lock poisoned")?;
    *owner = None;
    drop(owner);
    Ok(())
}

fn owner_is_present(handler: &ProductionParticipantHandler) -> Result<bool, Box<dyn Error>> {
    let cell = handler.cell(CONVERSATION)?;
    let owner = cell.lock().map_err(|_| "W3 owner lock poisoned")?;
    let present = owner.is_some();
    drop(owner);
    Ok(present)
}

fn restore_limits() -> Result<ConversationOutboxLimits, Box<dyn Error>> {
    let config = test_participant_config();
    Ok(ConversationOutboxLimits::try_new(
        config.max_retained_record_rows,
        config.identity_slots,
    )?)
}

fn ordered_pushes(
    handler: &ProductionParticipantHandler,
    connection: ConnectionIncarnation,
    conversation_id: u64,
) -> Result<Vec<ParticipantDelivery>, Box<dyn Error>> {
    let mut offered = None;
    let mut pushes = Vec::new();
    let limit = usize::try_from(marker_fixture_config().max_retained_record_rows)?;
    for _ in 0..limit {
        let Some(publication) = handler.next_publication(connection, conversation_id, offered)?
        else {
            break;
        };
        offered = Some(ParticipantOfferedProgress {
            binding_epoch: publication.binding_epoch,
            through_seq: publication.delivery_seq(),
        });
        pushes.push(publication.delivery);
    }
    Ok(pushes)
}

#[test]
fn outbox_restore_peak_decoded_rows_never_exceeds_one_page() -> Result<(), Box<dyn Error>> {
    let store = new_store()?;
    let handler = seed_enrollment(&store)?;
    drop(handler);
    let row_count = more_than_two_pages()?;
    duplicate_extension_to(&store, row_count)?;

    let accounting = RestorePageAccountingGuard::start();
    let restored =
        ProductionParticipantHandler::new_restored(Arc::clone(&store), test_participant_config())?;
    let snapshot = accounting.snapshot();
    assert_eq!(
        snapshot.cursors_started, 1,
        "the extension is read in one pass"
    );
    assert_eq!(snapshot.pages, pages_for(row_count));
    assert!(snapshot.peak_rows <= UNIT2_OUTBOX_RESTORE_BATCH_ROWS);
    assert_eq!(snapshot.current_rows, 0);
    assert!(!snapshot.cursor_overlap_observed);
    assert!(!snapshot.counter_overflow_observed);
    assert!(owner_is_present(&restored)?);
    assert_eq!(stream_payloads(&store, &extension_key())?.len(), row_count);
    Ok(())
}

#[test]
fn midstream_outbox_decode_failure_preserves_typed_error_and_publishes_no_state()
-> Result<(), Box<dyn Error>> {
    let store = new_store()?;
    let handler = seed_enrollment(&store)?;
    duplicate_extension_to(&store, UNIT2_OUTBOX_RESTORE_BATCH_ROWS)?;
    append_payload(
        &store,
        &extension_key(),
        vec![OUTBOX_SCHEMA_VERSION],
        u64::try_from(UNIT2_OUTBOX_RESTORE_BATCH_ROWS)?,
    )?;
    let before = stream_payloads(&store, &extension_key())?;
    clear_owner(&handler)?;

    // The validation pass reads the stream where its commits put it: the
    // flat prefix, then the routed continuation on the conversation's route,
    // the cursor a checkpoint-less restore reads. The flat cursor alone saw
    // none of these rows.
    let log = OutboxLog::new(Arc::clone(&store), CONVERSATION);
    let route = checkpoint_conversation_route(CONVERSATION);
    let mut validation = log.restore_rebuild_cursor(&route);
    let Err(typed) = block_on(validation.drain())? else {
        return Err("validation unexpectedly accepted a malformed suffix".into());
    };
    drop(validation);
    assert!(matches!(
        typed,
        OutboxLogError::UnexpectedEnd { field: "row_kind" }
    ));
    let operation_log = OperationLog::new(Arc::clone(&store), CONVERSATION);
    let error = handler
        .replay_and_repair(CONVERSATION, &operation_log)
        .err()
        .ok_or("production restore unexpectedly accepted a malformed suffix")?;
    let message = error.to_string();
    assert!(
        message.contains(&malformed_suffix_refusal()),
        "the restore did not refuse by the malformed row: {message}"
    );
    assert!(!message.contains("participant production operation failed"));
    assert!(!owner_is_present(&handler)?);
    assert_eq!(stream_payloads(&store, &extension_key())?, before);
    Ok(())
}

#[test]
fn restore_retained_authority_counts_are_measured_and_ledgered() -> Result<(), Box<dyn Error>> {
    let participant_id = 7_u64;
    let pair_count = UNIT2_OUTBOX_RESTORE_BATCH_ROWS
        .checked_add(1)
        .ok_or("retained fixture pair count overflowed")?;
    let mut rows = Vec::with_capacity(
        pair_count
            .checked_mul(2)
            .ok_or("retained fixture row capacity overflowed")?,
    );
    let mut physical_sequence = 0_u64;
    for index in 0..pair_count {
        let index = u64::try_from(index)?;
        let delivery_seq = index.checked_add(1).ok_or("delivery sequence overflowed")?;
        let produced_source = index.checked_mul(2).ok_or("source sequence overflowed")?;
        let ack_source = produced_source
            .checked_add(1)
            .ok_or("ack source overflowed")?;
        let record = ProjectedRecord::try_new(
            CONVERSATION,
            delivery_seq,
            ParticipantRecord::OrdinaryRecord {
                sender_participant_id: participant_id
                    .checked_add(1)
                    .ok_or("sender participant overflowed")?,
                payload: index.to_be_bytes().to_vec(),
            },
            vec![participant_id],
            participant_id.checked_add(1),
        )?;
        rows.push((
            physical_sequence,
            OutboxRow::Produced(ProducedBatch::new(
                produced_source,
                ProducedSourceKind::RecordAdmission,
                vec![record],
            )),
        ));
        physical_sequence = physical_sequence
            .checked_add(1)
            .ok_or("physical overflowed")?;
        rows.push((
            physical_sequence,
            OutboxRow::AckAdvanced {
                source_log_sequence: ack_source,
                participant_id,
                through_seq: delivery_seq,
            },
        ));
        physical_sequence = physical_sequence
            .checked_add(1)
            .ok_or("physical overflowed")?;
    }
    assert!(
        rows.len()
            > more_than_two_pages()?
                .checked_sub(1)
                .ok_or("page underflow")?
    );
    let owner = ConversationOutbox::restore(CONVERSATION, rows, restore_limits()?)?;
    assert_eq!(owner.live_record_count(), 0);
    assert_eq!(owner.live_recipient_obligation_count(), 0);
    assert_eq!(owner.charged_bytes(), 0);

    // "Retained owned bytes" means logical deep-owned element bytes: map keys,
    // tuple scalars, set elements, and canonical Vec payload lengths. It excludes
    // allocator/node overhead and spare capacity so W7 reruns are comparable.
    // Every record is acknowledged and reclaimed, so the owner holds only the
    // latest source batch, the one ack source that set the frontier, and no
    // obligation sequence, whatever the number of pairs.
    let measured = owner.retained_authority_measurements()?;
    assert_eq!(measured.source_batch_count, 1);
    assert_eq!(measured.ack_source_count, 1);
    assert_eq!(measured.obligation_participant_count, 0);
    assert_eq!(measured.obligation_sequence_count, 0);
    assert!(measured.source_batch_owned_bytes > 0);
    assert!(measured.ack_source_owned_bytes > 0);
    assert_eq!(measured.all_obligations_owned_bytes, 0);
    eprintln!(
        "{{\"w7_retained\":{{\"source_batches\":{{\"count\":{},\"owned_bytes\":{}}},\"ack_sources\":{{\"count\":{},\"owned_bytes\":{}}},\"all_obligations\":{{\"participants\":{},\"sequences\":{},\"owned_bytes\":{}}}}}}}",
        measured.source_batch_count,
        measured.source_batch_owned_bytes,
        measured.ack_source_count,
        measured.ack_source_owned_bytes,
        measured.obligation_participant_count,
        measured.obligation_sequence_count,
        measured.all_obligations_owned_bytes,
    );
    Ok(())
}

#[test]
fn observer_recovery_restores_absent_owner_page_wise_with_exact_errors()
-> Result<(), Box<dyn Error>> {
    let valid_store = new_store()?;
    let valid = seed_enrollment(&valid_store)?;
    let valid_rows = more_than_two_pages()?;
    duplicate_extension_to(&valid_store, valid_rows)?;
    clear_owner(&valid)?;
    let accounting = RestorePageAccountingGuard::start();
    let value = dispatch(
        &valid,
        ConnectionIncarnation::new(CONVERSATION, 2),
        ClientRequest::ObserverRecovery(ObserverRecoveryHandshake {
            observer_refusals: vec![ObserverRefusal {
                conversation_id: CONVERSATION,
                refused_epoch: 1,
            }],
        }),
    )?;
    assert!(matches!(
        value,
        ServerValue::InvalidObserverEpoch(InvalidObserverEpoch::EpochAhead { .. })
    ));
    assert!(owner_is_present(&valid)?);
    let snapshot = accounting.snapshot();
    assert_eq!(
        snapshot.cursors_started, 1,
        "the extension is read in one pass"
    );
    assert_eq!(snapshot.pages, pages_for(valid_rows));
    assert!(!snapshot.cursor_overlap_observed);
    assert!(!snapshot.counter_overflow_observed);
    drop(accounting);

    let malformed_store = new_store()?;
    let malformed = seed_enrollment(&malformed_store)?;
    duplicate_extension_to(&malformed_store, UNIT2_OUTBOX_RESTORE_BATCH_ROWS)?;
    append_payload(
        &malformed_store,
        &extension_key(),
        vec![OUTBOX_SCHEMA_VERSION],
        u64::try_from(UNIT2_OUTBOX_RESTORE_BATCH_ROWS)?,
    )?;
    clear_owner(&malformed)?;
    let before = stream_payloads(&malformed_store, &extension_key())?;
    let error = dispatch(
        &malformed,
        ConnectionIncarnation::new(CONVERSATION, 3),
        ClientRequest::ObserverRecovery(ObserverRecoveryHandshake {
            observer_refusals: vec![ObserverRefusal {
                conversation_id: CONVERSATION,
                refused_epoch: 1,
            }],
        }),
    )
    .err()
    .ok_or("malformed observer recovery unexpectedly classified")?;
    let message = error.to_string();
    assert!(
        message.contains(&malformed_suffix_refusal()),
        "the observer recovery's load did not refuse by the malformed row: {message}"
    );
    assert!(!message.contains("participant production operation failed"));
    assert!(!owner_is_present(&malformed)?);
    assert_eq!(stream_payloads(&malformed_store, &extension_key())?, before);
    Ok(())
}

#[test]
fn later_page_decode_failure_takes_precedence_over_earlier_semantic_conflict()
-> Result<(), Box<dyn Error>> {
    let store = new_store()?;
    let handler = seed_enrollment(&store)?;
    let valid_payload = stream_payloads(&store, &extension_key())?
        .first()
        .cloned()
        .ok_or("seed extension missing")?;
    let conflict = encode_row(&OutboxRow::AckAdvanced {
        source_log_sequence: 1,
        participant_id: 0,
        through_seq: 1,
    })?;
    append_payload(&store, &extension_key(), conflict, 1)?;
    let mut sequence = 2_u64;
    let page = u64::try_from(UNIT2_OUTBOX_RESTORE_BATCH_ROWS)?;
    while sequence < page {
        append_payload(&store, &extension_key(), valid_payload.clone(), sequence)?;
        sequence = sequence
            .checked_add(1)
            .ok_or("dual-fault sequence overflowed")?;
    }
    append_payload(&store, &extension_key(), vec![OUTBOX_SCHEMA_VERSION], page)?;
    clear_owner(&handler)?;
    let log = OperationLog::new(Arc::clone(&store), CONVERSATION);
    let error = handler
        .replay_and_repair(CONVERSATION, &log)
        .err()
        .ok_or("dual-fault restore unexpectedly succeeded")?;
    let message = error.to_string();
    assert!(
        message.contains(&malformed_suffix_refusal()),
        "the later decode failure did not take precedence: {message}"
    );
    assert!(!message.contains("projection at physical sequence"));
    assert!(!message.contains("participant production operation failed"));
    assert!(!owner_is_present(&handler)?);
    Ok(())
}

/// Extends the fixture's one same-head group beyond a page with exact
/// idempotent copies of its Produced projection. The two `MarkerAck` commits
/// that follow keep this unchanged base head and receive distinct physical
/// extension sequences.
fn extend_same_head_group(
    fixture: &MarkerFixture,
    outbox_log: &OutboxLog,
    projection: &OutboxRow,
) -> Result<(), Box<dyn Error>> {
    let cell = fixture
        .handler
        .cell(fixture.marker_delivery.conversation_id)?;
    let mut owner = cell
        .lock()
        .map_err(|_| "marker group owner lock poisoned")?;
    let authority = owner.as_mut().ok_or("marker group owner was absent")?;
    let outbox = authority
        .outbox
        .as_mut()
        .ok_or("marker group outbox was absent")?;
    for _ in 0..UNIT2_OUTBOX_RESTORE_BATCH_ROWS {
        let physical_sequence = outbox.next_extension_sequence();
        block_on(outbox_log.append(projection, physical_sequence))??;
        outbox.apply_row(physical_sequence, projection.clone())?;
    }
    drop(owner);
    Ok(())
}

/// One committed stream by key, as the restore reads it.
type CommittedStream = (String, Vec<Vec<u8>>);

/// The fixture conversation's operation and extension streams as their
/// commits left them (`extension` is the latter, already read), after
/// proving the fixture holds no checkpoint: a restore over a checkpoint would
/// not read the extension from its first page.
fn committed_streams_without_checkpoint(
    fixture: &MarkerFixture,
    extension: &[Vec<u8>],
) -> Result<[CommittedStream; 2], Box<dyn Error>> {
    let conversation_id = fixture.marker_delivery.conversation_id;
    let route = checkpoint_conversation_route(conversation_id);
    let locator = block_on(
        fixture
            .store
            .get_routed(&route, &CheckpointLocator::routed_key(conversation_id)),
    )??;
    if locator.is_some() {
        return Err(
            "the marker fixture holds a checkpoint, so a restore would not read the extension \
             from its first page"
                .into(),
        );
    }
    let operation_key = format!("{STREAM_PREFIX}{conversation_id}");
    let operation = stream_payloads(&fixture.store, &operation_key)?;
    Ok([
        (operation_key, operation),
        (
            format!("{OUTBOX_STREAM_PREFIX}{conversation_id}"),
            extension.to_vec(),
        ),
    ])
}

#[test]
fn same_base_head_group_straddling_page_cut_replays_without_repair() -> Result<(), Box<dyn Error>> {
    // A pin of the flat read contract: a cut page is a flat read that comes
    // back short, while a routed read is bounded by its head and never pages
    // short. The fixture commits its rows routed, so each restore below reads
    // its own copy of the store in which the conversation's two streams hold
    // those same committed rows flat, as a store written before pruning holds
    // them. A copy per cut, because a checkpoint-less restore publishes the
    // checkpoint it rebuilt, and a restore over that checkpoint would read no
    // extension page at all.
    let memory = Arc::new(MemoryStore::default());
    let fixture = prepare_marker_fixture_with_store(Arc::clone(&memory) as Arc<dyn DurableStore>)?;
    let conversation_id = fixture.marker_delivery.conversation_id;
    let extension_key = format!("{OUTBOX_STREAM_PREFIX}{conversation_id}");
    let outbox_log = OutboxLog::new(Arc::clone(&fixture.store), conversation_id);
    let rows_before_group = block_on(outbox_log.read_all())??;
    let (_, expected_projection) = rows_before_group
        .last()
        .cloned()
        .ok_or("marker fixture extension was empty")?;
    if !matches!(expected_projection, OutboxRow::Produced(_)) {
        return Err("marker group did not start with its exact Produced row".into());
    }
    let group_head = expected_projection
        .base_log_head()
        .ok_or("marker group Produced boundary overflowed")?;
    let group_start = rows_before_group
        .len()
        .checked_sub(1)
        .ok_or("marker group start underflowed")?;

    extend_same_head_group(&fixture, &outbox_log, &expected_projection)?;

    record_exact_marker_offer(&fixture)?;
    let stored_marker = commit_exact_marker_ack(&fixture)?;
    assert_eq!(stored_marker.base_log_head, group_head);

    let rows = block_on(outbox_log.read_all())??;
    let Some((physical, OutboxRow::MarkerAckCommitted(marker))) = rows.last() else {
        return Err("same-head group did not end in MarkerAckCommitted".into());
    };
    assert_eq!(marker.base_log_head, group_head);
    assert_eq!(marker.extension_sequence, *physical);

    let before = stream_payloads(&fixture.store, &extension_key)?;
    let live_target = marker_protocol_snapshot(
        &fixture.handler,
        conversation_id,
        fixture.record_participant,
    )?;
    let live_catchup = marker_protocol_snapshot(
        &fixture.handler,
        conversation_id,
        fixture.catchup_participant,
    )?;
    let live_target_pushes =
        ordered_pushes(&fixture.handler, fixture.record_connection, conversation_id)?;
    let live_catchup_pushes = ordered_pushes(
        &fixture.handler,
        fixture.catchup_connection,
        conversation_id,
    )?;

    let route = checkpoint_conversation_route(conversation_id);
    let committed = committed_streams_without_checkpoint(&fixture, &before)?;

    for cut in 1..UNIT2_OUTBOX_RESTORE_BATCH_ROWS {
        let flat: Arc<dyn DurableStore> =
            Arc::new(memory.copy_with_flat_streams(&route, &committed)?);
        let cut_store: Arc<dyn DurableStore> = Arc::new(CutPageStore::new(
            Arc::clone(&flat),
            u64::try_from(group_start)?,
            cut,
        ));
        let restored = ProductionParticipantHandler::new(cut_store, marker_fixture_config())?;
        assert_eq!(
            marker_protocol_snapshot(&restored, conversation_id, fixture.record_participant)?,
            live_target
        );
        assert_eq!(
            marker_protocol_snapshot(&restored, conversation_id, fixture.catchup_participant)?,
            live_catchup
        );
        assert_eq!(
            ordered_pushes(&restored, fixture.record_connection, conversation_id)?,
            live_target_pushes
        );
        assert_eq!(
            ordered_pushes(&restored, fixture.catchup_connection, conversation_id)?,
            live_catchup_pushes
        );
        assert!(
            restored
                .observer
                .lock()
                .map_err(|_| "observer owner lock poisoned")?
                .is_some()
        );
        assert_eq!(
            stream_payloads(&flat, &extension_key)?,
            before,
            "the restore across a cut at {cut} wrote a repair row"
        );
    }
    assert_eq!(stream_payloads(&fixture.store, &extension_key)?, before);
    Ok(())
}

#[test]
fn short_nonempty_pages_with_remaining_rows_do_not_terminate_restore() -> Result<(), Box<dyn Error>>
{
    // A pin of the flat read contract: a routed read is bounded by its head
    // and never pages short, so the rows are written flat on purpose, as a
    // store written before pruning holds them, and the restore walks them.
    let _legacy = legacy_rows::arm();
    let inner = new_store()?;
    let seeded = seed_enrollment(&inner)?;
    drop(seeded);
    let row_count = more_than_two_pages()?;
    duplicate_extension_to(&inner, row_count)?;
    let expected_bytes = stream_payloads(&inner, &extension_key())?;

    // The boot over the short-page store is a checkpoint-less rebuild; it
    // publishes the checkpoint it rebuilt through that store before the
    // owner is ready, so the second boot below restores from that checkpoint.
    let short = Arc::new(ShortPageStore::new(Arc::clone(&inner)));
    let short_store: Arc<dyn DurableStore> = short.clone();
    let restored = ProductionParticipantHandler::new(short_store, test_participant_config())?;
    assert!(owner_is_present(&restored)?);
    // one traversal, ended by exactly one confirming empty read
    assert_eq!(short.empty_reads_per_traversal()?, vec![1]);
    assert_eq!(stream_payloads(&inner, &extension_key())?, expected_bytes);
    drop(restored);

    let production_shape =
        ProductionParticipantHandler::new(Arc::clone(&inner), test_participant_config())?;
    assert!(owner_is_present(&production_shape)?);
    assert_eq!(stream_payloads(&inner, &extension_key())?, expected_bytes);
    Ok(())
}

struct RepairFailureResult {
    error: String,
    extension_bytes: Vec<Vec<u8>>,
    owner_present: bool,
}

/// A store whose second enrollment committed its operation row and lost its
/// extension row, then a later base row that cannot apply, restored by the
/// frozen aggregate reference or by the production load.
///
/// A live commit stages both rows in one routed batch, so only a store
/// written before pruning, flat and one append per row, can hold a lost
/// extension row: the rows are written under the legacy-rows arm, which also
/// keeps them where the frozen reference, a flat reader, reads them.
fn repair_then_later_failure(
    aggregate_reference: bool,
) -> Result<RepairFailureResult, Box<dyn Error>> {
    let legacy = legacy_rows::arm();
    let inner = new_store()?;
    let faults = Arc::new(OutboxAppendFaultStore::new(Arc::clone(&inner)));
    let store: Arc<dyn DurableStore> = faults.clone();
    let handler = seed_enrollment(&store)?;
    faults.set_fail(true);
    let lost = dispatch(
        &handler,
        ConnectionIncarnation::new(CONVERSATION, 2),
        enrollment(2),
    );
    faults.set_fail(false);
    // The operation row was issued before its extension row failed, so the
    // live path answers the enrollment's outcome as unknown.
    let lost = lost?;
    assert!(
        enrollment_outcome_unknown(&lost, 2),
        "the lost extension row did not leave the enrollment's outcome unknown: {lost:?}"
    );
    assert!(!owner_is_present(&handler)?);
    let base = stream_payloads(&store, &operation_key())?;
    let later_failure = base.first().cloned().ok_or("base stream missing genesis")?;
    append_payload(
        &store,
        &operation_key(),
        later_failure,
        u64::try_from(base.len())?,
    )?;
    drop(legacy);
    let before = stream_payloads(&store, &extension_key())?;
    assert_eq!(before.len(), 1);
    let log = OperationLog::new(Arc::clone(&store), CONVERSATION);
    let result = if aggregate_reference {
        handler.replay_aggregate_reference(CONVERSATION, &log)
    } else {
        handler.replay_and_repair(CONVERSATION, &log)
    };
    let error = result
        .err()
        .ok_or("later invalid base unexpectedly replayed")?;
    let after = stream_payloads(&store, &extension_key())?;
    Ok(RepairFailureResult {
        error: error.to_string(),
        extension_bytes: after,
        owner_present: owner_is_present(&handler)?,
    })
}

#[test]
fn missing_tail_repair_persists_before_later_base_failure_byte_identically()
-> Result<(), Box<dyn Error>> {
    let reference = repair_then_later_failure(true)?;
    let w3 = repair_then_later_failure(false)?;
    // Each seam words the refusal its own way: the frozen reference as a
    // production operation (a base failure, not an extension one), the load
    // as a checkpoint restore. The refusal after those words is the same.
    let reference_refusal = reference
        .error
        .strip_prefix(
            "participant semantic service failed: participant production operation failed: ",
        )
        .ok_or_else(|| {
            format!(
                "the reference refused as no base failure: {}",
                reference.error
            )
        })?;
    let w3_refusal = w3
        .error
        .strip_prefix(&format!(
            "participant semantic service failed: {CHECKPOINT_RESTORE_FAILED}"
        ))
        .ok_or_else(|| format!("the load refused as no checkpoint restore: {}", w3.error))?;
    assert_eq!(w3_refusal, reference_refusal);
    assert_eq!(w3.extension_bytes, reference.extension_bytes);
    assert_eq!(w3.extension_bytes.len(), 2);
    assert!(!reference.owner_present && !w3.owner_present);
    Ok(())
}

fn direct_w3_error(store: Arc<dyn DurableStore>) -> Result<RestoreError, Box<dyn Error>> {
    let outbox_log = OutboxLog::new(Arc::clone(&store), CONVERSATION);
    let operation_log = OperationLog::new(store, CONVERSATION);
    block_on(ConversationAuthority::replay_rewriting(
        CONVERSATION,
        &operation_log,
        &outbox_log,
        &test_participant_config(),
        restore_limits()?,
        None,
    ))?
    .err()
    .ok_or_else(|| "armed direct W3 restore unexpectedly succeeded".into())
}

/// The typed refusal of the production load itself, `restore_at_boot`, the
/// load a boot and a request's first touch both run: over this store, with no
/// checkpoint locator, the full rebuild that walks the extension from zero.
fn production_load_error(
    store: &Arc<dyn DurableStore>,
) -> Result<BootCheckpointError, Box<dyn Error>> {
    let outbox_log = OutboxLog::new(Arc::clone(store), CONVERSATION);
    let operation_log = OperationLog::new(Arc::clone(store), CONVERSATION);
    block_on(restore_at_boot(
        store.as_ref(),
        &operation_log,
        &outbox_log,
        &test_participant_config(),
        restore_limits()?,
    ))?
    .err()
    .ok_or_else(|| "armed production load unexpectedly succeeded".into())
}

#[test]
fn extension_cursor_failure_maps_through_extension_log_error_and_drops_staged_state()
-> Result<(), Box<dyn Error>> {
    // NonEof and EmptyEof are pins of the flat read contract: a routed read
    // is bounded by its head and has no empty read to confirm an end, so the
    // seed writes its rows flat on purpose and the restore walks them.
    let legacy = legacy_rows::arm();
    let inner = new_store()?;
    let seeded = seed_enrollment(&inner)?;
    drop(seeded);
    drop(legacy);
    let phase = Arc::new(PhaseAwareStore::new(
        Arc::clone(&inner),
        CursorFaultArm::NonEof,
    ));
    let store: Arc<dyn DurableStore> = phase.clone();
    let handler =
        ProductionParticipantHandler::new_restored(Arc::clone(&store), test_participant_config())?;
    let before = stream_payloads(&inner, &extension_key())?;
    let log = OperationLog::new(Arc::clone(&store), CONVERSATION);

    for arm in [CursorFaultArm::NonEof, CursorFaultArm::EmptyEof] {
        phase.reset_reference(arm)?;
        assert!(
            handler
                .replay_aggregate_reference(CONVERSATION, &log)
                .is_ok()
        );

        phase.reset_w3(arm)?;
        let typed = direct_w3_error(Arc::clone(&store))?;
        assert!(matches!(typed, Extension(OutboxLogError::Durability(_))));

        // The production load maps the cursor fault through the extension
        // log's typed error too, and the request seam reports it as the
        // load's refusal, never as a base operation's.
        phase.reset_w3(arm)?;
        let loaded = production_load_error(&store)?;
        assert!(
            matches!(
                loaded,
                BootCheckpointError::Replay(Extension(OutboxLogError::Durability(_)))
            ),
            "{arm:?}: the load refused as {loaded:?}"
        );

        clear_owner(&handler)?;
        phase.reset_w3(arm)?;
        let error = handler
            .replay_and_repair(CONVERSATION, &log)
            .err()
            .ok_or("armed production W3 restore unexpectedly succeeded")?;
        let message = error.to_string();
        assert!(
            message.contains(&format!(
                "{CHECKPOINT_RESTORE_FAILED}sequence conflict: expected "
            )),
            "{arm:?}: the cursor fault did not reach the request as the load's refusal: {message}"
        );
        assert!(!message.contains("participant production operation failed"));
        assert!(!owner_is_present(&handler)?);
        assert_eq!(stream_payloads(&inner, &extension_key())?, before);
    }
    Ok(())
}