zakura-network 7.1.1

//! zakura-network initialization tests using fixed configs.
//!
//! ## Failures due to Port Conflicts
//!
//! If the test has a port conflict with another test, or another process, then it will fail.
//! If these conflicts cause test failures, run the tests in an isolated environment.
//!
//! ## Failures due to Configured Network Interfaces
//!
//! If your test environment does not have any IPv6 interfaces configured, skip IPv6 tests
//! by setting the `SKIP_IPV6_TESTS` environmental variable.
//!
//! If it does not have any IPv4 interfaces, or IPv4 localhost is not on `127.0.0.1`,
//! skip all the network tests by setting the `SKIP_NETWORK_TESTS` environmental variable.

use std::{
    net::{IpAddr, Ipv4Addr, SocketAddr},
    sync::{
        atomic::{AtomicBool, AtomicUsize, Ordering},
        Arc,
    },
    time::{Duration, Instant},
};

use chrono::Utc;
use futures::{channel::mpsc, FutureExt, StreamExt};
use indexmap::IndexSet;
use tokio::{
    io::AsyncWriteExt,
    net::{TcpSocket, TcpStream},
    task::JoinHandle,
};
use tower::{service_fn, Layer, Service, ServiceExt};

use zakura_chain::{chain_tip::NoChainTip, parameters::Network, serialization::DateTime32};

use crate::{
    address_book_updater::AddressBookUpdater,
    config::CacheDir,
    constants, init,
    meta_addr::{MetaAddr, PeerAddrState},
    peer::{self, ClientTestHarness, ConnectedAddr, HandshakeRequest, OutboundConnectorRequest},
    peer_cache_updater::update_peer_cache_once,
    peer_set::{
        initialize::{
            accept_inbound_connections, add_initial_peers, batch_misbehavior_reports,
            crawl_and_dial, open_listener, outbound_peer_replenishment_demand, DiscoveredPeer,
            MISBEHAVIOR_BATCH_INTERVAL, OUTBOUND_PEER_REPLENISHMENT_TARGET_DENOMINATOR,
            OUTBOUND_PEER_REPLENISHMENT_TARGET_NUMERATOR,
        },
        set::MorePeers,
        ActiveConnectionCounter, CandidateSet, ConnectionTracker,
    },
    protocol::types::PeerServices,
    AddressBook, BannedIps, BoxError, Config, PeerSocketAddr, Request, Response,
};

use Network::*;

/// The maximum wall-clock time to wait for expected crawler test progress.
const CRAWLER_TEST_TIMEOUT: Duration = Duration::from_secs(10);

/// The maximum wall-clock time to wait for peer-cache startup progress.
const PEER_CACHE_TEST_TIMEOUT: Duration = Duration::from_secs(10);

/// A crawler peer limit large enough to exercise multi-peer behavior without
/// flooding the test runtime with hundreds of immediate fake handshakes.
const CRAWLER_MANY_PEER_LIMIT_FOR_TESTS: usize = 15;

/// The peer set target size used by the multi-peer listener tests.
///
/// The default target size would need hundreds of real TCP connections to go
/// over the inbound limit, which is too slow and too resource-hungry for a unit
/// test. The default limit arithmetic is covered by the `zakura_network::config`
/// tests instead.
const LISTENER_MANY_PEER_TARGET_SIZE_FOR_TESTS: usize = 10;

/// A small peer target that keeps replenishment race tests fast.
const CRAWLER_REPLENISHMENT_TARGET_SIZE_FOR_TESTS: usize = 10;

/// The outbound connection limit used by replenishment race tests.
const CRAWLER_REPLENISHMENT_OUTBOUND_LIMIT_FOR_TESTS: usize =
    CRAWLER_REPLENISHMENT_TARGET_SIZE_FOR_TESTS * constants::OUTBOUND_PEER_LIMIT_MULTIPLIER
        / constants::OUTBOUND_PEER_LIMIT_DIVISOR;

/// The outbound connection target used by replenishment race tests.
const CRAWLER_REPLENISHMENT_CONNECTION_TARGET_FOR_TESTS: usize = {
    let target = CRAWLER_REPLENISHMENT_TARGET_SIZE_FOR_TESTS
        .saturating_mul(OUTBOUND_PEER_REPLENISHMENT_TARGET_NUMERATOR)
        .div_ceil(OUTBOUND_PEER_REPLENISHMENT_TARGET_DENOMINATOR);

    // `Ord::min` is not const, and the bound matters: the target must never
    // exceed the outbound connection limit these tests configure.
    if target < CRAWLER_REPLENISHMENT_OUTBOUND_LIMIT_FOR_TESTS {
        target
    } else {
        CRAWLER_REPLENISHMENT_OUTBOUND_LIMIT_FOR_TESTS
    }
};

/// The maximum time to wait for the listener tests to make expected progress.
const LISTENER_TEST_DURATION: Duration = Duration::from_secs(10);

/// The amount of time to run zcashd-compat listener tests.
const ZCASHD_COMPAT_LISTENER_TEST_DURATION: Duration = Duration::from_millis(500);

/// The amount of time to make the inbound connection acceptor wait between peer connections.
const MIN_INBOUND_PEER_CONNECTION_INTERVAL_FOR_TESTS: Duration = Duration::from_millis(25);

/// Misbehavior reports are coalesced per peer and flushed after five seconds.
#[tokio::test(start_paused = true)]
async fn misbehavior_reports_are_batched_for_five_seconds() {
    let peer_addr: PeerSocketAddr = "127.0.0.1:8233".parse().unwrap();
    let (misbehavior_tx, misbehavior_rx) = tokio::sync::mpsc::channel(1);
    let (address_book_updater, mut address_book_updates) = tokio::sync::mpsc::channel(1);
    let batcher = tokio::spawn(batch_misbehavior_reports(
        misbehavior_rx,
        address_book_updater,
    ));

    misbehavior_tx
        .send((peer_addr, 40))
        .await
        .expect("misbehavior batcher is running");
    misbehavior_tx
        .send((peer_addr, 60))
        .await
        .expect("misbehavior batcher is running");
    // The third send can complete only after the first two reports have been
    // received from this capacity-one channel.
    misbehavior_tx
        .send((peer_addr, 0))
        .await
        .expect("misbehavior batcher is running");

    tokio::time::advance(MISBEHAVIOR_BATCH_INTERVAL - Duration::from_millis(1)).await;
    tokio::task::yield_now().await;
    assert!(
        address_book_updates.try_recv().is_err(),
        "misbehavior reports must remain batched before the interval elapses"
    );

    tokio::time::advance(Duration::from_millis(1)).await;
    let update = address_book_updates
        .recv()
        .await
        .expect("misbehavior batcher should flush an address book update");
    assert_eq!(update.addr(), peer_addr);
    assert_eq!(
        update.misbehavior_score(),
        constants::MAX_PEER_MISBEHAVIOR_SCORE
    );

    drop(misbehavior_tx);
    batcher
        .await
        .expect("misbehavior batcher task should not panic");
}

/// Test that zakura-network discovers dynamic bind-to-all-interfaces listener ports,
/// and sends them to the `AddressBook`.
///
/// Note: This test doesn't cover local interface or public IP address discovery.
#[tokio::test]
async fn local_listener_unspecified_port_unspecified_addr_v4() {
    let _init_guard = zakura_test::init();

    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    // these tests might fail on machines with no configured IPv4 addresses
    // (localhost should be enough)
    for network in Network::iter() {
        local_listener_port_with("0.0.0.0:0".parse().unwrap(), network).await;
    }
}

/// Test that zakura-network discovers dynamic bind-to-all-interfaces listener ports,
/// and sends them to the `AddressBook`.
///
/// Note: This test doesn't cover local interface or public IP address discovery.
#[tokio::test]
async fn local_listener_unspecified_port_unspecified_addr_v6() {
    let _init_guard = zakura_test::init();

    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    if zakura_test::net::zebra_skip_ipv6_tests() {
        return;
    }

    // these tests might fail on machines with no configured IPv6 addresses
    for network in Network::iter() {
        local_listener_port_with("[::]:0".parse().unwrap(), network).await;
    }
}

/// Test that zakura-network discovers dynamic localhost listener ports,
/// and sends them to the `AddressBook`.
#[tokio::test]
async fn local_listener_unspecified_port_localhost_addr_v4() {
    let _init_guard = zakura_test::init();

    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    // these tests might fail on machines with unusual IPv4 localhost configs
    for network in Network::iter() {
        local_listener_port_with("127.0.0.1:0".parse().unwrap(), network).await;
    }
}

/// Test that zakura-network discovers dynamic localhost listener ports,
/// and sends them to the `AddressBook`.
#[tokio::test]
async fn local_listener_unspecified_port_localhost_addr_v6() {
    let _init_guard = zakura_test::init();

    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    if zakura_test::net::zebra_skip_ipv6_tests() {
        return;
    }

    // these tests might fail on machines with no configured IPv6 addresses
    for network in Network::iter() {
        local_listener_port_with("[::1]:0".parse().unwrap(), network).await;
    }
}

/// Test that zakura-network propagates fixed localhost listener ports to the `AddressBook`.
#[tokio::test]
#[cfg(not(target_os = "windows"))]
async fn local_listener_fixed_port_localhost_addr_v4() {
    let _init_guard = zakura_test::init();

    let localhost_v4 = "127.0.0.1".parse().unwrap();

    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    for network in Network::iter() {
        local_listener_port_with(
            SocketAddr::new(localhost_v4, random_known_listener_port(localhost_v4)),
            network,
        )
        .await;
    }
}

/// Test that zakura-network propagates fixed localhost listener ports to the `AddressBook`.
#[tokio::test]
#[cfg(not(target_os = "windows"))]
async fn local_listener_fixed_port_localhost_addr_v6() {
    let _init_guard = zakura_test::init();

    let localhost_v6 = "::1".parse().unwrap();

    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    if zakura_test::net::zebra_skip_ipv6_tests() {
        return;
    }

    for network in Network::iter() {
        local_listener_port_with(
            SocketAddr::new(localhost_v6, random_known_listener_port(localhost_v6)),
            network,
        )
        .await;
    }
}

#[cfg(not(target_os = "windows"))]
fn random_known_listener_port(ip: IpAddr) -> u16 {
    std::net::TcpListener::bind(SocketAddr::new(
        ip,
        zakura_test::net::random_unallocated_port(),
    ))
    .expect("test needs an available listener port")
    .local_addr()
    .expect("listener has a local address")
    .port()
}

/// Test zakura-network with a peer limit of zero peers on mainnet.
/// (Zebra does not support this mode of operation.)
#[tokio::test]
#[should_panic]
async fn peer_limit_zero_mainnet() {
    let _init_guard = zakura_test::init();

    // This test should not require network access, because the connection limit is zero.

    let unreachable_inbound_service =
        service_fn(|_| async { unreachable!("inbound service should never be called") });

    let address_book =
        init_with_peer_limit(0, unreachable_inbound_service, Mainnet, None, None).await;
    assert_eq!(
        address_book.lock().unwrap().peers().count(),
        0,
        "expected no peers in Mainnet address book, but got: {:?}",
        address_book.lock().unwrap().address_metrics(Utc::now())
    );
}

/// Test zakura-network with a peer limit of zero peers on testnet.
/// (Zebra does not support this mode of operation.)
#[tokio::test]
#[should_panic]
async fn peer_limit_zero_testnet() {
    let _init_guard = zakura_test::init();

    // This test should not require network access, because the connection limit is zero.

    let unreachable_inbound_service =
        service_fn(|_| async { unreachable!("inbound service should never be called") });

    let address_book = init_with_peer_limit(
        0,
        unreachable_inbound_service,
        Network::new_default_testnet(),
        None,
        None,
    )
    .await;

    assert_eq!(
        address_book.lock().unwrap().peers().count(),
        0,
        "expected no peers in Testnet address book, but got: {:?}",
        address_book.lock().unwrap().address_metrics(Utc::now())
    );
}

/// Test zakura-network with a peer limit of one inbound and one outbound peer on mainnet.
#[tokio::test]
async fn peer_limit_one_mainnet() {
    let _init_guard = zakura_test::init();

    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    let nil_inbound_service = service_fn(|_| async { Ok(Response::Nil) });

    let _ = init_with_peer_limit(1, nil_inbound_service, Mainnet, None, None).await;
}

/// Test zakura-network with a peer limit of one inbound and one outbound peer on testnet.
#[tokio::test]
async fn peer_limit_one_testnet() {
    let _init_guard = zakura_test::init();

    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    let nil_inbound_service = service_fn(|_| async { Ok(Response::Nil) });

    let _ = init_with_peer_limit(
        1,
        nil_inbound_service,
        Network::new_default_testnet(),
        None,
        None,
    )
    .await;
}

/// Test zakura-network with a peer limit of two inbound and three outbound peers on mainnet.
#[tokio::test]
async fn peer_limit_two_mainnet() {
    let _init_guard = zakura_test::init();

    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    let nil_inbound_service = service_fn(|_| async { Ok(Response::Nil) });

    let _ = init_with_peer_limit(2, nil_inbound_service, Mainnet, None, None).await;
}

/// Test zakura-network with a peer limit of two inbound and three outbound peers on testnet.
#[tokio::test]
async fn peer_limit_two_testnet() {
    let _init_guard = zakura_test::init();

    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    let nil_inbound_service = service_fn(|_| async { Ok(Response::Nil) });

    let _ = init_with_peer_limit(
        2,
        nil_inbound_service,
        Network::new_default_testnet(),
        None,
        None,
    )
    .await;
}

/// Test zakura-network writes a peer cache file, and can read it back manually.
#[tokio::test]
async fn written_peer_cache_can_be_read_manually() {
    let _init_guard = zakura_test::init();

    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    let nil_inbound_service = service_fn(|_| async { Ok(Response::Nil) });

    let cache_dir = tempfile::tempdir().expect("temporary peer cache directory creation failed");
    let config = Config {
        initial_mainnet_peers: Default::default(),
        cache_dir: CacheDir::custom_path(cache_dir.path()),
        ..Config::default()
    };
    let address_book =
        init_with_peer_limit(1, nil_inbound_service, Mainnet, None, config.clone()).await;
    let expected_cached_peer = add_cacheable_peer(&address_book);

    update_peer_cache_once(&config, &address_book)
        .await
        .expect("writing the peer cache should succeed");

    let cached_peers = config
        .load_peer_cache()
        .await
        .expect("unexpected error reading peer cache");

    let peer_cache_file = config
        .cache_dir
        .peer_cache_file_path(&config.network)
        .expect("test cache directory is enabled");
    let disk_entry = tokio::fs::read_to_string(peer_cache_file)
        .await
        .expect("written peer cache is readable");
    assert_eq!(
        disk_entry.trim(),
        expected_cached_peer
            .remove_socket_addr_privacy()
            .to_string(),
        "peer cache entries must stay unredacted so they remain reconnectable"
    );

    assert!(
        cached_peers.contains(&expected_cached_peer),
        "expected peer missing from manual cache load: {:?}",
        config.cache_dir.peer_cache_file_path(&config.network)
    );
}

/// Test zakura-network writes a peer cache file, and reads it back automatically.
#[tokio::test]
async fn written_peer_cache_is_automatically_read_on_startup() {
    let _init_guard = zakura_test::init();

    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    let nil_inbound_service = service_fn(|_| async { Ok(Response::Nil) });

    let cache_dir = tempfile::tempdir().expect("temporary peer cache directory creation failed");
    let config = Config {
        initial_mainnet_peers: Default::default(),
        cache_dir: CacheDir::custom_path(cache_dir.path()),
        ..Config::default()
    };
    let address_book =
        init_with_peer_limit(1, nil_inbound_service, Mainnet, None, config.clone()).await;
    let expected_cached_peer = add_cacheable_peer(&address_book);

    update_peer_cache_once(&config, &address_book)
        .await
        .expect("writing the peer cache should succeed");

    let address_book =
        init_with_peer_limit(1, nil_inbound_service, Mainnet, None, config.clone()).await;

    tokio::time::timeout(PEER_CACHE_TEST_TIMEOUT, async {
        loop {
            if address_book
                .lock()
                .expect("previous thread panicked while holding address book lock")
                .get(expected_cached_peer)
                .is_some()
            {
                break;
            }

            tokio::task::yield_now().await;
        }
    })
    .await
    .unwrap_or_else(|_| {
        panic!(
            "expected peer missing from address book loaded from cache: {:?}",
            config.cache_dir.peer_cache_file_path(&config.network)
        )
    });
}

/// Adds a recent gossiped peer that is eligible for the disk cache.
fn add_cacheable_peer(address_book: &Arc<std::sync::Mutex<AddressBook>>) -> PeerSocketAddr {
    let peer = "127.0.0.1:8233".parse().expect("valid test peer address");
    let change =
        MetaAddr::new_gossiped_meta_addr(peer, PeerServices::NODE_NETWORK, DateTime32::now())
            .new_gossiped_change()
            .expect("recent gossiped peer creates an address book change");

    address_book
        .lock()
        .expect("previous thread panicked while holding address book lock")
        .update(change)
        .expect("test peer is valid for the mainnet address book");

    peer
}

#[test]
fn crawler_replenishes_outbound_peers_below_eighty_percent_of_target_size() {
    // (active connections, peer set target size, outbound connection limit, demand)
    let cases = [
        (0, 0, 0, 0),
        (0, 1, 100, 1),
        (1, 1, 100, 0),
        (0, 2, 100, 2),
        (2, 2, 100, 0),
        (0, 3, 100, 3),
        (3, 3, 100, 0),
        (0, 4, 100, 4),
        (1, 4, 100, 3),
        (4, 4, 100, 0),
        (0, 10, 100, 8),
        (2, 10, 100, 6),
        (8, 10, 100, 0),
        (0, 100, 150, 80),
        (34, 100, 150, 46),
        (79, 100, 150, 1),
        (80, 100, 150, 0),
        (100, 100, 150, 0),
    ];

    for (active, target_size, limit, expected) in cases {
        assert_eq!(
            outbound_peer_replenishment_demand(active, target_size, limit),
            expected,
            "unexpected replenishment decision for {active} active peers, \
             target size {target_size}, and limit {limit}",
        );
    }
}

/// The replenishment target must depend on the configured peer set size alone.
///
/// Keying it to the outbound connection limit made halving that limit move the
/// steady-state outbound target from 81 to 41, cutting the per-crawl dial budget
/// roughly sevenfold on a node holding 34 outbound peers.
#[test]
fn crawler_replenishment_ignores_outbound_connection_limit_changes() {
    const PEERSET_TARGET_SIZE: usize = 100;
    const HALVED_OUTBOUND_LIMIT: usize = PEERSET_TARGET_SIZE * 3 / 2;
    const ORIGINAL_OUTBOUND_LIMIT: usize = PEERSET_TARGET_SIZE * 3;

    for active in [0, 1, 34, 40, 41, 79, 80, 100] {
        assert_eq!(
            outbound_peer_replenishment_demand(
                active,
                PEERSET_TARGET_SIZE,
                ORIGINAL_OUTBOUND_LIMIT,
            ),
            outbound_peer_replenishment_demand(active, PEERSET_TARGET_SIZE, HALVED_OUTBOUND_LIMIT),
            "changing only the outbound connection limit moved the replenishment \
             demand for {active} active peers",
        );
    }
}

/// An operator who lowers the outbound connection limit below the replenishment
/// target must not get a crawler that dials forever against the ceiling.
#[test]
fn crawler_replenishment_is_bounded_by_the_outbound_connection_limit() {
    let cases = [
        (0, 100, 50, 50),
        (25, 100, 50, 25),
        (50, 100, 50, 0),
        (0, 100, 0, 0),
        (0, usize::MAX, 10, 10),
    ];

    for (active, target_size, limit, expected) in cases {
        assert_eq!(
            outbound_peer_replenishment_demand(active, target_size, limit),
            expected,
            "replenishment demand exceeded the outbound connection limit for \
             {active} active peers, target size {target_size}, and limit {limit}",
        );
    }
}

#[tokio::test(start_paused = true)]
async fn crawler_startup_demand_satisfies_replenishment() {
    let mut harness = spawn_replenishment_crawler(
        0,
        CRAWLER_REPLENISHMENT_TARGET_SIZE_FOR_TESTS,
        constants::DEFAULT_CRAWL_NEW_PEER_INTERVAL,
    )
    .await;

    harness.wait_for_crawl_start().await;
    harness.release_crawl();
    harness
        .assert_connection_attempts(CRAWLER_REPLENISHMENT_TARGET_SIZE_FOR_TESTS)
        .await;
}

#[tokio::test(start_paused = true)]
async fn crawler_partial_queued_demand_is_completed_to_target() {
    let mut harness =
        spawn_replenishment_crawler(0, 3, constants::DEFAULT_CRAWL_NEW_PEER_INTERVAL).await;

    harness.wait_for_crawl_start().await;
    harness.release_crawl();
    harness
        .assert_connection_attempts(CRAWLER_REPLENISHMENT_CONNECTION_TARGET_FOR_TESTS)
        .await;
}

#[tokio::test(start_paused = true)]
async fn crawler_demand_arriving_during_crawl_counts_toward_target() {
    let mut harness =
        spawn_replenishment_crawler(0, 0, constants::DEFAULT_CRAWL_NEW_PEER_INTERVAL).await;

    harness.wait_for_crawl_start().await;
    harness.queue_demand(4);
    harness.release_crawl();
    harness
        .assert_connection_attempts(CRAWLER_REPLENISHMENT_CONNECTION_TARGET_FOR_TESTS)
        .await;
}

#[tokio::test(start_paused = true)]
async fn crawler_disconnects_during_crawl_are_replenished() {
    let mut harness = spawn_replenishment_crawler(
        CRAWLER_REPLENISHMENT_CONNECTION_TARGET_FOR_TESTS,
        0,
        constants::DEFAULT_CRAWL_NEW_PEER_INTERVAL,
    )
    .await;

    harness.wait_for_crawl_start().await;
    harness.drop_initial_connections(4);
    harness.release_crawl();
    harness.assert_connection_attempts(4).await;
}

#[tokio::test(start_paused = true)]
async fn crawler_overlapping_timer_crawls_do_not_duplicate_replenishment() {
    let mut harness = spawn_replenishment_crawler(0, 0, Duration::from_secs(1)).await;

    harness.wait_for_crawl_start().await;
    tokio::time::advance(Duration::from_secs(1)).await;
    tokio::task::yield_now().await;
    harness.release_crawl();
    harness
        .assert_connection_attempts(CRAWLER_REPLENISHMENT_CONNECTION_TARGET_FOR_TESTS)
        .await;
}

#[tokio::test(start_paused = true)]
async fn crawler_excess_queued_demand_is_preserved() {
    let mut harness =
        spawn_replenishment_crawler(0, 12, constants::DEFAULT_CRAWL_NEW_PEER_INTERVAL).await;

    harness.wait_for_crawl_start().await;
    harness.release_crawl();
    harness.assert_connection_attempts(12).await;
}

#[tokio::test(start_paused = true)]
async fn crawler_replenishment_respects_hard_outbound_limit() {
    let mut harness =
        spawn_replenishment_crawler(0, 40, constants::DEFAULT_CRAWL_NEW_PEER_INTERVAL).await;

    harness.wait_for_crawl_start().await;
    harness.release_crawl();
    harness
        .assert_connection_attempts(CRAWLER_REPLENISHMENT_OUTBOUND_LIMIT_FOR_TESTS)
        .await;
}

/// Failed dials restore channel demand and must not permanently consume the
/// timer replenishment budget. Without crediting, `failed_dials` attempts would
/// shrink the successful connection count below the target.
#[tokio::test(start_paused = true)]
async fn crawler_failed_dials_do_not_shrink_replenishment_budget() {
    const FAILED_DIALS: usize = 3;

    let mut harness = spawn_replenishment_crawler_with(
        0,
        0,
        constants::DEFAULT_CRAWL_NEW_PEER_INTERVAL,
        ReplenishmentCrawlerOptions {
            fail_first_dials: FAILED_DIALS,
            ..ReplenishmentCrawlerOptions::default()
        },
    )
    .await;

    harness.wait_for_crawl_start().await;
    harness.release_crawl();
    harness
        .assert_connection_attempts(
            CRAWLER_REPLENISHMENT_CONNECTION_TARGET_FOR_TESTS + FAILED_DIALS,
        )
        .await;
}

/// When demand cannot be restored (full channel), failed dials must not credit
/// replenishment budget. Crediting without a queued `MorePeers` would leave
/// `remaining` unchanged and keep local replenishment dialing past the target.
#[tokio::test(start_paused = true)]
async fn crawler_unrestored_failed_dials_do_not_credit_replenishment() {
    let mut harness = spawn_replenishment_crawler_with(
        0,
        0,
        constants::DEFAULT_CRAWL_NEW_PEER_INTERVAL,
        ReplenishmentCrawlerOptions {
            fail_first_dials: CRAWLER_REPLENISHMENT_CONNECTION_TARGET_FOR_TESTS,
            force_failed_dial_demand_restore_full: true,
        },
    )
    .await;

    harness.wait_for_crawl_start().await;
    harness.release_crawl();
    harness
        .assert_connection_attempts(CRAWLER_REPLENISHMENT_CONNECTION_TARGET_FOR_TESTS)
        .await;
}

#[test]
fn try_restore_demand_after_failed_dial_reports_full_channel() {
    // futures::mpsc capacity is buffer + number of senders, so buffer 0 still
    // has one sender slot. Filling that slot makes the next try_send Full.
    let (mut demand_tx, demand_rx) = mpsc::channel::<MorePeers>(0);
    demand_tx
        .try_send(MorePeers)
        .expect("test demand channel accepts the sender's reserved slot");

    let demand_restored =
        super::super::try_restore_demand_after_failed_dial(&mut demand_tx, &AtomicBool::new(false))
            .expect("full channel is not a fatal restore error");
    assert!(
        !demand_restored,
        "a full demand channel must not report a successful restore"
    );

    std::mem::drop(demand_rx);
}

#[test]
fn try_restore_demand_after_failed_dial_reports_success() {
    let (mut demand_tx, demand_rx) = mpsc::channel::<MorePeers>(1);

    let demand_restored =
        super::super::try_restore_demand_after_failed_dial(&mut demand_tx, &AtomicBool::new(false))
            .expect("empty channel accepts restored demand");
    assert!(
        demand_restored,
        "an empty demand channel must report a successful restore"
    );

    std::mem::drop(demand_rx);
}

#[test]
fn try_restore_demand_after_failed_dial_honors_force_full() {
    let (mut demand_tx, demand_rx) = mpsc::channel::<MorePeers>(1);

    let demand_restored =
        super::super::try_restore_demand_after_failed_dial(&mut demand_tx, &AtomicBool::new(true))
            .expect("forced full channel is not a fatal restore error");
    assert!(
        !demand_restored,
        "forced full-channel mode must not report a successful restore"
    );

    std::mem::drop(demand_rx);
}

/// Test the crawler with an outbound peer limit of zero peers, and a connector that panics.
#[tokio::test(start_paused = true)]
async fn crawler_peer_limit_zero_connect_panic() {
    let _init_guard = zakura_test::init();

    // This test does not require network access, because the outbound connector
    // and peer set are fake.

    let unreachable_outbound_connector = service_fn(|_| async {
        unreachable!("outbound connector should never be called with a zero peer limit")
    });

    let (_config, discovered_peers) = spawn_crawler_with_peer_limit(
        0,
        ExpectedCrawlerConnections::None,
        unreachable_outbound_connector,
    )
    .await;

    assert!(
        discovered_peers.is_empty(),
        "unexpected peer when outbound limit is zero: {discovered_peers:?}",
    );
}

/// Test the crawler with an outbound peer limit of one peer, and a connector that always errors.
#[tokio::test(start_paused = true)]
async fn crawler_peer_limit_one_connect_error() {
    let _init_guard = zakura_test::init();

    // This test does not require network access, because the outbound connector
    // and peer set are fake.

    let error_outbound_connector =
        service_fn(|_| async { Err("test outbound connector always returns errors".into()) });

    let (_config, discovered_peers) = spawn_crawler_with_peer_limit(
        1,
        ExpectedCrawlerConnections::OverLimit,
        error_outbound_connector,
    )
    .await;

    assert!(
        discovered_peers.is_empty(),
        "unexpected peer when all connections error: {discovered_peers:?}",
    );
}

/// Test the crawler with an outbound peer limit of one peer,
/// and a connector that returns success then disconnects the peer.
#[tokio::test(start_paused = true)]
async fn crawler_peer_limit_one_connect_ok_then_drop() {
    let _init_guard = zakura_test::init();

    // This test does not require network access, because the outbound connector
    // and peer set are fake.

    let success_disconnect_outbound_connector =
        service_fn(|req: OutboundConnectorRequest| async move {
            let OutboundConnectorRequest {
                addr,
                connection_tracker,
            } = req;

            let (fake_client, _harness) = ClientTestHarness::build().finish();

            // Fake the connection closing.
            std::mem::drop(connection_tracker);

            // Give the crawler time to get the message.
            tokio::task::yield_now().await;

            Ok((addr, fake_client))
        });

    let (config, discovered_peers) = spawn_crawler_with_peer_limit(
        1,
        ExpectedCrawlerConnections::OverLimit,
        success_disconnect_outbound_connector,
    )
    .await;

    let peer_count = discovered_peers.len();

    assert!(
        peer_count > config.peerset_outbound_connection_limit(),
        "unexpected number of peer connections {}, should be at least the limit of {}",
        peer_count,
        config.peerset_outbound_connection_limit(),
    );
}

/// Test the crawler with an outbound peer limit of one peer,
/// and a connector that returns success then holds the peer open.
#[tokio::test(start_paused = true)]
async fn crawler_peer_limit_one_connect_ok_stay_open() {
    let _init_guard = zakura_test::init();

    // This test does not require network access, because the outbound connector
    // and peer set are fake.

    let (peer_tracker_tx, mut peer_tracker_rx) = mpsc::unbounded();

    let success_stay_open_outbound_connector = service_fn(move |req: OutboundConnectorRequest| {
        let peer_tracker_tx = peer_tracker_tx.clone();
        async move {
            let OutboundConnectorRequest {
                addr,
                connection_tracker,
            } = req;

            let (fake_client, _harness) = ClientTestHarness::build().finish();

            // Make the connection staying open.
            peer_tracker_tx
                .unbounded_send(connection_tracker)
                .expect("unexpected error sending to unbounded channel");

            Ok((addr, fake_client))
        }
    });

    let (config, discovered_peers) = spawn_crawler_with_peer_limit(
        1,
        ExpectedCrawlerConnections::AtLimit,
        success_stay_open_outbound_connector,
    )
    .await;

    let peer_change_count = discovered_peers.len();

    let mut peer_tracker_count: usize = 0;
    loop {
        let peer_tracker_result = peer_tracker_rx.try_recv();
        match peer_tracker_result {
            // We held this peer tracker open until now.
            Ok(peer_tracker) => {
                std::mem::drop(peer_tracker);
                peer_tracker_count += 1;
            }

            // The channel is closed or there are no messages left in the channel.
            Err(_) => break,
        }
    }

    assert!(
        peer_change_count <= config.peerset_outbound_connection_limit(),
        "unexpected number of peer changes {}, over limit of {}, had {} peer trackers",
        peer_change_count,
        config.peerset_outbound_connection_limit(),
        peer_tracker_count,
    );

    assert!(
        peer_tracker_count <= config.peerset_outbound_connection_limit(),
        "unexpected number of peer trackers {}, over limit of {}, had {} peer changes",
        peer_tracker_count,
        config.peerset_outbound_connection_limit(),
        peer_change_count,
    );
}

/// Test the crawler with a multi-peer outbound limit, and a connector that always errors.
#[tokio::test(start_paused = true)]
async fn crawler_peer_limit_many_connect_error() {
    let _init_guard = zakura_test::init();

    // This test does not require network access, because the outbound connector
    // and peer set are fake.

    let error_outbound_connector =
        service_fn(|_| async { Err("test outbound connector always returns errors".into()) });

    let (_config, discovered_peers) = spawn_crawler_with_peer_limit(
        CRAWLER_MANY_PEER_LIMIT_FOR_TESTS,
        ExpectedCrawlerConnections::OverLimit,
        error_outbound_connector,
    )
    .await;

    assert!(
        discovered_peers.is_empty(),
        "unexpected peer when all connections error: {discovered_peers:?}",
    );
}

/// Test the crawler with a multi-peer outbound limit,
/// and a connector that returns success then disconnects the peer.
#[tokio::test(start_paused = true)]
async fn crawler_peer_limit_many_connect_ok_then_drop() {
    let _init_guard = zakura_test::init();

    // This test does not require network access, because the outbound connector
    // and peer set are fake.

    let success_disconnect_outbound_connector =
        service_fn(|req: OutboundConnectorRequest| async move {
            let OutboundConnectorRequest {
                addr,
                connection_tracker,
            } = req;

            let (fake_client, _harness) = ClientTestHarness::build().finish();

            // Fake the connection closing.
            std::mem::drop(connection_tracker);

            // Give the crawler time to get the message.
            tokio::task::yield_now().await;

            Ok((addr, fake_client))
        });

    let (config, discovered_peers) = spawn_crawler_with_peer_limit(
        CRAWLER_MANY_PEER_LIMIT_FOR_TESTS,
        ExpectedCrawlerConnections::OverLimit,
        success_disconnect_outbound_connector,
    )
    .await;

    let peer_count = discovered_peers.len();

    assert!(
        peer_count > config.peerset_outbound_connection_limit(),
        "unexpected number of peer connections {}, should be over the limit of {}",
        peer_count,
        config.peerset_outbound_connection_limit(),
    );
}

/// Test the crawler with a multi-peer outbound limit,
/// and a connector that returns success then holds the peer open.
#[tokio::test(start_paused = true)]
async fn crawler_peer_limit_many_connect_ok_stay_open() {
    let _init_guard = zakura_test::init();

    // This test does not require network access, because the outbound connector
    // and peer set are fake.

    let (peer_tracker_tx, mut peer_tracker_rx) = mpsc::unbounded();

    let success_stay_open_outbound_connector = service_fn(move |req: OutboundConnectorRequest| {
        let peer_tracker_tx = peer_tracker_tx.clone();
        async move {
            let OutboundConnectorRequest {
                addr,
                connection_tracker,
            } = req;

            let (fake_client, _harness) = ClientTestHarness::build().finish();

            // Make the connection staying open.
            peer_tracker_tx
                .unbounded_send(connection_tracker)
                .expect("unexpected error sending to unbounded channel");

            Ok((addr, fake_client))
        }
    });

    let (config, discovered_peers) = spawn_crawler_with_peer_limit(
        CRAWLER_MANY_PEER_LIMIT_FOR_TESTS,
        ExpectedCrawlerConnections::AtLimit,
        success_stay_open_outbound_connector,
    )
    .await;

    let peer_change_count = discovered_peers.len();

    let mut peer_tracker_count: usize = 0;
    loop {
        let peer_tracker_result = peer_tracker_rx.try_recv();
        match peer_tracker_result {
            // We held this peer tracker open until now.
            Ok(peer_tracker) => {
                std::mem::drop(peer_tracker);
                peer_tracker_count += 1;
            }

            // The channel is closed or there are no messages left in the channel.
            Err(_) => break,
        }
    }

    assert!(
        peer_change_count <= config.peerset_outbound_connection_limit(),
        "unexpected number of peer changes {}, over limit of {}, had {} peer trackers",
        peer_change_count,
        config.peerset_outbound_connection_limit(),
        peer_tracker_count,
    );

    assert!(
        peer_tracker_count <= config.peerset_outbound_connection_limit(),
        "unexpected number of peer trackers {}, over limit of {}, had {} peer changes",
        peer_tracker_count,
        config.peerset_outbound_connection_limit(),
        peer_change_count,
    );
}

/// Test the listener with an inbound peer limit of zero peers, and a handshaker that panics.
#[tokio::test]
async fn listener_peer_limit_zero_handshake_panic() {
    let _init_guard = zakura_test::init();

    // This test requires an IPv4 network stack with 127.0.0.1 as localhost.
    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    let unreachable_inbound_handshaker = service_fn(|_| async {
        unreachable!("inbound handshaker should never be called with a zero peer limit")
    });

    let (_config, discovered_peers) =
        spawn_inbound_listener_with_peer_limit(0, None, unreachable_inbound_handshaker).await;

    assert!(
        discovered_peers.is_empty(),
        "unexpected peer when inbound limit is zero: {discovered_peers:?}",
    );
}

/// Test the listener with an inbound peer limit of one peer, and a handshaker that always errors.
#[tokio::test]
async fn listener_peer_limit_one_handshake_error() {
    let _init_guard = zakura_test::init();

    // This test requires an IPv4 network stack with 127.0.0.1 as localhost.
    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    let error_inbound_handshaker =
        service_fn(|_| async { Err("test inbound handshaker always returns errors".into()) });

    let (_config, discovered_peers) =
        spawn_inbound_listener_with_peer_limit(1, None, error_inbound_handshaker).await;

    assert!(
        discovered_peers.is_empty(),
        "unexpected peer when all handshakes error: {discovered_peers:?}",
    );
}

/// Test the listener with an inbound peer limit of one peer,
/// and a handshaker that returns success then disconnects the peer.
#[tokio::test]
async fn listener_peer_limit_one_handshake_ok_then_drop() {
    let _init_guard = zakura_test::init();

    // This test requires an IPv4 network stack with 127.0.0.1 as localhost.
    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    let success_disconnect_inbound_handshaker =
        service_fn(|req: HandshakeRequest<TcpStream>| async move {
            let HandshakeRequest {
                data_stream: tcp_stream,
                connected_addr: _,
                connection_tracker,
            } = req;

            let (fake_client, _harness) = ClientTestHarness::build().finish();

            // Actually close the connection.
            std::mem::drop(connection_tracker);
            std::mem::drop(tcp_stream);

            // Give the crawler time to get the message.
            tokio::task::yield_now().await;

            Ok(fake_client)
        });

    let (config, discovered_peers) = spawn_inbound_listener_with_peer_limit(
        1,
        usize::MAX,
        success_disconnect_inbound_handshaker,
    )
    .await;

    let peer_count = discovered_peers.len();

    assert!(
        peer_count > config.peerset_inbound_connection_limit(),
        "unexpected number of peer connections {}, should be over the limit of {}",
        peer_count,
        config.peerset_inbound_connection_limit(),
    );
}

/// Test the listener with an inbound peer limit of one peer,
/// and a handshaker that returns success then holds the peer open.
#[tokio::test]
async fn listener_peer_limit_one_handshake_ok_stay_open() {
    let _init_guard = zakura_test::init();

    // This test requires an IPv4 network stack with 127.0.0.1 as localhost.
    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    let (peer_tracker_tx, mut peer_tracker_rx) = mpsc::unbounded();

    let success_stay_open_inbound_handshaker =
        service_fn(move |req: HandshakeRequest<TcpStream>| {
            let peer_tracker_tx = peer_tracker_tx.clone();
            async move {
                let HandshakeRequest {
                    data_stream: tcp_stream,
                    connected_addr: _,
                    connection_tracker,
                } = req;

                let (fake_client, _harness) = ClientTestHarness::build().finish();

                // Make the connection staying open.
                peer_tracker_tx
                    .unbounded_send((tcp_stream, connection_tracker))
                    .expect("unexpected error sending to unbounded channel");

                Ok(fake_client)
            }
        });

    let (config, discovered_peers) =
        spawn_inbound_listener_with_peer_limit(1, None, success_stay_open_inbound_handshaker).await;

    let peer_change_count = discovered_peers.len();

    let mut peer_tracker_count: usize = 0;
    loop {
        let peer_tracker_result = peer_tracker_rx.try_recv();
        match peer_tracker_result {
            // We held this peer connection and tracker open until now.
            Ok((peer_connection, peer_tracker)) => {
                std::mem::drop(peer_connection);
                std::mem::drop(peer_tracker);
                peer_tracker_count += 1;
            }

            // The channel is closed or there are no messages left in the channel.
            Err(_) => break,
        }
    }

    assert!(
        peer_change_count <= config.peerset_inbound_connection_limit(),
        "unexpected number of peer changes {}, over limit of {}, had {} peer trackers",
        peer_change_count,
        config.peerset_inbound_connection_limit(),
        peer_tracker_count,
    );

    assert!(
        peer_tracker_count <= config.peerset_inbound_connection_limit(),
        "unexpected number of peer trackers {}, over limit of {}, had {} peer changes",
        peer_tracker_count,
        config.peerset_inbound_connection_limit(),
        peer_change_count,
    );
}

/// Test the listener reserves one inbound slot for a configured zcashd-compat peer,
/// matching native and IPv4-mapped address representations canonically.
#[tokio::test]
async fn listener_reserves_one_zcashd_compat_inbound_slot() {
    let _init_guard = zakura_test::init();

    // This test requires an IPv4 network stack with 127.0.0.1 as localhost.
    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    let public_ip = Ipv4Addr::LOCALHOST;
    let zcashd_compat_ip = Ipv4Addr::new(127, 0, 0, 2);
    let (peer_tracker_tx, mut peer_tracker_rx) = mpsc::unbounded();

    let success_stay_open_inbound_handshaker =
        service_fn(move |req: HandshakeRequest<TcpStream>| {
            let peer_tracker_tx = peer_tracker_tx.clone();
            async move {
                let HandshakeRequest {
                    data_stream: tcp_stream,
                    connected_addr,
                    connection_tracker,
                } = req;

                let ConnectedAddr::InboundDirect { addr } = connected_addr else {
                    unreachable!("inbound listener handshakes use inbound direct addresses");
                };

                let (fake_client, _harness) = ClientTestHarness::build()
                    .with_connected_addr(connected_addr)
                    .finish();

                peer_tracker_tx
                    .unbounded_send((addr.ip(), tcp_stream, connection_tracker))
                    .expect("unexpected error sending to unbounded channel");

                Ok(fake_client)
            }
        });

    // The smallest peer set target size that still leaves public inbound slots
    // next to the reserved zcashd-compat slot.
    let mut config = Config {
        listen_addr: "127.0.0.1:0".parse().unwrap(),
        peerset_initial_target_size: 1,
        max_connections_per_ip: usize::MAX,
        ..Config::default()
    };
    let public_inbound_limit = config.peerset_inbound_connection_limit() - 1;
    let (tcp_listener, listen_addr) = open_listener(&config.clone()).await;
    config.listen_addr = listen_addr;

    let (peerset_tx, mut peerset_rx) =
        mpsc::channel::<DiscoveredPeer>(config.peerset_inbound_connection_limit() * 2);
    let bans = BannedIps::default();

    let listen_fut = accept_inbound_connections(
        config.clone(),
        tcp_listener,
        MIN_INBOUND_PEER_CONNECTION_INTERVAL_FOR_TESTS,
        success_stay_open_inbound_handshaker,
        peerset_tx,
        bans,
        vec![IpAddr::V6(zcashd_compat_ip.to_ipv6_mapped())],
    );
    let listen_task_handle = tokio::spawn(listen_fut);

    // Fill every public slot, then offer one more connection of each kind than
    // the listener can accept.
    let mut public_connections = Vec::new();
    for _ in 0..public_inbound_limit {
        public_connections.push(connect_from(public_ip, listen_addr).await);
    }
    let rejected_public_connection = connect_from(public_ip, listen_addr).await;
    let zcashd_compat_connection = connect_from(zcashd_compat_ip, listen_addr).await;
    let rejected_zcashd_compat_connection = connect_from(zcashd_compat_ip, listen_addr).await;

    tokio::time::sleep(ZCASHD_COMPAT_LISTENER_TEST_DURATION).await;

    listen_task_handle.abort();
    tokio::task::yield_now().await;
    assert_listener_task_cancelled(listen_task_handle);

    let accepted_ips = drain_accepted_inbound_ips(&mut peer_tracker_rx);
    let peer_change_count = drain_discovered_peers(&mut peerset_rx);

    assert_eq!(
        peer_change_count,
        config.peerset_inbound_connection_limit(),
        "accepted connections must stay within the total inbound limit"
    );
    assert_eq!(
        accepted_ips.iter().filter(|ip| **ip == public_ip).count(),
        public_inbound_limit,
        "ordinary inbound peers should only use the public slots"
    );
    assert_eq!(
        accepted_ips
            .iter()
            .filter(|ip| **ip == zcashd_compat_ip)
            .count(),
        1,
        "the canonically matched zcashd-compat peer should use the reserved slot"
    );

    std::mem::drop(public_connections);
    std::mem::drop(rejected_public_connection);
    std::mem::drop(zcashd_compat_connection);
    std::mem::drop(rejected_zcashd_compat_connection);
}

/// Test the listener does not add zcashd-compat peers to the recent-IP throttle.
#[tokio::test]
async fn listener_zcashd_compat_reconnect_bypasses_recent_ip_limit() {
    let _init_guard = zakura_test::init();

    // This test requires an IPv4 network stack with 127.0.0.1 as localhost.
    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    let zcashd_compat_ip = Ipv4Addr::new(127, 0, 0, 2);
    let success_disconnect_inbound_handshaker =
        service_fn(|req: HandshakeRequest<TcpStream>| async move {
            let HandshakeRequest {
                data_stream: tcp_stream,
                connected_addr,
                connection_tracker,
            } = req;

            let (fake_client, _harness) = ClientTestHarness::build()
                .with_connected_addr(connected_addr)
                .finish();

            std::mem::drop(connection_tracker);
            std::mem::drop(tcp_stream);
            tokio::task::yield_now().await;

            Ok(fake_client)
        });

    let mut config = Config {
        listen_addr: "127.0.0.1:0".parse().unwrap(),
        peerset_initial_target_size: 1,
        max_connections_per_ip: 1,
        ..Config::default()
    };
    let (tcp_listener, listen_addr) = open_listener(&config.clone()).await;
    config.listen_addr = listen_addr;

    let (peerset_tx, mut peerset_rx) = mpsc::channel::<DiscoveredPeer>(2);
    let bans = BannedIps::default();

    let listen_fut = accept_inbound_connections(
        config,
        tcp_listener,
        MIN_INBOUND_PEER_CONNECTION_INTERVAL_FOR_TESTS,
        success_disconnect_inbound_handshaker,
        peerset_tx,
        bans,
        vec![zcashd_compat_ip.into()],
    );
    let listen_task_handle = tokio::spawn(listen_fut);

    let first_connection = connect_from(zcashd_compat_ip, listen_addr).await;
    tokio::time::sleep(MIN_INBOUND_PEER_CONNECTION_INTERVAL_FOR_TESTS * 2).await;
    let second_connection = connect_from(zcashd_compat_ip, listen_addr).await;

    tokio::time::sleep(ZCASHD_COMPAT_LISTENER_TEST_DURATION).await;

    listen_task_handle.abort();
    tokio::task::yield_now().await;
    assert_listener_task_cancelled(listen_task_handle);

    assert_eq!(
        drain_discovered_peers(&mut peerset_rx),
        2,
        "rapid zcashd-compat reconnects should bypass the recent-IP throttle"
    );

    std::mem::drop(first_connection);
    std::mem::drop(second_connection);
}

/// Test configured zcashd-compat reconnects bypass the normal
/// successful-handshake delay.
#[tokio::test]
async fn listener_zcashd_compat_accept_bypasses_success_delay() {
    let _init_guard = zakura_test::init();

    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    let zcashd_compat_ip = Ipv4Addr::LOCALHOST;
    let (handshake_finished_tx, mut handshake_finished_rx) = tokio::sync::mpsc::unbounded_channel();
    let success_disconnect_inbound_handshaker =
        service_fn(move |req: HandshakeRequest<TcpStream>| {
            let handshake_finished_tx = handshake_finished_tx.clone();
            async move {
                let HandshakeRequest {
                    data_stream: tcp_stream,
                    connected_addr,
                    connection_tracker,
                } = req;
                let (fake_client, _harness) = ClientTestHarness::build()
                    .with_connected_addr(connected_addr)
                    .finish();
                handshake_finished_tx
                    .send(())
                    .expect("test receiver should remain open");
                std::mem::drop(connection_tracker);
                std::mem::drop(tcp_stream);

                Ok(fake_client)
            }
        });

    let mut config = Config {
        listen_addr: "127.0.0.1:0".parse().unwrap(),
        peerset_initial_target_size: 2,
        max_connections_per_ip: usize::MAX,
        ..Config::default()
    };
    let (tcp_listener, listen_addr) = open_listener(&config.clone()).await;
    config.listen_addr = listen_addr;
    let (peerset_tx, _peerset_rx) = mpsc::channel::<DiscoveredPeer>(2);

    // Queue two sidecar connections before starting the accept loop. The first
    // handshake must not impose the normal one-second delay on the reconnect.
    let first_connection = connect_from(zcashd_compat_ip, listen_addr).await;
    let second_connection = connect_from(zcashd_compat_ip, listen_addr).await;

    let listen_task_handle = tokio::spawn(accept_inbound_connections(
        config,
        tcp_listener,
        Duration::from_secs(1),
        success_disconnect_inbound_handshaker,
        peerset_tx,
        BannedIps::default(),
        vec![zcashd_compat_ip.into()],
    ));

    tokio::time::timeout(Duration::from_millis(250), async {
        for _ in 0..2 {
            handshake_finished_rx
                .recv()
                .await
                .expect("listener should finish both sidecar handshakes");
        }
    })
    .await
    .expect("configured sidecar reconnect should bypass the one-second delay");

    listen_task_handle.abort();
    tokio::task::yield_now().await;
    assert_listener_task_cancelled(listen_task_handle);
    std::mem::drop(first_connection);
    std::mem::drop(second_connection);
}

/// Test a queued burst of successful public handshakes is drained using short
/// pacing instead of applying the full success delay to every connection.
#[tokio::test]
async fn listener_drains_ready_accept_burst_with_short_pacing() {
    let _init_guard = zakura_test::init();

    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    const BURST_CONNECTIONS: usize = 8;

    let (handshake_finished_tx, mut handshake_finished_rx) = tokio::sync::mpsc::unbounded_channel();
    let success_disconnect_inbound_handshaker =
        service_fn(move |req: HandshakeRequest<TcpStream>| {
            let handshake_finished_tx = handshake_finished_tx.clone();
            async move {
                let HandshakeRequest {
                    data_stream: tcp_stream,
                    connected_addr,
                    connection_tracker,
                } = req;
                let (fake_client, _harness) = ClientTestHarness::build()
                    .with_connected_addr(connected_addr)
                    .finish();
                handshake_finished_tx
                    .send(())
                    .expect("test receiver should remain open");
                std::mem::drop(connection_tracker);
                std::mem::drop(tcp_stream);

                Ok(fake_client)
            }
        });

    let mut config = Config {
        listen_addr: "127.0.0.1:0".parse().unwrap(),
        peerset_initial_target_size: BURST_CONNECTIONS,
        max_connections_per_ip: usize::MAX,
        ..Config::default()
    };
    let (tcp_listener, listen_addr) = open_listener(&config.clone()).await;
    config.listen_addr = listen_addr;
    let (peerset_tx, _peerset_rx) = mpsc::channel::<DiscoveredPeer>(BURST_CONNECTIONS);

    let mut public_connections = Vec::new();
    for _ in 0..BURST_CONNECTIONS {
        public_connections.push(connect_from(Ipv4Addr::LOCALHOST, listen_addr).await);
    }

    let listen_task_handle = tokio::spawn(accept_inbound_connections(
        config,
        tcp_listener,
        Duration::from_secs(1),
        success_disconnect_inbound_handshaker,
        peerset_tx,
        BannedIps::default(),
        Vec::new(),
    ));

    tokio::time::timeout(Duration::from_millis(1_500), async {
        for _ in 0..BURST_CONNECTIONS {
            handshake_finished_rx
                .recv()
                .await
                .expect("listener should finish every queued handshake");
        }
    })
    .await
    .expect("ready accept burst should not incur one full delay per connection");

    listen_task_handle.abort();
    tokio::task::yield_now().await;
    assert_listener_task_cancelled(listen_task_handle);
    std::mem::drop(public_connections);
}

/// Test bans are applied before the zcashd-compat reserved slot.
#[tokio::test]
async fn listener_bans_zcashd_compat_peer_before_reserved_slot() {
    let _init_guard = zakura_test::init();

    // This test requires an IPv4 network stack with 127.0.0.1 as localhost.
    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    let zcashd_compat_ip = Ipv4Addr::new(127, 0, 0, 2);
    let unreachable_inbound_handshaker = service_fn(|_| async {
        unreachable!("banned zcashd-compat peer should never reach the handshaker")
    });

    let mut config = Config {
        listen_addr: "127.0.0.1:0".parse().unwrap(),
        peerset_initial_target_size: 1,
        ..Config::default()
    };
    let (tcp_listener, listen_addr) = open_listener(&config.clone()).await;
    config.listen_addr = listen_addr;

    let (peerset_tx, mut peerset_rx) = mpsc::channel::<DiscoveredPeer>(1);
    let bans = BannedIps::with_banned_ip(zcashd_compat_ip.into());

    let listen_fut = accept_inbound_connections(
        config,
        tcp_listener,
        MIN_INBOUND_PEER_CONNECTION_INTERVAL_FOR_TESTS,
        unreachable_inbound_handshaker,
        peerset_tx,
        bans,
        vec![zcashd_compat_ip.into()],
    );
    let listen_task_handle = tokio::spawn(listen_fut);

    let banned_connection = connect_from(zcashd_compat_ip, listen_addr).await;

    tokio::time::sleep(ZCASHD_COMPAT_LISTENER_TEST_DURATION).await;

    listen_task_handle.abort();
    tokio::task::yield_now().await;
    assert_listener_task_cancelled(listen_task_handle);

    assert_eq!(
        drain_discovered_peers(&mut peerset_rx),
        0,
        "banned zcashd-compat peers must not be admitted"
    );

    std::mem::drop(banned_connection);
}

/// Test the listener with a multi-peer inbound limit, and a handshaker that always errors.
#[tokio::test]
async fn listener_peer_limit_many_handshake_error() {
    let _init_guard = zakura_test::init();

    // This test requires an IPv4 network stack with 127.0.0.1 as localhost.
    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    let error_inbound_handshaker =
        service_fn(|_| async { Err("test inbound handshaker always returns errors".into()) });

    let (_config, discovered_peers) = spawn_inbound_listener_with_peer_limit(
        LISTENER_MANY_PEER_TARGET_SIZE_FOR_TESTS,
        None,
        error_inbound_handshaker,
    )
    .await;

    assert!(
        discovered_peers.is_empty(),
        "unexpected peer when all handshakes error: {discovered_peers:?}",
    );
}

/// Test the listener with a multi-peer inbound limit,
/// and a handshaker that returns success then disconnects the peer.
///
/// TODO: tweak the crawler timeouts and rate-limits so we get over the actual limit on macOS
///       (currently, getting over the limit can take 30 seconds or more)
#[cfg(not(target_os = "macos"))]
#[tokio::test]
async fn listener_peer_limit_many_handshake_ok_then_drop() {
    let _init_guard = zakura_test::init();

    // This test requires an IPv4 network stack with 127.0.0.1 as localhost.
    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    let success_disconnect_inbound_handshaker =
        service_fn(|req: HandshakeRequest<TcpStream>| async move {
            let HandshakeRequest {
                data_stream: tcp_stream,
                connected_addr: _,
                connection_tracker,
            } = req;

            let (fake_client, _harness) = ClientTestHarness::build().finish();

            // Actually close the connection.
            std::mem::drop(connection_tracker);
            std::mem::drop(tcp_stream);

            // Give the crawler time to get the message.
            tokio::task::yield_now().await;

            Ok(fake_client)
        });

    let (config, discovered_peers) = spawn_inbound_listener_with_peer_limit(
        LISTENER_MANY_PEER_TARGET_SIZE_FOR_TESTS,
        usize::MAX,
        success_disconnect_inbound_handshaker,
    )
    .await;

    let peer_count = discovered_peers.len();

    assert!(
        peer_count > config.peerset_inbound_connection_limit(),
        "unexpected number of peer connections {}, should be over the limit of {}",
        peer_count,
        config.peerset_inbound_connection_limit(),
    );
}

/// Test the listener with a multi-peer inbound limit,
/// and a handshaker that returns success then holds the peer open.
#[tokio::test]
async fn listener_peer_limit_many_handshake_ok_stay_open() {
    let _init_guard = zakura_test::init();

    // This test requires an IPv4 network stack with 127.0.0.1 as localhost.
    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    let (peer_tracker_tx, mut peer_tracker_rx) = mpsc::unbounded();

    let success_stay_open_inbound_handshaker =
        service_fn(move |req: HandshakeRequest<TcpStream>| {
            let peer_tracker_tx = peer_tracker_tx.clone();
            async move {
                let HandshakeRequest {
                    data_stream: tcp_stream,
                    connected_addr: _,
                    connection_tracker,
                } = req;

                let (fake_client, _harness) = ClientTestHarness::build().finish();

                // Make the connection staying open.
                peer_tracker_tx
                    .unbounded_send((tcp_stream, connection_tracker))
                    .expect("unexpected error sending to unbounded channel");

                Ok(fake_client)
            }
        });

    let (config, discovered_peers) = spawn_inbound_listener_with_peer_limit(
        LISTENER_MANY_PEER_TARGET_SIZE_FOR_TESTS,
        None,
        success_stay_open_inbound_handshaker,
    )
    .await;

    let peer_change_count = discovered_peers.len();

    let mut peer_tracker_count: usize = 0;
    loop {
        let peer_tracker_result = peer_tracker_rx.try_recv();
        match peer_tracker_result {
            // We held this peer connection and tracker open until now.
            Ok((peer_connection, peer_tracker)) => {
                std::mem::drop(peer_connection);
                std::mem::drop(peer_tracker);
                peer_tracker_count += 1;
            }

            // The channel is closed or there are no messages left in the channel.
            Err(_) => break,
        }
    }

    assert!(
        peer_change_count <= config.peerset_inbound_connection_limit(),
        "unexpected number of peer changes {}, over limit of {}, had {} peer trackers",
        peer_change_count,
        config.peerset_inbound_connection_limit(),
        peer_tracker_count,
    );

    assert!(
        peer_tracker_count <= config.peerset_inbound_connection_limit(),
        "unexpected number of peer trackers {}, over limit of {}, had {} peer changes",
        peer_tracker_count,
        config.peerset_inbound_connection_limit(),
        peer_change_count,
    );
}

/// Test if the initial seed peer connections is rate-limited.
#[tokio::test]
async fn add_initial_peers_is_rate_limited() {
    let _init_guard = zakura_test::init();

    // This test should not require network access.

    // We don't need to actually connect to the peers; we only need to check
    // if the connection attempts is rate-limited. Therefore, just return an error.
    let outbound_connector =
        service_fn(|_| async { Err("test outbound connector always returns errors".into()) });

    const PEER_COUNT: usize = 10;

    let before = Instant::now();

    let (initial_peers_task_handle, peerset_rx, address_book_updater_task_handle) =
        spawn_add_initial_peers(PEER_COUNT, outbound_connector).await;
    let connections = peerset_rx.take(PEER_COUNT).collect::<Vec<_>>().await;

    let elapsed = Instant::now() - before;

    // Errors are ignored, so we don't expect any peers here
    assert_eq!(connections.len(), 0);
    // Make sure the rate limiting worked by checking if it took long enough
    assert!(
        elapsed
            > constants::MIN_OUTBOUND_PEER_CONNECTION_INTERVAL
                .saturating_mul((PEER_COUNT - 1) as u32),
        "elapsed only {elapsed:?}"
    );

    let initial_peers_result = initial_peers_task_handle.await;
    assert!(
        matches!(initial_peers_result, Ok(Ok(_))),
        "unexpected error or panic in add_initial_peers task: {initial_peers_result:?}",
    );

    // Check for panics or errors in the address book updater task.
    let updater_result = address_book_updater_task_handle.now_or_never();
    assert!(
        updater_result.is_none()
            || matches!(updater_result, Some(Err(ref join_error)) if join_error.is_cancelled())
            // The task method only returns one kind of error.
            // We can't check for error equality due to type erasure,
            // and we can't downcast due to ownership.
            || matches!(updater_result, Some(Ok(Err(ref _all_senders_closed)))),
        "unexpected error or panic in address book updater task: {updater_result:?}",
    );
}

/// Test that self-connections fail.
//
// TODO:
// - add a unit test that makes sure the error is a nonce reuse error
// - add a unit test that makes sure connections that replay nonces also get rejected
#[tokio::test]
async fn self_connections_should_fail() {
    let _init_guard = zakura_test::init();

    // This test requires an IPv4 network stack with 127.0.0.1 as localhost.
    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    const TEST_PEERSET_INITIAL_TARGET_SIZE: usize = 3;
    const TEST_CRAWL_NEW_PEER_INTERVAL: Duration = Duration::from_secs(1);

    // If we get an inbound request from a peer, the test has a bug,
    // because self-connections should fail at the handshake stage.
    let unreachable_inbound_service =
        service_fn(|_| async { unreachable!("inbound service should never be called") });

    let force_listen_addr: SocketAddr = "127.0.0.1:0".parse().unwrap();

    let no_initial_peers_config = Config {
        crawl_new_peer_interval: TEST_CRAWL_NEW_PEER_INTERVAL,

        initial_mainnet_peers: IndexSet::new(),
        initial_testnet_peers: IndexSet::new(),
        cache_dir: CacheDir::disabled(),

        ..Config::default()
    };

    let address_book = init_with_peer_limit(
        TEST_PEERSET_INITIAL_TARGET_SIZE,
        unreachable_inbound_service,
        Mainnet,
        force_listen_addr,
        no_initial_peers_config,
    )
    .await;

    // Insert our own address into the address book, and make sure it works
    let (real_self_listener, updated_addr) = {
        let mut unlocked_address_book = address_book
            .lock()
            .expect("unexpected panic in address book");

        let real_self_listener = unlocked_address_book.local_listener_meta_addr(Utc::now());

        // Set a fake listener to get past the check for adding our own address
        unlocked_address_book.set_local_listener("192.168.0.0:1".parse().unwrap());

        let updated_addr = unlocked_address_book.update(
            real_self_listener
                .new_gossiped_change()
                .expect("change is valid"),
        );

        std::mem::drop(unlocked_address_book);

        (real_self_listener, updated_addr)
    };

    // Make sure we modified the address book correctly
    assert!(
        updated_addr.is_some(),
        "inserting our own address into the address book failed: {real_self_listener:?}"
    );
    assert_eq!(
        updated_addr.unwrap().addr(),
        real_self_listener.addr(),
        "wrong address inserted into address book"
    );
    assert_ne!(
        updated_addr.unwrap().addr().ip(),
        Ipv4Addr::UNSPECIFIED,
        "invalid address inserted into address book: ip must be valid for inbound connections"
    );
    assert_ne!(
        updated_addr.unwrap().addr().port(),
        0,
        "invalid address inserted into address book: port must be valid for inbound connections"
    );

    // Wait until the crawler has tried at least one self-connection
    tokio::time::sleep(TEST_CRAWL_NEW_PEER_INTERVAL * 3).await;

    // Check that the self-connection failed
    let self_connection_status = {
        let mut unlocked_address_book = address_book
            .lock()
            .expect("unexpected panic in address book");

        let self_connection_status = unlocked_address_book
            .get(real_self_listener.addr())
            .expect("unexpected dropped listener address in address book");

        std::mem::drop(unlocked_address_book);

        self_connection_status
    };

    // Make sure we fetched from the address book correctly
    assert_eq!(
        self_connection_status.addr(),
        real_self_listener.addr(),
        "wrong address fetched from address book"
    );

    // Make sure the self-connection failed
    assert_eq!(
        self_connection_status.last_connection_state,
        PeerAddrState::Failed,
        "self-connection should have failed"
    );
}

/// Test that the number of nonces is limited when peers send an invalid response or
/// if handshakes time out and are dropped.
#[tokio::test]
async fn remnant_nonces_from_outbound_connections_are_limited() {
    use tower::timeout::TimeoutLayer;

    let _init_guard = zakura_test::init();

    // This test requires an IPv4 network stack with 127.0.0.1 as localhost.
    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    const TEST_PEERSET_INITIAL_TARGET_SIZE: usize = 3;

    // Create a test config that listens on an unused port.
    let listen_addr = "127.0.0.1:0".parse().unwrap();
    let config = Config {
        listen_addr,
        peerset_initial_target_size: TEST_PEERSET_INITIAL_TARGET_SIZE,
        ..Config::default()
    };

    let hs_timeout = TimeoutLayer::new(constants::HANDSHAKE_TIMEOUT);
    let nil_inbound_service =
        tower::service_fn(|_req| async move { Ok::<Response, BoxError>(Response::Nil) });

    let hs = peer::Handshake::builder()
        .with_config(config.clone())
        .with_inbound_service(nil_inbound_service)
        .with_user_agent(("Test user agent").to_string())
        .with_latest_chain_tip(NoChainTip)
        .want_transactions(true)
        .finish()
        .expect("configured all required parameters");

    let mut outbound_connector = hs_timeout.layer(peer::Connector::new(hs.clone()));

    let mut active_outbound_connections = ActiveConnectionCounter::new_counter();

    let expected_max_nonces = config.peerset_total_connection_limit();
    let num_connection_attempts = 2 * expected_max_nonces;

    for i in 1..num_connection_attempts {
        let expected_nonce_count = expected_max_nonces.min(i);

        let (tcp_listener, addr) = open_listener(&config.clone()).await;

        tokio::spawn(async move {
            let (mut tcp_stream, _addr) = tcp_listener
                .accept()
                .await
                .expect("client connection should succeed");

            tcp_stream
                .shutdown()
                .await
                .expect("shutdown should succeed");
        });

        let outbound_connector = outbound_connector
            .ready()
            .await
            .expect("outbound connector never errors");

        let connection_tracker = active_outbound_connections.track_connection();

        let req = OutboundConnectorRequest {
            addr: addr.into(),
            connection_tracker,
        };

        outbound_connector
            .call(req)
            .await
            .expect_err("connection attempt should fail");

        let nonce_count = hs.nonce_count().await;

        assert!(
            expected_max_nonces >= nonce_count,
            "number of nonces should be limited to `peerset_total_connection_limit`"
        );

        assert!(
            expected_nonce_count == nonce_count,
            "number of nonces should be the lesser of the number of closed connections and `peerset_total_connection_limit`"
        )
    }
}

/// Test that [`init`] does not deadlock in `add_initial_peers`,
/// even if the seeders return a lot of peers.
#[tokio::test]
async fn add_initial_peers_deadlock() {
    // The `PEER_COUNT` is the amount of initial seed peers. The value is set so
    // that the peers fill up `PEERSET_INITIAL_TARGET_SIZE`, fill up the channel
    // for sending unused peers to the `AddressBook`, and so that there are
    // still some extra peers left.
    const PEER_COUNT: usize = 200;
    const PEERSET_INITIAL_TARGET_SIZE: usize = 2;
    const TIME_LIMIT: Duration = Duration::from_secs(20);

    let _init_guard = zakura_test::init();

    // This test requires an IPv4 network stack. Localhost should be enough.
    if zakura_test::net::zebra_skip_network_tests() {
        return;
    }

    // Create a list of dummy IPs, and initialize a config using them as the
    // initial peers. The amount of these peers will overflow
    // `PEERSET_INITIAL_TARGET_SIZE`.
    let mut peers = IndexSet::new();
    for address_number in 0..PEER_COUNT {
        peers.insert(
            SocketAddr::new(Ipv4Addr::new(127, 1, 1, address_number as _).into(), 1).to_string(),
        );
    }

    // This test might fail on machines with no configured IPv4 addresses
    // (localhost should be enough).
    let unused_v4 = "0.0.0.0:0".parse().unwrap();

    let config = Config {
        initial_mainnet_peers: peers,
        peerset_initial_target_size: PEERSET_INITIAL_TARGET_SIZE,

        network: Network::Mainnet,
        listen_addr: unused_v4,

        ..Config::default()
    };

    let nil_inbound_service = service_fn(|_| async { Ok(Response::Nil) });

    let init_future = init(
        config,
        nil_inbound_service,
        NoChainTip,
        "Test user agent".to_string(),
        PeerServices::NODE_NETWORK,
    );

    tokio::time::timeout(TIME_LIMIT, init_future)
        .await
        .expect("should not timeout");
}

/// Open a local listener on `listen_addr` for `network`.
/// Asserts that the local listener address works as expected.
async fn local_listener_port_with(listen_addr: SocketAddr, network: Network) {
    let config = Config {
        listen_addr,
        network,

        // Stop Zebra making outbound connections
        initial_mainnet_peers: IndexSet::new(),
        initial_testnet_peers: IndexSet::new(),
        cache_dir: CacheDir::disabled(),

        ..Config::default()
    };
    let inbound_service =
        service_fn(|_| async { unreachable!("inbound service should never be called") });

    let (_peer_service, address_book, _) = init(
        config,
        inbound_service,
        NoChainTip,
        "Test user agent".to_string(),
        PeerServices::NODE_NETWORK,
    )
    .await;
    let local_listener = address_book
        .lock()
        .unwrap()
        .local_listener_meta_addr(Utc::now());

    if listen_addr.port() == 0 {
        assert_ne!(
            local_listener.addr.port(),
            0,
            "dynamic ports are replaced with OS-assigned ports"
        );
    } else {
        assert_eq!(
            local_listener.addr.port(),
            listen_addr.port(),
            "fixed ports are correctly propagated"
        );
    }

    assert_eq!(
        local_listener.addr.ip(),
        listen_addr.ip(),
        "IP addresses are correctly propagated"
    );
}

/// Initialize a peer set with `peerset_initial_target_size`, `inbound_service`, and `network`.
///
/// If `force_listen_addr` is set, binds the network listener to that address.
/// Otherwise, binds the network listener to an unused port on all network interfaces.
/// Uses `default_config` or Zebra's defaults for the rest of the configuration.
///
/// Returns the newly created [`AddressBook`] for testing.
async fn init_with_peer_limit<S>(
    peerset_initial_target_size: usize,
    inbound_service: S,
    network: Network,
    force_listen_addr: impl Into<Option<SocketAddr>>,
    default_config: impl Into<Option<Config>>,
) -> Arc<std::sync::Mutex<AddressBook>>
where
    S: Service<Request, Response = Response, Error = BoxError> + Clone + Send + Sync + 'static,
    S::Future: Send + 'static,
{
    // This test might fail on machines with no configured IPv4 addresses
    // (localhost should be enough).
    let unused_v4 = "0.0.0.0:0".parse().unwrap();

    let default_config = default_config.into().unwrap_or_default();

    let config = Config {
        peerset_initial_target_size,

        network,
        listen_addr: force_listen_addr.into().unwrap_or(unused_v4),

        ..default_config
    };

    let (_peer_service, address_book, _) = init(
        config,
        inbound_service,
        NoChainTip,
        "Test user agent".to_string(),
        PeerServices::NODE_NETWORK,
    )
    .await;

    address_book
}

struct ReplenishmentCrawlerTestHarness {
    config: Config,
    demand_tx: mpsc::Sender<MorePeers>,
    crawl_started_rx: tokio::sync::mpsc::UnboundedReceiver<()>,
    crawl_release_tx: tokio::sync::watch::Sender<bool>,
    connection_tracker_rx: tokio::sync::mpsc::UnboundedReceiver<ConnectionTracker>,
    crawl_task_handle: JoinHandle<Result<(), BoxError>>,
    address_book_updater_guard: JoinHandle<Result<(), BoxError>>,
    _peerset_rx: mpsc::Receiver<DiscoveredPeer>,
    initial_connection_trackers: Vec<ConnectionTracker>,
}

impl ReplenishmentCrawlerTestHarness {
    async fn wait_for_crawl_start(&mut self) {
        wait_for_crawler_events(
            &mut self.crawl_started_rx,
            1,
            "periodic peer crawl should start before the timeout",
        )
        .await;
    }

    fn release_crawl(&self) {
        self.crawl_release_tx
            .send(true)
            .expect("controlled crawl receiver remains open");
    }

    fn queue_demand(&mut self, demand_count: usize) {
        for _ in 0..demand_count {
            self.demand_tx
                .try_send(MorePeers)
                .expect("replenishment test demand channel has capacity");
        }
    }

    fn drop_initial_connections(&mut self, connection_count: usize) {
        assert!(
            connection_count <= self.initial_connection_trackers.len(),
            "cannot drop more initial connections than the harness holds"
        );

        let keep_count = self.initial_connection_trackers.len() - connection_count;
        let dropped_connections = self.initial_connection_trackers.split_off(keep_count);
        std::mem::drop(dropped_connections);
    }

    async fn assert_connection_attempts(mut self, expected_connection_attempts: usize) {
        let mut connection_trackers = wait_for_crawler_events(
            &mut self.connection_tracker_rx,
            expected_connection_attempts,
            "crawler should start the expected replenishment attempts",
        )
        .await;

        self.demand_tx.close_channel();

        let shutdown_deadline = Instant::now() + CRAWLER_TEST_TIMEOUT;
        while !self.crawl_task_handle.is_finished() {
            assert!(
                Instant::now() < shutdown_deadline,
                "crawler should shut down after its demand channel closes"
            );
            tokio::task::spawn_blocking(|| {})
                .await
                .expect("crawler test blocking-task synchronization should not panic");
            tokio::task::yield_now().await;
        }

        let crawl_result = self.crawl_task_handle.await;
        match crawl_result {
            Ok(Err(error)) => assert!(
                error.to_string().contains("demand stream closed"),
                "unexpected peer crawler shutdown error: {error:?}"
            ),
            other => panic!("unexpected peer crawler shutdown result: {other:?}"),
        }

        let connection_deadline = Instant::now() + CRAWLER_TEST_TIMEOUT;
        loop {
            while let Ok(connection_tracker) = self.connection_tracker_rx.try_recv() {
                connection_trackers.push(connection_tracker);
            }

            if self.connection_tracker_rx.is_closed() {
                break;
            }

            assert!(
                Instant::now() < connection_deadline,
                "all spawned connection attempts should finish before the timeout"
            );
            tokio::time::advance(constants::MIN_OUTBOUND_PEER_CONNECTION_INTERVAL).await;
            tokio::task::spawn_blocking(|| {})
                .await
                .expect("crawler test blocking-task synchronization should not panic");
            tokio::task::yield_now().await;
        }

        assert_eq!(
            connection_trackers.len(),
            expected_connection_attempts,
            "crawler started an unexpected number of connection attempts"
        );
        assert!(
            connection_trackers.len() <= self.config.peerset_outbound_connection_limit(),
            "crawler exceeded the hard outbound connection limit"
        );

        self.address_book_updater_guard.abort();
    }
}

/// Options for [`spawn_replenishment_crawler_with`].
#[derive(Clone, Debug, Default)]
struct ReplenishmentCrawlerOptions {
    /// Fail the first N outbound dials before succeeding.
    fail_first_dials: usize,
    /// Report failed-dial demand restores as full-channel misses.
    force_failed_dial_demand_restore_full: bool,
}

async fn spawn_replenishment_crawler(
    initial_connection_count: usize,
    initial_demand_count: usize,
    crawl_new_peer_interval: Duration,
) -> ReplenishmentCrawlerTestHarness {
    spawn_replenishment_crawler_with(
        initial_connection_count,
        initial_demand_count,
        crawl_new_peer_interval,
        ReplenishmentCrawlerOptions::default(),
    )
    .await
}

async fn spawn_replenishment_crawler_with(
    initial_connection_count: usize,
    initial_demand_count: usize,
    crawl_new_peer_interval: Duration,
    options: ReplenishmentCrawlerOptions,
) -> ReplenishmentCrawlerTestHarness {
    let config = Config {
        peerset_initial_target_size: CRAWLER_REPLENISHMENT_TARGET_SIZE_FOR_TESTS,
        crawl_new_peer_interval,
        ..Config::default()
    };
    let outbound_connection_limit = config.peerset_outbound_connection_limit();

    assert!(initial_connection_count <= outbound_connection_limit);

    let (
        address_book,
        _bans_receiver,
        address_book_updater,
        _address_metrics,
        address_book_updater_guard,
    ) = AddressBookUpdater::spawn(&config, config.listen_addr, PeerServices::NODE_NETWORK);

    let candidate_count = outbound_connection_limit
        .saturating_mul(2)
        .saturating_add(1);
    for address_number in 0..candidate_count {
        let address_number =
            u32::try_from(address_number).expect("small replenishment test address count fits u32");
        let ip =
            Ipv4Addr::from(u32::from(Ipv4Addr::new(127, 2, 0, 0)).saturating_add(address_number));
        let addr = MetaAddr::new_gossiped_meta_addr(
            SocketAddr::new(ip.into(), 8233).into(),
            PeerServices::NODE_NETWORK,
            DateTime32::now(),
        )
        .new_gossiped_change()
        .expect("replenishment test peer has valid gossiped address fields");

        address_book
            .lock()
            .expect("previous test thread panicked while holding the address book")
            .update(addr)
            .expect("replenishment test peer is valid");
    }

    let (crawl_started_tx, crawl_started_rx) = tokio::sync::mpsc::unbounded_channel();
    let (crawl_release_tx, crawl_release_rx) = tokio::sync::watch::channel(false);
    let controlled_peer_set = service_fn(move |request| {
        let crawl_started_tx = crawl_started_tx.clone();
        let mut crawl_release_rx = crawl_release_rx.clone();

        async move {
            assert!(
                matches!(request, Request::Peers),
                "crawler should only request peer addresses"
            );
            crawl_started_tx
                .send(())
                .expect("replenishment test crawl observer remains open");
            crawl_release_rx
                .wait_for(|crawl_released| *crawl_released)
                .await
                .expect("replenishment test crawl controller remains open");

            Err::<Response, BoxError>("controlled peer crawl returns no addresses".into())
        }
    });
    let candidates = CandidateSet::new(address_book, controlled_peer_set);

    // Include room for failed-dial demand requeues plus the successful retries.
    let channel_capacity = candidate_count
        .max(initial_demand_count)
        .saturating_add(options.fail_first_dials)
        .saturating_add(CRAWLER_REPLENISHMENT_CONNECTION_TARGET_FOR_TESTS)
        .max(1);
    let (peerset_tx, peerset_rx) = mpsc::channel::<DiscoveredPeer>(channel_capacity);
    let (mut demand_tx, demand_rx) = mpsc::channel::<MorePeers>(channel_capacity);
    for _ in 0..initial_demand_count {
        demand_tx
            .try_send(MorePeers)
            .expect("initial replenishment test demand channel has capacity");
    }

    let mut active_outbound_connections = ActiveConnectionCounter::new_counter_with(
        outbound_connection_limit,
        "Replenishment Test Outbound Connections",
    );
    let initial_connection_trackers = (0..initial_connection_count)
        .map(|_| active_outbound_connections.track_connection())
        .collect();

    let (connection_tracker_tx, connection_tracker_rx) = tokio::sync::mpsc::unbounded_channel();
    let remaining_failures = Arc::new(AtomicUsize::new(options.fail_first_dials));
    let outbound_connector = service_fn(move |request: OutboundConnectorRequest| {
        let connection_tracker_tx = connection_tracker_tx.clone();
        let remaining_failures = remaining_failures.clone();

        async move {
            let OutboundConnectorRequest {
                addr,
                connection_tracker,
            } = request;

            connection_tracker_tx
                .send(connection_tracker)
                .expect("replenishment connection observer remains open");

            if remaining_failures
                .try_update(Ordering::SeqCst, Ordering::SeqCst, |failures| {
                    failures.checked_sub(1)
                })
                .is_ok()
            {
                return Err("controlled replenishment dial failure".into());
            }

            let (fake_client, _harness) = ClientTestHarness::build().finish();
            Ok((addr, fake_client))
        }
    });

    let crawl_task_handle = tokio::spawn(crawl_and_dial(
        config.clone(),
        demand_tx.clone(),
        demand_rx,
        candidates,
        outbound_connector,
        peerset_tx,
        active_outbound_connections,
        address_book_updater,
        Arc::new(AtomicBool::new(
            options.force_failed_dial_demand_restore_full,
        )),
    ));

    ReplenishmentCrawlerTestHarness {
        config,
        demand_tx,
        crawl_started_rx,
        crawl_release_tx,
        connection_tracker_rx,
        crawl_task_handle,
        address_book_updater_guard,
        _peerset_rx: peerset_rx,
        initial_connection_trackers,
    }
}

/// The number of connector calls a crawler peer-limit test expects to observe.
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
enum ExpectedCrawlerConnections {
    /// A zero connection limit must prevent every connector call.
    None,
    /// Held-open connections must stop at the configured limit.
    AtLimit,
    /// Failed or dropped connections must be replaced beyond the limit.
    OverLimit,
}

/// Wait for `event_count` crawler test events while advancing mocked time.
///
/// The crawler can keep runnable tasks queued, so Tokio's automatic paused-time
/// advancement is not sufficient for its connection rate-limit sleeps. It also
/// uses [`tokio::task::spawn_blocking`] in [`CandidateSet::next`], so each time
/// advance is paired with a blocking-pool round trip. The round trip is only a
/// progress aid; correctness comes from the observed events, not FIFO task
/// scheduling.
async fn wait_for_crawler_events<T>(
    events: &mut tokio::sync::mpsc::UnboundedReceiver<T>,
    event_count: usize,
    timeout_message: &str,
) -> Vec<T> {
    let deadline = Instant::now() + CRAWLER_TEST_TIMEOUT;
    let mut received_events = Vec::with_capacity(event_count);

    while received_events.len() < event_count {
        while let Ok(event) = events.try_recv() {
            received_events.push(event);
            if received_events.len() == event_count {
                return received_events;
            }
        }

        assert!(!events.is_closed(), "{timeout_message}: channel closed");
        assert!(Instant::now() < deadline, "{timeout_message}");

        tokio::time::advance(constants::MIN_OUTBOUND_PEER_CONNECTION_INTERVAL).await;
        tokio::task::spawn_blocking(|| {})
            .await
            .expect("crawler test blocking-task synchronization should not panic");
        tokio::task::yield_now().await;
    }

    received_events
}

/// Run a peer crawler with `peerset_initial_target_size` and `outbound_connector`.
///
/// Uses the default values for all other config fields.
/// Does not bind a local listener.
///
/// Returns the generated [`Config`], and the discovered peers.
async fn spawn_crawler_with_peer_limit<C>(
    peerset_initial_target_size: impl Into<Option<usize>>,
    expected_connections: ExpectedCrawlerConnections,
    outbound_connector: C,
) -> (Config, Vec<DiscoveredPeer>)
where
    C: Service<
            OutboundConnectorRequest,
            Response = (PeerSocketAddr, peer::Client),
            Error = BoxError,
        > + Clone
        + Send
        + 'static,
    C::Future: Send + 'static,
{
    // Create a test config.
    let mut config = Config::default();
    if let Some(peerset_initial_target_size) = peerset_initial_target_size.into() {
        config.peerset_initial_target_size = peerset_initial_target_size;
    }

    let (
        address_book,
        _bans_receiver,
        address_book_updater,
        _address_metrics,
        _address_book_updater_guard,
    ) = AddressBookUpdater::spawn(&config, config.listen_addr, PeerServices::NODE_NETWORK);

    // Add enough fake peers to go over the limit, even if the limit is zero.
    let over_limit_peers = config.peerset_outbound_connection_limit() * 2 + 1;
    let mut fake_peer = None;
    for address_number in 0..over_limit_peers {
        let addr = SocketAddr::new(Ipv4Addr::new(127, 1, 1, address_number as _).into(), 1);
        let addr = MetaAddr::new_gossiped_meta_addr(
            addr.into(),
            PeerServices::NODE_NETWORK,
            DateTime32::now(),
        );
        fake_peer = Some(addr);
        let addr = addr
            .new_gossiped_change()
            .expect("created MetaAddr contains enough information to represent a gossiped address");

        address_book
            .lock()
            .expect("panic in previous thread while accessing the address book")
            .update(addr);
    }

    // Create a fake peer set. The notification proves the crawler has started
    // an observable crawl action before this helper stops it. The zero-limit
    // response stays pending so its timer crawl cannot enqueue more demand
    // while the test closes the demand channel to establish a drain barrier.
    let (crawl_observed_tx, mut crawl_observed_rx) = tokio::sync::mpsc::unbounded_channel();
    let nil_peer_set = service_fn(move |req| {
        let crawl_observed_tx = crawl_observed_tx.clone();

        async move {
            let rsp = match req {
                // Return the correct response variant for Peers requests,
                // reusing one of the peers we already provided.
                Request::Peers => Response::Peers(vec![fake_peer.unwrap()]),
                _ => unreachable!("unexpected request: {:?}", req),
            };

            let _ = crawl_observed_tx.send(());

            if expected_connections == ExpectedCrawlerConnections::None {
                std::future::pending::<Result<Response, BoxError>>().await
            } else {
                Ok(rsp)
            }
        }
    });

    // Make the channels large enough to hold all the peers.
    let (peerset_tx, mut peerset_rx) = mpsc::channel::<DiscoveredPeer>(over_limit_peers);
    let (mut demand_tx, demand_rx) = mpsc::channel::<MorePeers>(over_limit_peers);

    let candidates = CandidateSet::new(address_book.clone(), nil_peer_set);

    // In zakura_network::initialize() the counter would already have some initial peer connections,
    // but in this test we start with an empty counter.
    let active_outbound_connections = ActiveConnectionCounter::new_counter();

    // Add fake demand over the limit.
    for _ in 0..over_limit_peers {
        let _ = demand_tx.try_send(MorePeers);
    }
    let mut demand_shutdown_tx = demand_tx.clone();

    // Observe connector completion rather than sleeping for a fixed duration.
    let (connection_finished_tx, mut connection_finished_rx) =
        tokio::sync::mpsc::unbounded_channel();
    let outbound_connector = service_fn(move |request| {
        let connector = outbound_connector.clone();
        let connection_finished_tx = connection_finished_tx.clone();

        async move {
            let result = connector.oneshot(request).await;
            let _ = connection_finished_tx.send(result.is_ok());
            result
        }
    });

    let expected_connection_count = match expected_connections {
        ExpectedCrawlerConnections::None => 0,
        ExpectedCrawlerConnections::AtLimit => config.peerset_outbound_connection_limit(),
        ExpectedCrawlerConnections::OverLimit => config.peerset_outbound_connection_limit() + 1,
    };

    // Start the crawler.
    let crawl_fut = crawl_and_dial(
        config.clone(),
        demand_tx,
        demand_rx,
        candidates,
        outbound_connector,
        peerset_tx,
        active_outbound_connections,
        address_book_updater,
        Arc::new(AtomicBool::new(false)),
    );
    let crawl_task_handle = tokio::spawn(crawl_fut);

    wait_for_crawler_events(
        &mut crawl_observed_rx,
        1,
        "peer crawler should attempt a crawl before the timeout",
    )
    .await;

    let connection_results = wait_for_crawler_events(
        &mut connection_finished_rx,
        expected_connection_count,
        "peer crawler should make the expected connections before the timeout",
    )
    .await;
    let expected_discovered_peers = connection_results
        .into_iter()
        .filter(|connection_succeeded| *connection_succeeded)
        .count();

    let peer_deadline = Instant::now() + CRAWLER_TEST_TIMEOUT;
    let mut discovered_peers = Vec::with_capacity(expected_discovered_peers);
    while discovered_peers.len() < expected_discovered_peers {
        while let Ok(peer) = peerset_rx.try_recv() {
            discovered_peers.push(peer);
            if discovered_peers.len() == expected_discovered_peers {
                break;
            }
        }

        if discovered_peers.len() < expected_discovered_peers {
            assert!(
                Instant::now() < peer_deadline,
                "successful crawler connections should reach the peer set"
            );
            tokio::time::advance(constants::MIN_OUTBOUND_PEER_CONNECTION_INTERVAL).await;
            tokio::task::spawn_blocking(|| {})
                .await
                .expect("crawler test blocking-task synchronization should not panic");
            tokio::task::yield_now().await;
        }
    }

    if matches!(
        expected_connections,
        ExpectedCrawlerConnections::None | ExpectedCrawlerConnections::AtLimit
    ) {
        // Closing demand after the expected connections makes the crawler drain
        // every queued excess signal before it observes channel shutdown. This
        // is a deterministic barrier proving zero-limit and held-open cases do
        // not start an additional connection.
        demand_shutdown_tx.close_channel();

        let shutdown_deadline = Instant::now() + CRAWLER_TEST_TIMEOUT;
        while !crawl_task_handle.is_finished() {
            assert!(
                Instant::now() < shutdown_deadline,
                "peer crawler should drain queued demand before the timeout"
            );
            tokio::task::spawn_blocking(|| {})
                .await
                .expect("crawler test blocking-task synchronization should not panic");
            tokio::task::yield_now().await;
        }

        let crawl_result = crawl_task_handle.await;
        match crawl_result {
            Ok(Err(error)) => assert!(
                error.to_string().contains("demand stream closed"),
                "unexpected peer crawler shutdown error: {error:?}"
            ),
            other => panic!("unexpected peer crawler shutdown result: {other:?}"),
        }

        assert!(
            connection_finished_rx.is_closed(),
            "queued excess demand must not leave a peer connection in flight"
        );
        assert!(
            connection_finished_rx.try_recv().is_err(),
            "queued excess demand must not start another peer connection"
        );
    } else {
        // Error and disconnected-connection tests only need to prove that
        // replacement attempts exceed the configured connection limit.
        crawl_task_handle.abort();

        let crawl_result = crawl_task_handle.await;
        assert!(
            matches!(crawl_result, Err(ref error) if error.is_cancelled()),
            "unexpected error or panic in peer crawler task: {crawl_result:?}",
        );
    }

    while let Ok(peer) = peerset_rx.try_recv() {
        discovered_peers.push(peer);
    }

    // Check the final address book contents.
    assert_eq!(
        address_book.lock().unwrap().peers().count(),
        over_limit_peers,
        "expected {} peers in Mainnet address book, but got: {:?}",
        over_limit_peers,
        address_book.lock().unwrap().address_metrics(Utc::now())
    );

    (config, discovered_peers)
}

async fn connect_from(source_ip: Ipv4Addr, listen_addr: SocketAddr) -> TcpStream {
    let socket = TcpSocket::new_v4().expect("test should create an IPv4 TCP socket");
    socket
        .bind(SocketAddr::new(source_ip.into(), 0))
        .expect("test should bind to a loopback source address");
    socket
        .connect(listen_addr)
        .await
        .expect("test should connect to the inbound listener")
}

fn drain_accepted_inbound_ips(
    peer_tracker_rx: &mut mpsc::UnboundedReceiver<(IpAddr, TcpStream, ConnectionTracker)>,
) -> Vec<Ipv4Addr> {
    let mut accepted_ips = Vec::new();

    while let Ok((ip, peer_connection, peer_tracker)) = peer_tracker_rx.try_recv() {
        let IpAddr::V4(ip) = ip else {
            panic!("zcashd-compat listener tests only use IPv4 peers");
        };

        accepted_ips.push(ip);
        std::mem::drop(peer_connection);
        std::mem::drop(peer_tracker);
    }

    accepted_ips
}

fn drain_discovered_peers(peerset_rx: &mut mpsc::Receiver<DiscoveredPeer>) -> usize {
    let mut peer_count = 0;

    while peerset_rx.try_recv().is_ok() {
        peer_count += 1;
    }

    peer_count
}

fn assert_listener_task_cancelled(listen_task_handle: JoinHandle<Result<(), BoxError>>) {
    let listen_result = listen_task_handle.now_or_never();
    assert!(
        listen_result.is_none() || matches!(listen_result, Some(Err(ref e)) if e.is_cancelled()),
        "unexpected error or panic in inbound peer listener task: {listen_result:?}",
    );
}

/// Run an inbound peer listener with `peerset_initial_target_size` and `handshaker`.
///
/// Binds the local listener to an unused localhost port.
/// Uses the default values for all other config fields.
///
/// Returns the generated [`Config`], and the discovered peers.
async fn spawn_inbound_listener_with_peer_limit<S>(
    peerset_initial_target_size: impl Into<Option<usize>>,
    max_connections_per_ip: impl Into<Option<usize>>,
    listen_handshaker: S,
) -> (Config, Vec<DiscoveredPeer>)
where
    S: Service<peer::HandshakeRequest<TcpStream>, Response = peer::Client, Error = BoxError>
        + Clone
        + Send
        + Sync
        + 'static,
    S::Future: Send + 'static,
{
    // Create a test config that listens on an unused port.
    let listen_addr = "127.0.0.1:0".parse().unwrap();
    let mut config = Config {
        listen_addr,
        max_connections_per_ip: max_connections_per_ip
            .into()
            .unwrap_or(constants::DEFAULT_MAX_CONNS_PER_IP),
        ..Config::default()
    };

    if let Some(peerset_initial_target_size) = peerset_initial_target_size.into() {
        config.peerset_initial_target_size = peerset_initial_target_size;
    }

    // Open the listener port.
    let (tcp_listener, listen_addr) = open_listener(&config.clone()).await;

    // Make enough inbound connections to go over the limit, even if the limit is zero.
    // Make the channels large enough to hold all the connections.
    let over_limit_connections = config.peerset_inbound_connection_limit() * 2 + 1;
    let (peerset_tx, mut peerset_rx) = mpsc::channel::<DiscoveredPeer>(over_limit_connections);

    // Observe completed handshakes so the helper can deterministically wait for
    // every spawned handshake task before inspecting the peer set channel.
    let (handshake_finished_tx, mut handshake_finished_rx) = tokio::sync::mpsc::unbounded_channel();
    let listen_handshaker = service_fn(move |request| {
        let handshaker = listen_handshaker.clone();
        let handshake_finished_tx = handshake_finished_tx.clone();

        async move {
            let result = handshaker.oneshot(request).await;
            let _ = handshake_finished_tx.send(());
            result
        }
    });

    let bans = BannedIps::default();

    // Start listening for connections.
    let listen_fut = accept_inbound_connections(
        config.clone(),
        tcp_listener,
        MIN_INBOUND_PEER_CONNECTION_INTERVAL_FOR_TESTS,
        listen_handshaker,
        peerset_tx,
        bans,
        Vec::new(),
    );
    let listen_task_handle = tokio::spawn(listen_fut);

    // Open inbound connections.
    let (connection_finished_tx, mut connection_finished_rx) =
        tokio::sync::mpsc::unbounded_channel();
    let mut outbound_task_handles = Vec::new();
    for _ in 0..over_limit_connections {
        let connection_finished_tx = connection_finished_tx.clone();
        let outbound_fut = async move {
            let outbound_result = TcpStream::connect(listen_addr).await;
            // Let other tasks run before we block on reading.
            tokio::task::yield_now().await;

            if let Ok(outbound_stream) = outbound_result {
                // Wait until the listener closes the connection.
                // The handshaker is fake, so it never sends any data.
                let readable_result = outbound_stream.readable().await;
                debug!(
                    ?readable_result,
                    "outbound connection became readable or errored: \
                     closing connection to test inbound listener"
                );
            } else {
                // If the connection is closed quickly, we might get errors here.
                debug!(
                    ?outbound_result,
                    "outbound connection error in inbound listener test"
                );
            }

            let _ = connection_finished_tx.send(());
        };

        let outbound_task_handle = tokio::spawn(outbound_fut);
        outbound_task_handles.push(outbound_task_handle);
    }
    std::mem::drop(connection_finished_tx);

    // All test connections use the same IP, so at most this many connections
    // can stay open. Once every other connection has closed, the listener has
    // exercised both its accepted and over-limit paths.
    let maximum_open_connections = config
        .peerset_inbound_connection_limit()
        .min(config.max_connections_per_ip);
    let expected_closed_connections = over_limit_connections - maximum_open_connections;

    tokio::time::timeout(LISTENER_TEST_DURATION, async {
        for _ in 0..expected_closed_connections {
            connection_finished_rx
                .recv()
                .await
                .expect("inbound connection tasks remain active until they report completion");
        }
    })
    .await
    .expect("inbound listener should process the test connections before the timeout");

    // Stop the listener and outbound tasks, and wait for their cancellation.
    listen_task_handle.abort();
    for outbound_task_handle in &outbound_task_handles {
        outbound_task_handle.abort();
    }

    let listen_result = listen_task_handle.await;
    assert!(
        matches!(listen_result, Err(ref error) if error.is_cancelled()),
        "unexpected error or panic in inbound peer listener task: {listen_result:?}",
    );

    for outbound_task_handle in outbound_task_handles {
        let outbound_result = outbound_task_handle.await;
        assert!(
            outbound_result.is_ok()
                || matches!(outbound_result, Err(ref error) if error.is_cancelled()),
            "unexpected panic in inbound test connection task: {outbound_result:?}",
        );
    }

    let handshake_count = tokio::time::timeout(LISTENER_TEST_DURATION, async move {
        let mut handshake_count = 0;
        while handshake_finished_rx.recv().await.is_some() {
            handshake_count += 1;
        }
        handshake_count
    })
    .await
    .expect("inbound handshake tasks should finish before the timeout");

    if config.peerset_inbound_connection_limit() == 0 {
        assert_eq!(
            handshake_count, 0,
            "the handshaker must not run when the inbound peer limit is zero"
        );
    } else {
        assert!(
            handshake_count > 0,
            "the listener must exercise the configured handshaker"
        );
    }

    let discovered_peers = tokio::time::timeout(
        LISTENER_TEST_DURATION,
        peerset_rx.by_ref().collect::<Vec<_>>(),
    )
    .await
    .expect("inbound handshake peer changes should finish before the timeout");

    (config, discovered_peers)
}

/// Initialize a task that connects to `peer_count` initial peers using the
/// given connector.
///
/// Connects to IP addresses in the IPv4 localhost range.
/// Does not open a local listener port.
///
/// Returns the initial peers task [`JoinHandle`], the peer set receiver,
/// and the address book updater task join handle.
async fn spawn_add_initial_peers<C>(
    peer_count: usize,
    outbound_connector: C,
) -> (
    JoinHandle<Result<ActiveConnectionCounter, BoxError>>,
    mpsc::Receiver<DiscoveredPeer>,
    JoinHandle<Result<(), BoxError>>,
)
where
    C: Service<
            OutboundConnectorRequest,
            Response = (PeerSocketAddr, peer::Client),
            Error = BoxError,
        > + Clone
        + Send
        + 'static,
    C::Future: Send + 'static,
{
    // Create a list of dummy IPs and initialize a config using them as the
    // initial peers.
    let mut peers = IndexSet::new();
    for address_number in 0..peer_count {
        peers.insert(
            SocketAddr::new(Ipv4Addr::new(127, 1, 1, address_number as _).into(), 1).to_string(),
        );
    }

    // This address isn't actually bound - it just gets passed to the address book.
    let unused_v4 = "0.0.0.0:0".parse().unwrap();

    let config = Config {
        initial_mainnet_peers: peers,
        // We want exactly the above list of peers, without any cached peers.
        cache_dir: CacheDir::disabled(),

        network: Network::Mainnet,
        listen_addr: unused_v4,

        ..Config::default()
    };

    let (peerset_tx, peerset_rx) = mpsc::channel::<DiscoveredPeer>(peer_count + 1);

    let (
        _address_book,
        _bans_receiver,
        address_book_updater,
        _address_metrics,
        address_book_updater_guard,
    ) = AddressBookUpdater::spawn(&config, unused_v4, PeerServices::NODE_NETWORK);

    let add_fut = add_initial_peers(config, outbound_connector, peerset_tx, address_book_updater);
    let add_task_handle = tokio::spawn(add_fut);

    (add_task_handle, peerset_rx, address_book_updater_guard)
}