liminal-server 0.24.9

//! DUR-024: a checkpoint over the bound is published in batches, each under
//! the bound, the locator last (R1); the bound is the store's own, and one
//! item over it is refused by name before anything is submitted (R2); a
//! refused publication leaves the conversation serving, no write waits
//! behind a batch, the next crossing retries, and a contained conversation
//! is restored once per process (R3); the census counts what a publication
//! would write without writing (R5).
//!
//! Every count here is read from the memory store's commit log or the
//! handler's own counters, after the publisher thread is joined; nothing is
//! timed and nothing sleeps.

use std::collections::BTreeSet;
use std::error::Error;
use std::io;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Condvar, Mutex};

use liminal::durability::bridge::block_on;
use liminal::durability::{
    DurabilityError, DurableStore, RoutedBatchEntry, RoutedBatchItem, ScanCursor, ScanPage,
    StoredEntry,
};
use liminal_protocol::wire::{
    ClientRequest, ConnectionIncarnation, EnrollmentRequest, EnrollmentToken, Generation,
    RecordAdmission, RecordAdmissionAttemptToken, ServerValue,
};

use super::ProductionParticipantHandler;
use super::checkpoint_batches::{
    CHECKPOINT_BATCH_BYTES, CheckpointPublication, CheckpointPublishError, PublicationState,
    checkpoint_census, commit_count, commit_publication, publication_bound, split_history,
};
use super::operation_batch::OperationBatch;

#[path = "tests_checkpoint_batches_reads.rs"]
mod reads;
use super::restore_checkpoint_codec;
use super::restore_checkpoint_history::{CheckpointHistoryRoot, ROOT_ENCODED_BYTES};
use super::restore_checkpoint_locator::{CheckpointLocator, checkpoint_conversation_route};
use super::restore_from_checkpoint::{CheckpointDecision, load_checkpoint};
use super::routed_stream_codec;
use super::state::ConversationAuthority;
use super::state::checkpoint_owner::ConversationAuthorityRestoreContext;
use super::tests::{dispatch, test_participant_config};
use super::tests_boot_containment::{
    FaultMode, HEALTHY_CONVERSATION, OneConversationFault, POISONED_CONVERSATION, enroll,
    healthy_neighbour_is_served, seed_two_conversations,
};
use super::tests_memory_store::{CommitRecord, MemoryStore};
use crate::server::participant::ParticipantSemanticHandler;
use crate::server::participant::incarnation_stream::{
    ConnectionFateClass, IncarnationAllocation, IncarnationStartup, IncarnationStream,
};
use reads::{NodeReads, RowReads};

type TestResult = Result<(), Box<dyn Error>>;

const CONVERSATION: u64 = 10;
const NODE_KEY_PREFIX: &[u8] = b"checkpoint/history/node/v2/";

fn is_row_key(key: &[u8]) -> bool {
    routed_stream_codec::row_stream(key).is_some()
}

fn is_node_key(key: &[u8]) -> bool {
    key.starts_with(NODE_KEY_PREFIX)
}

fn is_locator_key(key: &[u8]) -> bool {
    key == CheckpointLocator::routed_key(CONVERSATION)
}

/// A publication batch carries history nodes or the locator and no row of
/// an operation; an operation's own commit carries its rows.
fn is_publication_batch(items: &[RoutedBatchItem]) -> bool {
    !items.iter().any(|item| is_row_key(&item.key))
        && items
            .iter()
            .any(|item| is_node_key(&item.key) || is_locator_key(&item.key))
}

/// The memory store with a script over publication batches: the Nth
/// publication batch is refused once, and every publication batch waits at
/// the gate while it is closed. Row reads are counted for the replay pin.
#[derive(Debug)]
struct Scripted {
    inner: MemoryStore,
    refuse_at: Mutex<Option<usize>>,
    publication_batches: AtomicUsize,
    gate_closed: Mutex<bool>,
    gate: Condvar,
    row_reads: AtomicUsize,
}

impl Scripted {
    fn new(inner: MemoryStore) -> Self {
        Self {
            inner,
            refuse_at: Mutex::new(None),
            publication_batches: AtomicUsize::new(0),
            gate_closed: Mutex::new(false),
            gate: Condvar::new(),
            row_reads: AtomicUsize::new(0),
        }
    }

    fn refuse_publication_batch(&self, index: usize) -> TestResult {
        *self.refuse_at.lock().map_err(|_| "script lock poisoned")? = Some(index);
        Ok(())
    }

    fn close_gate(&self) -> TestResult {
        *self.gate_closed.lock().map_err(|_| "gate lock poisoned")? = true;
        Ok(())
    }

    fn open_gate(&self) -> TestResult {
        *self.gate_closed.lock().map_err(|_| "gate lock poisoned")? = false;
        self.gate.notify_all();
        Ok(())
    }

    fn wait_at_gate(&self) -> Result<(), DurabilityError> {
        let mut closed = self
            .gate_closed
            .lock()
            .map_err(|_| DurabilityError::ConfigError("gate lock poisoned".into()))?;
        while *closed {
            closed = self
                .gate
                .wait(closed)
                .map_err(|_| DurabilityError::ConfigError("gate lock poisoned".into()))?;
        }
        drop(closed);
        Ok(())
    }

    /// Counts a publication batch, holds it at the gate and refuses the
    /// scripted one by name.
    fn admit(&self, items: &[RoutedBatchItem]) -> Result<(), DurabilityError> {
        if !is_publication_batch(items) {
            return Ok(());
        }
        let index = self.publication_batches.fetch_add(1, Ordering::SeqCst) + 1;
        self.wait_at_gate()?;
        let refused = {
            let mut refuse_at = self
                .refuse_at
                .lock()
                .map_err(|_| DurabilityError::ConfigError("script lock poisoned".into()))?;
            if *refuse_at == Some(index) {
                *refuse_at = None;
                true
            } else {
                false
            }
        };
        if refused {
            let bytes: usize = items.iter().map(|item| item.value.len()).sum();
            return Err(DurabilityError::ConfigError(format!(
                "scripted shard_queue_full: shard=1 queued_bytes=0 request_bytes={bytes} bound=scripted"
            )));
        }
        Ok(())
    }

    fn commits(&self) -> Result<Vec<CommitRecord>, Box<dyn Error>> {
        Ok(self.inner.commits()?)
    }

    fn publication_commits(&self) -> Result<Vec<CommitRecord>, Box<dyn Error>> {
        Ok(self
            .commits()?
            .into_iter()
            .filter(|record| {
                !record.carries_any(is_row_key)
                    && record.carries_any(|key| is_node_key(key) || is_locator_key(key))
            })
            .collect())
    }
}

#[async_trait::async_trait]
impl DurableStore for Scripted {
    fn max_request_bytes(&self) -> Option<usize> {
        self.inner.max_request_bytes()
    }

    async fn apply_routed_batch(
        &self,
        route_key: &str,
        items: Vec<RoutedBatchItem>,
    ) -> Result<(), DurabilityError> {
        self.admit(&items)?;
        self.inner.apply_routed_batch(route_key, items).await
    }

    /// A publication with deletes (DUR-028 R3) is scripted by its puts, as
    /// a put-only publication is.
    async fn apply_routed_entries(
        &self,
        route_key: &str,
        entries: Vec<RoutedBatchEntry>,
    ) -> Result<(), DurabilityError> {
        let puts: Vec<RoutedBatchItem> = entries
            .iter()
            .filter_map(|entry| match entry {
                RoutedBatchEntry::Put(item) => Some(item.clone()),
                RoutedBatchEntry::Delete { .. } => None,
            })
            .collect();
        self.admit(&puts)?;
        self.inner.apply_routed_entries(route_key, entries).await
    }

    async fn get_routed(
        &self,
        route_key: &str,
        key: &[u8],
    ) -> Result<Option<Vec<u8>>, DurabilityError> {
        if is_row_key(key) {
            self.row_reads.fetch_add(1, Ordering::SeqCst);
        }
        self.inner.get_routed(route_key, key).await
    }

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

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

    async fn read_from(
        &self,
        stream_key: &str,
        offset: u64,
        limit: usize,
    ) -> Result<Vec<StoredEntry>, DurabilityError> {
        let entries = self.inner.read_from(stream_key, offset, limit).await?;
        self.row_reads.fetch_add(entries.len(), Ordering::SeqCst);
        Ok(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_routed(
        &self,
        route_key: &str,
        from: &[u8],
        to: &[u8],
    ) -> Result<Vec<(Vec<u8>, Vec<u8>)>, DurabilityError> {
        self.inner.scan_routed(route_key, from, to).await
    }

    async fn remove_routed_range(
        &self,
        route_key: &str,
        from: &[u8],
        to: &[u8],
    ) -> Result<u64, DurabilityError> {
        self.inner.remove_routed_range(route_key, from, to).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
    }
}

/// A serving conversation: two enrolled participants on conversation 10,
/// the clock pinned, admissions numbered from 0.
struct Fixture {
    store: Arc<Scripted>,
    handler: ProductionParticipantHandler,
    connection: ConnectionIncarnation,
    participant_id: u64,
}

fn fixture(store: Scripted) -> Result<Fixture, Box<dyn Error>> {
    let store = Arc::new(store);
    let handler = ProductionParticipantHandler::new(
        Arc::clone(&store) as Arc<dyn DurableStore>,
        test_participant_config(),
    )?;
    handler.pin_clock_ms(1_000);
    let connection = ConnectionIncarnation::new(CONVERSATION, 1);
    let enrolled = dispatch(
        &handler,
        connection,
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id: CONVERSATION,
            enrollment_token: EnrollmentToken::new([1; 16]),
        }),
    )?;
    let ServerValue::EnrollBound(bound) = enrolled else {
        return Err(format!("enrollment did not bind: {enrolled:?}").into());
    };
    let recipient = dispatch(
        &handler,
        ConnectionIncarnation::new(CONVERSATION, 2),
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id: CONVERSATION,
            enrollment_token: EnrollmentToken::new([2; 16]),
        }),
    )?;
    assert!(matches!(recipient, ServerValue::EnrollBound(_)));
    Ok(Fixture {
        store,
        handler,
        connection,
        participant_id: bound.participant_id(),
    })
}

impl Fixture {
    fn admit(&self, operation: u64) -> Result<ServerValue, Box<dyn Error>> {
        let mut token = [0; 16];
        token[..8].copy_from_slice(&operation.to_be_bytes());
        dispatch(
            &self.handler,
            self.connection,
            ClientRequest::RecordAdmission(RecordAdmission {
                conversation_id: CONVERSATION,
                participant_id: self.participant_id,
                capability_generation: Generation::ONE,
                record_admission_attempt_token: RecordAdmissionAttemptToken::new(token),
                payload: operation.to_be_bytes().to_vec(),
            }),
        )
    }

    /// Admits `operations`, each answered as a live commit.
    fn admit_all(&self, operations: std::ops::Range<u64>) -> Result<usize, Box<dyn Error>> {
        let mut committed = 0;
        for operation in operations {
            let answer = self.admit(operation)?;
            if !matches!(answer, ServerValue::RecordCommitted(_)) {
                return Err(format!("operation {operation} was not committed: {answer:?}").into());
            }
            committed += 1;
        }
        Ok(committed)
    }

    fn join(&self) -> TestResult {
        self.handler.publisher.join_all()?;
        Ok(())
    }

    fn lock_takes(&self) -> u64 {
        self.handler.owner_lock_takes.load(Ordering::SeqCst)
    }

    fn state(&self) -> Option<PublicationState> {
        self.handler.publisher.ledger().state(CONVERSATION)
    }

    fn locator(&self) -> Result<Option<CheckpointLocator>, Box<dyn Error>> {
        let route = checkpoint_conversation_route(CONVERSATION);
        let bytes = block_on(
            self.store
                .get_routed(&route, &CheckpointLocator::routed_key(CONVERSATION)),
        )??;
        bytes
            .map(|bytes| CheckpointLocator::decode_canonical(&bytes, CONVERSATION))
            .transpose()
            .map_err(Into::into)
    }

    fn body_at_locator(&self) -> Result<Vec<u8>, Box<dyn Error>> {
        let locator = self.locator()?.ok_or("no locator is published")?;
        let route = checkpoint_conversation_route(CONVERSATION);
        block_on(
            self.store
                .get_routed(&route, &locator.body_key().routed_key(CONVERSATION)),
        )??
        .ok_or_else(|| "the locator's body is absent".into())
    }

    fn load_decision(&self) -> Result<Result<u64, String>, Box<dyn Error>> {
        let config = test_participant_config();
        let loaded = block_on(
            load_checkpoint::<ConversationAuthority, ROOT_ENCODED_BYTES>(
                self.store.as_ref(),
                CONVERSATION,
                |positions, root| {
                    ConversationAuthorityRestoreContext::new(*positions, root, &config)
                },
                |locator, root| {
                    CheckpointHistoryRoot::decode_canonical(root, locator.conversation_id())
                        .map(|_| ())
                        .map_err(
                            |source| super::restore_checkpoint::CheckpointError::LiveState {
                                source: Box::new(source),
                            },
                        )
                },
            ),
        )??;
        Ok(match loaded.decision {
            CheckpointDecision::Restore(checkpoint) => {
                Ok(checkpoint.positions().covered_log_sequence)
            }
            CheckpointDecision::Rebuild(required) => Err(required.refusal.to_string()),
        })
    }
}

/// What the first crossing of an unbounded conversation writes: the history
/// bytes and the body-plus-locator bytes, read from the one commit that
/// carried the locator.
struct Measured {
    history_bytes: usize,
    body_bytes: usize,
}

fn measure() -> Result<Measured, Box<dyn Error>> {
    let fixture = fixture(Scripted::new(MemoryStore::default()))?;
    fixture.admit_all(0..256)?;
    fixture.join()?;
    let locator = fixture
        .locator()?
        .ok_or("the unbounded crossing published no locator")?;
    let body_key = locator.body_key().routed_key(CONVERSATION);
    let commit = fixture
        .store
        .commits()?
        .into_iter()
        .find(|record| record.carries(&CheckpointLocator::routed_key(CONVERSATION)))
        .ok_or("no commit carried the locator")?;
    let history_bytes = commit
        .items
        .iter()
        .filter(|(key, _)| is_node_key(key))
        .map(|(_, bytes)| bytes)
        .sum();
    let body_bytes = commit
        .items
        .iter()
        .filter(|(key, _)| *key == body_key || is_locator_key(key))
        .map(|(_, bytes)| bytes)
        .sum();
    Ok(Measured {
        history_bytes,
        body_bytes,
    })
}

/// A batch target the measured history is two and a half times: three
/// history batches (two full, one half), then the final batch with the body
/// and the locator on its own, since the body is larger than the target.
fn bound_for(measured: &Measured) -> Result<usize, Box<dyn Error>> {
    let bound = measured.history_bytes * 2 / 5;
    if measured.body_bytes * 4 > STORE_LIMIT {
        return Err(format!(
            "the fixture's body ({} bytes) is too close to the store limit {STORE_LIMIT}",
            measured.body_bytes
        )
        .into());
    }
    Ok(bound)
}

/// What the bounded fixtures' store admits in one request: room for the
/// body (DUR-024: one item may use the store's whole bound).
const STORE_LIMIT: usize = 4 * 1024 * 1024;

fn bounded(bound: usize) -> Scripted {
    super::checkpoint_batches::TEST_BATCH_BYTES.store(bound, Ordering::SeqCst);
    Scripted::new(MemoryStore::with_request_bound(STORE_LIMIT))
}

/// A global log capture: the publisher's lines come from its own thread, so
/// a thread-local subscriber would miss them. nextest runs one test per
/// process, so installing the global subscriber once per test is exact.
#[derive(Clone, Default)]
struct Captured(Arc<Mutex<Vec<u8>>>);

impl Captured {
    fn install() -> Result<Self, Box<dyn Error>> {
        let captured = Self::default();
        let subscriber = tracing_subscriber::fmt()
            .without_time()
            .with_ansi(false)
            .with_max_level(tracing::Level::INFO)
            .with_writer(captured.clone())
            .finish();
        tracing::subscriber::set_global_default(subscriber)
            .map_err(|_| "a global subscriber is already installed; run under nextest")?;
        Ok(captured)
    }

    fn text(&self) -> Result<String, Box<dyn Error>> {
        Ok(String::from_utf8(
            self.0.lock().map_err(|_| "captured log poisoned")?.clone(),
        )?)
    }
}

impl io::Write for Captured {
    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
        self.0
            .lock()
            .map_err(|_| io::Error::other("captured log poisoned"))?
            .extend_from_slice(bytes);
        Ok(bytes.len())
    }

    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for Captured {
    type Writer = Self;

    fn make_writer(&'a self) -> Self {
        self.clone()
    }
}

// ─── R1 ─────────────────────────────────────────────────────────────────────

#[test]
fn checkpoint_over_the_bound_publishes_in_bounded_batches() -> TestResult {
    let measured = measure()?;
    let bound = bound_for(&measured)?;
    let fixture = fixture(bounded(bound))?;
    assert_eq!(fixture.admit_all(0..256)?, 256);
    fixture.join()?;
    let commits = fixture.store.publication_commits()?;
    assert_eq!(
        commits.len(),
        4,
        "history {} bytes at a bound of {bound} with a {}-byte body: {:?}",
        measured.history_bytes,
        measured.body_bytes,
        commits
            .iter()
            .map(|record| record.value_bytes)
            .collect::<Vec<_>>()
    );
    for (index, record) in commits.iter().enumerate() {
        let limit = if index == 3 { STORE_LIMIT } else { bound };
        assert!(
            record.value_bytes <= limit,
            "batch {} carries {} bytes over its limit {limit}",
            index + 1,
            record.value_bytes
        );
        let carries_locator = record.carries(&CheckpointLocator::routed_key(CONVERSATION));
        assert_eq!(
            carries_locator,
            index == 3,
            "the locator is in the fourth commit and absent before it: batch {}",
            index + 1
        );
    }
    assert!(matches!(
        fixture.state(),
        Some(PublicationState::Published(_))
    ));
    assert_eq!(
        fixture.locator()?.map(CheckpointLocator::covered_sequence),
        Some(255)
    );
    Ok(())
}

#[test]
fn interrupted_publication_resumes_with_absent_nodes_only() -> TestResult {
    let measured = measure()?;
    let bound = bound_for(&measured)?;
    let fixture = fixture(bounded(bound))?;
    fixture.store.refuse_publication_batch(3)?;
    fixture.admit_all(0..256)?;
    fixture.join()?;
    let first = fixture.store.publication_commits()?;
    assert_eq!(
        first.len(),
        2,
        "two batches landed before the third was refused"
    );
    let Some(PublicationState::Owed(refusal)) = fixture.state() else {
        return Err(format!("the checkpoint is not owed: {:?}", fixture.state()).into());
    };
    assert_eq!(refusal.batch_index, 3);
    assert_eq!(
        fixture.locator()?,
        None,
        "no locator before the final batch"
    );
    let written: BTreeSet<Vec<u8>> = first
        .iter()
        .flat_map(|record| record.items.iter().map(|(key, _)| key.clone()))
        .collect();
    assert!(!written.is_empty());

    // The owner kept serving (R3); the next crossing publishes the rest.
    assert_eq!(fixture.admit_all(256..512)?, 256);
    fixture.join()?;
    assert!(
        matches!(fixture.state(), Some(PublicationState::Published(_))),
        "{:?}",
        fixture.state()
    );
    let all = fixture.store.publication_commits()?;
    let second = &all[first.len()..];
    assert!(!second.is_empty());
    let rewritten = second
        .iter()
        .flat_map(|record| record.items.iter().map(|(key, _)| key))
        .filter(|key| written.contains(*key))
        .count();
    assert_eq!(
        rewritten, 0,
        "a node the first two commits wrote is never staged again"
    );
    assert_eq!(
        fixture.locator()?.map(CheckpointLocator::covered_sequence),
        Some(511)
    );
    assert_eq!(fixture.load_decision()?, Ok(511));

    // The same checkpoint as an uninterrupted publication of the same
    // operations: the same positions and the same history root (the same
    // node set), and a load answers the same covered sequence. The live
    // bodies are not compared byte for byte: two clean runs of this fixture
    // already differ in one 32-byte field the lifecycle capsule carries for
    // participant 1 (measured at offset 302 of a 145,245-byte body), which
    // is per run, not per path.
    let clean = fixture_at_512(bound)?;
    assert_eq!(
        clean.locator()?.map(CheckpointLocator::covered_sequence),
        Some(511)
    );
    assert_eq!(clean.load_decision()?, Ok(511));
    let (interrupted_body, clean_body) = (fixture.body_at_locator()?, clean.body_at_locator()?);
    let (interrupted_positions, interrupted_root, interrupted_live) =
        restore_checkpoint_codec::decode::<ROOT_ENCODED_BYTES>(&interrupted_body, CONVERSATION)?;
    let (clean_positions, clean_root, clean_live) =
        restore_checkpoint_codec::decode::<ROOT_ENCODED_BYTES>(&clean_body, CONVERSATION)?;
    assert_eq!(interrupted_positions, clean_positions);
    assert_eq!(
        interrupted_root, clean_root,
        "the resumed publication's history root is the uninterrupted one"
    );
    assert_eq!(interrupted_live.len(), clean_live.len());
    Ok(())
}

fn fixture_at_512(bound: usize) -> Result<Fixture, Box<dyn Error>> {
    let clean = fixture(bounded(bound))?;
    clean.admit_all(0..512)?;
    clean.join()?;
    Ok(clean)
}

#[test]
fn a_checkpoint_under_the_bound_is_one_commit_as_today() -> TestResult {
    let fixture = fixture(Scripted::new(MemoryStore::default()))?;
    assert_eq!(
        publication_bound(fixture.store.as_ref()).bound,
        CHECKPOINT_BATCH_BYTES
    );
    let before = fixture.store.commits()?.len();
    assert_eq!(fixture.admit_all(0..256)?, 256);
    fixture.join()?;
    let commits = fixture.store.commits()?;
    assert_eq!(
        commits.len() - before,
        256,
        "every operation commits once, the crossing included"
    );
    assert!(
        fixture.store.publication_commits()?.is_empty(),
        "nothing is published apart from the operation's own commit"
    );
    let crossing = commits
        .iter()
        .find(|record| record.carries(&CheckpointLocator::routed_key(CONVERSATION)))
        .ok_or("no commit carried the locator")?;
    assert!(
        crossing.carries_any(is_row_key),
        "the crossing's rows ride with its checkpoint"
    );
    assert!(crossing.carries_any(is_node_key), "and its history");
    assert_eq!(fixture.state(), None, "nothing was handed off");
    Ok(())
}

#[test]
fn locator_is_never_written_before_its_nodes() -> TestResult {
    let measured = measure()?;
    let bound = bound_for(&measured)?;
    let fixture = fixture(bounded(bound))?;
    fixture.store.refuse_publication_batch(4)?;
    fixture.admit_all(0..256)?;
    fixture.join()?;
    let commits = fixture.store.publication_commits()?;
    assert_eq!(commits.len(), 3);
    assert!(
        commits
            .iter()
            .all(|record| !record.carries(&CheckpointLocator::routed_key(CONVERSATION))),
        "the locator rides in the final batch only"
    );
    assert_eq!(
        fixture.load_decision()?,
        Err("checkpoint locator is absent".to_owned()),
        "a load finds no locator, never a locator whose nodes are missing"
    );
    Ok(())
}

// ─── R2 ─────────────────────────────────────────────────────────────────────

#[test]
fn store_bound_is_read_not_assumed() -> TestResult {
    const MIB: usize = 1024 * 1024;
    let captured = Captured::install()?;
    let store = MemoryStore::with_request_bound(8 * MIB);
    let bound = publication_bound(&store);
    assert_eq!(bound.bound, 8 * MIB);
    assert_eq!(bound.pinned, 16 * MIB);
    assert_eq!(bound.store, Some(8 * MIB));
    let ledger = super::checkpoint_batches::PublicationLedger::default();
    ledger.log_bound_once(bound);
    ledger.log_bound_once(bound);
    let route = checkpoint_conversation_route(CONVERSATION);
    let nodes: std::collections::BTreeMap<Vec<u8>, Vec<u8>> = (0..20_u8)
        .map(|index| {
            let mut key = NODE_KEY_PREFIX.to_vec();
            key.push(index);
            (key, vec![index; MIB])
        })
        .collect();
    let mut final_batch = OperationBatch::new(&route);
    final_batch.stage(RoutedBatchItem {
        key: b"checkpoint/tail".to_vec(),
        expected: None,
        value: vec![1; 64],
        ttl: None,
    })?;
    let publication = CheckpointPublication::new(CONVERSATION, &route, nodes, final_batch, bound)?;
    assert_eq!(
        publication.commits(),
        3,
        "8 + 8 + 4 MiB, the tail beside the last"
    );
    let published = block_on(commit_publication(&store, publication))?
        .map_err(|refusal| refusal.to_string())?;
    assert_eq!(published.commits, 3);
    let commits = store.commits()?;
    assert_eq!(commits.len(), 3);
    assert!(commits.iter().all(|record| record.value_bytes <= 8 * MIB));
    let log = captured.text()?;
    assert_eq!(
        log.matches("the store's bound is used").count(),
        1,
        "said once: {log}"
    );
    assert!(log.contains("pinned_batch_bytes=16777216"), "{log}");
    assert!(log.contains("store_max_request_bytes=8388608"), "{log}");
    Ok(())
}

#[test]
fn one_item_over_the_bound_is_refused_by_name() -> TestResult {
    let bound = 4_096;
    let store = MemoryStore::with_request_bound(bound);
    let mut nodes = std::collections::BTreeMap::new();
    let mut key = NODE_KEY_PREFIX.to_vec();
    key.extend_from_slice(b"one");
    nodes.insert(key, vec![9; bound + 1]);
    let refusal = match split_history(nodes, bound, bound) {
        Ok(batches) => {
            return Err(format!("an item over the bound was laid out: {batches:?}").into());
        }
        Err(refusal) => refusal,
    };
    assert_eq!(
        refusal,
        CheckpointPublishError::ItemExceedsStoreBound {
            key: "checkpoint/history/node/v2/one".to_owned(),
            bytes: bound + 1,
            bound,
        }
    );
    assert_eq!(
        refusal.to_string(),
        format!(
            "checkpoint_item_exceeds_store_bound: key=checkpoint/history/node/v2/one bytes={} bound={bound}",
            bound + 1
        )
    );
    assert_eq!(store.commits()?.len(), 0, "nothing was submitted");
    Ok(())
}

// ─── R3 ─────────────────────────────────────────────────────────────────────

#[test]
fn refused_publication_keeps_the_conversation_live() -> TestResult {
    let captured = Captured::install()?;
    let measured = measure()?;
    let bound = bound_for(&measured)?;
    let fixture = fixture(bounded(bound))?;
    fixture.store.close_gate()?;
    fixture.store.refuse_publication_batch(1)?;
    fixture.admit_all(0..256)?;
    // The publisher is at the gate with batch 1; the owner is free.
    assert!(matches!(fixture.state(), Some(PublicationState::InFlight)));
    let takes_before = fixture.lock_takes();
    assert_eq!(
        fixture.admit_all(256..274)?,
        18,
        "eighteen live commits during the publication"
    );
    let takes_during = fixture.lock_takes() - takes_before;
    assert!(
        takes_during >= 18,
        "the eighteen writes took the owner's lock: {takes_during}"
    );
    fixture.store.open_gate()?;
    fixture.join()?;
    assert_eq!(
        fixture.lock_takes() - takes_before,
        takes_during,
        "the publisher took the owner's lock 0 times while submitting"
    );
    let Some(PublicationState::Owed(refusal)) = fixture.state() else {
        return Err(format!("the checkpoint is not owed: {:?}", fixture.state()).into());
    };
    assert_eq!(refusal.batch_index, 1);
    assert!(refusal.bytes > 0 && refusal.bytes <= bound);
    assert_eq!(refusal.bound, bound);
    assert!(
        fixture
            .handler
            .publisher
            .ledger()
            .checkpoint_owed(CONVERSATION)
    );
    assert_eq!(fixture.store.publication_commits()?.len(), 0);
    let log = captured.text()?;
    let refusals: Vec<&str> = log
        .lines()
        .filter(|line| line.contains("checkpoint publication refused by the store"))
        .collect();
    assert_eq!(refusals.len(), 1, "one line names the refusal: {log}");
    assert!(
        refusals[0].contains("conversation_id=10"),
        "{}",
        refusals[0]
    );
    assert!(refusals[0].contains("batch_index=1"), "{}", refusals[0]);
    assert!(
        refusals[0].contains(&format!("bytes={}", refusal.bytes)),
        "{}",
        refusals[0]
    );
    assert!(
        refusals[0].contains(&format!("bound={bound}")),
        "{}",
        refusals[0]
    );
    assert_eq!(log.matches("outcome is unknown").count(), 0, "{log}");
    assert_eq!(log.matches("CHECKPOINT_REBUILD").count(), 0, "{log}");
    assert_eq!(log.matches("CONTAINMENT").count(), 0, "{log}");
    Ok(())
}

#[test]
fn next_crossing_retries_and_succeeds() -> TestResult {
    let measured = measure()?;
    let bound = bound_for(&measured)?;
    let fixture = fixture(bounded(bound))?;
    fixture.store.refuse_publication_batch(1)?;
    fixture.admit_all(0..256)?;
    fixture.join()?;
    assert!(
        fixture
            .handler
            .publisher
            .ledger()
            .checkpoint_owed(CONVERSATION)
    );
    assert_eq!(fixture.locator()?, None);
    assert_eq!(fixture.admit_all(256..512)?, 256);
    fixture.join()?;
    assert!(matches!(
        fixture.state(),
        Some(PublicationState::Published(_))
    ));
    assert_eq!(
        fixture.locator()?.map(CheckpointLocator::covered_sequence),
        Some(511)
    );
    assert_eq!(fixture.load_decision()?, Ok(511));

    // The following boot reads the tail only.
    let store = Arc::clone(&fixture.store);
    drop(fixture);
    store.row_reads.store(0, Ordering::SeqCst);
    let handler = ProductionParticipantHandler::new(
        Arc::clone(&store) as Arc<dyn DurableStore>,
        test_participant_config(),
    )?;
    let cell = handler.existing_cell(CONVERSATION)?;
    assert!(cell.is_some(), "the conversation restored");
    let rows_read = store.row_reads.load(Ordering::SeqCst);
    assert!(
        rows_read < 256,
        "a boot from the locator reads the tail and the drain, never the {} rows behind it: read {rows_read}",
        512
    );
    Ok(())
}

#[test]
fn publication_never_blocks_a_live_write() -> TestResult {
    let measured = measure()?;
    let bound = bound_for(&measured)?;
    let fixture = fixture(bounded(bound))?;
    // The takes one live write costs, measured on a write with nothing in
    // flight.
    let takes = fixture.lock_takes();
    fixture.admit_all(0..1)?;
    let per_write = fixture.lock_takes() - takes;
    assert!(per_write >= 1);
    fixture.store.close_gate()?;
    fixture.admit_all(1..256)?;
    assert!(matches!(fixture.state(), Some(PublicationState::InFlight)));
    let takes_before = fixture.lock_takes();
    assert_eq!(
        fixture.admit_all(256..257)?,
        1,
        "a live write commits while the batch is held"
    );
    assert_eq!(
        fixture.lock_takes() - takes_before,
        per_write,
        "the live write took the lock as any write does; the publication held it 0 times"
    );
    fixture.store.open_gate()?;
    fixture.join()?;
    assert_eq!(
        fixture.lock_takes() - takes_before,
        per_write,
        "0 takes while submitting"
    );
    assert!(matches!(
        fixture.state(),
        Some(PublicationState::Published(_))
    ));
    Ok(())
}

#[test]
fn refusal_does_not_cause_a_full_replay() -> TestResult {
    let captured = Captured::install()?;
    let measured = measure()?;
    let bound = bound_for(&measured)?;
    let fixture = fixture(bounded(bound))?;
    fixture.admit_all(0..256)?;
    fixture.join()?;
    let first = fixture.store.publication_commits()?.len();
    assert!(first >= 2);
    assert_eq!(fixture.load_decision()?, Ok(255));
    fixture.store.refuse_publication_batch(first + 1)?;
    assert_eq!(fixture.admit_all(256..512)?, 256);
    fixture.join()?;
    assert!(
        fixture
            .handler
            .publisher
            .ledger()
            .checkpoint_owed(CONVERSATION)
    );
    assert_eq!(
        fixture.load_decision()?,
        Ok(255),
        "the checkpoint on the owner is the one before the refusal, not RebuildRequired"
    );
    let cell = fixture
        .handler
        .existing_cell(CONVERSATION)?
        .ok_or("the cell is gone")?;
    assert!(
        cell.lock().map_err(|_| "owner lock poisoned")?.is_some(),
        "the owner was never discarded"
    );
    let log = captured.text()?;
    assert_eq!(log.matches("CHECKPOINT_REBUILD").count(), 0, "{log}");
    assert_eq!(log.matches("CONTAINMENT").count(), 0, "{log}");
    Ok(())
}

#[test]
fn a_contained_conversation_is_restored_once_per_process() -> TestResult {
    let captured = Captured::install()?;
    // Under R3 a refused publication no longer contains a conversation, so
    // the containment is driven by a load that refuses: the conversation's
    // first record made undecodable, the fault tests_boot_containment pins.
    let faulted: Arc<dyn DurableStore> = Arc::new(OneConversationFault::new(
        seed_two_conversations()?,
        POISONED_CONVERSATION,
        FaultMode::CorruptRecordNow(0),
    ));
    let store = Arc::new(RowReads {
        inner: faulted,
        reads: AtomicUsize::new(0),
    });
    let handler = ProductionParticipantHandler::new(
        Arc::clone(&store) as Arc<dyn DurableStore>,
        test_participant_config(),
    )?;
    let touch = |incarnation: u64| {
        dispatch(
            &handler,
            ConnectionIncarnation::new(900, incarnation),
            ClientRequest::Enrollment(EnrollmentRequest {
                conversation_id: POISONED_CONVERSATION,
                enrollment_token: EnrollmentToken::new([0xD1; 16]),
            }),
        )
    };
    // The one restore attempt: the replay runs once and refuses.
    let first = touch(1).map(|value| format!("{value:?}"));
    let first = match first {
        Ok(value) => return Err(format!("the poisoned conversation answered: {value}").into()),
        Err(error) => error.to_string(),
    };
    assert!(
        first.contains(&POISONED_CONVERSATION.to_string()),
        "{first}"
    );
    let rows_read = store.reads.load(Ordering::SeqCst);
    assert!(rows_read > 0, "the first restore replayed the log");

    // Every later touch is answered by the containment at once.
    let post = touch(2).map(|value| format!("{value:?}"));
    assert!(
        post.is_err(),
        "a post on the contained conversation was answered: {post:?}"
    );
    let open = handler.existing_cell(POISONED_CONVERSATION);
    assert!(
        open.is_err(),
        "a connection-fate open reached the cell: {open:?}"
    );
    let warm = handler.ensure_restored(POISONED_CONVERSATION);
    assert!(warm.is_err(), "the warm restored it again: {warm:?}");
    assert_eq!(
        store.reads.load(Ordering::SeqCst),
        rows_read,
        "0 further rows read: no second authoritative replay"
    );
    let log = captured.text()?;
    assert_eq!(
        log.matches("participant conversation refused at restore")
            .count(),
        1,
        "one restore line: {log}"
    );
    assert!(log.contains("conversation_elapsed_ms="), "{log}");
    Ok(())
}

/// R3, second line: a contained conversation must not fail the boot.
///
/// The door's 23:56 boot of 10 Oct died in connection-fate recovery: an
/// unmatched Open tracked conversation 2, the conversation was contained,
/// and `connection_fate_dispatch.rs` latched the process-wide fatal on the
/// containment refusal, so the runtime shut down on "Open 44 failed before
/// Complete". This drives the same recovery the boot runs
/// (`connection/incarnation.rs: complete_unmatched_opens`, one step per
/// Open, then Complete, then the resumed startup) over a store whose
/// previous process left an Open tracking a contained conversation and a
/// healthy one.
#[test]
fn a_contained_conversation_does_not_fail_the_boot() -> TestResult {
    let captured = Captured::install()?;
    let config = test_participant_config();
    let reference_bound = usize::try_from(config.max_semantic_conversations_per_connection)?;
    // The previous process: both conversations enrolled, an Open written for
    // the connection that tracked them, never Completed.
    let clean = seed_two_conversations()?;
    let startup = block_on(IncarnationStream::new(Arc::clone(&clean), reference_bound).startup())??;
    let IncarnationStartup::Started(mut started) = startup else {
        return Err("fresh incarnation stream did not start".into());
    };
    let IncarnationAllocation::Allocated {
        connection_incarnation,
        ..
    } = block_on(started.allocate(&[]))??
    else {
        return Err("fresh incarnation allocation was exhausted".into());
    };
    let intent = block_on(started.open_connection_fate(
        connection_incarnation,
        ConnectionFateClass::ConnectionLost,
        reference_bound,
        &[POISONED_CONVERSATION, HEALTHY_CONVERSATION],
    ))??;
    drop(started);

    // This process: the poisoned conversation's first record no longer
    // decodes, so its restore refuses and contains it.
    let store: Arc<dyn DurableStore> = Arc::new(OneConversationFault::new(
        clean,
        POISONED_CONVERSATION,
        FaultMode::CorruptRecordNow(0),
    ));
    let handler = ProductionParticipantHandler::new(Arc::clone(&store), config)?;
    let startup = block_on(IncarnationStream::new(Arc::clone(&store), reference_bound).startup())??;
    let IncarnationStartup::RecoveryRequired(mut recovery) = startup else {
        return Err("the unmatched Open did not require recovery".into());
    };
    let intents = recovery.intents();
    assert_eq!(intents.len(), 1, "one unmatched Open");
    assert_eq!(intents[0].open_sequence, intent.open_sequence);
    let work_item = intents[0].work_item();
    assert_eq!(
        work_item.tracked_conversations,
        vec![POISONED_CONVERSATION, HEALTHY_CONVERSATION],
        "the contained conversation is applied first, so the healthy one is a step AFTER the refusal"
    );

    // The recovery step: refused for the contained conversation, finished
    // for the other, and the Open completes.
    let applied = handler.handle_connection_fate(work_item);
    assert!(
        applied.is_ok(),
        "the contained conversation failed the Open and would fail the boot: {applied:?}"
    );
    block_on(recovery.complete(intent.open_sequence))??;
    let resumed = block_on(recovery.finish_startup())??;
    assert!(
        matches!(resumed, IncarnationStartup::Started(_)),
        "the resumed startup did not start"
    );
    handler.ensure_service_live()?;

    // Readiness: the other conversation is served; the contained one is
    // refused by the containment, by name, and nothing replayed it again.
    healthy_neighbour_is_served(&handler, 702)?;
    let refused = enroll(
        &handler,
        POISONED_CONVERSATION,
        [0xC9; 16],
        ConnectionIncarnation::new(703, 1),
    );
    let refused = match refused {
        Ok(value) => return Err(format!("the contained conversation answered: {value:?}").into()),
        Err(error) => error.to_string(),
    };
    assert!(
        refused.contains(&POISONED_CONVERSATION.to_string())
            && refused.to_lowercase().contains("unloadable"),
        "{refused}"
    );
    let log = captured.text()?;
    // Recovery restored neither conversation: both fates were held, and
    // each applies inside its conversation's own restore after readiness.
    let held: Vec<&str> = log
        .lines()
        .filter(|line| {
            line.contains("connection-fate step held for a conversation not yet restored")
        })
        .collect();
    assert_eq!(held.len(), 2, "both steps held at recovery: {log}");
    assert!(
        held.iter()
            .all(|line| line.contains(&format!("open_sequence={}", intent.open_sequence))),
        "{log}"
    );
    assert!(
        log.contains("a connection fate held at boot was applied in the conversation's restore"),
        "the healthy conversation's held fate applied at its restore: {log}"
    );
    assert_eq!(
        log.matches("participant conversation refused at restore")
            .count(),
        1,
        "one restore attempt for the process: {log}"
    );
    // Two conversations without a checkpoint each log their rebuild line;
    // nothing logs a fatal, and the service answered live above.
    assert!(!log.contains("fatal"), "{log}");
    Ok(())
}

// ─── Restore reads the history root alone ───────────────────────────────────

/// The short history: one `CHECKPOINT_EVERY` (256) period, so its one
/// crossing publishes a locator and a history index.
const SHORT_HISTORY: u64 = 256;

/// The long history: four periods, so four crossings, each extending the
/// index, and the last one's locator is the one a restore reads.
const LONG_HISTORY: u64 = 1_024;

/// What one restore read, beside what the store held before it.
#[derive(Debug)]
struct RestoreReads {
    /// History-node point reads while `new_restored` ran.
    node_reads: usize,
    /// Operation-log rows read while `new_restored` ran.
    row_reads: usize,
    /// Distinct history-node keys every commit before the restore wrote.
    stored_nodes: usize,
}

/// Admits `admissions` operations on a fresh conversation, confirms the last
/// crossing published its locator, then restores the store in a new handler
/// (no warm, so nothing but the restore reads) and answers what it read.
fn restore_reads(admissions: u64) -> Result<RestoreReads, Box<dyn Error>> {
    let fixture = fixture(Scripted::new(MemoryStore::default()))?;
    assert_eq!(
        fixture.admit_all(0..admissions)?,
        usize::try_from(admissions)?
    );
    fixture.join()?;
    assert_eq!(
        fixture.locator()?.map(CheckpointLocator::covered_sequence),
        Some(admissions - 1),
        "a checkpoint is published after {admissions} admissions"
    );
    let commits = fixture.store.commits()?;
    let stored_nodes = commits
        .iter()
        .flat_map(|record| record.items.iter().map(|(key, _)| key))
        .filter(|key| is_node_key(key))
        .collect::<BTreeSet<_>>()
        .len();

    let store = Arc::clone(&fixture.store);
    drop(fixture);
    store.row_reads.store(0, Ordering::SeqCst);
    let counted = Arc::new(NodeReads {
        inner: Arc::clone(&store) as Arc<dyn DurableStore>,
        reads: AtomicUsize::new(0),
    });
    let handler = ProductionParticipantHandler::new_restored(
        Arc::clone(&counted) as Arc<dyn DurableStore>,
        test_participant_config(),
    )?;
    let node_reads = counted.reads.load(Ordering::SeqCst);
    let row_reads = store.row_reads.load(Ordering::SeqCst);
    assert!(
        handler.existing_cell(CONVERSATION)?.is_some(),
        "the conversation restored"
    );
    Ok(RestoreReads {
        node_reads,
        row_reads,
        stored_nodes,
    })
}

/// A checkpoint restore reads the history index's root node alone, the same
/// at any history length.
///
/// The door's restore of conversation 2 (50,176 rows) took 194,942 ms from
/// its checkpoint with no replay (Waffles, 11 October 08:57): the restore
/// cross-checked every identity by a root-to-leaf lookup, one point read per
/// node, so its reads grew as n·log n with history. Commit 82f2e73d made the
/// restore read and authenticate the root alone
/// (`CheckpointHistoryIndex::verify_root`), and moved the full cross-check to
/// the server's background warm, which `new_restored` does not run. This pins
/// that: two histories, the long one storing more nodes than the short one,
/// read the same number of history nodes at restore, and that number is at
/// most one (the root; an empty trie reads none). The row reads show each
/// restore came from its checkpoint, so zero node reads cannot mean a rebuild.
#[test]
fn checkpoint_restore_reads_the_history_root_alone_at_any_length() -> TestResult {
    let short = restore_reads(SHORT_HISTORY)?;
    let long = restore_reads(LONG_HISTORY)?;
    assert!(
        long.stored_nodes > short.stored_nodes,
        "the long history stores more nodes than the short one: {short:?} {long:?}"
    );
    for (admissions, reads) in [(SHORT_HISTORY, &short), (LONG_HISTORY, &long)] {
        assert!(
            reads.row_reads < 256,
            "the restore at {admissions} admissions read the tail from its checkpoint, \
             not the log: {reads:?}"
        );
    }
    assert_eq!(
        short.node_reads, long.node_reads,
        "history-node reads at restore are the same at either length: {short:?} {long:?}"
    );
    assert!(
        long.node_reads <= 1,
        "a restore reads the root node alone: {long:?}"
    );
    Ok(())
}

// ─── R5 ─────────────────────────────────────────────────────────────────────

#[test]
fn census_counts_match_the_capture() -> TestResult {
    let fixture = fixture(Scripted::new(MemoryStore::default()))?;
    assert_eq!(fixture.admit_all(0..100)?, 100);
    let commits_before = fixture.store.commits()?.len();
    let store: Arc<dyn DurableStore> = Arc::clone(&fixture.store) as Arc<dyn DurableStore>;
    let census = checkpoint_census(&store, &test_participant_config(), CONVERSATION)?;
    assert_eq!(fixture.store.commits()?.len(), commits_before, "0 writes");
    assert_eq!(census.conversation_id, CONVERSATION);
    assert_eq!(census.tokens, 2);
    assert_eq!(census.committed_admissions, 100);
    assert_eq!(census.slots, 2);
    assert_eq!(census.retired, 0);
    assert_eq!(census.locator, None);
    assert_eq!(census.log_rows_behind, 0);
    assert!(census.log_rows_ahead >= 102, "{census:?}");
    let (count, bytes) = census
        .nodes
        .values()
        .fold((0, 0), |(count, bytes), (c, b)| (count + c, bytes + b));
    assert_eq!(count, census.history_nodes);
    assert_eq!(bytes, census.history_bytes);
    assert!(
        census.nodes["retained history values"].0 >= 102,
        "{census:?}"
    );
    assert_eq!(census.nodes["retired history"], (0, 0));
    assert!(census.nodes["index pages"].0 > 0, "{census:?}");
    assert!(matches!(census.body, Ok(bytes) if bytes > 0), "{census:?}");
    assert_eq!(census.bound.bound, CHECKPOINT_BATCH_BYTES);
    assert_eq!(census.batches, 1);
    let lines = census.lines();
    assert!(
        lines.contains(&"committed admissions: 100".to_owned()),
        "{lines:?}"
    );
    assert!(
        lines.contains(&"latest locator: absent".to_owned()),
        "{lines:?}"
    );
    Ok(())
}

#[test]
fn census_batches_match_r1() -> TestResult {
    let measured = measure()?;
    let bound = bound_for(&measured)?;
    let published = fixture(bounded(bound))?;
    published.admit_all(0..256)?;
    published.join()?;
    let commits = published.store.publication_commits()?.len();
    assert!(commits >= 2);

    // The same operations with every publication batch refused: the store
    // holds exactly what the census counts before a publication.
    let counted = fixture(bounded(bound))?;
    counted.store.refuse_publication_batch(1)?;
    counted.admit_all(0..256)?;
    counted.join()?;
    assert_eq!(counted.store.publication_commits()?.len(), 0);
    let store: Arc<dyn DurableStore> = Arc::clone(&counted.store) as Arc<dyn DurableStore>;
    let census = checkpoint_census(&store, &test_participant_config(), CONVERSATION)?;
    assert_eq!(census.bound.bound, bound);
    assert_eq!(
        census.batches, commits,
        "the census counts the commits R1 made: {census:?}"
    );
    assert_eq!(
        commit_count([census.history_bytes], census.body.unwrap_or(0), usize::MAX).max(1),
        1,
        "one commit at an unbounded store"
    );
    Ok(())
}

#[test]
fn the_pinned_numbers_are_the_ceilings_file_s() -> TestResult {
    let ceilings: toml::Value =
        toml::from_str(include_str!("../../../../tests/ceilings/checkpoint.toml"))?;
    assert_eq!(
        ceilings["checkpoint_batch_bytes"]["value"].as_integer(),
        Some(i64::try_from(CHECKPOINT_BATCH_BYTES)?)
    );
    assert_eq!(
        ceilings["default_keep_acknowledged_behind_checkpoint"]["value"].as_integer(),
        Some(i64::try_from(
            super::prune::DEFAULT_KEEP_ACKNOWLEDGED_BEHIND_CHECKPOINT
        )?)
    );
    Ok(())
}