liminal-server 0.20.1

//! ONE PASS, NOT TWO: the refusal guarantee of the merged replay.
//!
//! `ConversationAuthority::replay` used to walk the whole log twice — a
//! schema-and-contiguity pass, then the apply pass re-reading every page.
//! The passes are merged; this file pins that nothing the two-pass shape
//! promised was given up in the merge:
//!
//! - a log whose schema audit fails at entry N is refused WHOLE, with the
//!   SAME typed error naming the SAME entry the standalone pass named;
//! - the node still boots, the refusal is attributed to the conversation by
//!   id and entry, and no owner is ever installed for it — nothing partial
//!   reaches the handler — while the neighbour on the same store is served;
//! - an apply failure EARLIER in the log does not outrank that audit
//!   failure, because the two-pass replay audited everything first;
//! - a log refused by its audit gets no durable write, even when the pass
//!   had already reached an extension repair before the refused entry;
//! - and the point of the merge: a cold replay with no checkpoint to load
//!   from (the full rebuild a boot runs for a conversation with no checkpoint
//!   locator) reads each operation-log row EXACTLY ONCE, with a control
//!   proving the counter sees a second pass;
//! - and so does a checkpoint-less restore of the conversation's extension
//!   stream, which used to be decoded in full by a validation pass before the
//!   merge read it again;
//! - and a boot from the checkpoint a commit wrote reads NO operation-log row
//!   and NO extension row at all, with a control proving the same counters
//!   see a rebuild's every row;
//! - and the price of writing nothing before the audit clears is bounded: the
//!   pass stages one page of owed repairs at most, and a log owing more is
//!   audited whole, then replayed once more writing page by page.
//!
//! The replay these pins call is the production rebuild
//! (`ConversationAuthority::replay_for_rebuild`), which reads every committed
//! row where its commit put it: the flat rows below the native head, then the
//! routed rows at and after it on the conversation's route. A store whose
//! rows a pin compares against the frozen aggregate reference, a flat reader,
//! is written under the legacy-rows arm, flat, as a store written before
//! pruning holds them.

use std::error::Error;
use std::fs;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Instant;

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

use crate::server::participant::{ParticipantConnectionConversations, ParticipantSemanticError};

use super::ProductionParticipantHandler;
use super::fenced_attach_terminal::ComposedTerminalValidation;
use super::log::{
    FencedAttachProofRefusal, OperationLog, OperationLogError, OperationSchemaPhase,
    READ_BATCH_SIZE, STREAM_PREFIX, StoredBindingEpoch, StoredOperation, StoredRecoveredFate,
    StoredRecoveredPresentation, legacy_rows,
};
use super::operation_batch::CHECKPOINT_EVERY;
use super::outbox::ConversationOutboxLimits;
use super::outbox_log::{OUTBOX_STREAM_PREFIX, OutboxLog, UNIT2_OUTBOX_RESTORE_BATCH_ROWS};
use super::outbox_replay::{RestoreError, take_staged_repair_high_water};
use super::restore_checkpoint_locator::{CheckpointLocator, checkpoint_conversation_route};
use super::routed_stream::read_head;
use super::routed_stream_codec::row_stream;
use super::state::{ConversationAuthority, StateError};
use super::tests::{
    dispatch, dispatch_tracked, open_disk_store_for_tests, test_participant_config,
};
use super::tests_boot_containment::{
    HEALTHY_CONVERSATION, POISONED_CONVERSATION, seed_two_conversations,
};
use super::tests_memory_store::MemoryStore;
use super::tests_seed::{Shared, seeded_copy};
use super::tests_w3_restore_fixture::{
    CONVERSATION as REPAIR_CONVERSATION, OutboxAppendFaultStore, append_payload,
    duplicate_extension_to, enrollment, enrollment_outcome_unknown, extension_key, new_store,
    operation_key, seed_enrollment, stream_payloads,
};

const fn epoch(seed: u64) -> StoredBindingEpoch {
    StoredBindingEpoch {
        server_incarnation: seed,
        connection_ordinal: seed,
        capability_generation: seed,
    }
}

/// A `Recovered` reservation whose `died_source_sequence` is not behind it.
///
/// Decode and `validate_durable` both accept it; the composed-terminal audit
/// refuses it on sight, at exactly `at`, so it isolates the audit.
fn audit_refused_row(at: u64) -> StoredOperation {
    StoredOperation::Recovered {
        row: StoredRecoveredFate {
            participant_id: 0,
            last_dead_binding_epoch: epoch(1),
            died_source_sequence: at,
            fenced_attached_source_sequence: at,
            prior_binding_epoch: epoch(1),
            marker_delivery_seq: 1,
            resulting_floor: 1,
            presentation: StoredRecoveredPresentation::RecoveredOwnsAndReservesFinalizer,
        },
        event: at.to_be_bytes().to_vec(),
    }
}

/// Appends [`audit_refused_row`] at `conversation_id`'s current head and
/// returns that head: one routed commit, the way a live commit writes it, or,
/// while a test has armed legacy rows, the flat row a store written before
/// pruning holds, where the frozen flat readers read it.
fn append_audit_refused_row(
    store: &Arc<dyn DurableStore>,
    conversation_id: u64,
) -> Result<u64, Box<dyn Error>> {
    let head =
        u64::try_from(stream_payloads(store, &format!("{STREAM_PREFIX}{conversation_id}"))?.len())?;
    let log = OperationLog::new(Arc::clone(store), conversation_id);
    if legacy_rows::armed() {
        block_on(log.append_flat(&audit_refused_row(head), head))??;
    } else {
        block_on(log.append(&audit_refused_row(head), head))??;
    }
    Ok(head)
}

fn is_the_audit_refusal_at(error: &StateError, sequence: u64) -> bool {
    matches!(
        error,
        StateError::Log(OperationLogError::FencedAttachProof {
            sequence: refused,
            reason: FencedAttachProofRefusal::ComposedRecoveredReservationMismatch,
        }) if *refused == sequence
    )
}

/// The production single-pass replay with no checkpoint to load from, called
/// directly: the full rebuild a boot runs for a conversation with no
/// checkpoint locator. It reads the flat rows below the native head, then the
/// routed rows at and after it on the conversation's route.
fn one_pass(
    store: &Arc<dyn DurableStore>,
    conversation_id: u64,
) -> Result<Result<ConversationAuthority, RestoreError>, Box<dyn Error>> {
    let config = test_participant_config();
    let log = OperationLog::new(Arc::clone(store), conversation_id);
    let outbox_log = OutboxLog::new(Arc::clone(store), conversation_id);
    let limits =
        ConversationOutboxLimits::try_new(config.max_retained_record_rows, config.identity_slots)?;
    Ok(block_on(ConversationAuthority::replay_for_rebuild(
        conversation_id,
        &log,
        &outbox_log,
        &config,
        limits,
    ))?
    .map(|(authority, _, _, _)| authority))
}

/// The former standalone audit pass, kept as the two-pass reference: the
/// contiguity check and the composed-terminal audit, row by row, over the
/// whole log.
///
/// It is `validate_operation_schema`'s walk with one change: it reads every
/// committed row where its commit put it, the flat rows below the native head
/// and then the routed rows at and after it, where the frozen pass walks the
/// flat stream alone and so audits none of the rows a live commit stages
/// routed. Each row is audited by the same `ComposedTerminalValidation`.
fn two_pass_audit(
    store: &Arc<dyn DurableStore>,
    conversation_id: u64,
) -> Result<Result<(), StateError>, Box<dyn Error>> {
    let log = OperationLog::new(Arc::clone(store), conversation_id);
    Ok(block_on(standalone_audit(
        &log,
        test_participant_config().identity_slots,
    ))?)
}

async fn standalone_audit(log: &OperationLog, identity_slots: u64) -> Result<(), StateError> {
    let route = checkpoint_conversation_route(log.conversation_id());
    let native_head = log.read_stream_next_seq().await?.unwrap_or(0);
    let page_rows = u64::try_from(READ_BATCH_SIZE)
        .map_err(|_| StateError::invariant("the audit page exceeds the sequence space"))?;
    let mut composed = ComposedTerminalValidation::for_identity_slots(identity_slots)?;
    let mut phase = OperationSchemaPhase::V2Prefix;
    let mut sequence = 0_u64;
    loop {
        let page = if let Some(below_head) =
            native_head.checked_sub(sequence).filter(|rows| *rows != 0)
        {
            let limit = usize::try_from(below_head.min(page_rows))
                .map_err(|_| StateError::invariant("the audit page exceeds the address space"))?;
            log.read_bounded_page(sequence, phase, limit).await?
        } else {
            let Some(page) = log.read_rebuild_page(&route, sequence, phase).await? else {
                return Ok(());
            };
            page
        };
        phase = page.next_phase;
        if page.rows.is_empty() {
            return Ok(());
        }
        for decoded in page.rows {
            if decoded.sequence != sequence {
                return Err(OperationLogError::Sequence {
                    expected: sequence,
                    actual: decoded.sequence,
                }
                .into());
            }
            composed.validate(log, sequence, &decoded.operation).await?;
            sequence = sequence
                .checked_add(1)
                .ok_or(StateError::AllocationExhausted {
                    domain: "log sequence",
                })?;
        }
    }
}

/// The production load and the frozen aggregate reference refused the same
/// log with the same refusal, each in its own seam's words: the load as a
/// checkpoint restore, the reference as a production operation.
fn assert_same_refusal(
    production: &ParticipantSemanticError,
    reference: &ParticipantSemanticError,
    refusal: &StateError,
) {
    let failed = "participant semantic service failed:";
    assert_eq!(
        production.to_string(),
        format!("{failed} participant checkpoint restore failed: {refusal}")
    );
    assert_eq!(
        reference.to_string(),
        format!("{failed} participant production operation failed: {refusal}")
    );
}

/// The refusal a single-pass replay gave, or why it was not an audit refusal.
fn audit_refusal(
    store: &Arc<dyn DurableStore>,
    conversation_id: u64,
) -> Result<StateError, Box<dyn Error>> {
    match one_pass(store, conversation_id)? {
        Err(RestoreError::Semantic(refusal)) => Ok(refusal),
        Err(RestoreError::Extension(other)) => {
            Err(format!("the replay refused on its extension log instead: {other}").into())
        }
        Ok(_) => Err("the audit-refused log replayed".into()),
    }
}

fn enroll(
    handler: &ProductionParticipantHandler,
    conversation_id: u64,
    token: u8,
    connection: ConnectionIncarnation,
) -> Result<ServerValue, Box<dyn Error>> {
    dispatch_tracked(
        handler,
        connection,
        &mut ParticipantConnectionConversations::default(),
        ClientRequest::Enrollment(EnrollmentRequest {
            conversation_id,
            enrollment_token: EnrollmentToken::new([token; 16]),
        }),
    )
}

/// A store whose second enrollment committed its operation row but lost its
/// Unit 2 extension row, so the next replay owes exactly one repair row. A
/// live commit stages both rows in one routed batch, so only a store written
/// before pruning, flat and one append per row, can hold this state: the
/// seed writes it under the legacy-rows arm.
struct LostProjection {
    store: Arc<dyn DurableStore>,
    /// The extension stream as the lost append left it.
    extension: Vec<Vec<u8>>,
}

/// Builds a [`LostProjection`] through the production seam.
fn lost_projection_store() -> Result<LostProjection, Box<dyn Error>> {
    let legacy = legacy_rows::arm();
    let inner = new_store()?;
    let faults = Arc::new(OutboxAppendFaultStore::new(Arc::clone(&inner)));
    let store: Arc<dyn DurableStore> = faults.clone();
    let handler = seed_enrollment(&store)?;
    faults.set_fail(true);
    let refused = dispatch(
        &handler,
        ConnectionIncarnation::new(REPAIR_CONVERSATION, 2),
        enrollment(2),
    );
    faults.set_fail(false);
    // The flat operation row was issued and flushed before its extension row
    // failed, so the live path answers the enrollment's outcome as unknown:
    // a response on the open connection, not an error.
    let refused = refused?;
    if !enrollment_outcome_unknown(&refused, 2) {
        return Err(format!(
            "the outbox append fault did not leave the enrollment's outcome unknown: {refused:?}"
        )
        .into());
    }
    drop(handler);
    drop(legacy);
    let extension = stream_payloads(&store, &extension_key())?;
    Ok(LostProjection { store, extension })
}

/// The one-pass replay refuses at entry N with the exact typed error, and the
/// exact text, the standalone two-pass audit gives for the same log.
#[test]
fn schema_failure_at_entry_n_is_refused_whole_with_the_two_pass_error() -> Result<(), Box<dyn Error>>
{
    let store = seed_two_conversations()?;
    let poisoned_at = append_audit_refused_row(&store, POISONED_CONVERSATION)?;
    assert!(poisoned_at > 0, "the seed wrote no rows to poison behind");

    let Err(two_pass) = two_pass_audit(&store, POISONED_CONVERSATION)? else {
        return Err("the standalone audit accepted the poisoned log".into());
    };
    assert!(
        is_the_audit_refusal_at(&two_pass, poisoned_at),
        "standalone audit refused elsewhere: {two_pass:?}"
    );

    let merged = audit_refusal(&store, POISONED_CONVERSATION)?;
    assert!(
        is_the_audit_refusal_at(&merged, poisoned_at),
        "one-pass replay refused elsewhere: {merged:?}"
    );
    assert_eq!(merged.to_string(), two_pass.to_string());
    Ok(())
}

/// Booting over the poisoned store installs no owner for the refused
/// conversation, attributes the refusal to it by id AND entry, and keeps
/// serving the neighbour.
#[test]
fn schema_failure_at_entry_n_registers_no_conversation_and_names_it() -> Result<(), Box<dyn Error>>
{
    let store = seed_two_conversations()?;
    let poisoned_at = append_audit_refused_row(&store, POISONED_CONVERSATION)?;

    let handler = ProductionParticipantHandler::new(Arc::clone(&store), test_participant_config())?;

    let unloadable = handler.unloadable_conversations();
    let reason = unloadable
        .get(&POISONED_CONVERSATION)
        .ok_or("the poisoned conversation was not recorded as unloadable")?;
    assert!(
        reason.contains(&format!("sequence {poisoned_at}")),
        "the refusal does not name entry {poisoned_at}: {reason}"
    );
    assert!(
        !unloadable.contains_key(&HEALTHY_CONVERSATION),
        "the healthy neighbour was refused: {unloadable:?}"
    );

    let cell = handler.cell(POISONED_CONVERSATION)?;
    let owner = cell
        .lock()
        .map_err(|_| "poisoned conversation owner lock is poisoned")?;
    let installed = owner.is_some();
    drop(owner);
    assert!(
        !installed,
        "a partial authority was installed for the refused conversation"
    );

    let refused = enroll(
        &handler,
        POISONED_CONVERSATION,
        0xD1,
        ConnectionIncarnation::new(801, 1),
    );
    let Err(refused) = refused else {
        return Err(format!("the refused conversation answered a request: {refused:?}").into());
    };
    let refused = refused.to_string();
    assert!(
        refused.contains(&POISONED_CONVERSATION.to_string()),
        "the request refusal does not name conversation {POISONED_CONVERSATION}: {refused}"
    );

    let served = enroll(
        &handler,
        HEALTHY_CONVERSATION,
        0xD2,
        ConnectionIncarnation::new(802, 1),
    )?;
    assert!(
        matches!(served, ServerValue::EnrollBound(_)),
        "the healthy neighbour stopped serving: {served:?}"
    );
    Ok(())
}

/// An entry M that audits clean but cannot be applied, followed by an entry
/// N the audit refuses: the refusal names N, as the audit-first replay did,
/// and matches the frozen two-pass reference byte for byte.
///
/// The rows are written flat under the legacy-rows arm: the frozen aggregate
/// reference the handler pin below compares against reads the flat stream
/// alone.
#[test]
fn an_earlier_apply_failure_does_not_outrank_a_later_audit_failure() -> Result<(), Box<dyn Error>> {
    let legacy = legacy_rows::arm();
    let store = new_store()?;
    drop(seed_enrollment(&store)?);
    let base = stream_payloads(&store, &operation_key())?;
    let genesis = base
        .first()
        .cloned()
        .ok_or("seeded log has no genesis row")?;
    let apply_failure_at = u64::try_from(base.len())?;
    append_payload(&store, &operation_key(), genesis, apply_failure_at)?;

    // Control: a duplicated genesis audits clean and fails only when
    // applied, so M is an apply-phase failure and nothing else.
    assert!(
        matches!(two_pass_audit(&store, REPAIR_CONVERSATION)?, Ok(())),
        "the duplicated genesis row failed the audit, so it cannot stand for an apply failure"
    );
    let apply_only = match one_pass(&store, REPAIR_CONVERSATION)? {
        Ok(_) => return Err("a duplicated genesis row replayed".into()),
        Err(error) => error.to_string(),
    };

    let audit_failure_at = append_audit_refused_row(&store, REPAIR_CONVERSATION)?;
    drop(legacy);
    assert!(audit_failure_at > apply_failure_at);
    let Err(two_pass) = two_pass_audit(&store, REPAIR_CONVERSATION)? else {
        return Err("the standalone audit accepted the poisoned log".into());
    };
    assert!(
        is_the_audit_refusal_at(&two_pass, audit_failure_at),
        "standalone audit refused elsewhere: {two_pass:?}"
    );
    let merged = audit_refusal(&store, REPAIR_CONVERSATION)?;
    assert!(
        is_the_audit_refusal_at(&merged, audit_failure_at),
        "the apply failure at {apply_failure_at} outranked the audit failure at \
         {audit_failure_at}: {merged:?}"
    );
    assert_eq!(merged.to_string(), two_pass.to_string());
    assert_ne!(
        merged.to_string(),
        apply_only,
        "the fixture cannot tell the two refusals apart"
    );

    let handler = ProductionParticipantHandler::new(Arc::clone(&store), test_participant_config())?;
    let log = OperationLog::new(Arc::clone(&store), REPAIR_CONVERSATION);
    let reference = handler
        .replay_aggregate_reference(REPAIR_CONVERSATION, &log)
        .err()
        .ok_or("the two-pass reference replayed the poisoned log")?;
    let production = handler
        .replay_and_repair(REPAIR_CONVERSATION, &log)
        .err()
        .ok_or("the production replay replayed the poisoned log")?;
    assert_same_refusal(&production, &reference, &two_pass);
    Ok(())
}

/// The pass reaches an owed extension repair BEFORE the entry its audit
/// refuses: the repair stays staged and is never written, so the refused
/// conversation's store is byte-identical to what the audit-first replay
/// left.
#[test]
fn an_audit_refusal_after_a_staged_repair_writes_nothing() -> Result<(), Box<dyn Error>> {
    // Control: the same lost projection with nothing after it IS repaired,
    // so a repair really is owed in the arm below.
    let LostProjection {
        store: control,
        extension: control_before,
    } = lost_projection_store()?;
    assert_eq!(control_before.len(), 1);
    if let Err(refused) = one_pass(&control, REPAIR_CONVERSATION)? {
        return Err(format!("the unpoisoned control refused: {refused}").into());
    }
    assert_eq!(
        stream_payloads(&control, &extension_key())?.len(),
        2,
        "the control owed no repair, so the arm below proves nothing"
    );

    let LostProjection {
        store,
        extension: before,
    } = lost_projection_store()?;
    // Flat, behind the flat rows the lost projection left, where the frozen
    // aggregate reference the handler pin below compares against reads it.
    let legacy = legacy_rows::arm();
    let poisoned_at = append_audit_refused_row(&store, REPAIR_CONVERSATION)?;
    drop(legacy);
    let refusal = audit_refusal(&store, REPAIR_CONVERSATION)?;
    assert!(
        is_the_audit_refusal_at(&refusal, poisoned_at),
        "refused elsewhere: {refusal:?}"
    );
    assert_eq!(
        stream_payloads(&store, &extension_key())?,
        before,
        "a log refused by its audit had a repair row written"
    );

    let handler = ProductionParticipantHandler::new(Arc::clone(&store), test_participant_config())?;
    assert!(
        handler
            .unloadable_conversations()
            .contains_key(&REPAIR_CONVERSATION),
        "the boot did not refuse the poisoned conversation"
    );
    let log = OperationLog::new(Arc::clone(&store), REPAIR_CONVERSATION);
    let reference = handler
        .replay_aggregate_reference(REPAIR_CONVERSATION, &log)
        .err()
        .ok_or("the two-pass reference replayed the poisoned log")?;
    let production = handler
        .replay_and_repair(REPAIR_CONVERSATION, &log)
        .err()
        .ok_or("the production replay replayed the poisoned log")?;
    assert_same_refusal(&production, &reference, &refusal);
    assert_eq!(
        stream_payloads(&store, &extension_key())?,
        before,
        "a boot or a request over the refused log wrote a repair row"
    );
    Ok(())
}

/// The conversation the read-count pin writes; its own id, so no fixture
/// sharing a store can land rows in the stream it counts.
const READ_ONCE_CONVERSATION: u64 = 0x0E_1C_E0;

/// Counts the rows served from ONE conversation's operation-log stream, by
/// paged read and by point read alike, flat or routed: a routed row is read
/// by its own key, so each routed row key that answers counts one.
#[derive(Debug)]
struct OperationRowCounter {
    inner: Arc<dyn DurableStore>,
    stream_key: String,
    rows: AtomicU64,
}

impl OperationRowCounter {
    fn new(inner: Arc<dyn DurableStore>, conversation_id: u64) -> Self {
        Self::for_stream(inner, format!("{STREAM_PREFIX}{conversation_id}"))
    }

    /// Counts the rows served from the stream named `stream_key` instead.
    fn for_stream(inner: Arc<dyn DurableStore>, stream_key: String) -> Self {
        Self {
            inner,
            stream_key,
            rows: AtomicU64::new(0),
        }
    }

    fn take(&self) -> u64 {
        self.rows.swap(0, Ordering::SeqCst)
    }

    fn count(&self, stream_key: &str, rows: usize) -> Result<(), DurabilityError> {
        if stream_key == self.stream_key {
            let rows = u64::try_from(rows).map_err(|_| {
                DurabilityError::ConfigError(format!("{rows} rows exceed the row counter"))
            })?;
            self.rows.fetch_add(rows, Ordering::SeqCst);
        }
        Ok(())
    }
}

#[async_trait::async_trait]
impl DurableStore for OperationRowCounter {
    async fn apply_routed_batch(
        &self,
        route_key: &str,
        items: Vec<liminal::durability::RoutedBatchItem>,
    ) -> Result<(), liminal::durability::DurabilityError> {
        self.inner.apply_routed_batch(route_key, items).await
    }

    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?;
        if let Some(stream) = row_stream(key) {
            self.count(stream, usize::from(value.is_some()))?;
        }
        Ok(value)
    }

    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.count(stream_key, entries.len())?;
        Ok(entries)
    }

    async fn read_at(
        &self,
        stream_key: &str,
        sequence: u64,
    ) -> Result<Option<StoredEntry>, DurabilityError> {
        let entry = self.inner.read_at(stream_key, sequence).await?;
        self.count(stream_key, usize::from(entry.is_some()))?;
        Ok(entry)
    }

    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 enrolled_participant(
    handler: &ProductionParticipantHandler,
    token: u8,
    connection: ConnectionIncarnation,
) -> Result<u64, Box<dyn Error>> {
    let value = enroll(handler, READ_ONCE_CONVERSATION, token, connection)?;
    let ServerValue::EnrollBound(receipt) = value else {
        return Err(format!("read-count enrollment {token:#x} did not bind: {value:?}").into());
    };
    Ok(receipt.participant_id())
}

fn acknowledge(
    handler: &ProductionParticipantHandler,
    connection: ConnectionIncarnation,
    participant_id: u64,
    through_seq: u64,
) -> Result<(), Box<dyn Error>> {
    let value = dispatch(
        handler,
        connection,
        ClientRequest::ParticipantAck(ParticipantAck {
            conversation_id: READ_ONCE_CONVERSATION,
            participant_id,
            capability_generation: Generation::ONE,
            through_seq,
        }),
    )?;
    if !matches!(value, ServerValue::AckCommitted(_)) {
        return Err(
            format!("read-count ack through {through_seq} did not commit: {value:?}").into(),
        );
    }
    Ok(())
}

/// Writes records and acks through the production seam until the operation
/// log spans more than two replay pages, so the pass crosses page boundaries.
fn seed_multi_page_log(store: &Arc<dyn DurableStore>) -> Result<u64, Box<dyn Error>> {
    let target = u64::try_from(READ_BATCH_SIZE)?
        .checked_mul(2)
        .and_then(|rows| rows.checked_add(1))
        .ok_or("read-count target overflowed")?;
    seed_log_past(store, target)
}

/// Writes records and acks through the production seam until the operation
/// log holds more than `target` rows; every record and every ack owes one
/// extension row, which the live commit writes.
fn seed_log_past(store: &Arc<dyn DurableStore>, target: u64) -> Result<u64, Box<dyn Error>> {
    let recipient_connection = ConnectionIncarnation::new(0xE1, 1);
    let sender_connection = ConnectionIncarnation::new(0xE1, 2);
    let handler = ProductionParticipantHandler::new(Arc::clone(store), test_participant_config())?;
    let recipient = enrolled_participant(&handler, 0xE1, recipient_connection)?;
    let sender = enrolled_participant(&handler, 0xE2, sender_connection)?;
    acknowledge(&handler, recipient_connection, recipient, 2)?;
    let mut nonce = 0_u8;
    loop {
        let rows = u64::try_from(
            stream_payloads(store, &format!("{STREAM_PREFIX}{READ_ONCE_CONVERSATION}"))?.len(),
        )?;
        if rows > target {
            return Ok(rows);
        }
        let value = dispatch(
            &handler,
            sender_connection,
            ClientRequest::RecordAdmission(RecordAdmission {
                conversation_id: READ_ONCE_CONVERSATION,
                participant_id: sender,
                capability_generation: Generation::ONE,
                record_admission_attempt_token: RecordAdmissionAttemptToken::new([nonce; 16]),
                payload: vec![nonce],
            }),
        )?;
        let ServerValue::RecordCommitted(committed) = value else {
            return Err(format!("read-count record {nonce} did not commit: {value:?}").into());
        };
        acknowledge(
            &handler,
            recipient_connection,
            recipient,
            committed.delivery_seq(),
        )?;
        nonce = nonce.checked_add(1).ok_or("read-count nonce overflowed")?;
    }
}

/// The point of the merge: a cold replay of a log spanning several pages,
/// with no checkpoint to load from, reads each of its rows exactly once. The
/// live seed commits no checkpoint (it never crosses a checkpoint boundary)
/// and the replay is the full rebuild a boot runs for such a conversation,
/// reading the routed rows the commits staged. The control runs the removed
/// audit pass in front of the same replay over the same counter and sees
/// exactly twice the rows, so the single count is a measurement, not a
/// counter that cannot see a second pass.
#[test]
fn a_cold_replay_reads_each_operation_row_exactly_once() -> Result<(), Box<dyn Error>> {
    let (seed_copy, inner) = seeded_copy("multi-page-log", |store| {
        seed_multi_page_log(store).map(drop)
    })?;
    let rows = u64::try_from(
        stream_payloads(&inner, &format!("{STREAM_PREFIX}{READ_ONCE_CONVERSATION}"))?.len(),
    )?;
    let counter = Arc::new(OperationRowCounter::new(inner, READ_ONCE_CONVERSATION));
    let store: Arc<dyn DurableStore> = counter.clone();

    let replayed = match one_pass(&store, READ_ONCE_CONVERSATION)? {
        Ok(authority) => authority.next_log_sequence,
        Err(refused) => return Err(format!("the multi-page log refused: {refused}").into()),
    };
    assert_eq!(replayed, rows);
    let read_once = counter.take();
    assert_eq!(
        read_once, rows,
        "a cold replay of a {rows}-row log read {read_once} operation rows; every row must be \
         read exactly once"
    );

    if let Err(refused) = two_pass_audit(&store, READ_ONCE_CONVERSATION)? {
        return Err(format!("the control's audit refused: {refused}").into());
    }
    if let Err(refused) = one_pass(&store, READ_ONCE_CONVERSATION)? {
        return Err(format!("the control's replay refused: {refused}").into());
    }
    let read_twice = counter.take();
    assert_eq!(
        read_twice,
        rows.checked_mul(2).ok_or("control row count overflowed")?,
        "the counter did not see the control's second pass, so the single count above is not a \
         measurement"
    );
    drop(store);
    drop(seed_copy);
    Ok(())
}

/// A2: a cold restore with no checkpoint to load from reads each row of the
/// conversation's extension stream exactly once. The restore used to decode
/// the whole stream in a validation pass and then read it again to merge it,
/// so this pin was twice the rows. The seed commits no checkpoint, so the boot
/// below is the full rebuild. The control reads the stream once more on a
/// cursor of its own over the same counter, the rebuild's cursor (the flat
/// prefix, then the routed continuation), and sees exactly the rows again, so
/// the single count is a measurement, not a counter that cannot see a second
/// pass.
#[test]
fn a_cold_restore_reads_each_extension_row_exactly_once() -> Result<(), Box<dyn Error>> {
    let rows = UNIT2_OUTBOX_RESTORE_BATCH_ROWS
        .checked_mul(3)
        .and_then(|rows| rows.checked_add(1))
        .ok_or("extension fixture row count overflowed")?;
    let (seed_copy, inner) = seeded_copy("extension-stream", |store| {
        drop(seed_enrollment(store)?);
        duplicate_extension_to(store, rows).map(drop)
    })?;
    let rows = u64::try_from(rows)?;
    let counter = Arc::new(OperationRowCounter::for_stream(inner, extension_key()));
    let store: Arc<dyn DurableStore> = counter.clone();

    let handler = ProductionParticipantHandler::new(Arc::clone(&store), test_participant_config())?;
    let read_once = counter.take();
    assert_eq!(
        read_once, rows,
        "a cold restore of a {rows}-row extension stream read {read_once} extension rows; every \
         row must be read exactly once"
    );

    let outbox_log = OutboxLog::new(Arc::clone(&store), REPAIR_CONVERSATION);
    let route = checkpoint_conversation_route(REPAIR_CONVERSATION);
    let mut control_cursor = outbox_log.restore_rebuild_cursor(&route);
    block_on(control_cursor.drain())??;
    drop(control_cursor);
    let control = counter.take();
    assert_eq!(
        control, rows,
        "the counter did not see the control's pass, so the single count above is not a \
         measurement"
    );
    drop(handler);
    drop(store);
    drop(seed_copy);
    Ok(())
}

/// Commits a record and acknowledges it through the production seam, as
/// [`seed_log_past`] does, until a commit crosses the first checkpoint
/// boundary, and stops on that commit, so the checkpoint it wrote covers the
/// whole log. Every record is acknowledged before the next, because the test
/// configuration retains one unacknowledged ordinary record. Answers the
/// operation-log head.
fn seed_to_checkpoint_crossing(store: &Arc<dyn DurableStore>) -> Result<u64, Box<dyn Error>> {
    let recipient_connection = ConnectionIncarnation::new(0xE3, 1);
    let sender_connection = ConnectionIncarnation::new(0xE3, 2);
    let handler = ProductionParticipantHandler::new(Arc::clone(store), test_participant_config())?;
    let route = checkpoint_conversation_route(READ_ONCE_CONVERSATION);
    let stream = read_once_operation_key();
    let crossed = || -> Result<Option<u64>, Box<dyn Error>> {
        let head = block_on(read_head(store.as_ref(), &route, &stream))??
            .ok_or("a commit left no routed operation head")?
            .next_sequence;
        Ok((head >= CHECKPOINT_EVERY).then_some(head))
    };
    let recipient = enrolled_participant(&handler, 0xE3, recipient_connection)?;
    let sender = enrolled_participant(&handler, 0xE4, sender_connection)?;
    acknowledge(&handler, recipient_connection, recipient, 2)?;
    for record in 0..CHECKPOINT_EVERY {
        if let Some(head) = crossed()? {
            return Ok(head);
        }
        let mut token = [0; 16];
        token[..8].copy_from_slice(&record.to_be_bytes());
        let value = dispatch(
            &handler,
            sender_connection,
            ClientRequest::RecordAdmission(RecordAdmission {
                conversation_id: READ_ONCE_CONVERSATION,
                participant_id: sender,
                capability_generation: Generation::ONE,
                record_admission_attempt_token: RecordAdmissionAttemptToken::new(token),
                payload: record.to_be_bytes().to_vec(),
            }),
        )?;
        let ServerValue::RecordCommitted(committed) = value else {
            return Err(format!("checkpoint record {record} did not commit: {value:?}").into());
        };
        if let Some(head) = crossed()? {
            return Ok(head);
        }
        acknowledge(
            &handler,
            recipient_connection,
            recipient,
            committed.delivery_seq(),
        )?;
    }
    Err("the records never crossed the first checkpoint boundary".into())
}

/// Tom's millisecond boot: a boot from the checkpoint a commit wrote reads
/// no operation-log row and no extension row. The commit that crossed the
/// checkpoint boundary is the log's last, so its checkpoint covers every row
/// and the boot owes no tail. The control runs the checkpoint-less rebuild
/// over the same counters and sees every row of both streams once, so the
/// zero counts are a measurement, not counters that cannot see a read.
#[test]
fn a_boot_from_a_commit_written_checkpoint_reads_no_operation_or_extension_row()
-> Result<(), Box<dyn Error>> {
    let inner: Arc<dyn DurableStore> = Arc::new(MemoryStore::default());
    let head = seed_to_checkpoint_crossing(&inner)?;
    let route = checkpoint_conversation_route(READ_ONCE_CONVERSATION);
    let locator = block_on(inner.get_routed(&route, CheckpointLocator::routed_key()))??
        .ok_or("the crossing commit wrote no checkpoint locator")?;
    let locator = CheckpointLocator::decode_canonical(&locator, READ_ONCE_CONVERSATION)?;
    assert_eq!(
        locator.covered_sequence().checked_add(1),
        Some(head),
        "the commit-written checkpoint does not cover the whole {head}-row log"
    );
    let extension_key = format!("{OUTBOX_STREAM_PREFIX}{READ_ONCE_CONVERSATION}");
    let extension_rows = u64::try_from(stream_payloads(&inner, &extension_key)?.len())?;
    assert!(extension_rows > 0, "the seed wrote no extension row");

    let operations = Arc::new(OperationRowCounter::new(
        Arc::clone(&inner),
        READ_ONCE_CONVERSATION,
    ));
    let extensions = Arc::new(OperationRowCounter::for_stream(
        Arc::clone(&operations) as Arc<dyn DurableStore>,
        extension_key,
    ));
    let store: Arc<dyn DurableStore> = extensions.clone();

    let booted = ProductionParticipantHandler::new(Arc::clone(&store), test_participant_config())?;
    assert!(
        booted.unloadable_conversations().is_empty(),
        "the boot refused a conversation: {:?}",
        booted.unloadable_conversations()
    );
    let cell = booted.cell(READ_ONCE_CONVERSATION)?;
    let owner = cell
        .lock()
        .map_err(|_| "checkpoint conversation owner lock is poisoned")?;
    let installed = owner.is_some();
    drop(owner);
    assert!(installed, "the boot installed no owner from the checkpoint");
    let operation_reads = operations.take();
    let extension_reads = extensions.take();
    assert_eq!(
        operation_reads, 0,
        "a boot from the commit-written checkpoint read {operation_reads} operation rows"
    );
    assert_eq!(
        extension_reads, 0,
        "a boot from the commit-written checkpoint read {extension_reads} extension rows"
    );
    drop(booted);

    let replayed = match one_pass(&store, READ_ONCE_CONVERSATION)? {
        Ok(authority) => authority.next_log_sequence,
        Err(refused) => return Err(format!("the control's rebuild refused: {refused}").into()),
    };
    assert_eq!(replayed, head);
    assert_eq!(
        operations.take(),
        head,
        "the counter did not see the control's rebuild read the operation log"
    );
    assert_eq!(
        extensions.take(),
        extension_rows,
        "the counter did not see the control's rebuild read the extension stream"
    );
    Ok(())
}

/// DUR-008 A3 (D03): the replay's owed extension repairs are written as one
/// durable operation per page, never one `append` and one flush per row.
/// Since every commit is a routed batch, that one operation is one
/// `apply_routed_batch`, which publishes without a separate flush. The
/// replay writes nothing else, so every write it counts is the repair's.
#[test]
fn owed_extension_repairs_are_one_append_all_and_one_flush() -> Result<(), Box<dyn Error>> {
    let LostProjection { store, extension } = lost_projection_store()?;
    assert_eq!(extension.len(), 1);
    let counting = Arc::new(CountingStore::new(Shared(Arc::clone(&store))));
    let counts = counting.handle();
    let counted: Arc<dyn DurableStore> = counting;
    take_staged_repair_high_water();
    if let Err(refused) = one_pass(&counted, REPAIR_CONVERSATION)? {
        return Err(format!("the lost projection refused: {refused}").into());
    }
    assert_eq!(
        take_staged_repair_high_water(),
        1,
        "the one owed repair was not staged exactly once"
    );
    let after = stream_payloads(&store, &extension_key())?;
    assert_eq!(after.len(), 2, "the replay owed one repair row");
    let counted = counts.snapshot();
    assert_eq!(counted.append_calls, 0, "a repair was appended row by row");
    assert_eq!(counted.append_all_calls, 0, "a repair was appended flat");
    assert_eq!(
        counted.apply_routed_batch_calls, 1,
        "the repair page was not one routed commit"
    );
    assert_eq!(counted.flushes, 0, "the routed commit flushed separately");
    Ok(())
}

/// The operation-log stream of the conversation the paged-repair pins write.
fn read_once_operation_key() -> String {
    format!("{STREAM_PREFIX}{READ_ONCE_CONVERSATION}")
}

/// The extension stream of the conversation the paged-repair pins write.
fn read_once_extension_key() -> String {
    format!("{OUTBOX_STREAM_PREFIX}{READ_ONCE_CONVERSATION}")
}

/// A fresh store holding `operation` as the paged-repair conversation's whole
/// operation log and `extension` as its extension stream, each stream written
/// as one `append_all` and one flush.
fn store_holding(
    operation: &[Vec<u8>],
    extension: &[Vec<u8>],
) -> Result<Arc<dyn DurableStore>, Box<dyn Error>> {
    let store = new_store()?;
    for (stream_key, payloads) in [
        (read_once_operation_key(), operation),
        (read_once_extension_key(), extension),
    ] {
        if payloads.is_empty() {
            continue;
        }
        let next = block_on(store.append_all(&stream_key, payloads.to_vec(), 0))??;
        if next != u64::try_from(payloads.len())? {
            return Err(format!("fixture stream {stream_key} ended at {next}").into());
        }
        block_on(store.flush())??;
    }
    Ok(store)
}

/// The frozen in-line repair over `store`: the pre-W3 aggregate reference,
/// which appends each owed repair row the moment its replay reaches it.
/// Answers the extension stream it leaves.
fn in_line_repair(store: &Arc<dyn DurableStore>) -> Result<Vec<Vec<u8>>, Box<dyn Error>> {
    let config = test_participant_config();
    let log = OperationLog::new(Arc::clone(store), READ_ONCE_CONVERSATION);
    let outbox_log = OutboxLog::new(Arc::clone(store), READ_ONCE_CONVERSATION);
    let limits =
        ConversationOutboxLimits::try_new(config.max_retained_record_rows, config.identity_slots)?;
    let extension_rows = block_on(outbox_log.read_all())??;
    drop(block_on(
        ConversationAuthority::replay_aggregate_reference(
            READ_ONCE_CONVERSATION,
            &log,
            &outbox_log,
            extension_rows,
            &config,
            limits,
        ),
    )??);
    stream_payloads(store, &read_once_extension_key())
}

fn memory_holding(
    operation: &[Vec<u8>],
    extension: &[Vec<u8>],
) -> Result<Arc<dyn DurableStore>, Box<dyn Error>> {
    let store: Arc<dyn DurableStore> = Arc::new(MemoryStore::default());
    for (key, rows) in [
        (read_once_operation_key(), operation),
        (read_once_extension_key(), extension),
    ] {
        assert_eq!(
            block_on(store.append_all(&key, rows.to_vec(), 0))??,
            u64::try_from(rows.len())?
        );
    }
    Ok(store)
}

#[derive(Debug, serde::Serialize, serde::Deserialize)]
struct RepairSeedWork {
    target: u64,
    streams: u64,
    rows: u64,
    append_calls: u64,
    append_all_calls: u64,
    append_all_rows: u64,
    flushes: u64,
}

#[derive(Debug)]
struct RepairSeed {
    directory: tempfile::TempDir,
    store: Arc<dyn DurableStore>,
}

/// Seeds the paged-repair conversation through the production seam and
/// copies its two streams to disk FLAT, each as one `append_all`, under one
/// flush.
///
/// The live commits stage the rows routed; the copy reads each stream where
/// its commits put it and writes the committed rows flat, the shape a store
/// written before pruning holds and the one the frozen in-line reference, a
/// flat reader, reads. Only these two streams are read back from the copy.
fn batched_repair_copy(target: u64) -> Result<RepairSeed, Box<dyn Error>> {
    let directory = crate::test_fixture::fixture_copy("batched-paged-repair-log", |data| {
        let start = Instant::now();
        let memory = Arc::new(MemoryStore::default());
        let store: Arc<dyn DurableStore> = memory.clone();
        seed_log_past(&store, target)?;
        let mut streams = Vec::new();
        for key in [read_once_operation_key(), read_once_extension_key()] {
            streams.push((stream_payloads(&store, &key)?, key));
        }
        drop(store);
        let values = {
            let mut guard = memory
                .values
                .lock()
                .map_err(|_| "repair fixture values lock poisoned")?;
            let values = std::mem::take(&mut *guard);
            drop(guard);
            values
        };
        println!(
            "repair generation elapsed_ms={}",
            start.elapsed().as_secs_f64() * 1_000.0
        );
        let stream_count = u64::try_from(streams.len())?;
        let store = CountingStore::new(Shared(open_disk_store_for_tests(&data.join("store"))?));
        let counts = store.handle();
        let mut total_rows = 0_u64;
        let start = Instant::now();
        for (rows, key) in streams {
            let head = u64::try_from(rows.len())?;
            total_rows = total_rows
                .checked_add(head)
                .ok_or("repair fixture row count overflow")?;
            assert_eq!(block_on(store.append_all(&key, rows, 0))??, head);
        }
        for (key, value) in values {
            block_on(store.cas(&key, 0, value))??;
        }
        block_on(store.flush())??;
        let counts = counts.snapshot();
        println!(
            "repair durable seed elapsed_ms={} counts={counts:?}",
            start.elapsed().as_secs_f64() * 1_000.0
        );
        let work = RepairSeedWork {
            target,
            streams: stream_count,
            rows: total_rows,
            append_calls: counts.append_calls,
            append_all_calls: counts.append_all_calls,
            append_all_rows: counts.append_all_rows,
            flushes: counts.flushes,
        };
        fs::write(data.join("work.json"), serde_json::to_vec(&work)?)?;
        drop(store);
        Ok(())
    })?;
    let work: RepairSeedWork =
        serde_json::from_slice(&fs::read(directory.path().join("work.json"))?)?;
    assert_eq!(work.target, target);
    assert_eq!(work.append_calls, 0);
    assert_eq!(work.append_all_calls, work.streams);
    assert_eq!(work.append_all_rows, work.rows);
    assert_eq!(work.flushes, 1);
    let builds = crate::test_fixture::fixture_builds("batched-paged-repair-log")?;
    assert_eq!(builds, 1, "repair fixture is built once per binary");
    println!("repair fixture builds={builds} work={work:?}");
    let store = open_disk_store_for_tests(&directory.path().join("store"))?;
    Ok(RepairSeed { directory, store })
}

/// A conversation log long enough to owe several pages of extension repairs,
/// and the complete extension stream a full repair of it gives, both as flat
/// rows (see [`batched_repair_copy`]).
struct RepairFixture {
    operation: Vec<Vec<u8>>,
    operation_rows: u64,
    complete_extension: Vec<Vec<u8>>,
}

impl RepairFixture {
    /// Seeds (once per test binary) a log of more than four operation-log
    /// pages and repairs it in line from an extension stream as the live
    /// commits left it.
    fn new() -> Result<Self, Box<dyn Error>> {
        let start = Instant::now();
        let target = u64::try_from(READ_BATCH_SIZE)?
            .checked_mul(4)
            .ok_or("paged-repair target overflowed")?;
        let RepairSeed {
            directory: seed_copy,
            store: seeded,
        } = batched_repair_copy(target)?;
        println!(
            "repair seed/copy elapsed_ms={}",
            start.elapsed().as_secs_f64() * 1_000.0
        );
        let operation = stream_payloads(&seeded, &read_once_operation_key())?;
        let seeded_extension = stream_payloads(&seeded, &read_once_extension_key())?;
        drop(seeded);
        drop(seed_copy);
        let start = Instant::now();
        let complete_extension = in_line_repair(&memory_holding(&operation, &seeded_extension)?)?;
        println!(
            "repair fixture reference elapsed_ms={}",
            start.elapsed().as_secs_f64() * 1_000.0
        );
        if !complete_extension.starts_with(&seeded_extension) {
            return Err("the in-line repair rewrote a row the live commits had written".into());
        }
        Ok(Self {
            operation_rows: u64::try_from(operation.len())?,
            operation,
            complete_extension,
        })
    }

    /// The complete extension stream with its last `owed` rows lost.
    fn extension_owing(&self, owed: usize) -> Result<&[Vec<u8>], Box<dyn Error>> {
        let kept = self
            .complete_extension
            .len()
            .checked_sub(owed)
            .ok_or_else(|| {
                format!(
                    "the fixture's {} extension rows cannot lose {owed}",
                    self.complete_extension.len()
                )
            })?;
        self.complete_extension
            .get(..kept)
            .ok_or_else(|| "the kept extension prefix is out of range".into())
    }
}

/// What one production replay of a log owing repairs did.
struct PagedRepair {
    /// The extension stream the replay left: the flat rows it was given, then
    /// the repair rows it committed routed after them.
    extension: Vec<Vec<u8>>,
    /// The most repair rows the replay's merges held at once.
    staged_high_water: usize,
    /// Every write and flush the replay made.
    counts: StoreCounts,
    store_counts: StoreCounts,
    /// Operation-log rows the replay read.
    operation_reads: u64,
}

/// Runs the production replay over a fresh store holding `operation` and
/// `extension`, counting its writes, its operation-log reads and the most
/// repair rows it held at once.
fn production_repair(
    operation: &[Vec<u8>],
    extension: &[Vec<u8>],
    operation_rows: u64,
) -> Result<PagedRepair, Box<dyn Error>> {
    let inner = Arc::new(CountingStore::new(Shared(store_holding(
        operation, extension,
    )?)));
    let store_counts = inner.handle();
    let inner: Arc<dyn DurableStore> = inner;
    let reads = Arc::new(OperationRowCounter::new(
        Arc::clone(&inner),
        READ_ONCE_CONVERSATION,
    ));
    let read_counted: Arc<dyn DurableStore> = reads.clone();
    let counting = Arc::new(CountingStore::new(Shared(read_counted)));
    let counts = counting.handle();
    let counted: Arc<dyn DurableStore> = counting;
    let earlier_high_water = take_staged_repair_high_water();
    if earlier_high_water != 0 {
        return Err(format!(
            "{earlier_high_water} staged repair rows were measured before the replay ran"
        )
        .into());
    }
    let replayed = match one_pass(&counted, READ_ONCE_CONVERSATION)? {
        Ok(authority) => authority.next_log_sequence,
        Err(refused) => return Err(format!("the log owing repairs refused: {refused}").into()),
    };
    let staged_high_water = take_staged_repair_high_water();
    assert_eq!(replayed, operation_rows);
    Ok(PagedRepair {
        extension: stream_payloads(&inner, &read_once_extension_key())?,
        staged_high_water,
        counts: counts.snapshot(),
        store_counts: store_counts.snapshot(),
        operation_reads: reads.take(),
    })
}

fn pin_repair_forwarding(repair: &PagedRepair) {
    println!(
        "repair outer counts={:?} actual store counts={:?}",
        repair.counts, repair.store_counts
    );
    assert_eq!(
        repair.store_counts.append_calls, 0,
        "read-count wrapper must preserve batch appends"
    );
    assert_eq!(
        repair.store_counts.append_all_calls,
        repair.counts.append_all_calls
    );
    assert_eq!(
        repair.store_counts.apply_routed_batch_calls, repair.counts.apply_routed_batch_calls,
        "read-count wrapper must preserve routed commits"
    );
    assert_eq!(
        repair.store_counts.apply_routed_batch_items,
        repair.counts.apply_routed_batch_items
    );
    assert_eq!(repair.store_counts.flushes, repair.counts.flushes);
}

/// The items of the routed commits that write `rows` repair rows as `pages`
/// pages: each page stages its rows into one batch, and the commit adds the
/// extension stream's head once.
fn routed_repair_items(rows: usize, pages: u64) -> Result<u64, Box<dyn Error>> {
    u64::try_from(rows)?
        .checked_add(pages)
        .ok_or_else(|| "routed repair item count overflowed".into())
}

/// DUR-008 D03 (Hermes 20:18): staging is bounded to ONE page. A log whose
/// owed repairs span three pages is replayed as two passes — the first audits
/// the whole log and writes nothing, the second writes each page as it fills —
/// and:
///
/// - the extension stream it leaves is byte-identical to the in-line repair's,
///   rows and sequences alike;
/// - no merge ever held more than one page of repair rows, with a full page
///   observed so the bound is a measurement;
/// - every page is one durable operation, which since every commit is a
///   routed batch is one `apply_routed_batch` that publishes without a
///   separate flush, and nothing is appended row by row or flat;
/// - the operation log is read exactly twice, once per pass.
#[test]
fn owed_repairs_spanning_three_pages_never_stage_more_than_one_page() -> Result<(), Box<dyn Error>>
{
    let page = UNIT2_OUTBOX_RESTORE_BATCH_ROWS;
    let owed = page
        .checked_mul(2)
        .and_then(|rows| rows.checked_add(page / 2))
        .ok_or("owed repair count overflowed")?;
    let pages = owed.div_ceil(page);
    assert_eq!(pages, 3, "the owed repairs do not span three pages");
    let fixture = RepairFixture::new()?;
    let owing = fixture.extension_owing(owed)?;

    let in_line = in_line_repair(&memory_holding(&fixture.operation, owing)?)?;
    assert_eq!(
        in_line, fixture.complete_extension,
        "the in-line reference did not restore the {owed} lost rows"
    );

    let repaired = production_repair(&fixture.operation, owing, fixture.operation_rows)?;
    assert_eq!(
        repaired.extension, in_line,
        "the paged repair's extension stream differs from the in-line repair's"
    );
    assert!(
        repaired.staged_high_water <= page,
        "the replay held {} repair rows at once; the bound is one page of {page}",
        repaired.staged_high_water
    );
    assert_eq!(
        repaired.staged_high_water, page,
        "no full page was ever staged, so the bound above is not a measurement"
    );
    let pages = u64::try_from(pages)?;
    assert_eq!(
        repaired.counts.append_calls, 0,
        "a repair was appended row by row"
    );
    assert_eq!(
        repaired.counts.append_all_calls, 0,
        "a repair page was appended flat"
    );
    assert_eq!(
        repaired.counts.apply_routed_batch_calls, pages,
        "a repair page was not one routed commit"
    );
    assert_eq!(
        repaired.counts.apply_routed_batch_items,
        routed_repair_items(owed, pages)?
    );
    assert_eq!(
        repaired.counts.flushes, 0,
        "a routed repair page flushed separately"
    );
    assert_eq!(
        repaired.operation_reads,
        fixture
            .operation_rows
            .checked_mul(2)
            .ok_or("operation read count overflowed")?,
        "a log owing more than one page must be audited whole, then replayed once more"
    );
    pin_repair_forwarding(&repaired);
    Ok(())
}

/// The bound's edge: exactly one page of owed repairs is staged and written
/// in the ONE pass, as one routed commit and no separate flush, reading the
/// operation log once; one row more is the two-pass shape, and both leave the
/// in-line repair's bytes.
#[test]
fn one_page_of_repairs_is_one_pass_and_one_row_more_is_two() -> Result<(), Box<dyn Error>> {
    let page = UNIT2_OUTBOX_RESTORE_BATCH_ROWS;
    let fixture = RepairFixture::new()?;

    let one_page = production_repair(
        &fixture.operation,
        fixture.extension_owing(page)?,
        fixture.operation_rows,
    )?;
    assert_eq!(one_page.extension, fixture.complete_extension);
    assert_eq!(one_page.staged_high_water, page);
    assert_eq!(one_page.counts.append_calls, 0);
    assert_eq!(one_page.counts.append_all_calls, 0);
    assert_eq!(one_page.counts.apply_routed_batch_calls, 1);
    assert_eq!(
        one_page.counts.apply_routed_batch_items,
        routed_repair_items(page, 1)?
    );
    assert_eq!(one_page.counts.flushes, 0);
    assert_eq!(
        one_page.operation_reads, fixture.operation_rows,
        "one page of repairs fits the stage, so the replay must stay one pass"
    );

    let over = page.checked_add(1).ok_or("owed repair count overflowed")?;
    let one_more = production_repair(
        &fixture.operation,
        fixture.extension_owing(over)?,
        fixture.operation_rows,
    )?;
    assert_eq!(one_more.extension, fixture.complete_extension);
    assert_eq!(one_more.staged_high_water, page);
    assert_eq!(one_more.counts.append_calls, 0);
    assert_eq!(one_more.counts.append_all_calls, 0);
    assert_eq!(one_more.counts.apply_routed_batch_calls, 2);
    assert_eq!(
        one_more.counts.apply_routed_batch_items,
        routed_repair_items(over, 2)?
    );
    assert_eq!(one_more.counts.flushes, 0);
    assert_eq!(
        one_more.operation_reads,
        fixture
            .operation_rows
            .checked_mul(2)
            .ok_or("operation read count overflowed")?,
        "one row past the page must run the audit-then-write two passes"
    );
    pin_repair_forwarding(&one_page);
    pin_repair_forwarding(&one_more);
    Ok(())
}