urma-runtime 0.3.4

URMA publication planning, network transports and recovery
Documentation
use super::*;
use crate::config::{
    P2P_BLOCK_PEER_TIMEOUT_SECS, P2P_HEADERS_TIMEOUT_SECS, P2P_HEIGHT_SLACK, P2P_PIPELINE_PER_PEER,
};
use crate::p2p::headers::Anchor;
use bitcoin::Block;
use bitcoin::consensus::encode::serialize;
use std::sync::Arc;
use std::sync::atomic::Ordering;

#[allow(dead_code)]
#[path = "../../tests/support/mock_peer.rs"]
mod mock_peer;

use mock_peer::{Behaviour, MockPeer};

fn synced_shared(dir: &std::path::Path, chain: &[Block], known: usize) -> Shared {
    let anchor = Anchor {
        height: 0,
        header: chain[0].header,
        next_hash: chain[1].block_hash(),
    };
    let mut headers = HeaderChain::open_at(Chain::BitcoinRegtest, dir, anchor).unwrap();
    let held: Vec<_> = chain[1..=known].iter().map(|block| block.header).collect();
    headers
        .extend(&held, chain[known].header.time + 100)
        .unwrap();
    Shared {
        chain: Chain::BitcoinRegtest,
        pinned: Vec::new(),
        cache_dir: dir.to_owned(),
        headers: Mutex::new(headers),
        blocks: Mutex::new(BlockCache::new()),
        status: Mutex::new(Status {
            peers: 0,
            target: 0,
            synced: true,
            refreshed: Freshness::Never,
        }),
        stop: AtomicBool::new(false),
    }
}

fn connected(dir: &std::path::Path, mocks: &[&MockPeer]) -> Peers {
    let pinned = mocks.iter().map(|mock| mock.address).collect();
    let mut peers = Peers::new(Chain::BitcoinRegtest, pinned, dir);
    peers.discover();
    peers.top_up(0, Duration::from_secs(5));
    assert_eq!(peers.count(), mocks.len());
    peers
}

#[test]
fn a_refresh_walks_past_a_peer_at_our_height_to_the_tip() {
    let chain = Arc::new(mock_peer::regtest_chain(7));
    let lagging = mock_peer::spawn(Arc::new(chain[..3].to_vec()), Behaviour::Serve);
    let ahead = mock_peer::spawn(chain.clone(), Behaviour::Serve);
    let dir = tempfile::tempdir().unwrap();
    let shared = synced_shared(dir.path(), &chain, 2);
    let mut peers = connected(dir.path(), &[&lagging, &ahead]);
    assert_eq!(peers.connected[0].address, lagging.address);
    sync_round(&shared, &mut peers).unwrap();
    assert_eq!(locked(&shared.headers, "headers").tip_height(), 6);
    assert!(ahead.getheaders.load(Ordering::Acquire) >= 1);
    sync_round(&shared, &mut peers).unwrap();
    assert_eq!(peers.count(), 2, "an unchanged tip is not a failure");

    let lagging = mock_peer::spawn(Arc::new(chain[..3].to_vec()), Behaviour::Serve);
    let ahead = mock_peer::spawn(chain.clone(), Behaviour::Serve);
    let dir = tempfile::tempdir().unwrap();
    let shared = synced_shared(dir.path(), &chain, 2);
    let mut unpinned = Peers::new(Chain::BitcoinRegtest, Vec::new(), dir.path());
    for mock in [&lagging, &ahead] {
        let peer = Peer::connect(Chain::BitcoinRegtest, mock.address, 0).unwrap();
        unpinned.connected.push(peer);
    }
    assert_eq!(unpinned.connected[0].address, lagging.address);
    sync_round(&shared, &mut unpinned).unwrap();
    assert_eq!(locked(&shared.headers, "headers").tip_height(), 6);
    assert_eq!(
        lagging.getheaders.load(Ordering::Acquire),
        0,
        "unpinned refreshes ask the highest announced peer first and stop at the best height"
    );
}

#[test]
fn an_inflated_announcement_costs_one_question_and_then_ranks_by_proof() {
    let chain = Arc::new(mock_peer::regtest_chain(7));
    let liar = mock_peer::spawn(Arc::new(chain[..3].to_vec()), Behaviour::Announce(i32::MAX));
    let honest = mock_peer::spawn(chain.clone(), Behaviour::Serve);
    let dir = tempfile::tempdir().unwrap();
    let shared = synced_shared(dir.path(), &chain, 2);
    let mut peers = Peers::new(Chain::BitcoinRegtest, Vec::new(), dir.path());
    for mock in [&liar, &honest] {
        let peer = Peer::connect(Chain::BitcoinRegtest, mock.address, 0).unwrap();
        peers.connected.push(peer);
    }
    assert_eq!(peers.best_height(2), 2 + P2P_HEIGHT_SLACK);
    sync_round(&shared, &mut peers).unwrap();
    assert_eq!(locked(&shared.headers, "headers").tip_height(), 6);
    assert_eq!(
        peers.best_height(6),
        6,
        "the lie no longer outranks the tip"
    );
    let first = peers.order_for_headers(6)[0];
    assert_eq!(
        peers.connected[first].address, honest.address,
        "the proven peer ranks ahead of the inflated one"
    );
}

#[test]
fn a_quiet_refresh_skips_an_unproven_peer_announcing_less_than_our_tip() {
    let chain = Arc::new(mock_peer::regtest_chain(7));
    let syncing = mock_peer::spawn(Arc::new(chain[..3].to_vec()), Behaviour::Trickle);
    let synced = mock_peer::spawn(chain.clone(), Behaviour::Serve);
    let dir = tempfile::tempdir().unwrap();
    let shared = synced_shared(dir.path(), &chain, 6);
    let mut peers = Peers::new(Chain::BitcoinRegtest, Vec::new(), dir.path());
    for mock in [&syncing, &synced] {
        let peer = Peer::connect(Chain::BitcoinRegtest, mock.address, 0).unwrap();
        peers.connected.push(peer);
    }
    let started = Instant::now();
    sync_round(&shared, &mut peers).unwrap();
    assert!(
        started.elapsed() < Duration::from_secs(2),
        "{:?}",
        started.elapsed()
    );
    assert_eq!(syncing.getheaders.load(Ordering::Acquire), 0);
    assert!(synced.getheaders.load(Ordering::Acquire) >= 1);
    assert_eq!(peers.count(), 2);
}

#[test]
fn an_empty_answer_proves_nothing_and_a_tip_answer_proves_the_tip() {
    let chain = Arc::new(mock_peer::regtest_chain(7));
    let lagging = mock_peer::spawn(Arc::new(chain[..4].to_vec()), Behaviour::Serve);
    let synced = mock_peer::spawn(chain.clone(), Behaviour::Serve);
    let inflated = mock_peer::spawn(Arc::new(chain[..4].to_vec()), Behaviour::Announce(i32::MAX));
    let dir = tempfile::tempdir().unwrap();
    let shared = synced_shared(dir.path(), &chain, 6);
    let mut peers = Peers::new(Chain::BitcoinRegtest, Vec::new(), dir.path());
    for mock in [&lagging, &synced, &inflated] {
        let peer = Peer::connect(Chain::BitcoinRegtest, mock.address, 0).unwrap();
        peers.connected.push(peer);
    }
    assert!(!sync_from(&shared, &mut peers, 2).unwrap());
    assert_eq!(peers.connected[2].proven_height, None);
    assert_eq!(
        peers.connected[2].height_hint(6),
        5,
        "an empty answer to a tip-1 locator means the peer lacks our tip"
    );
    assert!(!peers.order_for_block(6, 6).contains(&2));
    assert!(!sync_from(&shared, &mut peers, 0).unwrap());
    assert_eq!(peers.connected[0].proven_height, None);
    assert_eq!(peers.connected[0].height_hint(6), 3);
    assert!(!sync_from(&shared, &mut peers, 1).unwrap());
    assert_eq!(peers.connected[1].proven_height, Some(6));
    let holders = peers.order_for_block(5, 6);
    assert_eq!(
        holders,
        vec![1, 2],
        "the lagging peer (index 0) holds nothing above its height; the inflated peer stays an unproven candidate up to tip-1"
    );
}

#[test]
fn a_refresh_ends_at_the_first_peer_that_reaches_the_best_height() {
    let chain = Arc::new(mock_peer::regtest_chain(5));
    let first = mock_peer::spawn(chain.clone(), Behaviour::Serve);
    let second = mock_peer::spawn(chain.clone(), Behaviour::Serve);
    let dir = tempfile::tempdir().unwrap();
    let shared = synced_shared(dir.path(), &chain, 1);
    let mut peers = connected(dir.path(), &[&first, &second]);
    let asked = second.getheaders.load(Ordering::Acquire);
    sync_round(&shared, &mut peers).unwrap();
    assert_eq!(locked(&shared.headers, "headers").tip_height(), 4);
    assert_eq!(
        second.getheaders.load(Ordering::Acquire),
        asked,
        "the second peer is not asked once the first reached the best height"
    );
}

#[test]
fn a_silent_block_holder_is_dropped_for_redial_and_low_peers_are_skipped() {
    let chain = Arc::new(mock_peer::regtest_chain(6));
    let lagging = mock_peer::spawn(Arc::new(chain[..3].to_vec()), Behaviour::Serve);
    let silent = mock_peer::spawn(chain.clone(), Behaviour::SilentOnGetData);
    let good = mock_peer::spawn(chain.clone(), Behaviour::Serve);
    let dir = tempfile::tempdir().unwrap();
    let shared = synced_shared(dir.path(), &chain, 5);
    let mut peers = connected(dir.path(), &[&lagging, &silent, &good]);
    let mut flight = InFlight::default();
    let started = Instant::now();
    let raw = fetch_block(&shared, &mut peers, &mut flight, chain[4].block_hash()).unwrap();
    assert_eq!(raw, serialize(&chain[4]));
    while peers.count() == 3 {
        assert!(
            started.elapsed() < Duration::from_secs(P2P_BLOCK_PEER_TIMEOUT_SECS + 4),
            "the silent holder was never dropped"
        );
        std::thread::sleep(Duration::from_millis(50));
        pipeline::settle(&shared, &mut peers, &mut flight);
    }
    assert!(
        started.elapsed() < Duration::from_secs(P2P_BLOCK_PEER_TIMEOUT_SECS + 4),
        "{:?}",
        started.elapsed()
    );
    assert_eq!(
        lagging.getdata.load(Ordering::Acquire),
        0,
        "a peer that announced a lower height is not asked"
    );
    assert_eq!(silent.getdata.load(Ordering::Acquire), 1);
    assert_eq!(peers.count(), 2);
    assert_eq!(
        peers.redial_pending(),
        1,
        "a silent peer is redialed, not banned"
    );
}

#[test]
fn a_prefetch_asks_a_lone_holder_once_and_keeps_the_other_peers() {
    let chain = Arc::new(mock_peer::regtest_chain(8));
    let first = mock_peer::spawn(Arc::new(chain[..3].to_vec()), Behaviour::Serve);
    let second = mock_peer::spawn(Arc::new(chain[..3].to_vec()), Behaviour::Serve);
    let holder = mock_peer::spawn(chain.clone(), Behaviour::SilentOnGetData);
    let dir = tempfile::tempdir().unwrap();
    let shared = synced_shared(dir.path(), &chain, 7);
    let mut peers = connected(dir.path(), &[&first, &second, &holder]);
    let mut flight = InFlight::default();
    let raw = fetch_block(&shared, &mut peers, &mut flight, chain[2].block_hash()).unwrap();
    assert_eq!(raw, serialize(&chain[2]));
    let started = Instant::now();
    while peers.count() == 3 {
        assert!(
            started.elapsed() < Duration::from_secs(P2P_BLOCK_PEER_TIMEOUT_SECS + 4),
            "the silent holder was never dropped"
        );
        std::thread::sleep(Duration::from_millis(50));
        pipeline::settle(&shared, &mut peers, &mut flight);
    }
    assert_eq!(holder.getdata.load(Ordering::Acquire), 1);
    assert_eq!(peers.count(), 2);
    let kept: Vec<_> = peers.connected.iter().map(|peer| peer.address).collect();
    assert_eq!(kept, vec![first.address, second.address]);
}

#[test]
fn unsolicited_blocks_are_dropped_and_only_requested_blocks_are_stashed() {
    let chain = Arc::new(mock_peer::regtest_chain(120));
    let flood = mock_peer::spawn(chain.clone(), Behaviour::Flood);
    let dir = tempfile::tempdir().unwrap();
    let shared = synced_shared(dir.path(), &chain, 119);
    let mut peers = connected(dir.path(), &[&flood]);
    let started = Instant::now();
    while started.elapsed() < Duration::from_secs(2) {
        peers.poll_idle();
        assert!(peers.connected[0].arrived.is_empty());
        assert!(peers.connected[0].lacking.is_empty());
        std::thread::sleep(Duration::from_millis(50));
    }
    assert_eq!(peers.count(), 1, "a flood is dropped, not the peer");
    let mut flight = InFlight::default();
    let raw = fetch_block(&shared, &mut peers, &mut flight, chain[5].block_hash()).unwrap();
    assert_eq!(raw, serialize(&chain[5]));
    for _round in 0..10 {
        peers.poll_idle();
        assert!(peers.connected[0].arrived.len() <= P2P_PIPELINE_PER_PEER + 1);
        assert!(peers.connected[0].outstanding() <= P2P_PIPELINE_PER_PEER + 1);
        std::thread::sleep(Duration::from_millis(20));
    }
}

#[test]
fn a_prefetch_notfound_keeps_the_peer() {
    let chain = Arc::new(mock_peer::regtest_chain(6));
    let first = mock_peer::spawn(Arc::new(chain[..3].to_vec()), Behaviour::Serve);
    let second = mock_peer::spawn(Arc::new(chain[..3].to_vec()), Behaviour::Serve);
    let dir = tempfile::tempdir().unwrap();
    let shared = synced_shared(dir.path(), &chain, 5);
    let mut peers = connected(dir.path(), &[&first, &second]);
    let mut flight = InFlight::default();
    fetch_block(&shared, &mut peers, &mut flight, chain[2].block_hash()).unwrap();
    let started = Instant::now();
    while second.getdata.load(Ordering::Acquire) == 0 {
        assert!(
            started.elapsed() < Duration::from_secs(2),
            "prefetch never asked"
        );
        std::thread::sleep(Duration::from_millis(20));
    }
    std::thread::sleep(Duration::from_millis(200));
    peers.poll_idle();
    pipeline::settle(&shared, &mut peers, &mut flight);
    assert_eq!(peers.count(), 2, "notfound to a prefetch is not a drop");
    assert_eq!(peers.redial_pending(), 0);
}

#[test]
fn a_silent_headers_request_drops_and_redials_but_never_bans() {
    let chain = Arc::new(mock_peer::regtest_chain(4));
    let trickle = mock_peer::spawn(chain.clone(), Behaviour::Trickle);
    let dir = tempfile::tempdir().unwrap();
    let shared = synced_shared(dir.path(), &chain, 1);
    let mut peers = connected(dir.path(), &[&trickle]);
    let started = Instant::now();
    let error = sync_round(&shared, &mut peers).unwrap_err();
    let elapsed = started.elapsed();
    assert!(
        elapsed >= Duration::from_secs(P2P_HEADERS_TIMEOUT_SECS - 1)
            && elapsed < Duration::from_secs(P2P_HEADERS_TIMEOUT_SECS + 4),
        "{elapsed:?}"
    );
    assert!(!matches!(error, Error::Invalid(_)), "{error}");
    assert_eq!(peers.count(), 0);
    assert_eq!(
        peers.redial_pending(),
        1,
        "a slow peer is redialed, not banned"
    );
}