zakura-state 11.0.0

//! StateService test vectors.

#![allow(clippy::unwrap_in_result)]

// TODO: move these tests into tests::vectors and tests::prop modules.

use std::{env, sync::Arc, time::Duration};

use tokio::{runtime::Runtime, time::timeout};
use tower::{buffer::Buffer, util::BoxService, Service, ServiceExt};

use zakura_chain::{
    block::{self, Block, CountedHeader, Height},
    chain_tip::ChainTip,
    fmt::SummaryDebug,
    parameters::{Network, NetworkUpgrade},
    serialization::{ZcashDeserialize, ZcashDeserializeInto},
    transaction, transparent,
    value_balance::ValueBalance,
};

use zakura_test::{prelude::*, transcript::Transcript};

use crate::tests::FakeChainHelper;

use crate::{
    arbitrary::Prepare,
    init_test,
    service::{
        arbitrary::populated_state,
        chain_tip::TipAction,
        finalized_state::{DiskWriteBatch, FinalizedState, FrontierArtifact, FrontierEntry},
        write::NonFinalizedWriteFailureKind,
        StateService,
    },
    tests::setup::{partial_nu5_chain_strategy, transaction_v4_from_coinbase},
    BlockAdmission, BoxError, CheckpointVerifiedBlock, CommitBlockError, Config,
    HistoricalTreeUnavailable, ParentInputs, PruningConfig, Request, Response,
    SemanticallyVerifiedBlock, StateInitError, StorageMode, ValidateContextError,
    CHAIN_TIP_UPDATE_WAIT_LIMIT, MAX_HISTORICAL_TREE_REPLAY_BLOCKS,
};

const LAST_BLOCK_HEIGHT: u32 = 10;

#[tokio::test]
async fn sent_sibling_removal_keeps_await_utxo_ready() {
    use super::write::{NonFinalizedWriteFailure, NonFinalizedWriteUpdate};
    use crate::tests::FakeChainHelper;

    let _init_guard = zakura_test::init();
    let block: Arc<Block> = zakura_test::vectors::BLOCK_MAINNET_419201_BYTES
        .zcash_deserialize_into()
        .unwrap();
    let siblings = block.make_fake_siblings(2);
    let removed = siblings[0].clone().prepare();
    let in_flight = siblings[1].clone().prepare();
    let (outpoint, expected_output) = in_flight
        .new_outputs
        .iter()
        .find(|(_, output)| !output.utxo.from_coinbase)
        .unwrap();

    for update in [
        NonFinalizedWriteUpdate::Evicted(vec![removed.hash]),
        NonFinalizedWriteUpdate::Failed(NonFinalizedWriteFailure {
            hash: removed.hash,
            kind: NonFinalizedWriteFailureKind::Invalid,
        }),
        NonFinalizedWriteUpdate::Failed(NonFinalizedWriteFailure {
            hash: removed.hash,
            kind: NonFinalizedWriteFailureKind::Retryable,
        }),
    ] {
        let (mut state, _, _, _) =
            StateService::new(Config::ephemeral(), &Network::Mainnet, Height::MAX, 0)
                .await
                .unwrap();
        state.non_finalized_block_write_sent_hashes.add(&removed);
        state.non_finalized_block_write_sent_hashes.add(&in_flight);
        state.handle_non_finalized_write_update(update);
        assert!(!state
            .non_finalized_block_write_sent_hashes
            .contains(&removed.hash));
        assert!(state
            .non_finalized_block_write_sent_hashes
            .contains(&in_flight.hash));

        let response = timeout(
            Duration::from_secs(5),
            state.call(Request::AwaitUtxo(*outpoint)),
        )
        .await
        .expect("a sent sibling still supplies this output before it commits")
        .unwrap();
        assert_eq!(response, Response::Utxo(expected_output.utxo.clone()));
        assert_eq!(state.pending_utxos.len(), 0);
    }
}

#[tokio::test(flavor = "multi_thread")]
async fn block_info_does_not_wait_for_a_queued_parent() {
    let _init_guard = zakura_test::init();
    use tower::Service;

    let (mut state, _, _, _) =
        StateService::new(Config::ephemeral(), &Network::Mainnet, Height(1), 0)
            .await
            .unwrap();
    let genesis: Arc<Block> = zakura_test::vectors::BLOCK_MAINNET_GENESIS_BYTES
        .zcash_deserialize_into()
        .unwrap();
    let parent: Arc<Block> = zakura_test::vectors::BLOCK_MAINNET_1_BYTES
        .zcash_deserialize_into()
        .unwrap();
    let hash = parent.hash();
    let limit = Duration::from_secs(10);

    assert!(matches!(
        timeout(limit, (&mut state).oneshot(Request::BlockInfo(hash)))
            .await
            .unwrap()
            .unwrap(),
        Response::BlockInfo(None)
    ));

    // Without genesis, the writer queues this parent but cannot commit it.
    let commit = state
        .ready()
        .await
        .unwrap()
        .call(Request::CommitCheckpointVerifiedBlock(parent.into()));
    assert_eq!(state.finalized_state_queued_blocks.len(), 1);
    assert!(matches!(
        (&mut state)
            .oneshot(Request::KnownBlock(hash))
            .await
            .unwrap(),
        Response::KnownBlock(Some(_))
    ));
    assert!(matches!(
        timeout(limit, (&mut state).oneshot(Request::BlockInfo(hash)))
            .await
            .unwrap()
            .unwrap(),
        Response::BlockInfo(None)
    ));

    timeout(
        limit,
        (&mut state).oneshot(Request::CommitCheckpointVerifiedBlock(genesis.into())),
    )
    .await
    .unwrap()
    .unwrap();
    timeout(limit, commit).await.unwrap().unwrap();
    assert!(matches!(
        timeout(limit, (&mut state).oneshot(Request::BlockInfo(hash)))
            .await
            .unwrap()
            .unwrap(),
        Response::BlockInfo(Some(_))
    ));
}

#[tokio::test(flavor = "multi_thread")]
async fn await_block_info_waits_for_checkpoint_commit() {
    let _init_guard = zakura_test::init();
    let state = init_test(&Network::Mainnet).await;
    let block0: Arc<Block> = zakura_test::vectors::BLOCK_MAINNET_GENESIS_BYTES
        .zcash_deserialize_into()
        .expect("genesis block deserializes");
    let block1: Arc<Block> = zakura_test::vectors::BLOCK_MAINNET_1_BYTES
        .zcash_deserialize_into()
        .expect("block 1 deserializes");
    let hash = block1.hash();
    let limit = Duration::from_secs(10);
    let mut wait = tokio::spawn(state.clone().oneshot(Request::AwaitBlockInfo(hash)));
    let second_wait = tokio::spawn(state.clone().oneshot(Request::AwaitBlockInfo(hash)));

    // A different block's commit must wake readers without answering this request.
    timeout(
        limit,
        state
            .clone()
            .oneshot(Request::CommitCheckpointVerifiedBlock(block0.into())),
    )
    .await
    .expect("genesis commit completes")
    .expect("genesis commits");
    assert!(timeout(Duration::from_millis(50), &mut wait).await.is_err());

    timeout(
        limit,
        state
            .clone()
            .oneshot(Request::CommitCheckpointVerifiedBlock(block1.into())),
    )
    .await
    .expect("parent commit completes")
    .expect("parent commits");
    for waiter in [wait, second_wait] {
        let response = timeout(limit, waiter)
            .await
            .expect("reader observes commit")
            .expect("reader task completes")
            .expect("lookup succeeds");
        assert!(matches!(response, Response::BlockInfo(Some(_))));
    }

    // A reader registered after the notification must also see the committed block.
    let response = timeout(limit, state.oneshot(Request::AwaitBlockInfo(hash)))
        .await
        .expect("known parent returns immediately")
        .expect("lookup succeeds");
    assert!(matches!(response, Response::BlockInfo(Some(_))));
}

/// Returns Mainnet blocks 0..=2 with v4 coinbases, so block 2 can reach the non-finalized state.
fn v4_coinbase_mainnet_chain() -> Vec<Arc<Block>> {
    let mut chain: Vec<Arc<Block>> = Vec::new();
    for (_height, block_bytes) in zakura_test::vectors::MAINNET_BLOCKS.range(0..=2) {
        let mut block = block_bytes
            .zcash_deserialize_into::<Block>()
            .expect("the mainnet block vector decodes");
        block.transactions = vec![Arc::new(transaction_v4_from_coinbase(
            &block.transactions[0],
        ))];
        if let Some(parent) = chain.last() {
            Arc::make_mut(&mut block.header).previous_block_hash = parent.hash();
        }
        chain.push(Arc::new(block));
    }
    chain
}

/// Returns a state with blocks 0 and 1 of `chain` finalized.
async fn state_with_finalized_parent(chain: &[Arc<Block>]) -> StateService {
    let (mut state, _, _, _) =
        StateService::new(Config::ephemeral(), &Network::Mainnet, Height::MAX, 0)
            .await
            .expect("ephemeral state initialization succeeds");
    for block in &chain[0..=1] {
        let result = state
            .queue_and_commit_to_finalized_state(CheckpointVerifiedBlock::from(block.clone()))
            .await;
        assert!(matches!(result, Ok(Ok(_))), "checkpoint commit: {result:?}");
    }
    state
}

fn await_block_info_error(result: Result<Response, BoxError>) -> crate::AwaitBlockInfoError {
    let error = result.expect_err("the wait fails");
    error
        .downcast_ref::<crate::AwaitBlockInfoError>()
        .unwrap_or_else(|| panic!("unexpected error: {error:?}"))
        .clone()
}

/// A block restored by `reconsiderblock` must wake readers waiting for it,
/// even though no block commit follows.
#[tokio::test(flavor = "multi_thread")]
async fn await_block_info_wakes_after_reconsider() {
    let _init_guard = zakura_test::init();
    let limit = Duration::from_secs(10);
    let chain = v4_coinbase_mainnet_chain();
    let mut state = state_with_finalized_parent(&chain).await;

    let child = chain[2].clone().prepare();
    let hash = child.hash;
    timeout(
        limit,
        state.queue_and_commit_to_non_finalized_state(child, None),
    )
    .await
    .expect("non-finalized commit completes")
    .expect("the response channel stays open")
    .expect("the non-finalized block commits");

    let state = Buffer::new(BoxService::new(state), 10);
    timeout(limit, state.clone().oneshot(Request::InvalidateBlock(hash)))
        .await
        .expect("invalidation completes")
        .expect("block is invalidated");

    let mut wait = tokio::spawn(state.clone().oneshot(Request::AwaitBlockInfo(hash)));
    assert!(
        timeout(Duration::from_millis(100), &mut wait)
            .await
            .is_err(),
        "an invalidated block has no block info"
    );

    timeout(limit, state.clone().oneshot(Request::ReconsiderBlock(hash)))
        .await
        .expect("reconsideration completes")
        .expect("block is reconsidered");

    let response = timeout(limit, wait)
        .await
        .expect("reader wakes without another block commit")
        .expect("reader task completes")
        .expect("lookup succeeds");
    assert!(matches!(response, Response::BlockInfo(Some(_))));
}

/// A reader waiting for a block stops waiting when the state rejects that block.
#[tokio::test(flavor = "multi_thread")]
async fn await_block_info_fails_when_the_block_is_rejected() {
    let _init_guard = zakura_test::init();
    let limit = Duration::from_secs(10);
    let chain = v4_coinbase_mainnet_chain();
    let mut state = state_with_finalized_parent(&chain).await;

    // A spend of an output this state does not have fails contextual validation.
    let later: Block = zakura_test::vectors::BLOCK_MAINNET_419201_BYTES
        .zcash_deserialize_into()
        .expect("the later block vector decodes");
    let spend = later
        .transactions
        .iter()
        .find(|transaction| !transaction.is_coinbase() && !transaction.inputs().is_empty())
        .expect("the later block has a transparent spend")
        .clone();
    let mut invalid = (*chain[2]).clone();
    invalid.transactions.push(spend);
    let invalid = Arc::new(invalid).prepare();
    let hash = invalid.hash;

    let wait = state.call(Request::AwaitBlockInfo(hash));
    let commit = state.queue_and_commit_to_non_finalized_state(invalid, None);
    timeout(limit, commit)
        .await
        .expect("non-finalized commit completes")
        .expect("the response channel stays open")
        .expect_err("the invalid block is rejected");

    let error = await_block_info_error(timeout(limit, wait).await.expect("reader wakes"));
    assert_eq!(error, crate::AwaitBlockInfoError::Rejected { hash });
}

/// A reader waiting for a block that never arrives stops at the wait limit.
#[tokio::test(flavor = "multi_thread")]
async fn await_block_info_times_out() {
    let _init_guard = zakura_test::init();
    let chain = v4_coinbase_mainnet_chain();
    let mut state = state_with_finalized_parent(&chain).await;
    let wait_limit = Duration::from_millis(200);
    state.await_block_info_timeout = wait_limit;

    let hash = chain[2].hash();
    let error = await_block_info_error(
        timeout(
            Duration::from_secs(10),
            state.call(Request::AwaitBlockInfo(hash)),
        )
        .await
        .expect("the wait limit ends the request"),
    );
    assert_eq!(
        error,
        crate::AwaitBlockInfoError::TimedOut {
            hash,
            limit: wait_limit,
        }
    );

    // A reader that starts after the block commits still succeeds.
    let child = chain[2].clone().prepare();
    timeout(
        Duration::from_secs(10),
        state.queue_and_commit_to_non_finalized_state(child, None),
    )
    .await
    .expect("non-finalized commit completes")
    .expect("the response channel stays open")
    .expect("the non-finalized block commits");
    let response = state
        .call(Request::AwaitBlockInfo(hash))
        .await
        .expect("lookup succeeds");
    assert!(matches!(response, Response::BlockInfo(Some(_))));
}

#[test]
fn mined_orphans_finish_without_entering_the_sync_queue() {
    let _init_guard = zakura_test::init();
    let network = Network::Mainnet;
    let runtime = Runtime::new().expect("the Tokio runtime starts");
    let (mut state_service, _, _, _) = runtime
        .block_on(StateService::new(
            Config::ephemeral(),
            &network,
            Height::MAX,
            0,
        ))
        .expect("ephemeral state initialization succeeds");
    let block = Arc::new(
        zakura_test::vectors::BLOCK_MAINNET_1_BYTES
            .zcash_deserialize_into::<Block>()
            .expect("the mainnet height-one block is valid"),
    )
    .prepare();

    for _ in 0..2 {
        let admission = BlockAdmission::pending();
        let response = state_service
            .queue_and_commit_to_non_finalized_state(block.clone(), Some(admission.clone()));
        assert!(!runtime.block_on(admission.wait()));
        assert!(response
            .blocking_recv()
            .expect("state responds immediately")
            .is_err());
        assert!(state_service
            .non_finalized_state_queued_blocks
            .get_mut(&block.hash)
            .is_none());
    }

    let mut response = state_service.queue_and_commit_to_non_finalized_state(block.clone(), None);
    assert!(matches!(
        response.try_recv(),
        Err(tokio::sync::oneshot::error::TryRecvError::Empty)
    ));
    assert!(state_service
        .non_finalized_state_queued_blocks
        .get_mut(&block.hash)
        .is_some());
}

/// A Heartwood-activation state the relay preflight tests share.
struct Fixture {
    network: Network,
    finalized: super::finalized_state::FinalizedState,
    non_finalized: super::non_finalized_state::NonFinalizedState,
    /// The block the test builds its candidate on.
    parent: Arc<Block>,
}

impl Fixture {
    /// A state whose best chain ends just after the Heartwood activation block.
    fn heartwood() -> Self {
        let network = Network::Mainnet;
        let heartwood_height = NetworkUpgrade::Heartwood
            .activation_height(&network)
            .expect("Heartwood activates")
            .0;
        let root = Arc::new(
            network.block_map()[&(heartwood_height - 1)]
                .zcash_deserialize_into::<Block>()
                .expect("pre-Heartwood test block is valid"),
        );
        let finalized = super::finalized_state::FinalizedState::new(&Config::ephemeral(), &network)
            .expect("ephemeral finalized state opens");
        let mut non_finalized = super::non_finalized_state::NonFinalizedState::new(&network);
        non_finalized
            .commit_new_chain(root.clone().prepare(), &finalized)
            .expect("root commits");
        let activation = root.make_fake_child().set_block_commitment([0; 32]);
        non_finalized
            .commit_block(activation.clone().prepare(), &finalized)
            .expect("Heartwood activation commits");

        Self {
            network,
            finalized,
            non_finalized,
            parent: activation,
        }
    }

    /// The Heartwood state, plus a heavier best child and the lighter side child it returns.
    fn with_side_chain() -> Self {
        let mut fixture = Self::heartwood();
        let activation = fixture.parent.clone();
        let sibling_commitment = fixture.best_chain_commitment();
        let best = activation
            .make_fake_child()
            .set_block_commitment(sibling_commitment)
            .set_work(100);
        let side = activation
            .make_fake_child()
            .set_block_commitment(sibling_commitment)
            .set_work(50);
        fixture
            .non_finalized
            .commit_block(best.prepare(), &fixture.finalized)
            .expect("best child commits");
        fixture
            .non_finalized
            .commit_block(side.clone().prepare(), &fixture.finalized)
            .expect("side child commits");
        fixture.parent = side;

        fixture
    }

    /// The Heartwood state extended by one full difficulty-adjustment window.
    fn with_difficulty_context() -> Self {
        use zakura_header_chain::POW_ADJUSTMENT_BLOCK_SPAN;

        let mut fixture = Self::heartwood();
        // The activation block is the first block of the window.
        for _ in 1..POW_ADJUSTMENT_BLOCK_SPAN {
            let commitment = fixture.best_chain_commitment();
            let mut child = fixture
                .parent
                .make_fake_child()
                .set_block_commitment(commitment);
            let child_height = child.coinbase_height().expect("the child has a height");
            Arc::make_mut(&mut Arc::make_mut(&mut child).header).time = fixture.parent.header.time
                + NetworkUpgrade::target_spacing_for_height(&fixture.network, child_height);
            fixture
                .non_finalized
                .commit_block(child.clone().prepare(), &fixture.finalized)
                .expect("difficulty context block commits");
            fixture.parent = child;
        }

        fixture
    }

    /// The history root the best chain commits to.
    fn best_chain_commitment(&self) -> [u8; 32] {
        self.non_finalized
            .best_chain()
            .expect("the best chain exists")
            .history_block_commitment_tree()
            .hash()
            .expect("the best chain has a history root")
            .into()
    }

    /// Runs the relay preflight for a candidate built on this fixture.
    fn preflight(
        &self,
        child: Arc<Block>,
    ) -> Result<crate::PreparedMinedRelayEligibility, BoxError> {
        super::check_prepared_mined_relay_eligibility_for_state(
            &self.network,
            &self.non_finalized,
            &self.finalized.db,
            crate::BlockCommitmentData {
                block: child,
                auth_data_root: None,
            },
        )
    }
}

#[test]
fn prepared_relay_preflight_authorizes_a_selected_tip_child() {
    let _init_guard = zakura_test::init();
    let fixture = Fixture::with_side_chain();
    let best = fixture
        .non_finalized
        .best_tip_block()
        .expect("the test state has a best tip")
        .block
        .clone();
    let child = best
        .make_fake_child()
        .set_block_commitment(fixture.best_chain_commitment());

    assert_eq!(
        fixture
            .preflight(child)
            .expect("the selected tip child passes the relay preflight"),
        crate::PreparedMinedRelayEligibility::Authorized
    );
}

#[test]
fn prepared_relay_preflight_uses_commit_first_for_a_side_chain() {
    let _init_guard = zakura_test::init();
    let fixture = Fixture::with_side_chain();
    let side = fixture.parent.clone();
    let parent_hash = side.hash();
    let parent_chain = fixture
        .non_finalized
        .find_chain(|chain| chain.contains_block_hash(parent_hash))
        .expect("side parent chain exists");
    let history_tree = super::read::tree::history_tree(
        Some(parent_chain),
        &fixture.finalized.db,
        parent_hash.into(),
    )
    .expect("side parent has a history tree");
    let commitment: [u8; 32] = history_tree
        .hash()
        .expect("the side chain has a history root")
        .into();
    let child = side.make_fake_child().set_block_commitment(commitment);

    assert_eq!(
        fixture
            .preflight(child)
            .expect("the side-chain child proves its expected work"),
        crate::PreparedMinedRelayEligibility::CommitFirst
    );
}

#[test]
fn prepared_relay_preflight_rejects_an_easier_claimed_target() {
    let _init_guard = zakura_test::init();
    let fixture = Fixture::with_difficulty_context();
    let tip = fixture.parent.clone();
    let mut child = tip
        .make_fake_child()
        .set_block_commitment(fixture.best_chain_commitment())
        .set_work(1);
    let child_height = child.coinbase_height().expect("the child has a height");
    Arc::make_mut(&mut Arc::make_mut(&mut child).header).time =
        tip.header.time + NetworkUpgrade::target_spacing_for_height(&fixture.network, child_height);

    let error = fixture
        .preflight(child)
        .expect_err("an easier claimed target fails the relay preflight");

    assert!(matches!(
        error.downcast_ref::<ValidateContextError>(),
        Some(ValidateContextError::InvalidDifficultyThreshold { .. })
    ));
}

#[test]
fn prepared_relay_preflight_rejects_time_at_or_below_median() {
    use zakura_header_chain::{AdjustedDifficulty, POW_ADJUSTMENT_BLOCK_SPAN};

    let _init_guard = zakura_test::init();
    let fixture = Fixture::with_difficulty_context();
    let tip = fixture.parent.clone();
    let parent_hash = tip.hash();
    let mut child = tip
        .make_fake_child()
        .set_block_commitment(fixture.best_chain_commitment());
    let too_early =
        super::any_ancestor_blocks(&fixture.non_finalized, &fixture.finalized.db, parent_hash)
            .take(11)
            .last()
            .expect("the context has a median-time window")
            .header
            .time;
    Arc::make_mut(&mut Arc::make_mut(&mut child).header).time = too_early;
    let relevant_data =
        super::any_ancestor_blocks(&fixture.non_finalized, &fixture.finalized.db, parent_hash)
            .take(POW_ADJUSTMENT_BLOCK_SPAN)
            .map(|block| (block.header.difficulty_threshold, block.header.time));
    let expected_target =
        AdjustedDifficulty::new_from_block(&child, &fixture.network, relevant_data)
            .expect("the context derives an expected target")
            .expected_difficulty_threshold();
    Arc::make_mut(&mut Arc::make_mut(&mut child).header).difficulty_threshold = expected_target;

    let error = fixture
        .preflight(child)
        .expect_err("a time at or below median-time-past fails the relay preflight");

    assert!(matches!(
        error.downcast_ref::<ValidateContextError>(),
        Some(ValidateContextError::TimeTooEarly { .. })
    ));
}

#[test]
fn prepared_relay_preflight_rejects_a_forged_commitment() {
    let _init_guard = zakura_test::init();
    let fixture = Fixture::with_side_chain();
    let child = fixture
        .parent
        .make_fake_child()
        .set_block_commitment([0x42; 32]);

    let error = fixture
        .preflight(child)
        .expect_err("a forged commitment fails the relay preflight");

    assert!(matches!(
        error.downcast_ref::<ValidateContextError>(),
        Some(ValidateContextError::InvalidBlockCommitment(_))
    ));
}

/// A full orphan queue must not strand the descendants that are waiting on the very block
/// that would release them.
///
/// The bound only applies to blocks that must wait for an absent parent. When the queue is
/// filled by descendants of a missing block `B`, `B` itself can extend the chain now, and
/// rejecting it leaves its descendants indexed under a hash that no drain ever visits.
#[tokio::test(flavor = "multi_thread")]
async fn a_full_orphan_queue_still_admits_a_block_whose_parent_is_available() -> Result<()> {
    let _init_guard = zakura_test::init();
    let network = Network::Mainnet;

    // Rewrite the early mainnet coinbases as v4, so these blocks can reach the non-finalized
    // state, and relink each block onto its rewritten parent.
    let mut chain: Vec<Arc<Block>> = Vec::new();
    for (_height, block_bytes) in zakura_test::vectors::MAINNET_BLOCKS.range(0..=2) {
        let mut block = block_bytes
            .zcash_deserialize_into::<Block>()
            .expect("the mainnet block vector decodes");
        block.transactions = vec![Arc::new(transaction_v4_from_coinbase(
            &block.transactions[0],
        ))];
        if let Some(parent) = chain.last() {
            Arc::make_mut(&mut block.header).previous_block_hash = parent.hash();
        }
        chain.push(Arc::new(block));
    }

    let (mut state_service, _, _, _) =
        StateService::new(Config::ephemeral(), &network, Height::MAX, 0)
            .await
            .expect("ephemeral state initialization succeeds");

    // Commit the first two blocks as checkpoint verified blocks, so the finalized tip becomes a
    // parent that `can_fork_chain_at` accepts.
    for block in &chain[0..=1] {
        let result = state_service
            .queue_and_commit_to_finalized_state(CheckpointVerifiedBlock::from(block.clone()))
            .await;
        assert!(
            matches!(result, Ok(Ok(_))),
            "the checkpoint verified block commits: {result:?}",
        );
    }

    // `child` extends the finalized tip, so it can be committed as soon as it is queued.
    // `grandchild` can only be released by `child`.
    let child = chain[2].clone().prepare();
    assert_eq!(child.block.header.previous_block_hash, chain[1].hash());

    let mut grandchild_block = (*child.block).clone();
    Arc::make_mut(&mut grandchild_block.header).previous_block_hash = child.hash;
    let grandchild = Arc::new(grandchild_block).prepare();
    let (grandchild_tx, _grandchild_rx) = tokio::sync::oneshot::channel();
    state_service.non_finalized_state_queued_blocks.queue((
        grandchild.clone(),
        grandchild_tx,
        None,
        0,
    ));

    // Fill the rest of the queue with blocks whose parents this state will never have.
    let mut orphan_count = 0_u64;
    while !state_service.non_finalized_state_queued_blocks.is_full() {
        let mut orphan_block = (*child.block).clone();
        Arc::make_mut(&mut orphan_block.header)
            .previous_block_hash
            .0[..8]
            .copy_from_slice(&orphan_count.to_le_bytes());
        let (orphan_tx, _orphan_rx) = tokio::sync::oneshot::channel();
        state_service.non_finalized_state_queued_blocks.queue((
            Arc::new(orphan_block).prepare(),
            orphan_tx,
            None,
            0,
        ));
        orphan_count += 1;
    }
    assert!(
        orphan_count > 0,
        "the queue reaches its bound through blocks with absent parents",
    );

    // The queue is full, but `child` extends the finalized tip, so it must still be admitted.
    let admission = BlockAdmission::pending();
    let _response = state_service
        .queue_and_commit_to_non_finalized_state(child.clone(), Some(admission.clone()));

    assert!(
        state_service
            .non_finalized_block_write_sent_hashes
            .contains(&child.hash),
        "a block whose parent is the finalized tip is sent for commit even when the queue is full",
    );
    assert!(
        state_service
            .non_finalized_block_write_sent_hashes
            .contains(&grandchild.hash),
        "the descendants waiting on that block are released with it",
    );
    assert!(
        state_service
            .non_finalized_state_queued_blocks
            .get_mut(&grandchild.hash)
            .is_none(),
        "the released descendant leaves the queue",
    );

    Ok(())
}

#[tokio::test]
async fn descendant_arriving_after_a_local_parent_failure_completes_immediately() {
    use super::write::{NonFinalizedWriteFailure, NonFinalizedWriteUpdate};

    let network = Network::Mainnet;
    let (mut state, _, _, _) = StateService::new(Config::ephemeral(), &network, Height::MAX, 0)
        .await
        .expect("the ephemeral state opens");
    let block: Arc<Block> = zakura_test::vectors::BLOCK_MAINNET_419201_BYTES
        .zcash_deserialize_into()
        .expect("the child block vector decodes");
    let block = block.prepare();
    let ancestor = block.block.header.previous_block_hash;
    state.remember_failed_ancestor(ancestor, ancestor, NonFinalizedWriteFailureKind::Retryable);
    let unrelated = block::Hash([0xff; 32]);
    state.remember_failed_ancestor(unrelated, unrelated, NonFinalizedWriteFailureKind::Invalid);
    state.handle_non_finalized_write_update(NonFinalizedWriteUpdate::Evicted(vec![ancestor]));
    assert!(!state.non_finalized_failed_ancestors.contains_key(&ancestor));
    assert_eq!(
        state.non_finalized_failed_ancestors.get(&unrelated),
        Some(&(unrelated, NonFinalizedWriteFailureKind::Invalid))
    );

    // A fresh failure sent after eviction must still reject the next child.
    state.handle_non_finalized_write_update(NonFinalizedWriteUpdate::Failed(
        NonFinalizedWriteFailure {
            hash: ancestor,
            kind: NonFinalizedWriteFailureKind::Retryable,
        },
    ));

    let response = state
        .queue_and_commit_to_non_finalized_state(block.clone(), None)
        .await
        .expect("the state keeps the response channel open")
        .expect_err("the failed parent prevents this request from waiting");

    assert!(matches!(
        response.inner(),
        CommitBlockError::HeaderChainError { error }
            if error.contains(&ancestor.to_string())
    ));
    assert_eq!(
        state.non_finalized_failed_ancestors.get(&block.hash),
        Some(&(ancestor, NonFinalizedWriteFailureKind::Retryable))
    );
}

#[tokio::test]
async fn fork_eviction_allows_new_and_replayed_parents_to_extend() {
    use super::write::{NonFinalizedWriteFailure, NonFinalizedWriteUpdate};
    use crate::tests::FakeChainHelper;

    let _init_guard = zakura_test::init();
    let network = Network::Mainnet;
    let (mut state, _, _, _) = StateService::new(Config::ephemeral(), &network, Height::MAX, 0)
        .await
        .expect("the ephemeral state opens");
    for (_, bytes) in zakura_test::vectors::MAINNET_BLOCKS.range(0..=1) {
        let block: Arc<Block> = bytes.zcash_deserialize_into().unwrap();
        state
            .queue_and_commit_to_finalized_state(CheckpointVerifiedBlock::from(block))
            .await
            .unwrap()
            .unwrap();
    }
    let mut template: Block = zakura_test::vectors::BLOCK_MAINNET_2_BYTES
        .zcash_deserialize_into()
        .unwrap();
    template.transactions = vec![Arc::new(transaction_v4_from_coinbase(
        &template.transactions[0],
    ))];
    let mut siblings =
        Arc::new(template).make_fake_siblings(crate::constants::MAX_NON_FINALIZED_CHAIN_FORKS + 1);
    // Receipt order deliberately opposes the hash tie-breaker.
    siblings.reverse();
    let evicted = &siblings[siblings.len() - 2];
    // Record a temporary failure before this block succeeds on its next attempt.
    state.handle_non_finalized_write_update(NonFinalizedWriteUpdate::Failed(
        NonFinalizedWriteFailure {
            hash: evicted.hash(),
            kind: NonFinalizedWriteFailureKind::Retryable,
        },
    ));
    for (order, block) in siblings.iter().enumerate() {
        let mut prepared = block.clone().prepare();
        prepared.receipt_order = Some(u64::try_from(order).unwrap());
        tokio::time::timeout(
            Duration::from_secs(10),
            state.queue_and_commit_to_non_finalized_state(prepared, None),
        )
        .await
        .unwrap()
        .unwrap()
        .unwrap();
        let retained = state.read_service.latest_non_finalized_state();
        assert_eq!(retained.best_tip().unwrap().1, siblings[0].hash());
        assert!(retained.any_chain_contains(&siblings[order].hash()));
        assert!(retained.chain_iter().count() <= crate::constants::MAX_NON_FINALIZED_CHAIN_FORKS);
    }
    state.drain_non_finalized_write_updates();
    assert!(!state.can_fork_chain_at(&evicted.hash()));
    assert!(!state
        .non_finalized_block_write_sent_hashes
        .contains(&evicted.hash()));
    assert!(!state
        .non_finalized_failed_ancestors
        .contains_key(&evicted.hash()));

    // The newest branch can win with a child. An evicted branch can then be
    // downloaded again and win with two children, without restarting the node.
    for (parent, mut depth) in [(siblings.last().unwrap(), 1), (evicted, 2)] {
        let mut tip = parent.clone();
        if parent.hash() == evicted.hash() {
            tip = tip.make_fake_child();
            let mut child_response =
                state.queue_and_commit_to_non_finalized_state(tip.clone().prepare(), None);
            assert!(
                matches!(
                    child_response.try_recv(),
                    Err(tokio::sync::oneshot::error::TryRecvError::Empty)
                ),
                "the child waits for its previously valid parent to be downloaded again"
            );
            tokio::time::timeout(
                Duration::from_secs(10),
                state.queue_and_commit_to_non_finalized_state(parent.clone().prepare(), None),
            )
            .await
            .unwrap()
            .unwrap()
            .unwrap();
            tokio::time::timeout(Duration::from_secs(10), child_response)
                .await
                .unwrap()
                .unwrap()
                .unwrap();
            depth -= 1;
        }
        for _ in 0..depth {
            tip = tip.make_fake_child();
            tokio::time::timeout(
                Duration::from_secs(10),
                state.queue_and_commit_to_non_finalized_state(tip.clone().prepare(), None),
            )
            .await
            .unwrap()
            .unwrap()
            .unwrap();
        }
        assert_eq!(
            state
                .read_service
                .latest_non_finalized_state()
                .best_tip()
                .unwrap()
                .1,
            tip.hash()
        );
    }
}

#[tokio::test]
async fn reconsideration_eviction_allows_legacy_replay() {
    assert_reconsideration_eviction_replay(false).await;
}

#[tokio::test]
async fn reconsideration_eviction_allows_header_integrated_replay() {
    assert_reconsideration_eviction_replay(true).await;
}

async fn assert_reconsideration_eviction_replay(header_runtime: bool) {
    use crate::tests::FakeChainHelper;

    let _init_guard = zakura_test::init();
    let limit = Duration::from_secs(10);
    let network = Network::new_regtest(Default::default());
    let config = Config {
        enable_zakura_header_seed_from_committed_blocks: header_runtime,
        ..Config::ephemeral()
    };
    let (mut state, _, _, _) = StateService::new(config, &network, Height::MAX, 0)
        .await
        .unwrap();
    let genesis = block::genesis::regtest_genesis_block();
    let mut parent = genesis.make_fake_child();
    Arc::make_mut(&mut Arc::make_mut(&mut parent).header).time += chrono::Duration::seconds(1);
    for block in [genesis, parent.clone()] {
        timeout(
            limit,
            state.queue_and_commit_to_finalized_state(CheckpointVerifiedBlock::from(block)),
        )
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    }
    let mut template = parent.make_fake_child();
    let transaction = transaction_v4_from_coinbase(&template.transactions[0]);
    Arc::make_mut(&mut template).transactions[0] = Arc::new(transaction);
    let merkle_root = template.transactions.iter().cloned().collect();
    let header = Arc::make_mut(&mut Arc::make_mut(&mut template).header);
    header.time += chrono::Duration::seconds(1);
    header.merkle_root = merkle_root;
    header.commitment_bytes =
        <[u8; 32]>::from(state.read_service.db.history_tree().hash().unwrap()).into();

    let mut siblings =
        template.make_fake_siblings(crate::constants::MAX_NON_FINALIZED_CHAIN_FORKS + 1);
    // Receipt order deliberately opposes the hash tie-breaker.
    siblings.reverse();
    for (order, block) in siblings.iter().enumerate() {
        if order == crate::constants::MAX_NON_FINALIZED_CHAIN_FORKS {
            timeout(limit, state.send_invalidate_block(siblings[0].hash()))
                .await
                .unwrap()
                .unwrap()
                .unwrap();
        }
        let mut prepared = block.clone().prepare();
        prepared.receipt_order = Some(u64::try_from(order).unwrap());
        timeout(
            limit,
            state.queue_and_commit_to_non_finalized_state(prepared, None),
        )
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    }
    assert_eq!(
        state
            .read_service
            .header_chain_reader_receiver
            .borrow()
            .is_some(),
        header_runtime
    );
    assert_eq!(
        state
            .read_service
            .latest_non_finalized_state()
            .chain_iter()
            .count(),
        crate::constants::MAX_NON_FINALIZED_CHAIN_FORKS
    );
    timeout(limit, state.send_reconsider_block(siblings[0].hash()))
        .await
        .unwrap()
        .unwrap()
        .unwrap();

    let evicted = siblings.last().unwrap();
    assert!(!state
        .read_service
        .latest_non_finalized_state()
        .any_chain_contains(&evicted.hash()));
    let known = timeout(limit, state.call(Request::KnownBlock(evicted.hash())))
        .await
        .unwrap()
        .unwrap();
    assert!(
        matches!(known, Response::KnownBlock(None)),
        "an evicted body must be downloadable again: {known:?}"
    );
    assert!(!state.can_fork_chain_at(&evicted.hash()));
    assert!(!state
        .non_finalized_failed_ancestors
        .contains_key(&evicted.hash()));

    let mut prepared = evicted.clone().prepare();
    prepared.receipt_order = Some(u64::try_from(siblings.len()).unwrap());
    timeout(
        limit,
        state.queue_and_commit_to_non_finalized_state(prepared, None),
    )
    .await
    .unwrap()
    .unwrap()
    .unwrap();
    let retained = state.read_service.latest_non_finalized_state();
    assert_eq!(retained.best_tip().unwrap().1, siblings[0].hash());
    let history_root = retained
        .find_chain(|chain| chain.contains_block_hash(evicted.hash()))
        .unwrap()
        .history_tree(crate::HashOrHeight::Hash(evicted.hash()))
        .unwrap()
        .hash()
        .unwrap();
    let mut child = evicted.make_fake_child();
    let merkle_root = child.transactions.iter().cloned().collect();
    let header = Arc::make_mut(&mut Arc::make_mut(&mut child).header);
    header.time += chrono::Duration::seconds(1);
    header.merkle_root = merkle_root;
    header.commitment_bytes = <[u8; 32]>::from(history_root).into();
    timeout(
        limit,
        state.queue_and_commit_to_non_finalized_state(child.clone().prepare(), None),
    )
    .await
    .unwrap()
    .unwrap()
    .unwrap();
    assert_eq!(
        state
            .read_service
            .latest_non_finalized_state()
            .best_tip()
            .unwrap()
            .1,
        child.hash()
    );
}

#[tokio::test]
async fn state_init_loads_the_embedded_mainnet_frontier_grid() {
    let network = Network::Mainnet;
    let config = Config::ephemeral();
    assert!(
        config.derive_historical_trees(false),
        "an ephemeral archive config derives, so this covers the deriving case"
    );

    assert!(
        super::init(config.clone(), &network, Height::MAX, 0)
            .await
            .is_ok(),
        "a deriving Mainnet node must start without a frontier grid path"
    );

    let loaded = super::load_historical_frontier_artifact(&network, &config, false)
        .expect("the embedded Mainnet grid loads");
    assert_eq!(
        loaded.last_checkpoint,
        Some(network.checkpoint_list().max_height()),
        "the default grid covers the embedded checkpoint handoff"
    );
}

#[tokio::test]
async fn historical_frontier_load_errors_are_returned_from_state_init() {
    let network = Network::Mainnet;
    let temp_dir = tempfile::tempdir().expect("temporary directory is created");
    let missing_path = temp_dir.path().join("missing.bin");
    let missing_config = Config {
        historical_frontier_artifact: Some(missing_path.clone()),
        ..Config::ephemeral()
    };

    assert!(matches!(
        super::init(missing_config, &network, Height::MAX, 0).await,
        Err(StateInitError::HistoricalFrontierArtifact { path, .. }) if path == missing_path
    ));

    let corrupt_path = temp_dir.path().join("corrupt.bin");
    std::fs::write(&corrupt_path, b"not a frontier artifact")
        .expect("corrupt test artifact is written");
    let corrupt_config = Config {
        historical_frontier_artifact: Some(corrupt_path.clone()),
        ..Config::ephemeral()
    };

    assert!(matches!(
        super::init(corrupt_config.clone(), &network, Height::MAX, 0).await,
        Err(StateInitError::HistoricalFrontierArtifact { path, .. }) if path == corrupt_path
    ));

    // A node that does not derive never reads the file, so the same broken path is a warning
    // rather than a refusal to start.
    let legacy_recompute = Config {
        vct_fast_sync: false,
        ..corrupt_config
    };
    assert!(
        super::load_historical_frontier_artifact(&network, &legacy_recompute, false).is_ok(),
        "an unusable grid must not stop a node that would never have read it"
    );
}

#[test]
fn durable_vct_marker_keeps_historical_derivation_enabled_after_config_change() {
    let network = Network::Mainnet;
    let initial_config = Config::ephemeral();
    let finalized_state =
        FinalizedState::new(&initial_config, &network).expect("ephemeral finalized state opens");
    let artifact_checkpoint = Height(11);
    let artifact_file =
        tempfile::NamedTempFile::new().expect("temporary frontier artifact is created");
    let artifact = FrontierArtifact {
        spacing: 1,
        last_checkpoint: artifact_checkpoint,
        entries: vec![FrontierEntry {
            height: Height(0),
            sapling: Arc::new(Default::default()),
            orchard: Arc::new(Default::default()),
            ironwood: Arc::new(Default::default()),
        }],
    };
    std::fs::write(artifact_file.path(), artifact.encode(&network))
        .expect("historical frontier artifact is written");

    let mut batch = DiskWriteBatch::new();
    batch.update_vct_sync_marker(&finalized_state.db, Height(10));
    finalized_state
        .db
        .write_batch(batch)
        .expect("VCT handoff marker is written");

    let reopened_config = Config {
        checkpoint_sync: false,
        vct_fast_sync: false,
        historical_frontier_artifact: Some(artifact_file.path().to_path_buf()),
        ..initial_config
    };
    assert!(
        !reopened_config.derive_historical_trees(false),
        "the current configuration does not start a VCT fast sync"
    );

    let loaded = super::load_historical_frontier_artifact(
        &network,
        &reopened_config,
        finalized_state.db.vct_synced_below().is_some(),
    )
    .expect("the durable marker loads the configured grid after sync settings change");
    assert_eq!(
        loaded.last_checkpoint,
        Some(artifact_checkpoint),
        "the reopened archive keeps the grid needed to serve its absent band"
    );
}

#[test]
fn historical_frontier_artifact_older_than_database_vct_handoff_is_ignored() {
    let network = Network::Mainnet;
    let temp_dir = tempfile::tempdir().expect("temporary directory is created");
    let state_config = Config::ephemeral();
    let finalized_state =
        FinalizedState::new(&state_config, &network).expect("ephemeral finalized state opens");
    let vct_handoff = Height(10);
    let mut batch = DiskWriteBatch::new();
    batch.update_vct_sync_marker(&finalized_state.db, vct_handoff);
    finalized_state
        .db
        .write_batch(batch)
        .expect("VCT handoff marker is written");

    let artifact_path = |checkpoint: Height| {
        let path = temp_dir
            .path()
            .join(format!("frontiers-{}.bin", checkpoint.0));
        let artifact = FrontierArtifact {
            spacing: 1,
            last_checkpoint: checkpoint,
            entries: vec![FrontierEntry {
                height: checkpoint,
                sapling: Arc::new(Default::default()),
                orchard: Arc::new(Default::default()),
                ironwood: Arc::new(Default::default()),
            }],
        };
        std::fs::write(&path, artifact.encode(&network))
            .expect("historical frontier artifact is written");
        path
    };

    let stale_path = artifact_path(Height(9));
    let stale_config = Config {
        historical_frontier_artifact: Some(stale_path.clone()),
        ..state_config.clone()
    };
    let stale_cache = super::load_historical_frontier_artifact(&network, &stale_config, false)
        .expect("the stale artifact decodes")
        .discard_if_before_vct_handoff(&stale_config, &finalized_state.db);
    assert_eq!(
        stale_cache
            .lock()
            .expect("historical tree cache lock is available")
            .last_checkpoint(),
        None,
        "a stale artifact is unavailable rather than preventing startup"
    );

    for checkpoint in [vct_handoff, Height(11)] {
        let config = Config {
            historical_frontier_artifact: Some(artifact_path(checkpoint)),
            ..state_config.clone()
        };
        let cache = super::load_historical_frontier_artifact(&network, &config, false)
            .expect("the covering artifact decodes")
            .discard_if_before_vct_handoff(&config, &finalized_state.db);
        assert_eq!(
            cache
                .lock()
                .expect("historical tree cache lock is available")
                .last_checkpoint(),
            Some(checkpoint),
            "an artifact at or above the VCT handoff must cover the absent band"
        );
    }
}

#[test]
fn historical_frontier_artifact_must_tile_the_band_within_the_replay_limit() {
    let network = Network::Mainnet;
    let temp_dir = tempfile::tempdir().expect("temporary directory is created");

    let write = |name: &str, last_checkpoint: Height, entries: Vec<Height>| {
        let path = temp_dir.path().join(name);
        let artifact = FrontierArtifact {
            spacing: 1,
            last_checkpoint,
            entries: entries
                .into_iter()
                .map(|height| FrontierEntry {
                    height,
                    sapling: Arc::new(Default::default()),
                    orchard: Arc::new(Default::default()),
                    ironwood: Arc::new(Default::default()),
                })
                .collect(),
        };
        std::fs::write(&path, artifact.encode(&network)).expect("artifact writes");
        path
    };

    let sparse_path = write("sparse.bin", Height(200_000), vec![Height(0)]);
    let sparse = Config {
        historical_frontier_artifact: Some(sparse_path.clone()),
        ..Config::ephemeral()
    };
    assert!(matches!(
        super::load_historical_frontier_artifact(&network, &sparse, false),
        Err(StateInitError::HistoricalFrontierArtifactTooSparse {
            path,
            blocks: 199_999,
            limit: MAX_HISTORICAL_TREE_REPLAY_BLOCKS,
        }) if path == sparse_path
    ));

    let empty_path = write("empty.bin", Height(200_000), vec![]);
    let empty = Config {
        historical_frontier_artifact: Some(empty_path.clone()),
        ..Config::ephemeral()
    };
    assert!(matches!(
        super::load_historical_frontier_artifact(&network, &empty, false),
        Err(StateInitError::HistoricalFrontierArtifactTooSparse {
            path,
            blocks: 200_000,
            limit: MAX_HISTORICAL_TREE_REPLAY_BLOCKS,
        }) if path == empty_path
    ));

    let covering = Config {
        historical_frontier_artifact: Some(write(
            "covering.bin",
            Height(10),
            vec![Height(0), Height(10)],
        )),
        ..Config::ephemeral()
    };
    assert!(
        super::load_historical_frontier_artifact(&network, &covering, false).is_ok(),
        "a grid whose gaps fit the serving replay limit must load"
    );

    let mid_chain_path = write("mid-chain.bin", Height(2_000_100), vec![Height(2_000_000)]);
    let mid_chain = Config {
        historical_frontier_artifact: Some(mid_chain_path.clone()),
        ..Config::ephemeral()
    };
    assert!(matches!(
        super::load_historical_frontier_artifact(&network, &mid_chain, false),
        Err(StateInitError::HistoricalFrontierArtifactTooSparse {
            path,
            blocks: 2_000_000,
            limit: MAX_HISTORICAL_TREE_REPLAY_BLOCKS,
        }) if path == mid_chain_path
    ));

    let unused_sparse = Config {
        historical_frontier_artifact: Some(sparse_path),
        storage_mode: StorageMode::Pruned(PruningConfig::default()),
        ..Config::ephemeral()
    };
    assert!(
        super::load_historical_frontier_artifact(&network, &unused_sparse, false).is_ok(),
        "a sparse grid is ignored when derivation is off"
    );
}

#[test]
fn frontier_grid_coverage_is_incomparable_until_both_sides_exist() {
    assert_eq!(
        super::frontier_grid_ends_before_vct_handoff(None, None),
        None,
        "neither side is comparable"
    );
    assert_eq!(
        super::frontier_grid_ends_before_vct_handoff(Some(Height(9)), None),
        None,
        "an unmarked database cannot fail coverage"
    );
    assert_eq!(
        super::frontier_grid_ends_before_vct_handoff(None, Some(Height(10))),
        None,
        "an unloaded grid cannot fail coverage"
    );
    assert_eq!(
        super::frontier_grid_ends_before_vct_handoff(Some(Height(9)), Some(Height(10))),
        Some((Height(9), Height(10))),
        "a grid that ends below the handoff is uncovered"
    );
    assert_eq!(
        super::frontier_grid_ends_before_vct_handoff(Some(Height(10)), Some(Height(10))),
        None,
        "a grid that ends on the handoff covers the band"
    );
    assert_eq!(
        super::frontier_grid_ends_before_vct_handoff(Some(Height(11)), Some(Height(10))),
        None,
        "a newer grid may cover an older handoff"
    );
}

#[test]
fn historical_frontier_coverage_is_rechecked_once_the_vct_marker_exists() {
    use zakura_node_services::sync_lifecycle::{
        HeaderRuntimeDetachedReason, HeaderRuntimeStatus, LifecycleEpoch,
    };

    use crate::service::{
        non_finalized_state::NonFinalizedState, watch_receiver::WatchReceiver,
        HeaderChainSubscriptions, ReadStateService, VctRootRepairStatus,
    };

    let network = Network::Mainnet;
    let temp_dir = tempfile::tempdir().expect("temporary directory is created");
    let artifact_path = temp_dir.path().join("frontiers.bin");
    let artifact = FrontierArtifact {
        spacing: 1,
        last_checkpoint: Height(9),
        entries: vec![FrontierEntry {
            height: Height(9),
            sapling: Arc::new(Default::default()),
            orchard: Arc::new(Default::default()),
            ironwood: Arc::new(Default::default()),
        }],
    };
    std::fs::write(&artifact_path, artifact.encode(&network))
        .expect("historical frontier artifact is written");

    let config = Config {
        historical_frontier_artifact: Some(artifact_path),
        ..Config::ephemeral()
    };
    let finalized_state =
        FinalizedState::new(&config, &network).expect("ephemeral finalized state opens");
    let historical_trees = super::load_historical_frontier_artifact(&network, &config, false)
        .expect("the historical frontier artifact loads")
        .discard_if_before_vct_handoff(&config, &finalized_state.db);

    let (_non_finalized_sender, non_finalized_receiver) =
        tokio::sync::watch::channel(NonFinalizedState::new(&network));
    let (_repair_sender, repair_receiver) =
        tokio::sync::watch::channel(VctRootRepairStatus::default());
    let (_header_chain_snapshot_sender, header_chain_snapshot_receiver) =
        tokio::sync::watch::channel(None);
    let (_header_chain_view_sender, header_chain_view_receiver) = tokio::sync::watch::channel(None);
    let (_header_runtime_status_sender, header_runtime_status_receiver) =
        tokio::sync::watch::channel(HeaderRuntimeStatus::Detached {
            epoch: LifecycleEpoch::INITIAL,
            reason: HeaderRuntimeDetachedReason::AwaitingSemanticHandoff,
        });
    let (_header_chain_reader_sender, header_chain_reader_receiver) =
        tokio::sync::watch::channel(None);

    let read_state = ReadStateService::new(
        &finalized_state,
        None,
        Arc::new(crate::service::write::BlockWriteFailure::default()),
        WatchReceiver::new(non_finalized_receiver),
        repair_receiver,
        HeaderChainSubscriptions {
            snapshots: header_chain_snapshot_receiver,
            views: header_chain_view_receiver,
            runtime_status: header_runtime_status_receiver,
            reader: header_chain_reader_receiver,
        },
        historical_trees,
    );

    assert!(
        read_state.has_usable_historical_frontier_grid(),
        "the grid is usable before the durable marker exists"
    );

    let mut batch = DiskWriteBatch::new();
    batch.update_vct_sync_marker(&finalized_state.db, Height(10));
    finalized_state
        .db
        .write_batch(batch)
        .expect("a newer VCT handoff marker is written");
    let unavailable = HistoricalTreeUnavailable {
        hash_or_height: Height(9).into(),
        last_checkpoint: Height(10),
    };
    let error = super::historical_frontiers(&read_state, Height(9).into(), unavailable.clone())
        .expect_err("a stale grid must not serve the uncovered band");
    assert_eq!(
        error.downcast_ref::<HistoricalTreeUnavailable>(),
        Some(&unavailable),
        "a stale grid makes historical trees unavailable without surfacing an artifact error"
    );
    assert_eq!(
        read_state
            .historical_trees
            .lock()
            .expect("historical tree cache lock is available")
            .last_checkpoint(),
        None,
        "the stale artifact is discarded after the first serving-time check"
    );
    assert!(!read_state.has_usable_historical_frontier_grid());
}

#[test]
fn block_sync_body_anchor_rolls_back_to_the_selected_fork_intersection() {
    let shared = block::Hash([1; 32]);
    let body_fork = block::Hash([2; 32]);
    let selected_fork = block::Hash([3; 32]);
    let anchor = super::highest_common_body_header_frontier(
        block::Height(2),
        block::Height(0),
        |height| match height.0 {
            1 => Some(shared),
            2 => Some(body_fork),
            _ => None,
        },
        |height| {
            Ok(match height.0 {
                1 => Some(shared),
                2 => Some(selected_fork),
                _ => None,
            })
        },
    )
    .expect("the selected and full-state forks share height one");

    assert_eq!(
        anchor,
        zakura_header_chain::Frontier::new(block::Height(1), shared)
    );
}

async fn test_populated_state_responds_correctly(
    mut state: Buffer<Request, futures::future::BoxFuture<'static, Result<Response, BoxError>>>,
) -> Result<()> {
    let blocks: Vec<Arc<Block>> = zakura_test::vectors::MAINNET_BLOCKS
        .range(0..=LAST_BLOCK_HEIGHT)
        .map(|(_, block_bytes)| block_bytes.zcash_deserialize_into().unwrap())
        .collect();

    let block_hashes: Vec<block::Hash> = blocks.iter().map(|block| block.hash()).collect();
    let block_headers: Vec<CountedHeader> = blocks
        .iter()
        .map(|block| CountedHeader {
            header: block.header.clone(),
        })
        .collect();

    for (ind, block) in blocks.into_iter().enumerate() {
        let mut transcript = vec![];
        let height = block.coinbase_height().unwrap();
        let hash = block.hash();

        // these requests don't have any arguments, so we just do them once
        if ind == LAST_BLOCK_HEIGHT as usize {
            transcript.push((Request::Tip, Ok(Response::Tip(Some((height, hash))))));

            let locator_hashes = vec![
                block_hashes[LAST_BLOCK_HEIGHT as usize],
                block_hashes[(LAST_BLOCK_HEIGHT - 1) as usize],
                block_hashes[(LAST_BLOCK_HEIGHT - 2) as usize],
                block_hashes[(LAST_BLOCK_HEIGHT - 4) as usize],
                block_hashes[(LAST_BLOCK_HEIGHT - 8) as usize],
                block_hashes[0],
            ];

            transcript.push((
                Request::BlockLocator,
                Ok(Response::BlockLocator(locator_hashes)),
            ));
        }

        transcript.push((
            Request::Block(hash.into()),
            Ok(Response::Block(Some(block.clone()))),
        ));

        transcript.push((
            Request::Block(height.into()),
            Ok(Response::Block(Some(block.clone()))),
        ));

        // Spec: transactions in the genesis block are ignored.
        if height.0 != 0 {
            for transaction in &block.transactions {
                let transaction_hash = transaction.hash();

                let from_coinbase = transaction.is_coinbase();
                for (index, output) in transaction.outputs().iter().cloned().enumerate() {
                    let outpoint = transparent::OutPoint::from_usize(transaction_hash, index);

                    let utxo = transparent::Utxo {
                        output,
                        height,
                        from_coinbase,
                    };

                    transcript.push((Request::AwaitUtxo(outpoint), Ok(Response::Utxo(utxo))));
                }
            }
        }

        let mut append_locator_transcript = |split_ind| {
            let block_hashes = block_hashes.clone();
            let (known_hashes, next_hashes) = block_hashes.split_at(split_ind);

            let block_headers = block_headers.clone();
            let (_, next_headers) = block_headers.split_at(split_ind);

            // no stop
            transcript.push((
                Request::FindBlockHashes {
                    known_blocks: known_hashes.iter().rev().cloned().collect(),
                    stop: None,
                },
                Ok(Response::BlockHashes(next_hashes.to_vec())),
            ));

            transcript.push((
                Request::FindBlockHeaders {
                    known_blocks: known_hashes.iter().rev().cloned().collect(),
                    stop: None,
                },
                Ok(Response::BlockHeaders(next_headers.to_vec())),
            ));

            // stop at the next block
            transcript.push((
                Request::FindBlockHashes {
                    known_blocks: known_hashes.iter().rev().cloned().collect(),
                    stop: next_hashes.first().cloned(),
                },
                Ok(Response::BlockHashes(
                    next_hashes.first().iter().cloned().cloned().collect(),
                )),
            ));

            transcript.push((
                Request::FindBlockHeaders {
                    known_blocks: known_hashes.iter().rev().cloned().collect(),
                    stop: next_hashes.first().cloned(),
                },
                Ok(Response::BlockHeaders(
                    next_headers.first().iter().cloned().cloned().collect(),
                )),
            ));

            // stop at a block that isn't actually in the chain
            // tests bug #2789
            transcript.push((
                Request::FindBlockHashes {
                    known_blocks: known_hashes.iter().rev().cloned().collect(),
                    stop: Some(block::Hash([0xff; 32])),
                },
                Ok(Response::BlockHashes(next_hashes.to_vec())),
            ));

            transcript.push((
                Request::FindBlockHeaders {
                    known_blocks: known_hashes.iter().rev().cloned().collect(),
                    stop: Some(block::Hash([0xff; 32])),
                },
                Ok(Response::BlockHeaders(next_headers.to_vec())),
            ));
        };

        // split before the current block, and locate the current block
        append_locator_transcript(ind);

        // split after the current block, and locate the next block
        append_locator_transcript(ind + 1);

        let transcript = Transcript::from(transcript);
        transcript.check(&mut state).await?;
    }

    Ok(())
}

#[tokio::main]
async fn populate_and_check(blocks: Vec<Arc<Block>>) -> Result<()> {
    let (state, _, _, _) = populated_state(blocks, &Network::Mainnet).await;
    test_populated_state_responds_correctly(state).await?;
    Ok(())
}

fn out_of_order_committing_strategy() -> BoxedStrategy<Vec<Arc<Block>>> {
    let blocks = zakura_test::vectors::MAINNET_BLOCKS
        .range(0..=LAST_BLOCK_HEIGHT)
        .map(|(_, block_bytes)| block_bytes.zcash_deserialize_into::<Arc<Block>>().unwrap())
        .collect::<Vec<_>>();

    Just(blocks).prop_shuffle().boxed()
}

#[tokio::test(flavor = "multi_thread")]
async fn empty_state_still_responds_to_requests() -> Result<()> {
    let _init_guard = zakura_test::init();

    let block =
        zakura_test::vectors::BLOCK_MAINNET_419200_BYTES.zcash_deserialize_into::<Arc<Block>>()?;

    let iter = vec![
        // No checks for SemanticallyVerifiedBlock or CommitCheckpointVerifiedBlock because empty state
        // precondition doesn't matter to them
        (Request::Tip, Ok(Response::Tip(None))),
        (Request::BlockLocator, Ok(Response::BlockLocator(vec![]))),
        (
            Request::Block(block.hash().into()),
            Ok(Response::Block(None)),
        ),
        (
            Request::Block(block.coinbase_height().unwrap().into()),
            Ok(Response::Block(None)),
        ),
        // No check for AwaitUTXO because it will wait if the UTXO isn't present
        (
            Request::FindBlockHashes {
                known_blocks: vec![block.hash()],
                stop: None,
            },
            Ok(Response::BlockHashes(Vec::new())),
        ),
        (
            Request::FindBlockHeaders {
                known_blocks: vec![block.hash()],
                stop: None,
            },
            Ok(Response::BlockHeaders(Vec::new())),
        ),
    ]
    .into_iter();
    let transcript = Transcript::from(iter);

    let network = Network::Mainnet;
    let state = init_test(&network).await;

    transcript.check(state).await?;

    Ok(())
}

/// Regression test for the checkpoint-to-non-finalized sync handoff stall.
///
/// The block write task switches from committing checkpoint verified blocks (finalized state) to
/// committing semantically verified blocks (non-finalized state) once the final checkpoint block is
/// durably written to disk. That handoff used to also require a semantically verified child to be
/// queued, so the pipeline could stall at the checkpoint boundary until the first fully-verified
/// block arrived (or the syncer restarted).
///
/// This test sets the maximum checkpoint height to the last finalized block, commits the checkpoint
/// blocks, and asserts that `poll_ready()` performs the handoff with an **empty** non-finalized
/// queue — i.e. it no longer waits for a semantically verified block to arrive.
///
/// It deliberately does not commit a semantically verified block afterwards: the first two Mainnet
/// blocks predate the Canopy checkpoint, and the non-finalized write path treats their transaction
/// versions as `unreachable!()` (pre-Canopy blocks are only ever checkpoint verified). The handoff
/// itself is what this test covers.
#[tokio::test(flavor = "multi_thread")]
async fn poll_ready_hands_off_at_max_checkpoint_height() -> Result<()> {
    use std::task::{Context, Waker};

    use tower::Service;

    let _init_guard = zakura_test::init();
    let network = Network::Mainnet;

    // Blocks 0 and 1 are committed as checkpoint verified (finalized) blocks.
    let blocks: Vec<Arc<Block>> = zakura_test::vectors::MAINNET_BLOCKS
        .range(0..=1)
        .map(|(_, block_bytes)| block_bytes.zcash_deserialize_into::<Arc<Block>>().unwrap())
        .collect();

    // Set the maximum checkpoint height to block 1, so the checkpoint phase ends once block 1 is
    // committed to the finalized state.
    let max_checkpoint_height = blocks[1].coinbase_height().unwrap();
    let mut config = Config::ephemeral();
    config.enable_zakura_header_seed_from_committed_blocks = true;
    // The state-only fixture commits bodies directly.
    // The fixture has no network-supplied auxiliary root.
    config.vct_fast_sync = false;
    let (mut state_service, read, _tip, _tip_change) =
        StateService::new(config, &network, max_checkpoint_height, 0)
            .await
            .expect("test state initialization succeeds");

    // Commit blocks 0 and 1 to the finalized state and wait for each write to land on disk, so the
    // finalized tip catches up to the maximum checkpoint height and the last block hash we sent.
    for (index, block) in blocks[0..=1].iter().enumerate() {
        let checkpoint = CheckpointVerifiedBlock::from(block.clone());
        let result = state_service
            .queue_and_commit_to_finalized_state(checkpoint)
            .await;
        assert!(
            matches!(result, Ok(Ok(_))),
            "checkpoint verified block should commit: {result:?}",
        );

        let expected_height = block.coinbase_height().expect("test block has a height");
        let expected_hash = block.hash();
        timeout(Duration::from_secs(5), async {
            loop {
                let snapshot = read.subscribe_header_chain_snapshots().borrow().clone();
                if snapshot.as_ref().is_some_and(|snapshot| {
                    snapshot.frontiers.finalized.height == expected_height
                        && snapshot.frontiers.finalized.hash == expected_hash
                        && snapshot.frontiers.verified_best == snapshot.frontiers.finalized
                }) {
                    break;
                }
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("checkpoint commits atomically advance the durable header runtime");
        if index == 0 {
            assert!(
                state_service.block_write_sender.finalized.is_some(),
                "the header runtime must become ready after genesis while checkpoint writes remain open",
            );
            // The writer publishes the attached runtime's snapshot before marking it ready.
            let mut status = read.subscribe_header_runtime_status();
            timeout(
                Duration::from_secs(5),
                status.wait_for(|status| status.is_ready()),
            )
            .await
            .expect("the header runtime becomes ready after genesis")
            .expect("the header runtime status sender stays open");
        }
    }

    let last_finalized_hash = blocks[1].hash();
    assert_eq!(
        state_service.read_service.db.finalized_tip_height(),
        Some(max_checkpoint_height),
        "finalized tip should have reached the maximum checkpoint height",
    );
    assert_eq!(
        state_service.read_service.db.finalized_tip_hash(),
        last_finalized_hash,
        "finalized tip on disk should have caught up to block 1",
    );

    // Preconditions: still in finalized-write mode, and crucially **no** semantically verified block
    // is queued. The old behavior would not hand off in this state.
    assert!(
        state_service.block_write_sender.finalized.is_some(),
        "write task should still be committing finalized blocks before the handoff",
    );
    assert!(
        !state_service
            .non_finalized_state_queued_blocks
            .has_queued_children(last_finalized_hash),
        "no semantically verified block should be queued before the handoff",
    );

    // Trigger the handoff. Nothing is queued, so this exercises the height-based path.
    let mut cx = Context::from_waker(Waker::noop());
    let _ = state_service.poll_ready(&mut cx);

    // The handoff should have happened purely because the final checkpoint write is durable, with no
    // semantically verified block queued. Before this fix, the finalized sender would still be open.
    assert!(
        state_service.block_write_sender.finalized.is_none(),
        "poll_ready should have handed off to non-finalized writes at the max checkpoint height",
    );

    timeout(Duration::from_secs(5), async {
        loop {
            let store = crate::service::finalized_state::header_chain::HeaderChainStore::new(
                state_service.read_service.db.header_chain_disk_db(),
            );
            if store
                .is_initialized()
                .expect("the header-chain format marker is readable")
            {
                break;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    })
    .await
    .expect("the production writer attaches the header runtime at handoff");

    Ok(())
}

/// The handoff request must release children queued before the final checkpoint became durable.
#[tokio::test(flavor = "multi_thread")]
async fn checkpoint_handoff_request_drains_waiting_children() -> Result<()> {
    use eyre::eyre;
    use tower::Service;

    let _init_guard = zakura_test::init();
    let blocks: Vec<Arc<Block>> = zakura_test::vectors::MAINNET_BLOCKS
        .range(0..=2)
        .map(|(_, bytes)| bytes.zcash_deserialize_into::<Arc<Block>>().unwrap())
        .collect();
    let (mut state, _read, _tip, _tip_change) =
        StateService::new(Config::ephemeral(), &Network::Mainnet, Height(1), 0)
            .await
            .expect("ephemeral state opens");
    state
        .queue_and_commit_to_finalized_state(blocks[0].clone().into())
        .await??;
    state
        .call(Request::CheckCheckpointHandoff)
        .await
        .map_err(|error| eyre!(error))?;
    assert!(
        state.block_write_sender.finalized.is_some(),
        "an intermediate checkpoint cannot hand off"
    );

    // Exercise the internal writer queue with the small historical vectors. Production consensus
    // submits semantic blocks after Canopy; this test only checks their release to the writer.
    let _child = state.queue_and_commit_to_non_finalized_state(blocks[2].clone().prepare(), None);
    state
        .queue_and_commit_to_finalized_state(blocks[1].clone().into())
        .await??;
    assert!(state
        .non_finalized_state_queued_blocks
        .has_queued_children(blocks[1].hash()));
    assert!(state.block_write_sender.finalized.is_some());

    // Deliberately call without poll_ready: Tip only reads, and the handoff request owns the check.
    state
        .call(Request::Tip)
        .await
        .map_err(|error| eyre!(error))?;
    assert!(state.block_write_sender.finalized.is_some());
    assert_eq!(
        state
            .call(Request::CheckCheckpointHandoff)
            .await
            .map_err(|error| eyre!(error))?,
        Response::CheckpointHandoffChecked
    );
    assert!(state.block_write_sender.finalized.is_none());
    assert!(!state
        .non_finalized_state_queued_blocks
        .has_queued_children(blocks[1].hash()));
    state
        .call(Request::CheckCheckpointHandoff)
        .await
        .map_err(|error| eyre!(error))?;
    assert!(
        state.block_write_sender.finalized.is_none(),
        "reconciliation is idempotent"
    );
    Ok(())
}

/// Legacy-only nodes must preserve the ordinary state handoff without creating or
/// reconstructing the native header runtime.
#[tokio::test(flavor = "multi_thread")]
async fn legacy_mode_handoff_keeps_header_runtime_detached() -> Result<()> {
    use std::task::{Context, Waker};

    use tower::Service;
    use zakura_node_services::sync_lifecycle::HeaderRuntimeStatus;

    let _init_guard = zakura_test::init();
    let network = Network::Mainnet;
    let genesis = zakura_test::vectors::MAINNET_BLOCKS
        .get(&0)
        .expect("the mainnet genesis vector is available")
        .zcash_deserialize_into::<Arc<Block>>()?;

    let (mut state, read, _tip, _tip_change) =
        StateService::new(Config::ephemeral(), &network, Height(0), 0)
            .await
            .expect("ephemeral state initialization succeeds");
    let result = state
        .queue_and_commit_to_finalized_state(CheckpointVerifiedBlock::from(genesis))
        .await;
    assert!(
        matches!(result, Ok(Ok(_))),
        "genesis should commit: {result:?}"
    );

    let mut cx = Context::from_waker(Waker::noop());
    let _ = state.poll_ready(&mut cx);
    assert!(
        state.block_write_sender.finalized.is_none(),
        "legacy state still hands off to semantic writes"
    );

    tokio::time::sleep(Duration::from_millis(50)).await;
    assert!(matches!(
        &*read.subscribe_header_runtime_status().borrow(),
        HeaderRuntimeStatus::Detached { .. }
    ));
    assert!(read.subscribe_header_chain_snapshots().borrow().is_none());
    assert!(
        !crate::service::finalized_state::header_chain::HeaderChainStore::new(
            state.read_service.db.header_chain_disk_db(),
        )
        .is_initialized()?,
        "legacy handoff must not create durable header-runtime state"
    );

    Ok(())
}

/// Micro-benchmark for the cost added to `poll_ready()` by the handoff trigger.
///
/// `poll_ready()` runs on essentially every state service readiness poll, so the added
/// `try_handoff_to_non_finalized_write()` call must be cheap. This measures three regimes:
///
/// - Raw `finalized_tip_hash()` DB read — a RocksDB seek-to-last. During checkpoint sync the
///   last-sent hash usually runs ahead of the on-disk tip, so the helper short-circuits after this
///   single read; it is the dominant per-poll cost in that phase.
/// - Full guard (still in finalized mode, on-disk tip matches the last-sent hash but below the max
///   checkpoint height and with no queued child): the helper runs the whole condition — two tip
///   reads plus a `HashMap::contains_key` — without transitioning. This is the most expensive
///   non-transitioning path, hit only at the checkpoint boundary.
/// - Steady state (after the handoff, `block_write_sender.finalized == None`): the call
///   short-circuits on a single `Option::is_some()` check and never touches the database. This is
///   what runs for the entire post-sync life of the node.
///
/// Run with:
/// `cargo test -p zakura-state --release -- --ignored --nocapture handoff_trigger_microbench`
#[ignore]
#[allow(clippy::print_stdout)]
#[tokio::test(flavor = "multi_thread")]
async fn handoff_trigger_microbench() -> Result<()> {
    use std::time::Instant;

    let _init_guard = zakura_test::init();
    let network = Network::Mainnet;

    let blocks: Vec<Arc<Block>> = zakura_test::vectors::MAINNET_BLOCKS
        .range(0..=1)
        .map(|(_, block_bytes)| block_bytes.zcash_deserialize_into::<Arc<Block>>().unwrap())
        .collect();

    // Use `Height::MAX` so the height condition is never met: the helper runs its full guard but
    // never transitions, which is exactly the non-transitioning path we want to measure.
    let (mut state_service, _read, _tip, _tip_change) =
        StateService::new(Config::ephemeral(), &network, Height::MAX, 0)
            .await
            .expect("ephemeral state initialization succeeds");

    for block in &blocks[0..=1] {
        let checkpoint = CheckpointVerifiedBlock::from(block.clone());
        state_service
            .queue_and_commit_to_finalized_state(checkpoint)
            .await
            .expect("commit channel open")
            .expect("checkpoint block commits");
    }

    const ITERS: u32 = 1_000_000;

    // Regime 1: raw `finalized_tip_hash()` DB read.
    let start = Instant::now();
    for _ in 0..ITERS {
        std::hint::black_box(state_service.read_service.db.finalized_tip_hash());
    }
    let tip_ns = start.elapsed().as_nanos() as f64 / f64::from(ITERS);

    // Regime 2: full guard cost. The on-disk tip equals the last-sent hash, the height condition is
    // false (`Height::MAX`), and no child is queued, so the helper evaluates every condition but
    // does not transition.
    let start = Instant::now();
    for _ in 0..ITERS {
        std::hint::black_box(state_service.try_handoff_to_non_finalized_write());
    }
    let guard_ns = start.elapsed().as_nanos() as f64 / f64::from(ITERS);
    assert!(
        state_service.block_write_sender.finalized.is_some(),
        "the benchmark must not transition: finalized sender should still be open",
    );

    // Regime 3: steady-state cost (post-handoff). Drop the finalized sender so the helper
    // short-circuits immediately, exactly as it does for the rest of the node's life.
    state_service.block_write_sender.finalized = None;
    let start = Instant::now();
    for _ in 0..ITERS {
        std::hint::black_box(state_service.try_handoff_to_non_finalized_write());
    }
    let steady_ns = start.elapsed().as_nanos() as f64 / f64::from(ITERS);

    println!("handoff trigger micro-benchmark ({ITERS} iters each):");
    println!("  finalized_tip_hash() DB read : {tip_ns:>8.2} ns/call");
    println!("  helper, full guard           : {guard_ns:>8.2} ns/call");
    println!("  helper, steady state         : {steady_ns:>8.2} ns/call");

    Ok(())
}

#[test]
fn state_behaves_when_blocks_are_committed_in_order() -> Result<()> {
    let _init_guard = zakura_test::init();

    let blocks = zakura_test::vectors::MAINNET_BLOCKS
        .range(0..=LAST_BLOCK_HEIGHT)
        .map(|(_, block_bytes)| block_bytes.zcash_deserialize_into::<Arc<Block>>().unwrap())
        .collect();

    populate_and_check(blocks)?;

    Ok(())
}

const DEFAULT_PARTIAL_CHAIN_PROPTEST_CASES: u32 = 2;

/// The legacy chain limit for tests.
const TEST_LEGACY_CHAIN_LIMIT: usize = 100;

/// Check more blocks than the legacy chain limit.
const OVER_LEGACY_CHAIN_LIMIT: u32 = TEST_LEGACY_CHAIN_LIMIT as u32 + 10;

/// Check fewer blocks than the legacy chain limit.
const UNDER_LEGACY_CHAIN_LIMIT: u32 = TEST_LEGACY_CHAIN_LIMIT as u32 - 10;

proptest! {
    #![proptest_config(
        proptest::test_runner::Config::with_cases(env::var("PROPTEST_CASES")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(DEFAULT_PARTIAL_CHAIN_PROPTEST_CASES))
    )]

    /// Test out of order commits of continuous block test vectors from genesis onward.
    #[test]
    fn state_behaves_when_blocks_are_committed_out_of_order(blocks in out_of_order_committing_strategy()) {
        let _init_guard = zakura_test::init();

        populate_and_check(blocks).unwrap();
    }

    /// Test blocks that are less than the NU5 activation height.
    #[test]
    fn some_block_less_than_network_upgrade(
        (network, nu_activation_height, chain) in partial_nu5_chain_strategy(4, true, UNDER_LEGACY_CHAIN_LIMIT, NetworkUpgrade::Canopy)
    ) {
        let response = crate::service::check::legacy_chain(nu_activation_height, chain.into_iter().rev(), &network, TEST_LEGACY_CHAIN_LIMIT)
            .map_err(|error| error.to_string());

        prop_assert_eq!(response, Ok(()));
    }

    /// Test the maximum amount of blocks to check before chain is declared to be legacy.
    #[test]
    fn no_transaction_with_network_upgrade(
        (network, nu_activation_height, chain) in partial_nu5_chain_strategy(4, true, OVER_LEGACY_CHAIN_LIMIT, NetworkUpgrade::Canopy)
    ) {
        let tip_height = chain
            .last()
            .expect("chain contains at least one block")
            .coinbase_height()
            .expect("chain contains valid blocks");

        let response = crate::service::check::legacy_chain(nu_activation_height, chain.into_iter().rev(), &network, TEST_LEGACY_CHAIN_LIMIT)
            .map_err(|error| error.to_string());

        prop_assert_eq!(
            response,
            Err(format!(
                "could not find any transactions in recent blocks: checked {TEST_LEGACY_CHAIN_LIMIT} blocks back from {tip_height:?}",
            ))
        );
    }

    /// Test the `Block.check_transaction_network_upgrade()` error inside the legacy check.
    #[test]
    fn at_least_one_transaction_with_inconsistent_network_upgrade(
        (network, nu_activation_height, chain) in partial_nu5_chain_strategy(5, false, OVER_LEGACY_CHAIN_LIMIT, NetworkUpgrade::Nu5)
    ) {
        // this test requires that an invalid block is encountered
        // before a valid block (and before the check gives up),
        // but setting `transaction_has_valid_network_upgrade` to false
        // sometimes generates blocks with all valid (or missing) network upgrades

        // we must check at least one block, and the first checked block must be invalid
        let first_checked_block = chain
            .iter()
            .rev()
            .take_while(|block| block.coinbase_height().unwrap() >= nu_activation_height)
            .take(100)
            .next();
        prop_assume!(first_checked_block.is_some());
        prop_assume!(
            first_checked_block
                .unwrap()
                .check_transaction_network_upgrade_consistency(&network)
                .is_err()
        );

        let response = crate::service::check::legacy_chain(
            nu_activation_height,
            chain.clone().into_iter().rev(),
            &network,
            TEST_LEGACY_CHAIN_LIMIT,
        ).map_err(|error| error.to_string());

        prop_assert_eq!(
            response,
            Err("inconsistent network upgrade found in transaction: WrongTransactionConsensusBranchId".into()),
            "first: {:?}, last: {:?}",
            chain.first().map(|block| block.coinbase_height()),
            chain.last().map(|block| block.coinbase_height()),
        );
    }

    /// Test there is at least one transaction with a valid `network_upgrade` in the legacy check.
    #[test]
    fn at_least_one_transaction_with_valid_network_upgrade(
        (network, nu_activation_height, chain) in partial_nu5_chain_strategy(5, true, UNDER_LEGACY_CHAIN_LIMIT, NetworkUpgrade::Nu5)
    ) {
        let response = crate::service::check::legacy_chain(nu_activation_height, chain.into_iter().rev(), &network, TEST_LEGACY_CHAIN_LIMIT)
            .map_err(|error| error.to_string());

        prop_assert_eq!(response, Ok(()));
    }

    /// Test that the value pool is updated accordingly.
    ///
    /// 1. Generate a finalized chain and some non-finalized blocks.
    /// 2. Check that initially the value pool is empty.
    /// 3. Commit the finalized blocks and check that the value pool is updated accordingly.
    /// 4. Commit the non-finalized blocks and check that the value pool is also updated
    ///    accordingly.
    #[test]
    fn value_pool_is_updated(
        (network, finalized_blocks, non_finalized_blocks)
            in continuous_empty_blocks_from_test_vectors(),
    ) {
        let _init_guard = zakura_test::init();
        let (mut state_service, _, _, _) = Runtime::new().unwrap().block_on(async {
            // We're waiting to verify each block here, so we don't need the maximum checkpoint height.
            StateService::new(Config::ephemeral(), &network, Height::MAX, 0).await
        }).expect("ephemeral state initialization succeeds");

        prop_assert_eq!(state_service.read_service.db.finalized_value_pool(), ValueBalance::zero());
        prop_assert_eq!(
            state_service.read_service.latest_non_finalized_state().best_chain().map(|chain| chain.chain_value_pools).unwrap_or_else(ValueBalance::zero),
            ValueBalance::zero()
        );

        // the slow start rate for the first few blocks, as in the spec
        const SLOW_START_RATE: i64 = 62500;
        // the expected transparent pool value, calculated using the slow start rate
        let mut expected_transparent_pool = ValueBalance::zero();

        let mut expected_finalized_value_pool = Ok(ValueBalance::zero());
        for block in finalized_blocks {
            // the genesis block has a zero-valued transparent output,
            // which is not included in the UTXO set
            if block.height > block::Height(0) {
                let utxos = &block.new_outputs.iter().map(|(k, ordered_utxo)| (*k, ordered_utxo.utxo.clone())).collect();
                let block_value_pool = &block.block.chain_value_pool_change(&network, utxos, None)?;
                expected_finalized_value_pool += *block_value_pool;
            }

            let result_receiver = state_service.queue_and_commit_to_finalized_state(block.clone());
            let result = result_receiver.blocking_recv();

            prop_assert!(result.is_ok(), "unexpected failed finalized block commit: {:?}", result);

            prop_assert_eq!(
                state_service.read_service.db.finalized_value_pool(),
                expected_finalized_value_pool.clone()?.constrain()?
            );

            let transparent_value = SLOW_START_RATE * i64::from(block.height.0);
            let transparent_value = transparent_value.try_into().unwrap();
            let transparent_value = ValueBalance::from_transparent_amount(transparent_value);
            expected_transparent_pool = (expected_transparent_pool + transparent_value).unwrap();
            prop_assert_eq!(
                state_service.read_service.db.finalized_value_pool(),
                expected_transparent_pool
            );
        }

        let mut expected_non_finalized_value_pool = Ok(expected_finalized_value_pool?);
        for block in non_finalized_blocks {
            let utxos = block.new_outputs.clone();
            let block_value_pool = &block.block.chain_value_pool_change(&network, &transparent::utxos_from_ordered_utxos(utxos), None)?;
            expected_non_finalized_value_pool += *block_value_pool;

            let result_receiver =
                state_service.queue_and_commit_to_non_finalized_state(block.clone(), None);
            let result = result_receiver.blocking_recv();

            prop_assert!(result.is_ok(), "unexpected failed non-finalized block commit: {:?}", result);

            prop_assert_eq!(
                state_service.read_service.latest_non_finalized_state().best_chain().unwrap().chain_value_pools,
                expected_non_finalized_value_pool.clone()?.constrain()?
            );

            let transparent_value = SLOW_START_RATE * i64::from(block.height.0);
            let transparent_value = transparent_value.try_into().unwrap();
            let transparent_value = ValueBalance::from_transparent_amount(transparent_value);
            expected_transparent_pool = (expected_transparent_pool + transparent_value).unwrap();
            prop_assert_eq!(
                state_service.read_service.latest_non_finalized_state().best_chain().unwrap().chain_value_pools,
                expected_transparent_pool
            );
        }
    }
}

// This test sleeps for every block, so we only ever want to run it once
proptest! {
    #![proptest_config(
        proptest::test_runner::Config::with_cases(1)
    )]

    /// Test that the best tip height is updated accordingly.
    ///
    /// 1. Generate a finalized chain and some non-finalized blocks.
    /// 2. Check that initially the best tip height is empty.
    /// 3. Commit the finalized blocks and check that the best tip height is updated accordingly.
    /// 4. Commit the non-finalized blocks and check that the best tip height is also updated
    ///    accordingly.
    #[test]
    fn chain_tip_sender_is_updated(
        (network, finalized_blocks, non_finalized_blocks)
            in continuous_empty_blocks_from_test_vectors(),
    ) {
        let _init_guard = zakura_test::init();

        let runtime = Runtime::new().unwrap();
        let (mut state_service, _read_only_state_service, latest_chain_tip, mut chain_tip_change) = runtime.block_on(async {
            // We're waiting to verify each block here, so we don't need the maximum checkpoint height.
            StateService::new(Config::ephemeral(), &network, Height::MAX, 0).await
        }).expect("ephemeral state initialization succeeds");

        prop_assert_eq!(latest_chain_tip.best_tip_height(), None);
        prop_assert_eq!(chain_tip_change.last_tip_change(), None);

        for block in finalized_blocks {
            let expected_block = block.clone();

            let expected_action = if expected_block.height == block::Height(0) {
                // Height 0 is reset by initialization. The BeforeOverwinter upgrade
                // (activation height 1) also resets at height 0 rather than at height 1,
                // because `ChainTipChange` resets one block *before* an activation height
                // (it checks `height.next()`, matching the height the mempool verifies
                // against). See `ChainTipChange::action`.
                TipAction::reset_with(expected_block.clone().into())
            } else {
                TipAction::grow_with(expected_block.clone().into())
            };

            let result_receiver = state_service.queue_and_commit_to_finalized_state(block);
            let result = result_receiver.blocking_recv();

            prop_assert!(result.is_ok(), "unexpected failed finalized block commit: {:?}", result);

            let actual_action = runtime
                .block_on(async {
                    timeout(
                        CHAIN_TIP_UPDATE_WAIT_LIMIT,
                        chain_tip_change.wait_for_tip_change(),
                    )
                    .await
                })
                .expect("tip change arrives because the committed block updates the channel")
                .expect("tip sender remains open while the state service is alive");

            prop_assert_eq!(latest_chain_tip.best_tip_height(), Some(expected_block.height));
            prop_assert_eq!(actual_action, expected_action);
        }

        for block in non_finalized_blocks {
            let expected_block = block.clone();

            // The genesis block (height 0) is always finalized, and the BeforeOverwinter
            // reset fires at height 0 (one block before its activation height of 1), so
            // every non-finalized block (height >= 1) grows the chain.
            let expected_action = TipAction::grow_with(expected_block.clone().into());

            let result_receiver =
                state_service.queue_and_commit_to_non_finalized_state(block, None);
            let result = result_receiver.blocking_recv();

            prop_assert!(result.is_ok(), "unexpected failed non-finalized block commit: {:?}", result);

            let actual_action = runtime
                .block_on(async {
                    timeout(
                        CHAIN_TIP_UPDATE_WAIT_LIMIT,
                        chain_tip_change.wait_for_tip_change(),
                    )
                    .await
                })
                .expect("tip change arrives because the committed block updates the channel")
                .expect("tip sender remains open while the state service is alive");

            prop_assert_eq!(latest_chain_tip.best_tip_height(), Some(expected_block.height));
            prop_assert_eq!(actual_action, expected_action);
        }
    }
}

/// Test strategy to generate a chain split in two from the test vectors.
///
/// Selects either the mainnet or testnet chain test vector and randomly splits the chain in two
/// lists of blocks. The first containing the blocks to be finalized (which always includes at
/// least the genesis block) and the blocks to be stored in the non-finalized state.
fn continuous_empty_blocks_from_test_vectors() -> impl Strategy<
    Value = (
        Network,
        SummaryDebug<Vec<CheckpointVerifiedBlock>>,
        SummaryDebug<Vec<SemanticallyVerifiedBlock>>,
    ),
> {
    any::<Network>()
        .prop_flat_map(|network| {
            // Select the test vector based on the network
            let raw_blocks = network.blockchain_map();

            // Transform the test vector's block bytes into a vector of `SemanticallyVerifiedBlock`s.
            let blocks: Vec<_> = raw_blocks
                .iter()
                .map(|(_height, &block_bytes)| {
                    let mut block_reader: &[u8] = block_bytes;
                    let mut block = Block::zcash_deserialize(&mut block_reader)
                        .expect("Failed to deserialize block from test vector");

                    let coinbase = transaction_v4_from_coinbase(&block.transactions[0]);
                    block.transactions = vec![Arc::new(coinbase)];

                    Arc::new(block).prepare()
                })
                .collect();

            // Always finalize the genesis block
            let finalized_blocks_count = 1..=blocks.len();

            (Just(network), Just(blocks), finalized_blocks_count)
        })
        .prop_map(|(network, mut blocks, finalized_blocks_count)| {
            let non_finalized_blocks = blocks.split_off(finalized_blocks_count);
            let finalized_blocks: Vec<_> =
                blocks.into_iter().map(CheckpointVerifiedBlock).collect();

            (
                network,
                finalized_blocks.into(),
                non_finalized_blocks.into(),
            )
        })
}

/// Opening a read-only state against an existing but empty cache directory (no database on
/// disk) must fail with [`StateInitError::ReadOnlyDatabaseNotFound`] rather than silently
/// creating a new, empty database.
#[test]
fn read_only_open_with_no_database_returns_error() {
    let network = Network::Mainnet;

    // An existing, readable, but empty cache directory: it contains no database.
    let cache_dir =
        tempfile::tempdir().expect("creating a temporary cache directory should succeed");
    let config = Config {
        cache_dir: cache_dir.path().to_path_buf(),
        ephemeral: false,
        ..Config::default()
    };

    match super::init_read_only(config, &network) {
        Err(crate::StateInitError::ReadOnlyDatabaseNotFound { .. }) => {}
        Err(other) => panic!("expected ReadOnlyDatabaseNotFound, got: {other:?}"),
        Ok(_) => panic!("expected an error when opening a read-only state with no database"),
    }
}

#[test]
fn read_only_secondary_workspace_is_deleted_on_drop() {
    let network = Network::Mainnet;
    let cache_dir =
        tempfile::tempdir().expect("creating a temporary cache directory should succeed");
    let config = Config {
        cache_dir: cache_dir.path().to_path_buf(),
        ephemeral: false,
        ..Config::default()
    };

    let mut finalized_state = super::finalized_state::FinalizedState::new(&config, &network)
        .expect("writable state creates the database");
    finalized_state.db.shutdown(true);
    drop(finalized_state);

    let (read_service, db, non_finalized_sender) =
        super::init_read_only(config, &network).expect("read-only state opens");
    let secondary_path = db
        .secondary_path()
        .expect("read-only state has a secondary workspace")
        .to_path_buf();
    assert!(secondary_path.is_dir());

    drop(read_service);
    drop(non_finalized_sender);
    drop(db);

    assert!(
        !secondary_path.exists(),
        "the secondary workspace is owned by the read-only database"
    );
}

/// Opening a read-only state against a missing or unreadable cache directory must fail with a
/// typed [`StateInitError::ReadOnlyCacheDirUnreadable`] rather than panicking while reading the
/// on-disk format version.
#[test]
fn read_only_open_with_unreadable_cache_dir_returns_error() {
    let network = Network::Mainnet;

    // A cache directory that does not exist. `read_dir` fails for a missing directory the same way
    // it does for an unreadable one, without depending on filesystem permissions (which `root`
    // ignores, so a chmod-based unreadable directory would not be a reliable test under CI).
    let parent = tempfile::tempdir().expect("creating a temporary directory should succeed");
    let config = Config {
        cache_dir: parent.path().join("missing"),
        ephemeral: false,
        ..Config::default()
    };

    match super::init_read_only(config, &network) {
        Err(crate::StateInitError::ReadOnlyCacheDirUnreadable { .. }) => {}
        Err(other) => panic!("expected ReadOnlyCacheDirUnreadable, got: {other:?}"),
        Ok(_) => {
            panic!("expected an error when opening a read-only state with an unreadable cache dir")
        }
    }
}

/// Opening a read-only state with an ephemeral database configured must fail with
/// [`StateInitError::ReadOnlyEphemeralConflict`]: a read-only secondary follows another
/// process's primary database and must never delete it, so it cannot also be ephemeral
/// (which would delete the primary's files on drop).
#[test]
fn read_only_open_with_ephemeral_config_returns_error() {
    let network = Network::Mainnet;

    let config = Config {
        ephemeral: true,
        ..Config::default()
    };

    match super::init_read_only(config, &network) {
        Err(crate::StateInitError::ReadOnlyEphemeralConflict) => {}
        Err(other) => panic!("expected ReadOnlyEphemeralConflict, got: {other:?}"),
        Ok(_) => {
            panic!("expected an error when opening a read-only state with an ephemeral config")
        }
    }
}

#[test]
fn read_only_open_with_malformed_version_returns_typed_error() {
    let network = Network::Mainnet;
    let cache_dir = tempfile::tempdir().expect("creating a temporary cache directory succeeds");
    let config = Config {
        cache_dir: cache_dir.path().to_path_buf(),
        ephemeral: false,
        ..Config::default()
    };
    let version_path = config.version_file_path(
        crate::constants::STATE_DATABASE_KIND,
        crate::state_database_format_version_in_code().major,
        &network,
    );
    std::fs::create_dir_all(
        version_path
            .parent()
            .expect("the version path has a cache-directory parent"),
    )
    .expect("the version-file parent is created");
    std::fs::write(&version_path, "not-a-semantic-version")
        .expect("the malformed version fixture is written");

    match super::init_read_only(config, &network) {
        Err(crate::StateInitError::DatabaseFormatVersion { path, .. }) => {
            assert_eq!(path, version_path);
        }
        Err(other) => panic!("expected DatabaseFormatVersion, got: {other:?}"),
        Ok(_) => panic!("expected malformed state version to fail closed"),
    }
}

/// While checkpoint writes are in flight the queue bound is hard, even for a block that names
/// the durable finalized tip.
///
/// The durable tip lags the last hash sent to the write task for as long as checkpoint writes are
/// outstanding. Nothing drains a block queued under the lagging tip: it does not complete the
/// handoff condition, and the eventual handoff walks forward from the last hash sent, not from
/// the tip that was durable when the block arrived. Admitting it past the bound would let a
/// caller grow the queue without limit for the length of the checkpoint sync.
#[tokio::test(flavor = "multi_thread")]
async fn checkpoint_write_lag_does_not_open_the_orphan_queue_bound() {
    let network = Network::Mainnet;
    let (mut state, _read, _tip, _height) =
        StateService::new(Config::ephemeral(), &network, Height::MAX, 0)
            .await
            .expect("an ephemeral state service is created");

    assert!(
        state.block_write_sender.finalized.is_some(),
        "the test starts while the write task still commits checkpoint blocks",
    );

    let durable_tip = state.read_service.db.finalized_tip_hash();

    // The write task has been sent a later checkpoint block that is not durable yet.
    state.finalized_block_write_last_sent_hash = block::Hash([9; 32]);

    assert!(
        !state.drains_the_non_finalized_queue_now(&durable_tip),
        "a block naming the lagging durable tip is not drained, so the bound must reject it",
    );
    assert!(
        !state.drains_the_non_finalized_queue_now(&block::Hash([9; 32])),
        "a child of the last hash sent is not drained either until that write is durable",
    );

    // Once the write task catches up, a child of the last hash sent completes the handoff
    // condition, and the handoff drains it on the same call.
    state.finalized_block_write_last_sent_hash = durable_tip;
    assert!(
        state.drains_the_non_finalized_queue_now(&durable_tip),
        "a child of a durably written last sent hash is drained by the handoff",
    );
    assert!(
        !state.drains_the_non_finalized_queue_now(&block::Hash([9; 32])),
        "any other parent is still not drained",
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn unpublished_writer_transitions_block_optimistic_relay_and_bound_bodies() {
    let _init_guard = zakura_test::init();
    let (mut state, _, _, _) =
        StateService::new(Config::ephemeral(), &Network::Mainnet, Height::MAX, 0)
            .await
            .expect("ephemeral state opens");
    let genesis: Arc<Block> = zakura_test::vectors::BLOCK_MAINNET_GENESIS_BYTES
        .zcash_deserialize_into()
        .unwrap();
    state
        .queue_and_commit_to_finalized_state(CheckpointVerifiedBlock::from(genesis))
        .await
        .unwrap()
        .unwrap();
    let (sender, mut writer) = tokio::sync::mpsc::unbounded_channel();
    state.block_write_sender.non_finalized = Some(sender);

    let candidate: Arc<Block> = zakura_test::vectors::BLOCK_MAINNET_1_BYTES
        .zcash_deserialize_into()
        .unwrap();
    let parent = candidate.header.previous_block_hash;
    let queue = |state: &mut StateService, nonce: u8, optimistic: bool| {
        let mut block = (*candidate).clone();
        Arc::make_mut(&mut block.header).nonce.0[0] = nonce;
        let admission = BlockAdmission::pending();
        if optimistic {
            admission.authorize_optimistic_relay();
        }
        let (tx, rx) = tokio::sync::oneshot::channel();
        state.non_finalized_state_queued_blocks.queue((
            Arc::new(block).prepare(),
            tx,
            Some(admission.clone()),
            0,
        ));
        state.send_ready_non_finalized_queued(parent);
        (admission, rx)
    };

    let (_, _first_response) = queue(&mut state, 1, false);
    let first = writer.try_recv().unwrap();
    let (sibling, _response) = queue(&mut state, 2, true);
    assert!(sibling.wait().await);
    assert!(
        !sibling.optimistic_relay_authorized(),
        "an ordinary earlier write can change the selected tip"
    );
    drop(first);
    drop(writer.try_recv().unwrap());

    let _reconsider_response = state.send_reconsider_block(block::Hash([99; 32]));
    let reconsider = writer.try_recv().unwrap();
    let (sibling, _response) = queue(&mut state, 3, true);
    assert!(sibling.wait().await);
    assert!(
        !sibling.optimistic_relay_authorized(),
        "reconsideration can restore a different best chain"
    );
    drop(reconsider);
    drop(writer.try_recv().unwrap());

    let (sibling, _response) = queue(&mut state, 8, true);
    assert!(sibling.wait().await);
    assert!(
        sibling.optimistic_relay_authorized(),
        "a published reconsideration releases its write slot, so relay is authorized again"
    );
    drop(writer.try_recv().unwrap());

    let _invalidate_response = state.send_invalidate_block(block::Hash([99; 32]));
    let invalidate = writer.try_recv().unwrap();
    let (sibling, _response) = queue(&mut state, 4, true);
    assert!(sibling.wait().await);
    assert!(
        !sibling.optimistic_relay_authorized(),
        "invalidation need not name the immediate parent"
    );
    drop(invalidate);
    drop(writer.try_recv().unwrap());

    let (sibling, _response) = queue(&mut state, 7, true);
    assert!(sibling.wait().await);
    assert!(
        sibling.optimistic_relay_authorized(),
        "a published invalidation releases its write slot, so relay is authorized again"
    );
    drop(writer.try_recv().unwrap());

    let capacity = state
        .non_finalized_write_slots
        .clone()
        .try_acquire_many_owned(u32::try_from(super::queued_blocks::MAX_QUEUED_BLOCKS).unwrap())
        .unwrap();
    let (rejected, response) = queue(&mut state, 5, true);
    assert!(!rejected.wait().await);
    assert!(response.await.unwrap().is_err());
    assert!(
        writer.try_recv().is_err(),
        "a full writer cannot retain another block body"
    );
    drop(capacity);

    let (admitted, _response) = queue(&mut state, 6, true);
    assert!(admitted.wait().await);
    assert!(
        admitted.optimistic_relay_authorized(),
        "an idle writer and live selected parent permit relay"
    );
    drop(writer.try_recv().unwrap());
    assert_eq!(
        state.non_finalized_write_slots.available_permits(),
        super::queued_blocks::MAX_QUEUED_BLOCKS
    );
}

/// The sent cache must not hide a completed commit from a same-hash retry.
#[tokio::test]
async fn known_block_prefers_committed_state_over_sent_cache() {
    use crate::KnownBlock;
    use tower::{Service, ServiceExt};

    let _init_guard = zakura_test::init();
    let mut config = Config::ephemeral();
    config.enable_zakura_header_seed_from_committed_blocks = true;
    config.vct_fast_sync = false;
    let (mut state, _, _, _) = StateService::new(config, &Network::Mainnet, Height::MAX, 0)
        .await
        .unwrap();
    let block: Arc<Block> = zakura_test::vectors::BLOCK_MAINNET_GENESIS_BYTES
        .zcash_deserialize_into()
        .unwrap();
    let hash = block.hash();
    let checkpoint = CheckpointVerifiedBlock::from(block);
    state
        .non_finalized_block_write_sent_hashes
        .add_finalized(&checkpoint);
    assert_eq!(
        state
            .ready()
            .await
            .unwrap()
            .call(Request::KnownBlock(hash))
            .await
            .unwrap(),
        Response::KnownBlock(Some(KnownBlock::WriteChannel))
    );

    timeout(
        Duration::from_secs(5),
        state.queue_and_commit_to_finalized_state(checkpoint),
    )
    .await
    .unwrap()
    .unwrap()
    .unwrap();
    assert!(state.non_finalized_block_write_sent_hashes.contains(&hash));
    assert_eq!(
        state
            .ready()
            .await
            .unwrap()
            .call(Request::KnownBlock(hash))
            .await
            .unwrap(),
        Response::KnownBlock(Some(KnownBlock::Finalized))
    );
}

#[tokio::test]
async fn parent_input_check_uses_the_parent_utxo_set() {
    let _init_guard = zakura_test::init();
    let mut config = Config::ephemeral();
    config.vct_fast_sync = false;
    let (mut state, _, _, _) = StateService::new(config, &Network::Mainnet, Height::MAX, 0)
        .await
        .expect("the test state opens");
    let blocks: Vec<Arc<Block>> = [
        zakura_test::vectors::BLOCK_MAINNET_GENESIS_BYTES.as_slice(),
        zakura_test::vectors::BLOCK_MAINNET_1_BYTES.as_slice(),
        zakura_test::vectors::BLOCK_MAINNET_2_BYTES.as_slice(),
        zakura_test::vectors::BLOCK_MAINNET_3_BYTES.as_slice(),
        zakura_test::vectors::BLOCK_MAINNET_4_BYTES.as_slice(),
        zakura_test::vectors::BLOCK_MAINNET_5_BYTES.as_slice(),
    ]
    .into_iter()
    .enumerate()
    .map(|(height, bytes)| {
        let mut block: Block = bytes
            .zcash_deserialize_into()
            .expect("the test block deserializes");
        // Non-finalized blocks need v4+ transactions. The header hash stays the same.
        if height >= 2 {
            block.transactions = vec![Arc::new(transaction_v4_from_coinbase(
                &block.transactions[0],
            ))];
        }
        Arc::new(block)
    })
    .collect();
    let coinbase = |i: usize| transparent::OutPoint {
        hash: blocks[i].transactions[0].hash(),
        index: 0,
    };
    let missing = transparent::OutPoint {
        hash: transaction::Hash([255; 32]),
        index: 0,
    };

    for block in &blocks[..2] {
        state
            .queue_and_commit_to_finalized_state(CheckpointVerifiedBlock::from(block.clone()))
            .await
            .expect("the writer responds")
            .expect("the checkpoint commits");
    }
    check_parent_inputs(
        &mut state,
        [
            (
                blocks[1].hash(),
                vec![coinbase(1)],
                ParentInputs::Inconclusive,
            ),
            (
                blocks[1].hash(),
                vec![coinbase(1), missing],
                ParentInputs::Missing(missing),
            ),
            (
                blocks[0].hash(),
                vec![missing],
                ParentInputs::ParentUnavailable,
            ),
            (
                block::Hash([255; 32]),
                vec![missing],
                ParentInputs::ParentUnavailable,
            ),
        ],
    )
    .await;

    for block in &blocks[2..] {
        state
            .queue_and_commit_to_non_finalized_state(block.clone().prepare(), None)
            .await
            .expect("the writer responds")
            .expect("the non-finalized block commits");
    }
    assert_eq!(
        state
            .read_service
            .latest_non_finalized_state()
            .best_chain()
            .map(|chain| chain.blocks.len()),
        Some(4),
        "blocks 2..=5 are non-finalized"
    );
    check_parent_inputs(
        &mut state,
        [
            (
                blocks[5].hash(),
                vec![coinbase(1), coinbase(3), coinbase(5)],
                ParentInputs::Inconclusive,
            ),
            (
                blocks[5].hash(),
                vec![coinbase(3), missing],
                ParentInputs::Missing(missing),
            ),
            // A parent below the best tip uses a fork that excludes the outputs above it.
            (
                blocks[4].hash(),
                vec![coinbase(3)],
                ParentInputs::Inconclusive,
            ),
            (
                blocks[4].hash(),
                vec![coinbase(5)],
                ParentInputs::Missing(coinbase(5)),
            ),
            // The finalized tip excludes non-finalized outputs.
            (
                blocks[1].hash(),
                vec![coinbase(1)],
                ParentInputs::Inconclusive,
            ),
            (
                blocks[1].hash(),
                vec![coinbase(2)],
                ParentInputs::Missing(coinbase(2)),
            ),
            (
                blocks[0].hash(),
                vec![missing],
                ParentInputs::ParentUnavailable,
            ),
        ],
    )
    .await;
}

async fn check_parent_inputs(
    state: &mut StateService,
    cases: impl IntoIterator<Item = (block::Hash, Vec<transparent::OutPoint>, ParentInputs)>,
) {
    use tower::{Service, ServiceExt};
    for (parent, outpoints, expected) in cases {
        let response = state
            .ready()
            .await
            .expect("state is ready")
            .call(Request::CheckParentInputs {
                parent,
                outpoints: outpoints.into(),
            })
            .await
            .expect("the input read succeeds");
        assert_eq!(
            response,
            Response::ParentInputs(expected),
            "parent {parent:?}"
        );
    }
}

/// The parent input check must not prove an input missing from a chain that leaves a gap above
/// the finalized tip or does not continue it. It must stay exact when a published chain overlaps
/// the database.
#[tokio::test]
async fn parent_input_check_requires_a_chain_that_continues_the_finalized_tip() {
    use crate::service::{non_finalized_state::NonFinalizedState, read};

    let _init_guard = zakura_test::init();
    let blocks: Vec<Arc<Block>> = [
        zakura_test::vectors::BLOCK_MAINNET_GENESIS_BYTES.as_slice(),
        zakura_test::vectors::BLOCK_MAINNET_1_BYTES.as_slice(),
        zakura_test::vectors::BLOCK_MAINNET_2_BYTES.as_slice(),
        zakura_test::vectors::BLOCK_MAINNET_3_BYTES.as_slice(),
        zakura_test::vectors::BLOCK_MAINNET_4_BYTES.as_slice(),
        zakura_test::vectors::BLOCK_MAINNET_5_BYTES.as_slice(),
    ]
    .into_iter()
    .enumerate()
    .map(|(height, bytes)| {
        let mut block: Block = bytes.zcash_deserialize_into().expect("vector");
        if height >= 2 {
            block.transactions = vec![Arc::new(transaction_v4_from_coinbase(
                &block.transactions[0],
            ))];
        }
        Arc::new(block)
    })
    .collect();
    let coinbase = |i: usize| transparent::OutPoint {
        hash: blocks[i].transactions[0].hash(),
        index: 0,
    };

    // State A: database holds 0..=1, the non-finalized state holds 2..=5.
    let mut config = Config::ephemeral();
    config.vct_fast_sync = false;
    let (mut state_a, _, _, _) =
        StateService::new(config.clone(), &Network::Mainnet, Height::MAX, 0)
            .await
            .expect("state opens");
    for block in &blocks[..2] {
        state_a
            .queue_and_commit_to_finalized_state(CheckpointVerifiedBlock::from(block.clone()))
            .await
            .expect("writer responds")
            .expect("checkpoint commits");
    }
    for block in &blocks[2..] {
        state_a
            .queue_and_commit_to_non_finalized_state(block.clone().prepare(), None)
            .await
            .expect("writer responds")
            .expect("non-finalized block commits");
    }
    let db_a = state_a.read_service.db.clone();
    let nfs_a = state_a.read_service.latest_non_finalized_state();
    let tip_a = db_a.tip().expect("tip");
    assert_eq!(tip_a.0, Height(1));

    // Sanity: the aligned view proves coinbase(2) present at block 5.
    assert_eq!(
        read::parent_inputs(&nfs_a, &db_a, tip_a, blocks[5].hash(), &[coinbase(2)]),
        ParentInputs::Inconclusive
    );

    // Gap: drop block 2 from the chain, so neither the chain nor the database has it.
    let mut gapped = (**nfs_a.best_chain().expect("best chain")).clone();
    let _ = gapped.pop_root();
    let mut nfs_gap = NonFinalizedState::new(&Network::Mainnet);
    nfs_gap.insert_test_chain(Arc::new(gapped));
    assert_eq!(
        read::parent_inputs(&nfs_gap, &db_a, tip_a, blocks[5].hash(), &[coinbase(2)]),
        ParentInputs::Inconclusive,
        "a chain with a gap above the finalized tip must not prove a missing input"
    );
    // The same with a fork parent below the chain tip.
    assert_eq!(
        read::parent_inputs(&nfs_gap, &db_a, tip_a, blocks[4].hash(), &[coinbase(2)]),
        ParentInputs::Inconclusive,
    );

    // Misaligned: the chain's root does not continue the claimed finalized tip hash.
    assert_eq!(
        read::parent_inputs(
            &nfs_a,
            &db_a,
            (Height(1), block::Hash([7; 32])),
            blocks[5].hash(),
            &[transparent::OutPoint {
                hash: transaction::Hash([9; 32]),
                index: 0
            }],
        ),
        ParentInputs::Inconclusive,
    );

    // State B: database holds 0..=2, so state A's published chain (root 2) overlaps it.
    let (mut state_b, _, _, _) = StateService::new(config, &Network::Mainnet, Height::MAX, 0)
        .await
        .expect("state opens");
    for block in &blocks[..3] {
        state_b
            .queue_and_commit_to_finalized_state(CheckpointVerifiedBlock::from(block.clone()))
            .await
            .expect("writer responds")
            .expect("checkpoint commits");
    }
    let db_b = state_b.read_service.db.clone();
    let tip_b = db_b.tip().expect("tip");
    assert_eq!(tip_b, (Height(2), blocks[2].hash()));
    for (parent, outpoints, expected) in [
        (
            blocks[5].hash(),
            vec![coinbase(1), coinbase(2), coinbase(5)],
            ParentInputs::Inconclusive,
        ),
        (
            blocks[4].hash(),
            vec![coinbase(2), coinbase(4)],
            ParentInputs::Inconclusive,
        ),
        (
            blocks[4].hash(),
            vec![coinbase(5)],
            ParentInputs::Missing(coinbase(5)),
        ),
        (
            blocks[3].hash(),
            vec![coinbase(4)],
            ParentInputs::Missing(coinbase(4)),
        ),
        (
            blocks[2].hash(),
            vec![coinbase(3)],
            ParentInputs::Missing(coinbase(3)),
        ),
        (
            blocks[2].hash(),
            vec![coinbase(2)],
            ParentInputs::Inconclusive,
        ),
        (
            blocks[1].hash(),
            vec![coinbase(1)],
            ParentInputs::ParentUnavailable,
        ),
    ] {
        assert_eq!(
            read::parent_inputs(&nfs_a, &db_b, tip_b, parent, &outpoints),
            expected,
            "overlap parent {parent:?}"
        );
    }
}

/// A failed checkpoint commit requests Tip during recovery. That request must release
/// a same-hash replacement queued before the writer reported the failure.
#[tokio::test(flavor = "multi_thread")]
async fn checkpoint_recovery_tip_request_releases_queued_replacement() {
    let _init_guard = zakura_test::init();
    let limit = Duration::from_secs(10);
    let (mut state, _, _, _) =
        StateService::new(Config::ephemeral(), &Network::Mainnet, Height(1), 0)
            .await
            .unwrap();
    let genesis: Arc<Block> = zakura_test::vectors::BLOCK_MAINNET_GENESIS_BYTES
        .zcash_deserialize_into()
        .unwrap();
    let child: Arc<Block> = zakura_test::vectors::BLOCK_MAINNET_1_BYTES
        .zcash_deserialize_into()
        .unwrap();
    let parent = genesis.hash();
    timeout(
        limit,
        state.queue_and_commit_to_finalized_state(genesis.into()),
    )
    .await
    .unwrap()
    .unwrap()
    .unwrap();

    // Keep the real writer alive but control its response and reset ordering.
    let real_writer = state.block_write_sender.finalized.take().unwrap();
    let (gate, mut intercepted) = tokio::sync::mpsc::unbounded_channel();
    state.block_write_sender.finalized = Some(gate);
    let (reset_tx, reset_rx) = tokio::sync::mpsc::unbounded_channel();
    let real_reset_rx = std::mem::replace(&mut state.invalid_block_write_reset_receiver, reset_rx);

    let old_response = state.queue_and_commit_to_finalized_state(child.clone().into());
    let old_write = intercepted.try_recv().unwrap();
    let replacement = state.queue_and_commit_to_finalized_state(child.clone().into());
    assert_eq!(state.finalized_state_queued_blocks.len(), 1);
    assert!(intercepted.try_recv().is_err());
    reset_tx.send(parent).unwrap();
    old_write
        .1
        .send(Err(CommitBlockError::WriteTaskExited.into()))
        .unwrap();
    assert!(old_response.await.unwrap().is_err());

    // The checkpoint task asks for Tip after receiving the error, even if its
    // generation has since been superseded or its original caller was dropped.
    let _ = state
        .ready()
        .await
        .unwrap()
        .call(Request::Tip)
        .await
        .unwrap();
    let replacement_write = timeout(limit, intercepted.recv())
        .await
        .expect("recovery must release the replacement without another block request")
        .expect("the writer channel stays open");
    assert!(state.finalized_state_queued_blocks.is_empty());
    real_writer.send(replacement_write).unwrap();
    timeout(limit, replacement).await.unwrap().unwrap().unwrap();
    state.invalid_block_write_reset_receiver = real_reset_rx;
}

#[tokio::test(flavor = "multi_thread")]
async fn await_block_info_waits_for_a_valid_retry_after_rejection() {
    let _init_guard = zakura_test::init();
    let limit = Duration::from_secs(10);
    let chain = v4_coinbase_mainnet_chain();
    let mut state = state_with_finalized_parent(&chain).await;
    let later: Block = zakura_test::vectors::BLOCK_MAINNET_419201_BYTES
        .zcash_deserialize_into()
        .unwrap();
    let spend = later
        .transactions
        .iter()
        .find(|tx| !tx.is_coinbase() && !tx.inputs().is_empty())
        .unwrap()
        .clone();
    let mut bad = (*chain[2]).clone();
    bad.transactions.push(spend);
    let bad = Arc::new(bad).prepare();
    let hash = bad.hash;
    assert_eq!(hash, chain[2].hash());
    timeout(
        limit,
        state.queue_and_commit_to_non_finalized_state(bad, None),
    )
    .await
    .unwrap()
    .unwrap()
    .unwrap_err();

    // Hold a valid retry before its writer runs.
    let real_writer = state.block_write_sender.non_finalized.take().unwrap();
    let (gate, mut intercepted) = tokio::sync::mpsc::unbounded_channel();
    state.block_write_sender.non_finalized = Some(gate);
    let valid_retry =
        state.queue_and_commit_to_non_finalized_state(chain[2].clone().prepare(), None);
    let message = timeout(limit, intercepted.recv()).await.unwrap().unwrap();
    let mut wait = state.call(Request::AwaitBlockInfo(hash));
    assert!(
        timeout(Duration::from_millis(50), &mut wait).await.is_err(),
        "a previous rejection must not fail a waiter for an admitted retry"
    );

    // Committing the retry must wake the same waiter.
    real_writer.send(message).unwrap();
    timeout(limit, valid_retry).await.unwrap().unwrap().unwrap();
    assert!(matches!(
        timeout(limit, wait).await.unwrap().unwrap(),
        Response::BlockInfo(Some(_))
    ));
}

/// A checkpoint body that fails its authorizing data commitment must not strand its
/// descendants behind a closed channel.
///
/// From NU5, a peer can serve a canonical header with altered authorizing data, so the
/// failure does not make the header or its descendants invalid. The writer fails the dropped
/// descendants with a retryable error that names the ancestor, so the syncer can request them
/// again instead of restarting. The next block at the right height clears that state.
#[tokio::test(flavor = "multi_thread")]
async fn rejected_checkpoint_body_fails_descendants_with_a_retryable_error() {
    use crate::tests::FakeChainHelper;
    use zakura_chain::{
        block::{ChainHistoryBlockTxAuthCommitmentHash, CHAIN_HISTORY_ACTIVATION_RESERVED},
        parameters::testnet::{ConfiguredActivationHeights, RegtestParameters},
    };

    let _init_guard = zakura_test::init();
    let network = Network::new_regtest(RegtestParameters {
        activation_heights: ConfiguredActivationHeights {
            nu5: Some(1),
            ..Default::default()
        },
        ..Default::default()
    });
    let (mut state, _, _, _) = StateService::new(Config::ephemeral(), &network, Height(10), 0)
        .await
        .expect("the ephemeral state opens");
    let limit = Duration::from_secs(10);

    let genesis = block::genesis::regtest_genesis_block();
    let honest = genesis.make_fake_child();
    let commitment: [u8; 32] = ChainHistoryBlockTxAuthCommitmentHash::from_commitments(
        &CHAIN_HISTORY_ACTIVATION_RESERVED.into(),
        &honest.auth_data_root(),
    )
    .into();
    let honest = honest.set_block_commitment(commitment);
    let forged = genesis.make_fake_child().set_block_commitment([0x42; 32]);
    let descendant = forged.make_fake_child();

    timeout(
        limit,
        (&mut state).oneshot(Request::CommitCheckpointVerifiedBlock(genesis.into())),
    )
    .await
    .expect("the genesis commit completes")
    .expect("genesis commits");

    let forged_hash = forged.hash();
    let forged_commit = state
        .ready()
        .await
        .expect("the state is ready")
        .call(Request::CommitCheckpointVerifiedBlock(forged.into()));
    let descendant_commit = state
        .ready()
        .await
        .expect("the state is ready")
        .call(Request::CommitCheckpointVerifiedBlock(descendant.into()));

    let forged_error = timeout(limit, forged_commit)
        .await
        .expect("the forged commit completes")
        .expect_err("the forged body fails its commitment");
    let forged_error = forged_error
        .downcast_ref::<crate::CommitCheckpointVerifiedError>()
        .expect("checkpoint commits fail with the checkpoint error type");
    assert!(
        forged_error.is_auth_commitment_mismatch(),
        "{forged_error:?}"
    );

    let descendant_error = timeout(limit, descendant_commit)
        .await
        .expect("the descendant commit completes")
        .expect_err("the descendant cannot commit without its parent");
    let descendant_error = descendant_error
        .downcast_ref::<crate::CommitCheckpointVerifiedError>()
        .expect("checkpoint commits fail with the checkpoint error type");
    assert_eq!(
        descendant_error.inner(),
        &CommitBlockError::ValidateContextError(Box::new(
            ValidateContextError::AncestorBodyRejected(forged_hash)
        )),
    );

    timeout(
        limit,
        (&mut state).oneshot(Request::CommitCheckpointVerifiedBlock(honest.into())),
    )
    .await
    .expect("the honest commit completes")
    .expect("the honest body for the same height commits");
}

#[tokio::test(flavor = "multi_thread")]
async fn precious_block_request_selects_the_preferred_tip_legacy() {
    assert_precious_block_request(false).await;
}

#[tokio::test(flavor = "multi_thread")]
async fn precious_block_request_selects_the_preferred_tip_header_integrated() {
    assert_precious_block_request(true).await;
}

/// `Request::PreciousBlock` changes the published best tip through the block write task.
async fn assert_precious_block_request(header_runtime: bool) {
    use crate::tests::FakeChainHelper;

    let _init_guard = zakura_test::init();
    let limit = Duration::from_secs(10);
    let network = Network::new_regtest(Default::default());
    let config = Config {
        enable_zakura_header_seed_from_committed_blocks: header_runtime,
        ..Config::ephemeral()
    };
    let (mut state, _, _, _) = StateService::new(config, &network, Height::MAX, 0)
        .await
        .unwrap();
    let genesis_hash = network.genesis_hash();
    let siblings = commit_precious_siblings(&mut state, limit).await;
    let best_tip = |state: &StateService| {
        state
            .read_service
            .latest_non_finalized_state()
            .best_tip()
            .unwrap()
            .1
    };
    assert_eq!(best_tip(&state), siblings[0].hash());

    // A finalized block is known, so its request succeeds without changing the tip.
    // The hash loser then becomes best when preferred.
    for hash in [genesis_hash, siblings[1].hash()] {
        timeout(limit, state.send_precious_block(hash))
            .await
            .unwrap()
            .unwrap()
            .unwrap();
    }
    assert_eq!(best_tip(&state), siblings[1].hash());

    let unknown = siblings[1].make_fake_child().hash();
    let error = timeout(limit, state.send_precious_block(unknown))
        .await
        .unwrap()
        .unwrap()
        .unwrap_err();
    assert!(matches!(error, crate::PreciousError::BlockNotFound(hash) if hash == unknown));
    assert_eq!(best_tip(&state), siblings[1].hash());

    // A full writer rejects the request instead of queueing it, and frees no slot it never held.
    wait_for_idle_writer(&state, limit).await;
    let all_slots = state
        .non_finalized_write_slots
        .clone()
        .try_acquire_many_owned(
            u32::try_from(super::queued_blocks::MAX_QUEUED_BLOCKS)
                .expect("the queued block limit fits in a semaphore permit count"),
        )
        .expect("no write is in flight");
    let error = timeout(limit, state.send_precious_block(siblings[0].hash()))
        .await
        .unwrap()
        .unwrap()
        .unwrap_err();
    assert!(matches!(error, crate::PreciousError::WriterFull));
    assert_eq!(best_tip(&state), siblings[1].hash());
    drop(all_slots);

    timeout(limit, state.send_precious_block(siblings[0].hash()))
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    assert_eq!(best_tip(&state), siblings[0].hash());
    wait_for_idle_writer(&state, limit).await;
}

/// Commits a regtest genesis block and its child to the finalized state, then two equal-work
/// children of that child to the non-finalized state.
///
/// Returns the non-finalized blocks by descending raw hash, so the first is the normal
/// equal-work winner and the second is the hash loser.
async fn commit_precious_siblings(state: &mut StateService, limit: Duration) -> Vec<Arc<Block>> {
    let genesis = block::genesis::regtest_genesis_block();
    let mut parent = genesis.make_fake_child();
    Arc::make_mut(&mut Arc::make_mut(&mut parent).header).time += chrono::Duration::seconds(1);
    for block in [genesis, parent.clone()] {
        timeout(
            limit,
            state.queue_and_commit_to_finalized_state(CheckpointVerifiedBlock::from(block)),
        )
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    }
    let mut template = parent.make_fake_child();
    let transaction = transaction_v4_from_coinbase(&template.transactions[0]);
    Arc::make_mut(&mut template).transactions[0] = Arc::new(transaction);
    let merkle_root = template.transactions.iter().cloned().collect();
    let header = Arc::make_mut(&mut Arc::make_mut(&mut template).header);
    header.time += chrono::Duration::seconds(1);
    header.merkle_root = merkle_root;
    header.commitment_bytes =
        <[u8; 32]>::from(state.read_service.db.history_tree().hash().unwrap()).into();

    let mut siblings = template.make_fake_siblings(2);
    siblings.sort_by_key(|block| std::cmp::Reverse(block.hash().0));
    for block in &siblings {
        timeout(
            limit,
            state.queue_and_commit_to_non_finalized_state(block.clone().prepare(), None),
        )
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    }
    siblings
}

/// Waits until the header chain publishes `verified_best` as its verified best tip.
async fn wait_for_verified_best(
    snapshots: &mut tokio::sync::watch::Receiver<Option<zakura_header_chain::EngineSnapshot>>,
    verified_best: block::Hash,
    limit: Duration,
) -> zakura_header_chain::EngineSnapshot {
    timeout(
        limit,
        snapshots.wait_for(|snapshot| {
            snapshot
                .as_ref()
                .is_some_and(|snapshot| snapshot.frontiers.verified_best.hash == verified_best)
        }),
    )
    .await
    .expect("the header chain publishes the expected verified tip")
    .expect("the header snapshot channel stays open")
    .clone()
    .expect("the waited-for snapshot is present")
}

/// The preference is local. A restart from the same database and non-finalized backup clears
/// it, so full state and the header chain both select the normal equal-work winner again.
#[test]
fn precious_block_preference_is_cleared_by_a_restart() {
    let _init_guard = zakura_test::init();
    let limit = Duration::from_secs(10);
    let network = Network::new_regtest(Default::default());
    let cache_dir = tempfile::tempdir().expect("the state cache directory is created");
    let config = Config {
        cache_dir: cache_dir.path().to_owned(),
        ephemeral: false,
        // Write the non-finalized backup synchronously before each publication.
        debug_skip_non_finalized_state_backup_task: true,
        enable_zakura_header_seed_from_committed_blocks: true,
        ..Config::default()
    };
    assert!(config.non_finalized_state_backup_dir(&network).is_some());
    // The finalized tip is at the last checkpoint, so the restart restores the backup.
    let max_checkpoint_height = Height(1);
    // Each node run has its own runtime. Shutting a runtime down drops the state's background
    // tasks, such as its metrics exporter, which hold database handles, so the next run can take
    // the database lock, like after a process exit.
    let node_run = || {
        tokio::runtime::Builder::new_multi_thread()
            .enable_all()
            .build()
            .expect("the node runtime starts")
    };

    let (winner, loser) = node_run().block_on(async {
        let (mut state, read, latest_chain_tip, _) =
            StateService::new(config.clone(), &network, max_checkpoint_height, 0)
                .await
                .unwrap();
        let siblings = commit_precious_siblings(&mut state, limit).await;
        let (winner, loser) = (siblings[0].hash(), siblings[1].hash());
        let mut snapshots = read.subscribe_header_chain_snapshots();
        wait_for_verified_best(&mut snapshots, winner, limit).await;

        timeout(limit, state.send_precious_block(loser))
            .await
            .unwrap()
            .unwrap()
            .unwrap();
        assert_eq!(read.best_tip().unwrap().1, loser);
        assert_eq!(latest_chain_tip.best_tip_hash(), Some(loser));
        let preferred = wait_for_verified_best(&mut snapshots, loser, limit).await;
        // Header-only selection keeps the raw-hash order.
        assert_eq!(preferred.frontiers.header_best.hash, winner);
        wait_for_idle_writer(&state, limit).await;
        (winner, loser)
    });

    node_run().block_on(async {
        let (state, read, latest_chain_tip, _) =
            StateService::new(config, &network, max_checkpoint_height, 0)
                .await
                .unwrap();
        let restored = read.latest_non_finalized_state();
        assert_eq!(
            restored.chain_count(),
            2,
            "the backup restores both siblings"
        );
        assert_eq!(restored.best_tip().unwrap().1, winner);
        assert_eq!(latest_chain_tip.best_tip_hash(), Some(winner));
        let mut snapshots = read.subscribe_header_chain_snapshots();
        let reopened = wait_for_verified_best(&mut snapshots, winner, limit).await;
        assert_eq!(reopened.frontiers.header_best.hash, winner);

        // The restored loser can still be preferred again.
        timeout(limit, state.send_precious_block(loser))
            .await
            .unwrap()
            .unwrap()
            .unwrap();
        assert_eq!(read.best_tip().unwrap().1, loser);
        wait_for_verified_best(&mut snapshots, loser, limit).await;
        wait_for_idle_writer(&state, limit).await;
    });
}

#[tokio::test(flavor = "multi_thread")]
async fn evicted_precious_block_is_downloaded_again_without_its_preference_legacy() {
    assert_evicted_precious_block_redelivery(false).await;
}

#[tokio::test(flavor = "multi_thread")]
async fn evicted_precious_block_is_downloaded_again_without_its_preference_header_integrated() {
    assert_evicted_precious_block_redelivery(true).await;
}

/// A fork-limit eviction forgets the evicted tip's preference. When the evicted body is
/// downloaded again, full state and the header chain select the normal equal-work winner.
async fn assert_evicted_precious_block_redelivery(header_runtime: bool) {
    let _init_guard = zakura_test::init();
    let limit = Duration::from_secs(10);
    let network = Network::new_regtest(Default::default());
    let config = Config {
        enable_zakura_header_seed_from_committed_blocks: header_runtime,
        ..Config::ephemeral()
    };
    let (mut state, read, latest_chain_tip, _) =
        StateService::new(config, &network, Height::MAX, 0)
            .await
            .unwrap();
    let siblings = commit_precious_siblings(&mut state, limit).await;
    let (winner, loser) = (siblings[0].clone(), siblings[1].clone());
    // Renumbering a sibling's nonce gives another child of the same parent.
    let branch = winner
        .make_fake_siblings(3)
        .into_iter()
        .find(|block| !siblings.contains(block))
        .unwrap();
    timeout(limit, state.send_precious_block(loser.hash()))
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    assert_eq!(read.best_tip().unwrap().1, loser.hash());
    assert!(read
        .latest_non_finalized_state()
        .precious_preferences()
        .1
        .contains_key(&loser.hash()));

    let child_of = |state: &StateService, parent: &Arc<Block>| {
        let history_root = state
            .read_service
            .latest_non_finalized_state()
            .find_chain(|chain| chain.contains_block_hash(parent.hash()))
            .unwrap()
            .history_tree(crate::HashOrHeight::Hash(parent.hash()))
            .unwrap()
            .hash()
            .unwrap();
        let mut child = parent.make_fake_child();
        let merkle_root = child.transactions.iter().cloned().collect();
        let header = Arc::make_mut(&mut Arc::make_mut(&mut child).header);
        header.time += chrono::Duration::seconds(1);
        header.merkle_root = merkle_root;
        header.commitment_bytes = <[u8; 32]>::from(history_root).into();
        child
    };
    let commit = |state: &mut StateService, block: &Arc<Block>| {
        let response = state.queue_and_commit_to_non_finalized_state(block.clone().prepare(), None);
        async move {
            timeout(limit, response).await.unwrap().unwrap().unwrap();
        }
    };

    // Give every other tip more work than `loser`, so it is the lowest chain when the last fork
    // arrives. Its parent is the finalized tip, so it can be downloaded again.
    let winner_child = child_of(&state, &winner);
    commit(&mut state, &winner_child).await;
    commit(&mut state, &branch).await;
    let forks = child_of(&state, &branch)
        .make_fake_siblings(crate::constants::MAX_NON_FINALIZED_CHAIN_FORKS - 1);
    for fork in &forks {
        commit(&mut state, fork).await;
    }
    let evicted = read.latest_non_finalized_state();
    assert!(!evicted.any_chain_contains(&loser.hash()));
    // The commit that evicted the tip forgets its preference, before any invalidation runs.
    assert!(!evicted.precious_preferences().1.contains_key(&loser.hash()));
    let known = timeout(limit, state.call(Request::KnownBlock(loser.hash())))
        .await
        .unwrap()
        .unwrap();
    assert!(
        matches!(known, Response::KnownBlock(None)),
        "an evicted body must be downloadable again: {known:?}"
    );

    // Remove every tip with more work, then download the evicted body again.
    for hash in [branch.hash(), winner_child.hash()] {
        timeout(limit, state.send_invalidate_block(hash))
            .await
            .unwrap()
            .unwrap()
            .unwrap();
    }
    commit(&mut state, &loser).await;

    assert_eq!(
        read.best_tip().unwrap().1,
        winner.hash(),
        "an evicted tip must not keep its operator preference"
    );
    assert_eq!(latest_chain_tip.best_tip_hash(), Some(winner.hash()));
    if header_runtime {
        let mut snapshots = read.subscribe_header_chain_snapshots();
        let snapshot = wait_for_verified_best(&mut snapshots, winner.hash(), limit).await;
        assert_eq!(snapshot.frontiers.header_best.hash, winner.hash());
    }
    wait_for_idle_writer(&state, limit).await;
}

/// Dropping the caller's response does not cancel an accepted preference. The writer commits and
/// publishes it, and keeps the request's write slot, which blocks optimistic relay, until the new
/// tip is published.
#[tokio::test(flavor = "multi_thread")]
async fn dropped_precious_block_response_still_publishes_before_releasing_its_slot() {
    let _init_guard = zakura_test::init();
    let limit = Duration::from_secs(10);
    let all_slots = super::queued_blocks::MAX_QUEUED_BLOCKS;
    let network = Network::new_regtest(Default::default());
    let config = Config {
        enable_zakura_header_seed_from_committed_blocks: true,
        ..Config::ephemeral()
    };
    let (mut state, read, latest_chain_tip, _) =
        StateService::new(config, &network, Height::MAX, 0)
            .await
            .unwrap();
    let siblings = commit_precious_siblings(&mut state, limit).await;
    let (winner, loser) = (siblings[0].hash(), siblings[1].hash());
    let mut snapshots = read.subscribe_header_chain_snapshots();
    wait_for_verified_best(&mut snapshots, winner, limit).await;
    wait_for_idle_writer(&state, limit).await;
    timeout(limit, state.ready()).await.unwrap().unwrap();

    // Gate: while a reader borrows the published non-finalized state, the writer cannot
    // publish a new one.
    let (held_tx, held_rx) = std::sync::mpsc::channel();
    let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
    let published = read.non_finalized_state_receiver.clone();
    let gate = std::thread::spawn(move || {
        published.borrow_mapped(|_published| {
            held_tx.send(()).expect("the test waits for the gate");
            release_rx
                .recv_timeout(Duration::from_secs(30))
                .expect("the test releases the gate");
        })
    });
    held_rx
        .recv_timeout(limit)
        .expect("the gate holds the published state");

    // The state accepts the request when it is called, so dropping the response future does not
    // withdraw it.
    drop(state.call(Request::PreciousBlock(loser)));
    assert_eq!(
        state.non_finalized_write_slots.available_permits(),
        all_slots - 1
    );

    // The header commit precedes publication, so the writer is now blocked on the gate.
    wait_for_verified_best(&mut snapshots, loser, limit).await;
    assert_eq!(
        state.non_finalized_write_slots.available_permits(),
        all_slots - 1,
        "the unpublished preference keeps its write slot"
    );
    assert_eq!(latest_chain_tip.best_tip_hash(), Some(winner));
    // The held slot makes the writer busy, so this check returns before it reads the published
    // state, which the gate blocks.
    let relay_authorized = |state: &StateService, parent| {
        let _candidate_slot = state
            .non_finalized_write_slots
            .clone()
            .try_acquire_owned()
            .expect("the writer has free slots");
        let admission = BlockAdmission::pending();
        admission.authorize_optimistic_relay();
        state.optimistic_relay_still_authorized(Some(&admission), parent)
    };
    assert!(
        !relay_authorized(&state, winner),
        "an unpublished preference blocks optimistic relay"
    );

    release_tx.send(()).expect("the gate is waiting");
    gate.join().expect("the gate thread does not panic");
    let capacity = timeout(
        limit,
        state.non_finalized_write_slots.clone().acquire_many_owned(
            u32::try_from(all_slots).expect("the queued block limit fits in a permit count"),
        ),
    )
    .await
    .expect("the writer releases the slot after publishing")
    .expect("the slot semaphore stays open");
    drop(capacity);
    assert_eq!(read.best_tip().unwrap().1, loser);
    assert_eq!(latest_chain_tip.best_tip_hash(), Some(loser));
    assert_eq!(
        snapshots
            .borrow()
            .as_ref()
            .unwrap()
            .frontiers
            .verified_best
            .hash,
        loser
    );
    assert!(relay_authorized(&state, loser));
}

/// Waits until the block writer releases every write slot. A request's response is sent
/// before the writer drops that request's slot.
async fn wait_for_idle_writer(state: &StateService, limit: Duration) {
    timeout(limit, async {
        while state.non_finalized_write_slots.available_permits()
            < super::queued_blocks::MAX_QUEUED_BLOCKS
        {
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    })
    .await
    .expect("the writer releases each request's slot after publishing");
}

/// During checkpoint sync the writer defers non-finalized messages, so precious requests fail
/// immediately instead of waiting for the checkpoint phase to end.
#[tokio::test(flavor = "multi_thread")]
async fn precious_block_request_fails_fast_during_checkpoint_sync() {
    let _init_guard = zakura_test::init();
    let network = Network::new_regtest(Default::default());
    let (state, _, _, _) = StateService::new(Config::ephemeral(), &network, Height::MAX, 0)
        .await
        .unwrap();
    assert!(state.block_write_sender.finalized.is_some());

    let error = timeout(
        Duration::from_secs(1),
        state.send_precious_block(network.genesis_hash()),
    )
    .await
    .expect("the request is answered without waiting for checkpoint sync")
    .unwrap()
    .unwrap_err();
    assert!(matches!(
        error,
        crate::PreciousError::ProcessingCheckpointedBlocks
    ));
}