zakura 1.6.0

//! Integration tests for block hash gossip.

#![allow(clippy::unwrap_in_result)]

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

use tokio::{task::JoinHandle, time::timeout};
use tower::{builder::ServiceBuilder, util::BoxService, Service, ServiceExt};
use tracing::Instrument;

use zakura_chain::{
    block::{Block, Height},
    fmt::humantime_seconds,
    parameters::Network::Mainnet,
    serialization::ZcashDeserializeInto,
};
use zakura_network::{Request, Response};
use zakura_rpc::{MinedBlockEvent, PendingBlockSignal, SubmitBlockChannel};
use zakura_state::{Config as StateConfig, CHAIN_TIP_UPDATE_WAIT_LIMIT};
use zakura_test::mock_service::{MockService, PanicAssertion};

use crate::components::sync::{self, BlockGossipError, SyncStatus, TIPS_RESPONSE_TIMEOUT};

const MAX_PEER_SET_REQUEST_DELAY: Duration = Duration::from_secs(30);

struct GossipTestSetup {
    peer_set: MockService<Request, Response, PanicAssertion>,
    submitblock_sender: tokio::sync::mpsc::UnboundedSender<MinedBlockEvent>,
    state_service: BoxService<zakura_state::Request, zakura_state::Response, crate::BoxError>,
    gossip_task_handle: JoinHandle<Result<(), BlockGossipError>>,
}

async fn setup_gossip_test() -> GossipTestSetup {
    let _init_guard = zakura_test::init();

    let network = Mainnet;
    let state_config = StateConfig::ephemeral();
    let (state, _read_only_state, _latest_chain_tip, mut chain_tip_change) =
        zakura_state::init(state_config, &network, Height::MAX, 0)
            .await
            .expect("ephemeral state initialization succeeds");

    let mut state_service = ServiceBuilder::new().buffer(1).service(state);

    let genesis_block: Arc<Block> = zakura_test::vectors::BLOCK_MAINNET_GENESIS_BYTES
        .zcash_deserialize_into()
        .unwrap();
    state_service
        .ready()
        .await
        .unwrap()
        .call(zakura_state::Request::CommitCheckpointVerifiedBlock(
            genesis_block.into(),
        ))
        .await
        .unwrap();

    if let Err(timeout_error) = timeout(
        CHAIN_TIP_UPDATE_WAIT_LIMIT,
        chain_tip_change.wait_for_tip_change(),
    )
    .await
    .map(|change_result| change_result.expect("unexpected chain tip update failure"))
    {
        panic!(
            "timeout waiting for genesis chain tip change after {}: {timeout_error:?}",
            humantime_seconds(CHAIN_TIP_UPDATE_WAIT_LIMIT),
        );
    }

    let block_one: Arc<Block> = zakura_test::vectors::BLOCK_MAINNET_1_BYTES
        .zcash_deserialize_into()
        .unwrap();
    state_service
        .clone()
        .oneshot(zakura_state::Request::CommitCheckpointVerifiedBlock(
            block_one.clone().into(),
        ))
        .await
        .unwrap();

    let (sync_status, mut recent_syncs) = SyncStatus::new();
    SyncStatus::sync_close_to_tip(&mut recent_syncs);

    let mut peer_set = MockService::build()
        .with_max_request_delay(MAX_PEER_SET_REQUEST_DELAY)
        .for_unit_tests();

    let submitblock_channel = SubmitBlockChannel::new();
    let submitblock_sender = submitblock_channel.sender();
    let gossip_task_handle = tokio::spawn(
        sync::gossip_best_tip_block_hashes(
            sync_status,
            chain_tip_change,
            peer_set.clone(),
            Some(submitblock_channel.receiver()),
        )
        .in_current_span(),
    );

    // The genesis block gossip is skipped because block 1 is committed before the task starts.
    peer_set
        .expect_request(Request::AdvertiseBlock(block_one.hash(), None))
        .await
        .respond(Response::Nil);

    GossipTestSetup {
        peer_set,
        submitblock_sender,
        state_service: BoxService::new(state_service),
        gossip_task_handle,
    }
}

/// After a successful mined block broadcast, the gossip task marks the tip as seen and does not
/// send a duplicate committed-tip gossip for the same hash.
#[tokio::test(flavor = "current_thread", start_paused = true)]
async fn mined_block_marks_tip_after_successful_broadcast() {
    let GossipTestSetup {
        mut peer_set,
        submitblock_sender,
        mut state_service,
        gossip_task_handle: _gossip_task_handle,
    } = setup_gossip_test().await;

    let block_two: Arc<Block> = zakura_test::vectors::BLOCK_MAINNET_2_BYTES
        .zcash_deserialize_into()
        .unwrap();

    state_service
        .ready()
        .await
        .unwrap()
        .call(zakura_state::Request::CommitCheckpointVerifiedBlock(
            block_two.clone().into(),
        ))
        .await
        .unwrap();

    submitblock_sender
        .send(MinedBlockEvent::Committed {
            hash: block_two.hash(),
            height: block_two.coinbase_height().unwrap(),
        })
        .expect("mined block notification should be accepted");

    peer_set
        .expect_request(Request::AdvertiseBlockToAll(block_two.hash()))
        .await
        .respond(Response::Nil);

    // The committed tip gossip path should not advertise the same hash again.
    peer_set.expect_no_requests().await;
}

/// A successful mined-block broadcast still suppresses the committed-tip fallback for that hash
/// even when another mined-block notification is already queued.
#[tokio::test(flavor = "current_thread", start_paused = true)]
async fn mined_block_mark_survives_pending_submit_queue() {
    let GossipTestSetup {
        mut peer_set,
        submitblock_sender,
        mut state_service,
        gossip_task_handle: _gossip_task_handle,
    } = setup_gossip_test().await;

    let block_two: Arc<Block> = zakura_test::vectors::BLOCK_MAINNET_2_BYTES
        .zcash_deserialize_into()
        .unwrap();
    let height = block_two.coinbase_height().unwrap();
    let hash = block_two.hash();

    state_service
        .ready()
        .await
        .unwrap()
        .call(zakura_state::Request::CommitCheckpointVerifiedBlock(
            block_two.clone().into(),
        ))
        .await
        .unwrap();

    // First mined notification — start AdvertiseBlockToAll but hold the response open.
    submitblock_sender
        .send(MinedBlockEvent::Committed { hash, height })
        .expect("mined block notification should be accepted");

    let first_broadcast = peer_set
        .expect_request(Request::AdvertiseBlockToAll(hash))
        .await;

    // Queue a second notification while the first broadcast is still in flight so the
    // submit-block channel is nonempty when the first mark arrives.
    submitblock_sender
        .send(MinedBlockEvent::Committed { hash, height })
        .expect("second mined block notification should be accepted");

    first_broadcast.respond(Response::Nil);

    // Second mined path also fires AdvertiseBlockToAll for the queued notification.
    peer_set
        .expect_request(Request::AdvertiseBlockToAll(hash))
        .await
        .respond(Response::Nil);

    // Without unconditional marking, the first mark would be dropped while the queue was
    // nonempty and the committed-tip path could still AdvertiseBlock(hash).
    peer_set.expect_no_requests().await;
}

/// If a mined block broadcast times out, the committed tip gossip path should still advertise the
/// hash as a fallback.
#[tokio::test(flavor = "current_thread", start_paused = true)]
async fn mined_block_broadcast_timeout_uses_committed_tip_fallback() {
    let GossipTestSetup {
        mut peer_set,
        submitblock_sender,
        mut state_service,
        gossip_task_handle: _gossip_task_handle,
    } = setup_gossip_test().await;

    let block_two: Arc<Block> = zakura_test::vectors::BLOCK_MAINNET_2_BYTES
        .zcash_deserialize_into()
        .unwrap();

    state_service
        .ready()
        .await
        .unwrap()
        .call(zakura_state::Request::CommitCheckpointVerifiedBlock(
            block_two.clone().into(),
        ))
        .await
        .unwrap();

    submitblock_sender
        .send(MinedBlockEvent::Committed {
            hash: block_two.hash(),
            height: block_two.coinbase_height().unwrap(),
        })
        .expect("mined block notification should be accepted");

    let slow_broadcast = peer_set
        .expect_request(Request::AdvertiseBlockToAll(block_two.hash()))
        .await;

    // Hold the mined block broadcast open past the gossip timeout so it fails without marking.
    tokio::time::sleep(TIPS_RESPONSE_TIMEOUT + Duration::from_secs(1)).await;
    drop(slow_broadcast);

    // The committed tip gossip path should advertise the same hash as a fallback.
    peer_set
        .expect_request(Request::AdvertiseBlock(block_two.hash(), None))
        .await
        .respond(Response::Nil);
}

/// An early broadcast advertises a hash whose body the node cannot serve yet, so it must not
/// suppress the committed-tip fallback.
///
/// A peer can follow the early inventory, exhaust `PENDING_BLOCK_WAIT` waiting for the body, and
/// give up. If the later committed broadcast then fails, the committed-tip gossip is the only
/// thing left that prompts that peer to ask again.
#[tokio::test(flavor = "current_thread", start_paused = true)]
async fn early_broadcast_does_not_suppress_the_committed_tip_fallback() {
    let GossipTestSetup {
        mut peer_set,
        submitblock_sender,
        mut state_service,
        gossip_task_handle: _gossip_task_handle,
    } = setup_gossip_test().await;

    let block_two: Arc<Block> = zakura_test::vectors::BLOCK_MAINNET_2_BYTES
        .zcash_deserialize_into()
        .unwrap();
    let hash = block_two.hash();
    let height = block_two.coinbase_height().unwrap();

    // The early broadcast succeeds, which is what would mark the tip as already gossiped.
    submitblock_sender
        .send(MinedBlockEvent::Early {
            hash,
            height,
            submitted_at: tokio::time::Instant::now().into_std(),
            pending: PendingBlockSignal::valid_for_tests(),
        })
        .expect("the early mined block notification is accepted");

    peer_set
        .expect_request(Request::AdvertiseBlockToAll(hash))
        .await
        .respond(Response::Nil);

    // Let the spawned early broadcast finish before the block commits.
    tokio::time::sleep(Duration::from_millis(200)).await;

    state_service
        .ready()
        .await
        .unwrap()
        .call(zakura_state::Request::CommitCheckpointVerifiedBlock(
            block_two.clone().into(),
        ))
        .await
        .unwrap();

    submitblock_sender
        .send(MinedBlockEvent::Committed { hash, height })
        .expect("the committed mined block notification is accepted");

    // Hold the committed broadcast open past the gossip timeout so it fails without marking.
    let slow_broadcast = peer_set
        .expect_request(Request::AdvertiseBlockToAll(hash))
        .await;
    tokio::time::sleep(TIPS_RESPONSE_TIMEOUT + Duration::from_secs(1)).await;
    drop(slow_broadcast);

    // Nothing has advertised a body this node can serve, so the fallback must still run.
    peer_set
        .expect_request(Request::AdvertiseBlock(hash, None))
        .await
        .respond(Response::Nil);
}

/// Consecutive committed blocks are gossiped as soon as they commit, with no delay between them.
///
/// This test uses real time: with a paused clock, the state commit auto-advances time by minutes,
/// which would hide a gossip delay.
#[tokio::test(flavor = "current_thread")]
async fn consecutive_committed_blocks_are_gossiped_without_delay() {
    /// Well below the removed 7 second delay, and far above the expected latency.
    const MAX_GOSSIP_LATENCY: Duration = Duration::from_secs(3);

    let GossipTestSetup {
        mut peer_set,
        submitblock_sender: _submitblock_sender,
        mut state_service,
        gossip_task_handle: _gossip_task_handle,
    } = setup_gossip_test().await;

    for block_bytes in [
        &*zakura_test::vectors::BLOCK_MAINNET_2_BYTES,
        &*zakura_test::vectors::BLOCK_MAINNET_3_BYTES,
    ] {
        let block: Arc<Block> = block_bytes.zcash_deserialize_into().unwrap();

        state_service
            .ready()
            .await
            .unwrap()
            .call(zakura_state::Request::CommitCheckpointVerifiedBlock(
                block.clone().into(),
            ))
            .await
            .unwrap();
        let committed_at = tokio::time::Instant::now();

        peer_set
            .expect_request(Request::AdvertiseBlock(block.hash(), None))
            .await
            .respond(Response::Nil);

        assert!(
            committed_at.elapsed() < MAX_GOSSIP_LATENCY,
            "block gossip waited {:?} after the commit",
            committed_at.elapsed(),
        );
    }
}

/// While a mined block broadcast is in flight, the committed-tip path does not advertise the same
/// hash.
#[tokio::test(flavor = "current_thread", start_paused = true)]
async fn in_flight_mined_block_broadcast_suppresses_committed_tip_gossip() {
    let GossipTestSetup {
        mut peer_set,
        submitblock_sender,
        mut state_service,
        gossip_task_handle: _gossip_task_handle,
    } = setup_gossip_test().await;

    let block_two: Arc<Block> = zakura_test::vectors::BLOCK_MAINNET_2_BYTES
        .zcash_deserialize_into()
        .unwrap();

    state_service
        .ready()
        .await
        .unwrap()
        .call(zakura_state::Request::CommitCheckpointVerifiedBlock(
            block_two.clone().into(),
        ))
        .await
        .unwrap();

    submitblock_sender
        .send(MinedBlockEvent::Committed {
            hash: block_two.hash(),
            height: block_two.coinbase_height().unwrap(),
        })
        .expect("mined block notification should be accepted");

    let in_flight_broadcast = peer_set
        .expect_request(Request::AdvertiseBlockToAll(block_two.hash()))
        .await;

    // Hold the broadcast open for less than its timeout, so it is still in flight when the
    // committed-tip path sees the tip change.
    tokio::time::sleep(TIPS_RESPONSE_TIMEOUT / 2).await;
    in_flight_broadcast.respond(Response::Nil);

    peer_set.expect_no_requests().await;
}