liminal-server 0.23.1

//! THE CONTAINMENT GATE, RED BY DESIGN.
//!
//! Tom's bar, 2026-08-06: *"this whole thing has to be robust beyond belief...
//! the fact that something can kill something on that... constitutes an
//! absolute failure of the project."*
//!
//! The property that bar reduces to, mechanically:
//!
//! > No single unloadable durable record can prevent the node from starting,
//! > and no single unloadable conversation can take down the others. Refuse it,
//! > NAME it, keep serving.
//!
//! DUR-020 (10 October 2026): construction no longer restores anything. The
//! loop body described below is now the per-conversation restore unit
//! (`restore_one_conversation`), run once per conversation by its first touch
//! or by the warm (`lazy_restore.rs`), and these tests reach it through
//! `new_restored`, which is construction and the warm. The line numbers below
//! are the pre-DUR-020 census. The warning at the end of this section has
//! fired: the unit now runs on a live shared handler, so the three guards it
//! calls unreachable are live routes. A poisoned map or cell refuses that
//! touch by name and records the conversation unloadable through
//! `ensure_restored`. No test here drives those three routes yet; that is a
//! named gap, not coverage.
//!
//! The property did not hold before the containment fix. Boot enumerated every registered
//! conversation and replays each one in a single loop:
//!
//! ```text
//! handler.rs:111 pub fn new -> :137 handler.restore_all_conversations()?;
//! handler.rs:237 restore_all_conversations -> the loop at :239-:280
//! ```
//!
//! Any `?` inside that loop body turns ONE conversation's failure into a
//! constructor failure at `:137`, and the node does not start. The per-request
//! gate at `handler.rs:350` already behaves correctly — it fails exactly one
//! conversation and the node keeps serving — so the defect is the boot loop
//! alone.
//!
//! # THE ROUTE AXIS, AND WHY THIS FILE HAS ONE
//!
//! A gate built only from fault SHAPES is blind to fault ROUTES. Every
//! corrupt-payload arm below trips `:250` first and never reaches the rest of
//! the loop body, so a fix applied at `:250` alone would turn every one of them
//! green while the remaining routes stayed fatal — the fix and its test would
//! agree with each other and both be wrong. So the arms below are chosen by
//! WHERE CONTROL GOES, one per propagation point, and each one proves it
//! actually drove its own route rather than trusting that it did.
//!
//! The loop body carries FIVE `?`-bearing statements. TWO of them are live at
//! boot and THREE cannot fire there; a sixth route, a post-drain rebuild from
//! the log, no longer exists, and its arm below proves boot reads nothing back:
//!
//! | site   | statement                    | at boot                            |
//! |--------|------------------------------|------------------------------------|
//! | `:240` | `self.cell(id)?`             | UNREACHABLE — see below            |
//! | `:243` | owner `lock()` poison        | UNREACHABLE — see below            |
//! | `:250` | `replay_and_repair(..)?`     | LIVE — driven by `CorruptRecordNow`|
//! | `:254` | `evict_uncommitted(..)?`     | UNREACHABLE — see below            |
//! | `:265` | `verdict.observe()?`         | LIVE — driven by `RefuseAppend`    |
//! | post-drain | commit tail, no read of the log | PROVED — `CorruptAfterAppend` never read |
//!
//! THE THREE UNREACHABLE ONES ARE NAMED, NOT DROPPED, because a gate that
//! silently omits coverage reads as though it covered everything. `:240` and
//! `:254` both fail only when `self.conversations` is poisoned; `:243` only
//! when a per-conversation cell mutex is poisoned. Every one of those mutexes
//! is constructed inside `new` itself — the map at `handler.rs:127`, each cell
//! at `handler.rs:303` in the same loop iteration that locks it — and
//! `restore_all_conversations` has EXACTLY ONE CALLER, `handler.rs:137`, which
//! runs before `new` returns and before any other thread can hold a reference.
//! A mutex is poisoned only by a panic while it is held, and no such panic can
//! have happened yet. So the three are guards over a state boot cannot be in.
//!
//! ⚠ THAT IS A CLAIM ABOUT THE CURRENT PROGRAM AND IT EXPIRES. The moment
//! anything calls `restore_all_conversations` on a live shared handler — a
//! re-restore, a reload, a second boot phase — all three become live routes and
//! this file is under-covering by half. The one-caller census above is the
//! thing to re-run, not this sentence.
//!
//! # ATTRIBUTION IS PART OF THE SAME PROPERTY, NOT A SECOND TICKET
//!
//! Containment without attribution ships a node that boots clean and silently
//! serves nothing on one conversation forever — worse than the crash, because
//! the crash is the only thing currently telling anyone.
//!
//! Attribution is presently SPLIT across the two live routes, which is itself
//! worth recording: `:265` refuses through
//! `ParticipantSemanticError::BootDrainRefused`, which CARRIES its
//! `conversation_id`; `:250` surfaces a bare decode failure
//! ("expected value at line 1 column 1") with no conversation anywhere in it.
//! The four sealed-binding-fate refusals one layer down (`ops_acks.rs:590`,
//! `ops_nonzero_ack.rs:424`, `binding_fate_completion.rs:83`,
//! `connection_fate_replay.rs:155`) do the same thing: they name the INVARIANT
//! and never the SUBJECT.
//!
//! # THE FAULTS
//!
//! Every fault here is scoped to EXACTLY ONE conversation's stream and nothing
//! else — the stream is present, its length and sequences are unchanged, and
//! every other conversation reads and writes through verbatim. That is the
//! shape of this week's failures: intact storage, one conversation the node
//! cannot accept. It is deliberately NOT an I/O outage; a store that is wholly
//! gone is a different property and fatal is the right answer to it.
//!
//! These tests are EXPECTED TO FAIL until the containment change lands. They
//! are the red arm for it. Do not weaken the assertions to make them pass.

use crate::config::types::ParticipantConfig;
use crate::server::participant::{ParticipantConnectionConversations, ParticipantSemanticError};

use liminal::durability::{
    DurabilityError, DurableStore, ScanCursor, ScanPage, StoredEntry, open_ephemeral,
};

use liminal_protocol::wire::{
    ClientRequest, ConnectionIncarnation, EnrollmentRequest, EnrollmentToken, ServerValue,
};

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

use super::ProductionParticipantHandler;
use super::log::STREAM_PREFIX;
use super::routed_stream_codec::{self, StreamKey};
use super::tests::{dispatch_tracked, test_participant_config};
use super::tests_w1b_pending_died_restart::pending_restart_fixture;

pub(super) const POISONED_CONVERSATION: u64 = 7_201;
pub(super) const HEALTHY_CONVERSATION: u64 = 7_202;
const POISONED_TOKEN: [u8; 16] = [0xC1; 16];
const HEALTHY_TOKEN: [u8; 16] = [0xC2; 16];

/// The conversation `pending_restart_fixture` writes. Asserted against the
/// fixture's own field rather than trusted, so a fixture reshape cannot leave
/// this file quietly targeting a stream that no longer exists.
const PENDING_CONVERSATION: u64 = 67;

/// Which single-conversation fault a [`OneConversationFault`] injects.
#[derive(Debug, Clone, Copy)]
pub(super) enum FaultMode {
    /// One record's payload is unloadable from the first read onward, so the
    /// row still reads but no longer decodes. Trips `handler.rs:250`.
    CorruptRecordNow(u64),
    /// Every append to this conversation's stream is refused, so the boot
    /// drain cannot persist the drain it decided on. Trips `handler.rs:265`.
    RefuseAppend,
    /// One record's payload becomes unloadable ONLY AFTER the first batch
    /// carrying a row of this conversation's stream lands. The boot drain's
    /// own commit is that trigger, so the load reads clean and any later read
    /// of the log would not; boot makes no such read.
    CorruptAfterAppend(u64),
}

/// A durable store that breaks exactly one conversation, exactly one way, and
/// records what it actually did so an arm can prove which route it drove.
#[derive(Debug)]
pub(super) struct OneConversationFault {
    inner: Arc<dyn DurableStore>,
    target_key: String,
    mode: FaultMode,
    /// Set once an append to the target stream, flat or in a routed batch,
    /// has landed.
    armed: Arc<AtomicBool>,
    /// Number of reads this store actually served a corrupted payload into.
    corrupted_reads: Arc<AtomicUsize>,
}

impl OneConversationFault {
    pub(super) fn new(inner: Arc<dyn DurableStore>, conversation_id: u64, mode: FaultMode) -> Self {
        Self {
            inner,
            target_key: format!("{STREAM_PREFIX}{conversation_id}"),
            mode,
            armed: Arc::new(AtomicBool::new(false)),
            corrupted_reads: Arc::new(AtomicUsize::new(0)),
        }
    }

    /// The record this store corrupts RIGHT NOW, if any. `CorruptAfterAppend`
    /// answers `None` until its append lands, which is what makes it a route
    /// discriminator rather than a second copy of `CorruptRecordNow`.
    fn armed_sequence(&self) -> Option<u64> {
        match self.mode {
            FaultMode::CorruptRecordNow(sequence) => Some(sequence),
            FaultMode::CorruptAfterAppend(sequence) if self.armed.load(Ordering::SeqCst) => {
                Some(sequence)
            }
            FaultMode::CorruptAfterAppend(_) | FaultMode::RefuseAppend => None,
        }
    }

    /// The sequence of the target stream's row this routed `key` names, if it
    /// names one: a commit writes the stream's name into every row key.
    fn target_row_sequence(&self, key: &[u8]) -> Option<u64> {
        match routed_stream_codec::decode_key(key) {
            Ok(StreamKey::Row { stream, sequence }) if stream == self.target_key => Some(sequence),
            Ok(StreamKey::Row { .. } | StreamKey::Head { .. }) | Err(_) => None,
        }
    }

    /// The routed row at `key`, with its payload made unloadable when the
    /// fault is armed for exactly that row: the row still reads, and no longer
    /// decodes, the shape the flat fault had.
    fn corrupt_routed(&self, key: &[u8], value: Vec<u8>) -> Vec<u8> {
        let Some(sequence) = self.target_row_sequence(key) else {
            return value;
        };
        if self.armed_sequence() != Some(sequence) {
            return value;
        }
        self.corrupted_reads.fetch_add(1, Ordering::SeqCst);
        let Ok(mut entry) = routed_stream_codec::decode_row(&value, sequence) else {
            return vec![0xFF; 16];
        };
        entry.payload = vec![0xFF; 16];
        routed_stream_codec::encode_row(&entry).unwrap_or_else(|_| vec![0xFF; 16])
    }

    fn corrupt(&self, stream_key: &str, mut entries: Vec<StoredEntry>) -> Vec<StoredEntry> {
        if stream_key != self.target_key {
            return entries;
        }
        let Some(target_sequence) = self.armed_sequence() else {
            return entries;
        };
        for entry in &mut entries {
            if entry.sequence == target_sequence {
                entry.payload = vec![0xFF; 16];
                self.corrupted_reads.fetch_add(1, Ordering::SeqCst);
            }
        }
        entries
    }
}

#[async_trait::async_trait]
impl DurableStore for OneConversationFault {
    async fn apply_routed_batch(
        &self,
        route_key: &str,
        items: Vec<liminal::durability::RoutedBatchItem>,
    ) -> Result<(), liminal::durability::DurabilityError> {
        // A commit is one routed batch, so the target stream's rows arrive
        // here; the flat arms below remain for the legacy-rows arming.
        let carries_target_row = items
            .iter()
            .any(|item| self.target_row_sequence(&item.key).is_some());
        if carries_target_row && matches!(self.mode, FaultMode::RefuseAppend) {
            return Err(DurabilityError::SequenceConflict {
                expected: 0,
                actual: 1,
            });
        }
        self.inner.apply_routed_batch(route_key, items).await?;
        if carries_target_row {
            self.armed.store(true, Ordering::SeqCst);
        }
        Ok(())
    }

    async fn get_routed(
        &self,
        route_key: &str,
        key: &[u8],
    ) -> Result<Option<Vec<u8>>, liminal::durability::DurabilityError> {
        let value = self.inner.get_routed(route_key, key).await?;
        Ok(value.map(|value| self.corrupt_routed(key, value)))
    }

    async fn append(
        &self,
        stream_key: &str,
        payload: Vec<u8>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        if stream_key == self.target_key && matches!(self.mode, FaultMode::RefuseAppend) {
            return Err(DurabilityError::SequenceConflict {
                expected: expected_seq,
                actual: expected_seq.saturating_add(1),
            });
        }
        let sequence = self.inner.append(stream_key, payload, expected_seq).await?;
        if stream_key == self.target_key {
            self.armed.store(true, Ordering::SeqCst);
        }
        Ok(sequence)
    }

    async fn append_all(
        &self,
        stream_key: &str,
        payloads: Vec<Vec<u8>>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        if stream_key == self.target_key && matches!(self.mode, FaultMode::RefuseAppend) {
            return Err(DurabilityError::SequenceConflict {
                expected: expected_seq,
                actual: expected_seq.saturating_add(1),
            });
        }
        let sequence = self
            .inner
            .append_all(stream_key, payloads, expected_seq)
            .await?;
        if stream_key == self.target_key {
            self.armed.store(true, Ordering::SeqCst);
        }
        Ok(sequence)
    }

    async fn read_from(
        &self,
        stream_key: &str,
        offset: u64,
        limit: usize,
    ) -> Result<Vec<StoredEntry>, DurabilityError> {
        let entries = self.inner.read_from(stream_key, offset, limit).await?;
        Ok(self.corrupt(stream_key, entries))
    }

    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
    }
}

/// Enrols one conversation through the production request seam.
pub(super) fn enroll(
    handler: &ProductionParticipantHandler,
    conversation_id: u64,
    token: [u8; 16],
    connection: ConnectionIncarnation,
) -> Result<ServerValue, Box<dyn Error>> {
    let mut conversations = ParticipantConnectionConversations::default();
    let value = dispatch_tracked(
        handler,
        connection,
        &mut conversations,
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id,
            enrollment_token: EnrollmentToken::new(token),
        }),
    )?;
    Ok(value)
}

/// Seeds two independent conversations through the production request seam and
/// returns the shared store they were written to.
pub(super) fn seed_two_conversations() -> Result<Arc<dyn DurableStore>, Box<dyn Error>> {
    let store: Arc<dyn DurableStore> = Arc::new(open_ephemeral(1)?);
    let handler =
        ProductionParticipantHandler::new_restored(Arc::clone(&store), test_participant_config())?;

    for (conversation_id, token, incarnation) in [
        (POISONED_CONVERSATION, POISONED_TOKEN, 1_u64),
        (HEALTHY_CONVERSATION, HEALTHY_TOKEN, 2_u64),
    ] {
        let enrolled = enroll(
            &handler,
            conversation_id,
            token,
            ConnectionIncarnation::new(701, incarnation),
        )?;
        if !matches!(enrolled, ServerValue::EnrollBound(_)) {
            return Err(format!("seed enrollment did not bind: {enrolled:?}").into());
        }
    }

    drop(handler);
    Ok(store)
}

/// The write-side shape `pending_restart_fixture` bound. A boot over its bytes
/// must present the same configuration for its replay audits to hold.
fn pending_boot_config() -> ParticipantConfig {
    let mut config = test_participant_config();
    config.max_retained_record_rows = 4;
    config
}

/// Seeds the store the two drain-route arms boot over: one conversation whose
/// immutable-candidate lane holds a pending binding terminal (so boot's drain
/// runs, reaches `:265`, and on success completes its commit tail), plus one ordinary
/// healthy neighbour on the same store to carry the containment assertion.
fn pending_lane_store_with_a_healthy_neighbour() -> Result<Arc<dyn DurableStore>, Box<dyn Error>> {
    let fixture = pending_restart_fixture()?;
    if fixture.conversation_id != PENDING_CONVERSATION {
        return Err(format!(
            "the pending-restart fixture moved to conversation {}; this file targets \
             {PENDING_CONVERSATION} and would otherwise fault a stream nothing reads",
            fixture.conversation_id
        )
        .into());
    }
    let enrolled = enroll(
        &fixture.handler,
        HEALTHY_CONVERSATION,
        HEALTHY_TOKEN,
        ConnectionIncarnation::new(711, 1),
    )?;
    if !matches!(enrolled, ServerValue::EnrollBound(_)) {
        return Err(format!("neighbour enrollment did not bind: {enrolled:?}").into());
    }
    let store = Arc::clone(&fixture.handler.store);
    drop(fixture);
    Ok(store)
}

/// Asserts the healthy neighbour still answers on a booted handler.
pub(super) fn healthy_neighbour_is_served(
    handler: &ProductionParticipantHandler,
    connection_id: u64,
) -> Result<(), Box<dyn Error>> {
    let served = enroll(
        handler,
        HEALTHY_CONVERSATION,
        [0xC5; 16],
        ConnectionIncarnation::new(connection_id, 1),
    )?;
    assert!(
        matches!(served, ServerValue::EnrollBound(_)),
        "CONTAINMENT: the healthy conversation {HEALTHY_CONVERSATION} stopped serving because \
         another conversation on the same store is unloadable: {served:?}"
    );
    Ok(())
}

// ─────────────────────────────────────────────────────────────────────────────
// ROUTE handler.rs:250 — replay refuses the conversation's own log
// ─────────────────────────────────────────────────────────────────────────────

/// The control: the identical fixture with NOTHING corrupted.
///
/// Without this arm a failure below would prove only that the fixture cannot
/// boot, not that corruption is what stops it.
#[test]
fn control_uncorrupted_store_boots_and_serves_both_conversations() -> Result<(), Box<dyn Error>> {
    let store = seed_two_conversations()?;
    let handler =
        ProductionParticipantHandler::new_restored(Arc::clone(&store), test_participant_config())?;
    let ids = handler.registered_conversation_ids()?;
    assert!(
        ids.contains(&POISONED_CONVERSATION) && ids.contains(&HEALTHY_CONVERSATION),
        "control boot lost a conversation: {ids:?}"
    );
    Ok(())
}

/// ASSERTION 1 — THE NODE STILL BOOTS.
///
/// Fails today at `handler.rs:250`'s `?`, which climbs to `:137` inside
/// `ProductionParticipantHandler::new`.
#[test]
fn one_unloadable_record_must_not_prevent_the_node_from_starting() -> Result<(), Box<dyn Error>> {
    let inner = seed_two_conversations()?;
    let poisoned: Arc<dyn DurableStore> = Arc::new(OneConversationFault::new(
        inner,
        POISONED_CONVERSATION,
        FaultMode::CorruptRecordNow(0),
    ));

    let outcome = ProductionParticipantHandler::new_restored(poisoned, test_participant_config());

    assert!(
        outcome.is_ok(),
        "CONTAINMENT: one unloadable record on conversation {POISONED_CONVERSATION} took the \
         whole node down. The boot loop at handler.rs:250 propagated a per-conversation replay \
         failure out of a loop over every conversation. Error: {:?}",
        outcome.err()
    );
    Ok(())
}

/// ASSERTION 2 — THE OTHER CONVERSATIONS ARE STILL SERVED.
#[test]
fn one_unloadable_conversation_must_not_take_down_the_others() -> Result<(), Box<dyn Error>> {
    let inner = seed_two_conversations()?;
    let poisoned: Arc<dyn DurableStore> = Arc::new(OneConversationFault::new(
        inner,
        POISONED_CONVERSATION,
        FaultMode::CorruptRecordNow(0),
    ));

    let booted = ProductionParticipantHandler::new_restored(poisoned, test_participant_config());
    let handler = booted
        .map_err(|error| format!("node did not boot (assertion 1 already covers this): {error}"))?;

    healthy_neighbour_is_served(&handler, 702)
}

/// ASSERTION 3 — THE REFUSAL NAMES THE CONVERSATION IT REFUSED.
///
/// This is the assertion that stops containment becoming a silent-success
/// property. A node that boots clean and serves nothing on one conversation,
/// without ever naming it, is worse than the crash it replaced.
#[test]
fn the_refusal_must_name_the_conversation_it_refused() -> Result<(), Box<dyn Error>> {
    let inner = seed_two_conversations()?;
    let poisoned: Arc<dyn DurableStore> = Arc::new(OneConversationFault::new(
        inner,
        POISONED_CONVERSATION,
        FaultMode::CorruptRecordNow(0),
    ));

    let booted = ProductionParticipantHandler::new_restored(poisoned, test_participant_config());
    let handler = booted
        .map_err(|error| format!("node did not boot (assertion 1 already covers this): {error}"))?;

    let refusal = enroll(
        &handler,
        POISONED_CONVERSATION,
        [0xC4; 16],
        ConnectionIncarnation::new(703, 1),
    );

    let error = match refusal {
        Ok(value) => {
            return Err(format!(
                "the poisoned conversation answered a request instead of refusing: {value:?}"
            )
            .into());
        }
        Err(error) => error.to_string(),
    };

    assert!(
        error.contains(&POISONED_CONVERSATION.to_string()),
        "ATTRIBUTION: the refusal names the invariant and not its subject. Nothing in this \
         message identifies conversation {POISONED_CONVERSATION}: {error}"
    );
    Ok(())
}

// ─────────────────────────────────────────────────────────────────────────────
// ROUTE handler.rs:265 — the boot drain's verdict refuses the boot
// ─────────────────────────────────────────────────────────────────────────────

/// The control for BOTH drain-route arms: the pending-terminal fixture plus its
/// healthy neighbour boots clean with no fault at all.
///
/// This is what makes the two reds below mean something. Without it they would
/// prove only that a lane-holding fixture and an ordinary enrolment cannot
/// share a store.
#[test]
fn control_pending_terminal_lane_and_neighbour_boot_clean() -> Result<(), Box<dyn Error>> {
    let store = pending_lane_store_with_a_healthy_neighbour()?;
    let handler =
        ProductionParticipantHandler::new_restored(Arc::clone(&store), pending_boot_config())?;
    let ids = handler.registered_conversation_ids()?;
    assert!(
        ids.contains(&PENDING_CONVERSATION) && ids.contains(&HEALTHY_CONVERSATION),
        "control boot lost a conversation: {ids:?}"
    );
    healthy_neighbour_is_served(&handler, 712)
}

/// ROUTE `handler.rs:265` — one conversation whose boot drain cannot persist
/// takes the whole node down.
///
/// The drain decides on a drain it cannot append, so the verdict is
/// `RefusedShape` and `verdict.observe()?` propagates out of the loop. A fix
/// applied at `:250` alone does not touch this route.
#[test]
fn a_refused_boot_drain_must_not_take_down_the_node() -> Result<(), Box<dyn Error>> {
    let inner = pending_lane_store_with_a_healthy_neighbour()?;
    let faulted: Arc<dyn DurableStore> = Arc::new(OneConversationFault::new(
        inner,
        PENDING_CONVERSATION,
        FaultMode::RefuseAppend,
    ));

    match ProductionParticipantHandler::new_restored(faulted, pending_boot_config()) {
        Ok(handler) => healthy_neighbour_is_served(&handler, 713),
        Err(ParticipantSemanticError::BootDrainRefused {
            conversation_id,
            refusal,
            reason,
            ..
        }) => {
            panic!(
                "CONTAINMENT: the boot drain refused conversation {conversation_id} \
                 ({refusal:?}: {reason}) and handler.rs:265 propagated that refusal out of a loop \
                 over every conversation, so the whole node did not start."
            )
        }
        Err(other) => Err(format!(
            "ROUTE AXIS BROKEN, NOT A CONTAINMENT FAILURE: this arm exists to drive \
             handler.rs:265, the boot-drain verdict, but the fault took a different route and \
             surfaced as: {other}. Redesign the fault — do not read this as coverage of :265."
        )
        .into()),
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// AFTER THE DRAIN — boot reads nothing back
// ─────────────────────────────────────────────────────────────────────────────

/// THE ROUTE DISCRIMINATOR, PROVED IN THE NEGATIVE DIRECTION.
///
/// `CorruptAfterAppend` must be INERT on a store where boot appends nothing:
/// no append, no arming, no corrupted read, clean boot. That is what licenses
/// the arm below to read "armed, never read" as boot appending and then never
/// reading its own log back.
#[test]
fn control_a_deferred_fault_is_inert_when_boot_appends_nothing() -> Result<(), Box<dyn Error>> {
    let inner = seed_two_conversations()?;
    let fault = Arc::new(OneConversationFault::new(
        inner,
        POISONED_CONVERSATION,
        FaultMode::CorruptAfterAppend(0),
    ));
    let armed = Arc::clone(&fault.armed);
    let corrupted_reads = Arc::clone(&fault.corrupted_reads);
    let store: Arc<dyn DurableStore> = fault;

    ProductionParticipantHandler::new_restored(store, test_participant_config()).map_err(
        |error| format!("a deferred fault that never armed still broke the boot: {error}"),
    )?;

    assert!(
        !armed.load(Ordering::SeqCst),
        "the deferred fault armed on a boot that was supposed to append nothing"
    );
    assert_eq!(
        corrupted_reads.load(Ordering::SeqCst),
        0,
        "the deferred fault served a corrupted read before it was armed"
    );
    Ok(())
}

/// Boot's drain completes each appended row's extension row in place and runs
/// only the commit tail after it, so nothing reads the conversation's log back
/// after the drain appends: a log that stops being readable after that append
/// cannot take the node down at boot, and the neighbour is served.
///
/// The fault arms on the drain's append and corrupts every later read of the
/// pending conversation's log; the proof is that it armed (the drain appended)
/// and that boot served no corrupted read at all.
#[test]
fn a_post_drain_boot_reads_nothing_back_and_serves_the_node() -> Result<(), Box<dyn Error>> {
    let inner = pending_lane_store_with_a_healthy_neighbour()?;
    let fault = Arc::new(OneConversationFault::new(
        inner,
        PENDING_CONVERSATION,
        FaultMode::CorruptAfterAppend(0),
    ));
    let armed = Arc::clone(&fault.armed);
    let corrupted_reads = Arc::clone(&fault.corrupted_reads);
    let store: Arc<dyn DurableStore> = fault;

    let handler = ProductionParticipantHandler::new_restored(store, pending_boot_config())?;
    assert!(
        armed.load(Ordering::SeqCst),
        "boot never appended to conversation {PENDING_CONVERSATION}, so this fixture did not \
         drain; redesign it"
    );
    assert_eq!(
        corrupted_reads.load(Ordering::SeqCst),
        0,
        "boot read conversation {PENDING_CONVERSATION}'s log back after its own drain append"
    );
    healthy_neighbour_is_served(&handler, 714)
}