zakura-state 11.0.0

//! Fixed test vectors for block queues.

use std::sync::Arc;

use tokio::sync::oneshot;

use zakura_chain::{block::Block, serialization::ZcashDeserializeInto};
use zakura_test::prelude::*;

use crate::{
    arbitrary::Prepare,
    service::queued_blocks::{
        QueuedBlocks, QueuedSemanticallyVerified, SentHashes, MAX_QUEUED_BLOCKS,
    },
    tests::FakeChainHelper,
    CommitBlockError, CommitSemanticallyVerifiedError,
};

// Quick helper trait for making queued blocks with throw away channels
trait IntoQueued {
    fn into_queued(self) -> QueuedSemanticallyVerified;
}

impl IntoQueued for Arc<Block> {
    fn into_queued(self) -> QueuedSemanticallyVerified {
        let (rsp_tx, _) = oneshot::channel();
        (self.prepare(), rsp_tx, None, 0)
    }
}

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

    let block1: Arc<Block> =
        zakura_test::vectors::BLOCK_MAINNET_419200_BYTES.zcash_deserialize_into()?;
    let child1: Arc<Block> =
        zakura_test::vectors::BLOCK_MAINNET_419201_BYTES.zcash_deserialize_into()?;
    let child2 = block1.make_fake_child();

    let parent = block1.header.previous_block_hash;

    let mut queue = QueuedBlocks::default();
    // Empty to start
    assert_eq!(0, queue.blocks.len());
    assert_eq!(0, queue.by_parent.len());
    assert_eq!(0, queue.by_height.len());
    assert_eq!(0, queue.known_utxos.len());

    // Inserting the first block gives us 1 in each table, and some UTXOs
    queue.queue(block1.clone().into_queued());
    assert_eq!(1, queue.blocks.len());
    assert_eq!(1, queue.by_parent.len());
    assert_eq!(1, queue.by_height.len());
    assert_eq!(2, queue.known_utxos.len());

    // The second gives us another in each table because its a child of the first,
    // and a lot of UTXOs
    queue.queue(child1.clone().into_queued());
    assert_eq!(2, queue.blocks.len());
    assert_eq!(2, queue.by_parent.len());
    assert_eq!(2, queue.by_height.len());
    assert_eq!(632, queue.known_utxos.len());

    // The 3rd only increments blocks, because it is also a child of the
    // first block, so for the second and third tables it gets added to the
    // existing HashSet value
    queue.queue(child2.clone().into_queued());
    assert_eq!(3, queue.blocks.len());
    assert_eq!(2, queue.by_parent.len());
    assert_eq!(2, queue.by_height.len());
    assert_eq!(634, queue.known_utxos.len());

    // Dequeueing the first block removes 1 block from each list
    let children = queue.dequeue_children(parent);
    assert_eq!(1, children.len());
    assert_eq!(block1, children[0].0.block);
    assert_eq!(2, queue.blocks.len());
    assert_eq!(1, queue.by_parent.len());
    assert_eq!(1, queue.by_height.len());
    assert_eq!(632, queue.known_utxos.len());

    // Dequeueing the children of the first block removes both of the other
    // blocks, and empties all lists
    let parent = children[0].0.block.hash();
    let children = queue.dequeue_children(parent);
    assert_eq!(2, children.len());
    assert!(children
        .iter()
        .any(|(block, _, _, _)| block.hash == child1.hash()));
    assert!(children
        .iter()
        .any(|(block, _, _, _)| block.hash == child2.hash()));
    assert_eq!(0, queue.blocks.len());
    assert_eq!(0, queue.by_parent.len());
    assert_eq!(0, queue.by_height.len());
    assert_eq!(0, queue.known_utxos.len());

    Ok(())
}

#[test]
fn same_hash_replacement_preserves_receipts_only_for_identical_bodies() -> Result<()> {
    let block: Arc<Block> =
        zakura_test::vectors::BLOCK_MAINNET_419200_BYTES.zcash_deserialize_into()?;
    for (original_order, different_body) in [(Some(1), false), (None, false), (Some(1), true)] {
        let mut replacement_block = (*block).clone();
        if different_body {
            replacement_block
                .transactions
                .push(block.transactions[0].clone());
        }
        let replacement_block = Arc::new(replacement_block);
        assert_eq!(block.hash(), replacement_block.hash());
        let mut queue = QueuedBlocks::default();
        let mut original = block.clone().into_queued();
        original.0.receipt_order = original_order;
        queue.queue(original);

        // Model a redelivery that received a new verifier receipt.
        let mut replacement = replacement_block.clone().into_queued();
        replacement.0.receipt_order = Some(9);
        let old = queue.replace(block.hash(), replacement);
        assert!(Arc::ptr_eq(&old.0.block, &block));
        let retained = queue.get_mut(&block.hash()).unwrap();
        assert!(Arc::ptr_eq(&retained.0.block, &replacement_block));
        assert_eq!(
            retained.0.receipt_order,
            if different_body {
                Some(9)
            } else {
                original_order
            }
        );
        let children = queue.dequeue_children(block.header.previous_block_hash);
        assert_eq!(children.len(), 1);
        assert_eq!(
            children[0].0.receipt_order,
            if different_body {
                Some(9)
            } else {
                original_order
            }
        );
    }
    Ok(())
}

/// The orphan queue holds a fixed number of entries.
///
/// `MAX_QUEUED_BLOCKS` also sizes the non-finalized write slots, and optimistic relay reads an
/// idle writer off that count, so this bound is load-bearing beyond the queue itself.
#[test]
fn orphan_queue_has_a_fixed_entry_bound() -> Result<()> {
    let block: Arc<Block> =
        zakura_test::vectors::BLOCK_MAINNET_419200_BYTES.zcash_deserialize_into()?;
    let mut queue = QueuedBlocks::default();
    assert!(!queue.is_full());

    for index in 0..MAX_QUEUED_BLOCKS {
        let mut queued = block.clone().into_queued();
        let index = u64::try_from(index).expect("the queue bound fits in u64");
        queued.0.hash.0[..8].copy_from_slice(&index.to_le_bytes());
        queue.blocks.insert(queued.0.hash, queued);
    }
    assert!(queue.is_full());
    Ok(())
}

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

    let block1: Arc<Block> =
        zakura_test::vectors::BLOCK_MAINNET_419200_BYTES.zcash_deserialize_into()?;
    let child1: Arc<Block> =
        zakura_test::vectors::BLOCK_MAINNET_419201_BYTES.zcash_deserialize_into()?;
    let child2 = block1.make_fake_child();

    let mut queue = QueuedBlocks::default();
    queue.queue(block1.clone().into_queued());
    queue.queue(child1.clone().into_queued());
    queue.queue(child2.clone().into_queued());
    assert_eq!(3, queue.blocks.len());
    assert_eq!(2, queue.by_parent.len());
    assert_eq!(2, queue.by_height.len());
    assert_eq!(634, queue.known_utxos.len());

    // Pruning the first height removes only block1
    queue.prune_by_height(block1.coinbase_height().unwrap());
    assert_eq!(2, queue.blocks.len());
    assert_eq!(1, queue.by_parent.len());
    assert_eq!(1, queue.by_height.len());
    assert!(queue.get_mut(&block1.hash()).is_none());
    assert!(queue.get_mut(&child1.hash()).is_some());
    assert!(queue.get_mut(&child2.hash()).is_some());
    assert_eq!(632, queue.known_utxos.len());

    // Pruning the children of the first block removes both of the other
    // blocks, and empties all lists
    queue.prune_by_height(child1.coinbase_height().unwrap());
    assert_eq!(0, queue.blocks.len());
    assert_eq!(0, queue.by_parent.len());
    assert_eq!(0, queue.by_height.len());
    assert!(queue.get_mut(&child1.hash()).is_none());
    assert!(queue.get_mut(&child2.hash()).is_none());
    assert_eq!(0, queue.known_utxos.len());

    Ok(())
}

/// `SentHashes::remove` must drop the hash, its unshared outpoints from `known_utxos`,
/// and the corresponding `(hash, height)` entry from `curr_buf` (or whichever
/// batch buffer holds it). Without this, a rejected same-hash block would
/// keep a later honest re-delivery of a block at the same hash locked out as
/// a "duplicate" forever.
#[test]
fn sent_hashes_remove_drops_rejected_hash_and_utxos() -> Result<()> {
    let _init_guard = zakura_test::init();

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

    let prepared1 = block1.clone().prepare();
    let prepared2 = block2.clone().prepare();

    let mut sent = SentHashes::default();
    sent.add(&prepared1);
    sent.add(&prepared2);

    // Both hashes are present, and `known_utxos` contains every outpoint from
    // both blocks' coinbase + transparent outputs.
    let utxos_after_add = sent.known_utxos.len();
    assert!(sent.contains(&prepared1.hash));
    assert!(sent.contains(&prepared2.hash));
    assert!(utxos_after_add > 0);

    // Remove block1. block1's hash disappears, block2's stays, and the
    // total number of known utxos shrinks by exactly block1's contribution.
    let block1_utxos = prepared1.new_outputs.len();
    sent.remove(&prepared1.hash);

    assert!(
        !sent.contains(&prepared1.hash),
        "removed hash must not satisfy contains()"
    );
    assert!(sent.contains(&prepared2.hash));
    assert_eq!(
        sent.known_utxos.len(),
        utxos_after_add - block1_utxos,
        "remove must drop only the removed block's outpoints"
    );

    // The (hash, height) entry must be gone from the batch buffer too,
    // otherwise a later `prune_by_height` could re-insert into `sent`.
    assert!(
        !sent.curr_buf.iter().any(|(h, _)| h == &prepared1.hash),
        "remove must drop the (hash, height) entry from curr_buf"
    );
    assert!(sent.curr_buf.iter().any(|(h, _)| h == &prepared2.hash));

    // Removing a hash that isn't tracked is a no-op.
    let block3 = block1.make_fake_child();
    sent.remove(&block3.hash());
    assert!(sent.contains(&prepared2.hash));

    Ok(())
}

#[test]
fn sent_hashes_remove_many_preserves_other_blocks_across_batches() -> Result<()> {
    let _init_guard = zakura_test::init();
    let earlier: Arc<Block> =
        zakura_test::vectors::BLOCK_MAINNET_419200_BYTES.zcash_deserialize_into()?;
    let block: Arc<Block> =
        zakura_test::vectors::BLOCK_MAINNET_419201_BYTES.zcash_deserialize_into()?;
    let earlier = earlier.prepare();
    let siblings = block.make_fake_siblings(3);
    let removed = siblings[0].clone().prepare();
    let retained = siblings[1].clone().prepare();
    let removed_current = siblings[2].clone().prepare();
    let unknown_hash = block.make_fake_child().hash();

    let mut sent = SentHashes::default();
    sent.add(&earlier);
    sent.finish_batch();
    sent.add(&removed);
    sent.add(&retained);
    sent.finish_batch();
    sent.add(&removed_current);

    // Empty input must leave both finished and unfinished batches untouched.
    sent.remove_many(&[]);
    assert_eq!(sent.sent.len(), 4);
    assert_eq!(sent.bufs.len(), 2);
    assert_eq!(sent.curr_buf.len(), 1);

    sent.remove_many(&[
        earlier.hash,
        removed.hash,
        removed_current.hash,
        removed.hash,
        unknown_hash,
    ]);

    assert_eq!(sent.sent.len(), 1);
    assert!(sent.contains(&retained.hash));
    assert_eq!(sent.known_utxos.len(), retained.new_outputs.len());
    for outpoint in retained.new_outputs.keys() {
        assert!(sent.utxo(outpoint).is_some());
    }
    assert!(sent.curr_buf.is_empty());
    assert_eq!(sent.bufs.len(), 1);
    assert_eq!(
        sent.bufs[0].iter().copied().collect::<Vec<_>>(),
        vec![(retained.hash, retained.height)]
    );

    // Redelivery acquires a new owner after the previous batch entry was removed.
    sent.add(&removed_current);
    sent.remove_many(&[retained.hash, removed_current.hash]);
    assert!(sent.sent.is_empty());
    assert!(sent.known_utxos.is_empty());
    assert!(sent.curr_buf.is_empty());
    assert!(sent.bufs.is_empty());
    Ok(())
}

#[test]
fn sent_hashes_remove_keeps_outputs_shared_with_sent_sibling() -> Result<()> {
    let _init_guard = zakura_test::init();
    let block: Arc<Block> =
        zakura_test::vectors::BLOCK_MAINNET_419201_BYTES.zcash_deserialize_into()?;
    let siblings = block.make_fake_siblings(2);
    let evicted = siblings[0].clone().prepare();
    let in_flight = siblings[1].clone().prepare();
    assert_ne!(evicted.hash, in_flight.hash);

    let mut sent = SentHashes::default();
    sent.add(&evicted);
    sent.add(&in_flight);
    sent.remove(&evicted.hash);

    assert!(sent.contains(&in_flight.hash));
    for outpoint in in_flight.new_outputs.keys() {
        assert!(
            sent.utxo(outpoint).is_some(),
            "removing one sibling must retain shared output {outpoint:?} for the other"
        );
    }
    Ok(())
}

#[test]
fn sent_hashes_shared_outputs_release_after_last_distinct_block() -> Result<()> {
    let _init_guard = zakura_test::init();
    let block: Arc<Block> =
        zakura_test::vectors::BLOCK_MAINNET_419201_BYTES.zcash_deserialize_into()?;
    let siblings = block.make_fake_siblings(2);
    let first = siblings[0].clone().prepare();
    let second = siblings[1].clone().prepare();
    let mut sent = SentHashes::default();
    sent.add(&first);
    sent.add(&first);
    sent.add_finalized(&crate::CheckpointVerifiedBlock::from(siblings[0].clone()));
    sent.add_finalized(&crate::CheckpointVerifiedBlock::from(siblings[1].clone()));
    sent.add(&second);
    sent.finish_batch();

    sent.remove(&first.hash);
    sent.remove(&first.hash);
    for outpoint in second.new_outputs.keys() {
        assert!(sent.utxo(outpoint).is_some());
    }
    sent.remove(&second.hash);
    assert!(sent.known_utxos.is_empty());
    assert!(sent.bufs.iter().all(|batch| batch.is_empty()));

    sent.add(&second);
    sent.prune_by_height(second.height);
    assert!(sent.known_utxos.is_empty());
    assert!(sent.sent.is_empty());
    Ok(())
}

#[test]
fn sent_hashes_pruning_keeps_outputs_owned_by_later_block() -> Result<()> {
    let _init_guard = zakura_test::init();
    let block: Arc<Block> =
        zakura_test::vectors::BLOCK_MAINNET_419201_BYTES.zcash_deserialize_into()?;
    let siblings = block.make_fake_siblings(2);
    let first = siblings[0].clone().prepare();
    // The other fork includes the shared transactions one block later.
    let mut other_parent = siblings[1].clone();
    Arc::make_mut(&mut other_parent).transactions.truncate(1);
    let mut later = other_parent.make_fake_child();
    Arc::make_mut(&mut later)
        .transactions
        .extend(block.transactions.iter().skip(1).cloned());
    let later = later.prepare();
    assert_eq!(later.height, (first.height + 1).unwrap());
    assert!(first.new_outputs.iter().any(|(outpoint, output)| {
        !output.utxo.from_coinbase && later.new_outputs.contains_key(outpoint)
    }));

    let mut sent = SentHashes::default();
    sent.add(&first);
    sent.finish_batch();
    sent.add(&later);
    sent.finish_batch();
    sent.prune_by_height(first.height);
    assert!(!sent.contains(&first.hash));
    assert!(sent.contains(&later.hash));
    for outpoint in later.new_outputs.keys() {
        assert!(sent.utxo(outpoint).is_some());
    }
    for outpoint in first.new_outputs.keys() {
        if !later.new_outputs.contains_key(outpoint) {
            assert!(sent.utxo(outpoint).is_none());
        }
    }

    sent.prune_by_height(later.height);
    assert!(sent.known_utxos.is_empty());
    assert!(sent.sent.is_empty());
    Ok(())
}

// Ensures `dequeue_children` does not remove same-height sibling blocks from other forks.
#[test]
fn dequeue_children_preserves_same_height_siblings() -> Result<()> {
    let _init_guard = zakura_test::init();

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

    let left_child: Arc<Block> =
        zakura_test::vectors::BLOCK_MAINNET_419201_BYTES.zcash_deserialize_into()?;
    let left_grandchild = left_child.make_fake_child();

    let right_child = root_block.make_fake_child();
    let right_grandchild = right_child.make_fake_child();

    let mut queue = QueuedBlocks::default();
    queue.queue(left_grandchild.clone().into_queued());
    queue.queue(right_grandchild.clone().into_queued());

    let height = left_grandchild.coinbase_height().unwrap();

    // Sanity check: both entries are indexed under the same height bucket
    assert_eq!(
        queue.by_height.get(&height).unwrap().len(),
        2,
        "expected both fork grandchildren to be in the same height bucket"
    );

    // Dequeue only one branch
    queue.dequeue_children(left_child.hash());

    assert!(
        queue.blocks.contains_key(&right_grandchild.hash()),
        "sibling block must remain in queue after unrelated dequeue"
    );

    assert!(
        queue
            .by_height
            .get(&height)
            .unwrap()
            .contains(&right_grandchild.hash()),
        "sibling must remain indexed by height after unrelated dequeue"
    );

    Ok(())
}

#[test]
fn dequeue_descendants_removes_the_complete_failed_subtree() -> Result<()> {
    let _init_guard = zakura_test::init();
    let root: Arc<Block> =
        zakura_test::vectors::BLOCK_MAINNET_419200_BYTES.zcash_deserialize_into()?;
    let failed_child: Arc<Block> =
        zakura_test::vectors::BLOCK_MAINNET_419201_BYTES.zcash_deserialize_into()?;
    let failed_grandchild = failed_child.make_fake_child();
    let sibling = root.make_fake_child();

    let mut queue = QueuedBlocks::default();
    let mut responses = Vec::new();
    for block in [failed_child, failed_grandchild, sibling] {
        let (response, receiver) = oneshot::channel();
        queue.queue((block.prepare(), response, None, 0));
        responses.push(receiver);
    }
    let error = CommitSemanticallyVerifiedError::from(CommitBlockError::HeaderChainError {
        error: format!("ancestor {} failed", root.hash()),
    });

    assert_eq!(queue.fail_descendants(root.hash(), error.clone()).len(), 3);
    for response in &mut responses {
        assert_eq!(response.try_recv(), Ok(Err(error.clone())));
    }
    assert!(queue.blocks.is_empty());
    assert!(queue.by_parent.is_empty());
    assert!(queue.by_height.is_empty());
    assert!(queue.known_utxos.is_empty());

    Ok(())
}