liminal-server 0.20.1

//! Extension/base restore, two-barrier crash cuts, and pre-publication refusal.

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

use liminal::durability::bridge::block_on;
use liminal::durability::{
    DurabilityError, DurableStore, ScanCursor, ScanPage, StoredEntry, open_ephemeral,
};
use liminal_protocol::wire::{
    AttachAttemptToken, ClientRequest, ConnectionIncarnation, CredentialAttachRequest, EnrollBound,
    EnrollmentRequest, EnrollmentToken, ReceiptReplay, ServerValue, WriteOutcomeUnknown,
};

use crate::server::connection::ReadyWaker;
use crate::server::mount::MountKind;
use crate::server::participant::{
    InstalledParticipantService, ParticipantConnectionContext, ParticipantConnectionConversations,
    ParticipantOfferedProgress, ParticipantSemanticError, ParticipantSemanticHandler,
};

use super::ProductionParticipantHandler;
use super::log::{STREAM_PREFIX, legacy_rows};
use super::observer::OBSERVER_STREAM_KEY;
use super::outbox_log::{OUTBOX_STREAM_PREFIX, OutboxLog, OutboxRow};
use super::routed_stream_codec::row_stream;
use super::tests::{dispatch, test_participant_config};
use super::tests_w3_restore_fixture::stream_payloads;

const CONVERSATION: u64 = 0xF0_C4;

/// Where a second-barrier fault strikes: after the enrollment's operation
/// rows, at its extension row.
///
/// A live commit stages both rows in one routed batch, so the batch is the
/// only barrier a live store has. A lost extension row under durable
/// operation rows is a state only a store written before pruning, flat and
/// one append per row, can hold; that cut runs under the legacy-rows arm.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum SecondBarrier {
    /// Legacy flat rows: the operation rows are appended and flushed, and
    /// the extension append is refused, so the extension row is missing.
    LostAppend,
    /// The live routed batch carrying the extension row lands, the store
    /// answers an unknown outcome, and the read-back that would settle it
    /// fails once, so the rows stand and the outcome stays unknown.
    UnknownUnread,
    /// The live routed batch carrying the extension row lands, the store
    /// answers an unknown outcome, and the read-back succeeds, so the settle
    /// finds the batch's own bytes and the commit stands as confirmed.
    UnknownLanded,
}

impl SecondBarrier {
    /// Holds the legacy-rows arm for the cut that needs it.
    fn legacy_rows(self) -> Option<legacy_rows::LegacyRows> {
        (self == Self::LostAppend).then(legacy_rows::arm)
    }
}

#[derive(Debug)]
struct OutboxFaultStore {
    inner: Arc<dyn DurableStore>,
    /// Refuses every flat extension append while set: live only under the
    /// legacy-rows arm, where the handler appends rows flat.
    fail_append: AtomicBool,
    /// Lets the next routed batch carrying an extension row land, then
    /// answers it as an unknown outcome.
    answer_unknown_after_extension_batch: AtomicBool,
    /// Whether the unknown answer is followed by a failed read-back: set for
    /// [`SecondBarrier::UnknownUnread`], clear for
    /// [`SecondBarrier::UnknownLanded`].
    fail_readback_after_unknown: AtomicBool,
    /// Fails the next routed read once: the settle's read-back.
    fail_next_readback: AtomicBool,
    /// Refuses the next flat append to the observer stream once: the
    /// post-commit `Track` row a first enrollment appends.
    refuse_observer_append: AtomicBool,
    /// Set when the armed observer refusal struck.
    observer_append_refused: AtomicBool,
    /// Fails the next flat read of the observer stream once: the read with
    /// which recovery resolves an uncertain observer head.
    fail_observer_read: AtomicBool,
    /// Set when the armed observer read failure struck.
    observer_read_failed: AtomicBool,
}

impl OutboxFaultStore {
    fn new(inner: Arc<dyn DurableStore>) -> Self {
        Self {
            inner,
            fail_append: AtomicBool::new(false),
            answer_unknown_after_extension_batch: AtomicBool::new(false),
            fail_readback_after_unknown: AtomicBool::new(false),
            fail_next_readback: AtomicBool::new(false),
            refuse_observer_append: AtomicBool::new(false),
            observer_append_refused: AtomicBool::new(false),
            fail_observer_read: AtomicBool::new(false),
            observer_read_failed: AtomicBool::new(false),
        }
    }

    fn arm(&self, cut: SecondBarrier) {
        match cut {
            SecondBarrier::LostAppend => self.fail_append.store(true, Ordering::SeqCst),
            SecondBarrier::UnknownUnread => {
                self.fail_readback_after_unknown
                    .store(true, Ordering::SeqCst);
                self.answer_unknown_after_extension_batch
                    .store(true, Ordering::SeqCst);
            }
            SecondBarrier::UnknownLanded => {
                self.fail_readback_after_unknown
                    .store(false, Ordering::SeqCst);
                self.answer_unknown_after_extension_batch
                    .store(true, Ordering::SeqCst);
            }
        }
    }

    /// Arms a one-shot refusal of the next flat observer-stream append.
    fn refuse_next_observer_append(&self) {
        self.observer_append_refused.store(false, Ordering::SeqCst);
        self.refuse_observer_append.store(true, Ordering::SeqCst);
    }

    /// Whether the armed observer refusal struck.
    fn observer_append_struck(&self) -> bool {
        self.observer_append_refused.load(Ordering::SeqCst)
    }

    /// Spends the armed observer refusal when `stream_key` is the observer
    /// stream, answering whether this append is the one refused.
    fn refuses_observer_append(&self, stream_key: &str) -> bool {
        let refused = stream_key == OBSERVER_STREAM_KEY
            && self.refuse_observer_append.swap(false, Ordering::SeqCst);
        if refused {
            self.observer_append_refused.store(true, Ordering::SeqCst);
        }
        refused
    }

    /// Arms a one-shot failure of the next flat observer-stream read.
    fn fail_next_observer_read(&self) {
        self.observer_read_failed.store(false, Ordering::SeqCst);
        self.fail_observer_read.store(true, Ordering::SeqCst);
    }

    /// Whether the armed observer read failure struck.
    fn observer_read_struck(&self) -> bool {
        self.observer_read_failed.load(Ordering::SeqCst)
    }

    /// Spends the armed observer read failure when `stream_key` is the
    /// observer stream, answering whether this read is the one failed.
    fn fails_observer_read(&self, stream_key: &str) -> bool {
        let failed = stream_key == OBSERVER_STREAM_KEY
            && self.fail_observer_read.swap(false, Ordering::SeqCst);
        if failed {
            self.observer_read_failed.store(true, Ordering::SeqCst);
        }
        failed
    }

    fn disarm(&self) {
        self.fail_append.store(false, Ordering::SeqCst);
        self.answer_unknown_after_extension_batch
            .store(false, Ordering::SeqCst);
        self.fail_readback_after_unknown
            .store(false, Ordering::SeqCst);
        self.fail_next_readback.store(false, Ordering::SeqCst);
        self.refuse_observer_append.store(false, Ordering::SeqCst);
        self.fail_observer_read.store(false, Ordering::SeqCst);
    }

    /// Whether the routed fault, once armed, has struck in full: the batch
    /// answered unknown and, when armed to, the read-back failed.
    fn routed_fault_spent(&self) -> bool {
        !self
            .answer_unknown_after_extension_batch
            .load(Ordering::SeqCst)
            && !self.fail_next_readback.load(Ordering::SeqCst)
    }
}

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

#[async_trait::async_trait]
impl DurableStore for OutboxFaultStore {
    async fn apply_routed_batch(
        &self,
        route_key: &str,
        items: Vec<liminal::durability::RoutedBatchItem>,
    ) -> Result<(), liminal::durability::DurabilityError> {
        let extension = carries_extension_row(&items);
        self.inner.apply_routed_batch(route_key, items).await?;
        if extension
            && self
                .answer_unknown_after_extension_batch
                .swap(false, Ordering::SeqCst)
        {
            // The first routed read after this answer is the settle's
            // read-back of the batch's keys; armed, it fails once.
            if self.fail_readback_after_unknown.load(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>>, liminal::durability::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> {
        if self.refuses_observer_append(stream_key) {
            return Err(DurabilityError::ConfigError(
                "the test refused the observer append".to_owned(),
            ));
        }
        if stream_key.starts_with(OUTBOX_STREAM_PREFIX) && self.fail_append.load(Ordering::SeqCst) {
            return Err(DurabilityError::SequenceConflict {
                expected: expected_seq,
                actual: u64::MAX,
            });
        }
        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> {
        if self.refuses_observer_append(stream_key) {
            return Err(DurabilityError::ConfigError(
                "the test refused the observer append".to_owned(),
            ));
        }
        if stream_key.starts_with(OUTBOX_STREAM_PREFIX) && self.fail_append.load(Ordering::SeqCst) {
            return Err(DurabilityError::SequenceConflict {
                expected: expected_seq,
                actual: u64::MAX,
            });
        }
        self.inner
            .append_all(stream_key, payloads, expected_seq)
            .await
    }

    async fn read_from(
        &self,
        stream_key: &str,
        offset: u64,
        limit: usize,
    ) -> Result<Vec<StoredEntry>, DurabilityError> {
        if self.fails_observer_read(stream_key) {
            return Err(DurabilityError::ConfigError(
                "the test failed the observer recovery read".to_owned(),
            ));
        }
        self.inner.read_from(stream_key, offset, limit).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 enrollment() -> ClientRequest {
    ClientRequest::Enrollment(EnrollmentRequest {
        conversation_id: CONVERSATION,
        enrollment_token: EnrollmentToken::new([0xC4; 16]),
    })
}

/// Every committed extension row, read where its commit put it: flat rows
/// below the native head, routed rows at and after it.
fn extension_count(store: &Arc<dyn DurableStore>) -> Result<usize, Box<dyn Error>> {
    let key = format!("{OUTBOX_STREAM_PREFIX}{CONVERSATION}");
    Ok(stream_payloads(store, &key)?.len())
}

/// Every committed operation row, read where its commit put it.
fn base_count(store: &Arc<dyn DurableStore>) -> Result<usize, Box<dyn Error>> {
    let key = format!("{STREAM_PREFIX}{CONVERSATION}");
    Ok(stream_payloads(store, &key)?.len())
}

/// Whether `value` is the write-outcome-unknown answer to `enrollment()`: the
/// dispatch seam answers an issued operation whose durability is not
/// confirmed on the open connection, as an `Ok` response, never an `Err`.
fn enrollment_outcome_unknown(value: &ServerValue) -> bool {
    matches!(
        value,
        ServerValue::WriteOutcomeUnknown(WriteOutcomeUnknown::Enrollment { request })
            if request.conversation_id == CONVERSATION
                && request.enrollment_token == EnrollmentToken::new([0xC4; 16])
    )
}

fn exercise_second_barrier_cut(
    cut: SecondBarrier,
    expected_rows_after_fault: usize,
) -> Result<(), Box<dyn Error>> {
    let legacy = cut.legacy_rows();
    let inner: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let faults = Arc::new(OutboxFaultStore::new(Arc::clone(&inner)));
    faults.arm(cut);
    let store: Arc<dyn DurableStore> = faults.clone();
    let handler = ProductionParticipantHandler::new(Arc::clone(&store), test_participant_config())?;
    let result = dispatch(&handler, ConnectionIncarnation::new(90, 1), enrollment())?;
    assert!(
        enrollment_outcome_unknown(&result),
        "second-barrier fault published a response: {result:?}"
    );
    assert!(
        faults.routed_fault_spent(),
        "the armed routed fault never struck in full"
    );

    assert_eq!(base_count(&store)?, 2);
    assert_eq!(extension_count(&store)?, expected_rows_after_fault);
    drop(handler);
    drop(legacy);

    faults.disarm();
    let repaired =
        ProductionParticipantHandler::new(Arc::clone(&store), test_participant_config())?;
    assert_eq!(extension_count(&store)?, 1);
    drop(repaired);

    let restored_again =
        ProductionParticipantHandler::new(Arc::clone(&store), test_participant_config())?;
    assert_eq!(extension_count(&store)?, 1);
    drop(restored_again);
    Ok(())
}

#[test]
fn postcommit_outbox_append_failure_repairs_before_publication() -> Result<(), Box<dyn Error>> {
    exercise_second_barrier_cut(SecondBarrier::LostAppend, 0)
}

#[test]
fn uncertain_outbox_flush_accepts_exact_existing_without_duplicate() -> Result<(), Box<dyn Error>> {
    exercise_second_barrier_cut(SecondBarrier::UnknownUnread, 1)
}

#[test]
fn malformed_unknown_and_mixed_extension_refuses_before_publication() -> Result<(), Box<dyn Error>>
{
    // A legacy flat stream mixed in version is refused: the enrollment writes
    // the flat rows a store from before pruning holds, so the unknown-version
    // row below follows them in the same flat stream.
    let legacy = legacy_rows::arm();
    let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let handler = ProductionParticipantHandler::new(Arc::clone(&store), test_participant_config())?;
    let enrolled = dispatch(&handler, ConnectionIncarnation::new(91, 1), enrollment())?;
    assert!(matches!(
        enrolled,
        liminal_protocol::wire::ServerValue::EnrollBound(_)
    ));
    drop(handler);
    drop(legacy);

    let key = format!("{OUTBOX_STREAM_PREFIX}{CONVERSATION}");
    let assigned = block_on(store.append(&key, vec![2], 1))??;
    assert_eq!(assigned, 1);
    block_on(store.flush())??;
    // CONTAINMENT: the node starts, and both properties this test protects are
    // asserted directly — no authority published, and the cause kept by name.
    let handler =
        ProductionParticipantHandler::new(store, test_participant_config()).map_err(|error| {
            format!("CONTAINMENT: one unloadable conversation took the whole node down: {error}")
        })?;
    assert_no_authority_and_named_cause(&handler, "mixed Unit 2 extension schema versions")
}

#[test]
fn impossible_extension_boundary_refuses_before_publication() -> Result<(), Box<dyn Error>> {
    let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let handler = ProductionParticipantHandler::new(Arc::clone(&store), test_participant_config())?;
    let enrolled = dispatch(&handler, ConnectionIncarnation::new(92, 1), enrollment())?;
    assert!(matches!(
        enrolled,
        liminal_protocol::wire::ServerValue::EnrollBound(_)
    ));
    drop(handler);

    let outbox = OutboxLog::new(Arc::clone(&store), CONVERSATION);
    block_on(outbox.append(
        &OutboxRow::AckAdvanced {
            source_log_sequence: 99,
            participant_id: 0,
            through_seq: 1,
        },
        1,
    ))??;
    let handler =
        ProductionParticipantHandler::new(store, test_participant_config()).map_err(|error| {
            format!("CONTAINMENT: one unloadable conversation took the whole node down: {error}")
        })?;
    assert_no_authority_and_named_cause(&handler, "impossible future boundary")
}

/// The successor to `let Err(..) = ProductionParticipantHandler::new(..)`.
///
/// Containment moved WHERE an unloadable conversation is refused, not WHETHER
/// it is. Both halves of the original property are checked here: the
/// conversation publishes no authority, and its refusal still carries the
/// specific cause instead of being flattened into "something went wrong".
fn assert_no_authority_and_named_cause(
    handler: &ProductionParticipantHandler,
    expected_cause: &str,
) -> Result<(), Box<dyn Error>> {
    let cell = handler.cell(CONVERSATION)?;
    let owner = cell
        .lock()
        .map_err(|_| "conversation owner lock is poisoned")?;
    assert!(
        owner.is_none(),
        "an unloadable conversation published an authority anyway"
    );
    drop(owner);

    let unloadable = handler.unloadable_conversations();
    let reason = unloadable
        .get(&CONVERSATION)
        .ok_or("the conversation was not recorded as unloadable")?;
    assert!(
        reason.contains(expected_cause),
        "the failure was contained but its cause was lost: expected {expected_cause:?} in {reason}"
    );
    Ok(())
}

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

/// Every committed extension payload, read where its commit put it.
fn raw_extension_payloads(store: &Arc<dyn DurableStore>) -> Result<Vec<Vec<u8>>, Box<dyn Error>> {
    let key = format!("{OUTBOX_STREAM_PREFIX}{CONVERSATION}");
    stream_payloads(store, &key)
}

fn installed_service(
    handler: Arc<ProductionParticipantHandler>,
    store: Arc<dyn DurableStore>,
) -> Result<InstalledParticipantService, Box<dyn Error>> {
    let config = test_participant_config();
    let semantic: Arc<dyn ParticipantSemanticHandler> = handler;
    InstalledParticipantService::new(semantic, store, config.wire_frame_limit)
        .map_err(|error| format!("repair service configuration failed: {error:?}").into())
}

fn apply_service(
    service: &InstalledParticipantService,
    incarnation: ConnectionIncarnation,
    request: ClientRequest,
) -> Result<ServerValue, Box<dyn Error>> {
    service
        .handle(
            ParticipantConnectionContext::new(incarnation, MountKind::Tcp),
            &mut ParticipantConnectionConversations::default(),
            request,
        )
        .map_err(Into::into)
}

/// The second enrollment's extension row as a healthy run writes it, in the
/// row shape `cut` runs in: flat under the legacy-rows arm, routed live.
fn expected_second_enrollment_payload(cut: SecondBarrier) -> Result<Vec<u8>, Box<dyn Error>> {
    let legacy = cut.legacy_rows();
    let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let handler = ProductionParticipantHandler::new(Arc::clone(&store), test_participant_config())?;
    let first = dispatch(
        &handler,
        ConnectionIncarnation::new(0xC4, 1),
        enrollment_with(0xC1),
    )?;
    assert!(matches!(first, ServerValue::EnrollBound(_)));
    let second = dispatch(
        &handler,
        ConnectionIncarnation::new(0xC4, 2),
        enrollment_with(0xC2),
    )?;
    assert!(matches!(second, ServerValue::EnrollBound(_)));
    drop(handler);
    drop(legacy);
    raw_extension_payloads(&store)?
        .into_iter()
        .nth(1)
        .ok_or_else(|| "healthy second enrollment omitted its outbox row".into())
}

fn durable_ack_through(
    handler: &ProductionParticipantHandler,
    participant_id: u64,
) -> Result<u64, Box<dyn Error>> {
    let cell = handler.cell(CONVERSATION)?;
    let owner = cell
        .lock()
        .map_err(|_| "repaired conversation owner lock was poisoned")?;
    let authority = owner
        .as_ref()
        .ok_or("repaired conversation owner was absent")?;
    let outbox = authority
        .outbox
        .as_ref()
        .ok_or("repaired conversation outbox was absent")?;
    let cursor = outbox.durable_ack_through(participant_id);
    drop(owner);
    Ok(cursor)
}

fn replay_then_commit_rebind(
    repaired: &InstalledParticipantService,
    repaired_handler: &ProductionParticipantHandler,
    first_bound: &EnrollBound,
    first_incarnation: ConnectionIncarnation,
    second_incarnation: ConnectionIncarnation,
    exact_request: ClientRequest,
) -> Result<(), Box<dyn Error>> {
    // Construction returns only after cold first-touch reconciliation. Neither
    // startup, registration, nor an exact-token replay commits an impact or tell.
    let repaired_wakes = Arc::new(AtomicU64::new(u64::from(false)));
    let repaired_inbox = repaired.new_publication_inbox();
    repaired.publication_registry().register(
        first_incarnation,
        &repaired_inbox,
        ReadyWaker::for_test(Arc::clone(&repaired_wakes)),
    )?;
    assert!(!repaired_inbox.has_pending()?);
    assert_eq!(repaired_wakes.load(Ordering::SeqCst), u64::from(false));
    let retry = apply_service(repaired, second_incarnation, exact_request)?;
    let ServerValue::Bound(ReceiptReplay::Enrollment(bound)) = retry else {
        return Err(format!("exact-token retry lost its terminal answer: {retry:?}").into());
    };
    assert_eq!(bound.token(), EnrollmentToken::new([0xC2; 16]));
    assert_eq!(bound.conversation_id(), CONVERSATION);
    assert!(!repaired_inbox.has_pending()?);
    assert_eq!(repaired_wakes.load(Ordering::SeqCst), u64::from(false));

    let reconciled_cursor = durable_ack_through(repaired_handler, first_bound.participant_id())?;
    let rebound_incarnation = ConnectionIncarnation::new(0xC4, 3);
    let rebound_inbox = repaired.new_publication_inbox();
    repaired.publication_registry().register(
        rebound_incarnation,
        &rebound_inbox,
        ReadyWaker::for_test(Arc::clone(&repaired_wakes)),
    )?;
    assert!(!rebound_inbox.has_pending()?);
    assert_eq!(repaired_wakes.load(Ordering::SeqCst), u64::from(false));

    let rebound = apply_service(
        repaired,
        rebound_incarnation,
        ClientRequest::CredentialAttach(CredentialAttachRequest {
            conversation_id: CONVERSATION,
            participant_id: first_bound.participant_id(),
            capability_generation: first_bound.capability_generation(),
            attach_secret: first_bound.attach_secret(),
            attach_attempt_token: AttachAttemptToken::new([0xC3; 16]),
            accept_marker_delivery_seq: None,
        }),
    )?;
    if !matches!(rebound, ServerValue::AttachBound(_)) {
        return Err(format!("committed rebind did not bind: {rebound:?}").into());
    }
    assert_eq!(repaired_wakes.load(Ordering::SeqCst), u64::from(true));
    assert!(!repaired_inbox.has_pending()?);
    let ready = rebound_inbox.take_ready()?;
    assert_eq!(ready.conversations, vec![CONVERSATION]);
    assert!(ready.observer_progressed.is_empty());

    let repaired_obligation = repaired
        .next_publication(rebound_incarnation, CONVERSATION, None)?
        .ok_or("committed rebind did not select the repaired obligation")?;
    assert_eq!(
        repaired_obligation.participant_id,
        first_bound.participant_id()
    );
    assert!(repaired_obligation.delivery_seq() > reconciled_cursor);
    let offered = ParticipantOfferedProgress {
        binding_epoch: repaired_obligation.binding_epoch,
        through_seq: repaired_obligation.delivery_seq(),
    };
    assert!(
        repaired
            .next_publication(rebound_incarnation, CONVERSATION, Some(offered))?
            .is_none(),
        "repaired obligation was not dispatched exactly once"
    );
    Ok(())
}

fn exercise_repair_and_retry(
    cut: SecondBarrier,
    expected_rows_after_fault: usize,
    expected_second_payload: &[u8],
) -> Result<(), Box<dyn Error>> {
    let legacy = cut.legacy_rows();
    let inner: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let faults = Arc::new(OutboxFaultStore::new(Arc::clone(&inner)));
    let store: Arc<dyn DurableStore> = faults.clone();
    let config = test_participant_config();
    let handler = Arc::new(ProductionParticipantHandler::new(
        Arc::clone(&store),
        config,
    )?);
    let service = installed_service(handler, Arc::clone(&store))?;
    let first_incarnation = ConnectionIncarnation::new(0xC4, 1);
    let second_incarnation = ConnectionIncarnation::new(0xC4, 2);
    let first = apply_service(&service, first_incarnation, enrollment_with(0xC1))?;
    let ServerValue::EnrollBound(first_bound) = first else {
        return Err(format!("first enrollment did not bind: {first:?}").into());
    };

    let wake_count = Arc::new(AtomicU64::new(0));
    let inbox = service.new_publication_inbox();
    service.publication_registry().register(
        first_incarnation,
        &inbox,
        ReadyWaker::for_test(Arc::clone(&wake_count)),
    )?;
    faults.arm(cut);
    let exact_request = enrollment_with(0xC2);
    // The semantic service answers the unconfirmed write as its error; the
    // dispatch seam is what turns that into the 0x0128 value.
    assert!(
        apply_service(&service, second_incarnation, exact_request.clone()).is_err(),
        "barrier-2 fault fabricated a successful terminal answer"
    );
    assert!(
        faults.routed_fault_spent(),
        "the armed routed fault never struck in full"
    );
    assert!(!inbox.has_pending()?);
    assert_eq!(wake_count.load(Ordering::SeqCst), 0);

    assert_eq!(
        base_count(&store)?,
        3,
        "both enrollments plus genesis remain durable after barrier-2 failure"
    );
    assert_eq!(
        raw_extension_payloads(&store)?.len(),
        expected_rows_after_fault
    );
    drop(service);
    drop(legacy);

    faults.disarm();
    let repaired_handler = Arc::new(ProductionParticipantHandler::new(
        Arc::clone(&store),
        config,
    )?);
    let repaired_payloads = raw_extension_payloads(&store)?;
    assert_eq!(repaired_payloads.len(), 2);
    assert_eq!(repaired_payloads[1], expected_second_payload);

    let repaired = installed_service(Arc::clone(&repaired_handler), Arc::clone(&store))?;
    replay_then_commit_rebind(
        &repaired,
        &repaired_handler,
        &first_bound,
        first_incarnation,
        second_incarnation,
        exact_request,
    )?;

    let payloads_after_rebind = raw_extension_payloads(&store)?;
    drop(repaired);
    let restored_again =
        ProductionParticipantHandler::new(Arc::clone(&store), test_participant_config())?;
    assert_eq!(raw_extension_payloads(&store)?, payloads_after_rebind);
    drop(restored_again);
    Ok(())
}

#[test]
fn postcommit_outbox_failure_is_repaired_not_rolled_back() -> Result<(), Box<dyn Error>> {
    let lost = expected_second_enrollment_payload(SecondBarrier::LostAppend)?;
    exercise_repair_and_retry(SecondBarrier::LostAppend, 1, &lost)?;
    let unknown = expected_second_enrollment_payload(SecondBarrier::UnknownUnread)?;
    exercise_repair_and_retry(SecondBarrier::UnknownUnread, 2, &unknown)
}

/// A second conversation, whose enrollment strands the shared observer head.
const OTHER_CONVERSATION: u64 = CONVERSATION + 1;

/// A handler whose committed lifecycle receipts reach the events subject,
/// over a store whose faults the test arms.
struct ReceiptRig {
    faults: Arc<OutboxFaultStore>,
    store: Arc<dyn DurableStore>,
    handler: ProductionParticipantHandler,
    events: liminal::channel::SubscriptionHandle,
    /// Holds the services and their store open for the rig's life.
    _fixture: crate::server::self_streams::tests::R3Fixture,
}

fn receipt_rig() -> Result<ReceiptRig, Box<dyn Error>> {
    use crate::server::self_streams::cost::ServerMetadata;
    use crate::server::self_streams::sink::SelfStreams;
    use crate::server::shutdown::ShutdownHandle;

    let fixture = crate::server::self_streams::tests::R3Fixture::new()?;
    let faults = Arc::new(OutboxFaultStore::new(Arc::clone(&fixture.store)));
    let store: Arc<dyn DurableStore> = faults.clone();
    let handler = ProductionParticipantHandler::new(Arc::clone(&store), test_participant_config())?;
    let services = &fixture.services;
    let events = services.subscribe_handle_for_test("liminal.events")?;
    let signals = services.signals();
    signals.install(Arc::new(SelfStreams::new(
        ServerMetadata::new("local".to_owned(), 90)?,
        services,
        signals.wake(),
        ShutdownHandle::new(),
        crate::config::types::SelfStreamsConfig::default(),
    )))?;
    handler.install_signals(signals.clone())?;
    Ok(ReceiptRig {
        faults,
        store,
        handler,
        events,
        _fixture: fixture,
    })
}

/// Asserts the events subject holds exactly one lifecycle receipt, naming
/// `conversation` and `transition`.
fn assert_one_receipt(
    events: &liminal::channel::SubscriptionHandle,
    conversation: u64,
    transition: &str,
) -> Result<(), Box<dyn Error>> {
    let event = events
        .try_next()?
        .ok_or("the committed lifecycle receipt was not emitted")?;
    let value: serde_json::Value = serde_json::from_slice(&event.payload)?;
    assert_eq!(value["details"]["conversation"], conversation);
    assert_eq!(value["details"]["transition"], transition);
    assert!(
        events.try_next()?.is_none(),
        "the committed lifecycle receipt was duplicated"
    );
    Ok(())
}

/// The enrollment's routed batch is durable; the post-commit observer `Track`
/// append is refused. The commit stands, so its receipt is emitted once while
/// the request answers that its outcome is unknown.
#[test]
fn committed_lifecycle_receipt_survives_outbox_append_failure() -> Result<(), Box<dyn Error>> {
    let rig = receipt_rig()?;
    rig.faults.refuse_next_observer_append();

    let result = dispatch(
        &rig.handler,
        ConnectionIncarnation::new(90, 1),
        enrollment(),
    )?;
    assert!(
        rig.faults.observer_append_struck(),
        "the injected observer Track failure did not fire"
    );
    assert!(
        enrollment_outcome_unknown(&result),
        "a refused post-commit Track was answered {result:?}"
    );
    assert_eq!(base_count(&rig.store)?, 2);
    assert_eq!(extension_count(&rig.store)?, 1);
    let event = rig
        .events
        .try_next()?
        .ok_or("committed lifecycle receipt was lost on error")?;
    let value: serde_json::Value = serde_json::from_slice(&event.payload)?;
    assert_eq!(value["details"]["conversation"], CONVERSATION);
    assert_eq!(value["details"]["transition"], "enrolled");
    assert!(
        rig.events.try_next()?.is_none(),
        "committed receipt was duplicated"
    );
    Ok(())
}

/// Enrolls and attaches `CONVERSATION`'s first member on `incarnation`,
/// draining both receipts.
fn enroll_and_attach(
    rig: &ReceiptRig,
    incarnation: ConnectionIncarnation,
    conversations: &mut ParticipantConnectionConversations,
) -> Result<(), Box<dyn Error>> {
    use super::tests::dispatch_tracked;
    use liminal_protocol::wire::Generation;

    let enrolled = dispatch_tracked(&rig.handler, incarnation, conversations, enrollment())?;
    let ServerValue::EnrollBound(receipt) = enrolled else {
        return Err("reconciliation fixture did not enroll".into());
    };
    assert_one_receipt(&rig.events, CONVERSATION, "enrolled")?;
    let attached = dispatch_tracked(
        &rig.handler,
        incarnation,
        conversations,
        ClientRequest::CredentialAttach(CredentialAttachRequest {
            conversation_id: CONVERSATION,
            participant_id: receipt.participant_id(),
            capability_generation: Generation::ONE,
            attach_secret: receipt.attach_secret(),
            attach_attempt_token: AttachAttemptToken::new([0xC5; 16]),
            accept_marker_delivery_seq: None,
        }),
    )?;
    assert!(matches!(attached, ServerValue::AttachBound(_)));
    assert_one_receipt(&rig.events, CONVERSATION, "attached")?;
    assert_eq!(extension_count(&rig.store)?, 2);
    Ok(())
}

/// Leaves the shared observer head uncertain: a second conversation's
/// enrollment commits, and its post-commit `Track` append is refused, so the
/// head names a row that never landed.
fn strand_observer_head(rig: &ReceiptRig) -> Result<(), Box<dyn Error>> {
    rig.faults.refuse_next_observer_append();
    let other = dispatch(
        &rig.handler,
        ConnectionIncarnation::new(91, 1),
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id: OTHER_CONVERSATION,
            enrollment_token: EnrollmentToken::new([0xC6; 16]),
        }),
    )?;
    assert!(
        rig.faults.observer_append_struck(),
        "the injected observer Track failure did not fire"
    );
    assert!(
        matches!(
            other,
            ServerValue::WriteOutcomeUnknown(WriteOutcomeUnknown::Enrollment { .. })
        ),
        "a refused post-commit Track was answered {other:?}"
    );
    assert_one_receipt(&rig.events, OTHER_CONVERSATION, "enrolled")
}

/// The detach's routed batch is durable; its post-commit observer
/// reconciliation must first recover the shared observer head an earlier
/// refused `Track` left uncertain, and the recovery's read of the observer
/// stream fails. The fate stands, so its receipt is emitted once while the
/// reconciliation answers an unknown outcome and the owner is unloaded.
#[test]
fn committed_lifecycle_receipt_survives_reconciliation_failure() -> Result<(), Box<dyn Error>> {
    use crate::server::participant::{ConnectionFateClass, ConnectionFateWorkItem};

    let rig = receipt_rig()?;
    let incarnation = ConnectionIncarnation::new(90, 1);
    let mut conversations = ParticipantConnectionConversations::default();
    enroll_and_attach(&rig, incarnation, &mut conversations)?;
    strand_observer_head(&rig)?;
    // Armed after the strand, so the first observer read is the fate's own
    // post-commit recovery of the uncertain head.
    rig.faults.fail_next_observer_read();

    let work = ConnectionFateWorkItem {
        open_sequence: 31,
        connection_incarnation: incarnation,
        class: ConnectionFateClass::CleanDisconnect,
        tracked_conversations: conversations.tracked_conversations(),
    };
    let result =
        rig.handler
            .with_conversation_fate_source(CONVERSATION, None, |authority, appender, _| {
                authority
                    .prepare_connection_fate_transaction(&work)
                    .complete(authority, appender)
            });
    assert!(
        rig.faults.observer_read_struck(),
        "the injected observer recovery read failure did not fire"
    );
    assert!(
        matches!(
            result,
            Err(ParticipantSemanticError::WriteOutcomeUnknown { .. })
        ),
        "the post-commit reconciliation failure did not fire: {result:?}"
    );
    assert_eq!(base_count(&rig.store)?, 4);
    assert_eq!(extension_count(&rig.store)?, 3);
    let event = rig
        .events
        .try_next()?
        .ok_or("committed fate receipt was lost during reconciliation")?;
    let value: serde_json::Value = serde_json::from_slice(&event.payload)?;
    assert_eq!(value["details"]["conversation"], CONVERSATION);
    assert_eq!(value["details"]["transition"], "detached");
    assert!(
        rig.events.try_next()?.is_none(),
        "committed fate receipt was duplicated"
    );
    let cell = rig.handler.cell(CONVERSATION)?;
    assert!(
        cell.lock()
            .map_err(|_| "conversation lock poisoned")?
            .is_none()
    );
    Ok(())
}

/// The routed batch carrying the extension row lands and is answered as an
/// unknown outcome; the settle's read-back finds the batch's own bytes, so
/// the commit is confirmed: the enrollment binds and its receipt is emitted
/// exactly once.
#[test]
fn committed_lifecycle_receipt_emitted_once_when_unknown_batch_reads_back_landed()
-> Result<(), Box<dyn Error>> {
    let rig = receipt_rig()?;
    rig.faults.arm(SecondBarrier::UnknownLanded);

    let result = dispatch(
        &rig.handler,
        ConnectionIncarnation::new(92, 1),
        enrollment(),
    )?;
    assert!(
        rig.faults.routed_fault_spent(),
        "the armed unknown answer never struck"
    );
    assert!(
        matches!(result, ServerValue::EnrollBound(_)),
        "a settled landed batch was answered {result:?}"
    );
    assert_eq!(base_count(&rig.store)?, 2);
    assert_eq!(extension_count(&rig.store)?, 1);
    assert_one_receipt(&rig.events, CONVERSATION, "enrolled")
}

/// The routed batch carrying the extension row lands and is answered as an
/// unknown outcome whose read-back fails: nothing confirms the commit, so no
/// receipt is emitted and the enrollment answers unknown. Re-presented, the
/// enrollment answers its committed receipt, emits exactly one lifecycle
/// receipt, and the extension row stands once.
#[test]
fn unread_unknown_batch_emits_no_receipt_until_the_enrollment_is_re_presented()
-> Result<(), Box<dyn Error>> {
    let rig = receipt_rig()?;
    rig.faults.arm(SecondBarrier::UnknownUnread);

    let result = dispatch(
        &rig.handler,
        ConnectionIncarnation::new(93, 1),
        enrollment(),
    )?;
    assert!(
        rig.faults.routed_fault_spent(),
        "the armed routed fault never struck in full"
    );
    assert!(
        enrollment_outcome_unknown(&result),
        "an unconfirmed batch was answered {result:?}"
    );
    assert!(
        rig.events.try_next()?.is_none(),
        "an unconfirmed commit emitted its receipt"
    );
    assert_eq!(extension_count(&rig.store)?, 1);

    rig.faults.disarm();
    let represented = dispatch(
        &rig.handler,
        ConnectionIncarnation::new(93, 2),
        enrollment(),
    )?;
    let ServerValue::Bound(ReceiptReplay::Enrollment(bound)) = represented else {
        return Err(format!("the re-presented enrollment lost its commit: {represented:?}").into());
    };
    assert_eq!(bound.conversation_id(), CONVERSATION);
    assert_eq!(bound.token(), EnrollmentToken::new([0xC4; 16]));
    // The commit owes its receipt, not the answer that first reported it:
    // the load that finds the landed rows emits it, exactly once.
    assert_one_receipt(&rig.events, CONVERSATION, "enrolled")?;
    assert_eq!(base_count(&rig.store)?, 2);
    assert_eq!(extension_count(&rig.store)?, 1);
    Ok(())
}