zebra_network/peer_set/

set.rs

//! Abstractions that represent "the rest of the network".
//!
//! # Implementation
//!
//! The [`PeerSet`] implementation is adapted from the one in [tower::Balance][tower-balance].
//!
//! As described in Tower's documentation, it:
//!
//! > Distributes requests across inner services using the [Power of Two Choices][p2c].
//! >
//! > As described in the [Finagle Guide][finagle]:
//! >
//! > > The algorithm randomly picks two services from the set of ready endpoints and
//! > > selects the least loaded of the two. By repeatedly using this strategy, we can
//! > > expect a manageable upper bound on the maximum load of any server.
//! > >
//! > > The maximum load variance between any two servers is bound by `ln(ln(n))` where
//! > > `n` is the number of servers in the cluster.
//!
//! The Power of Two Choices should work well for many network requests, but not all of them.
//! Some requests should only be made to a subset of connected peers.
//! For example, a request for a particular inventory item
//! should be made to a peer that has recently advertised that inventory hash.
//! Other requests require broadcasts, such as transaction diffusion.
//!
//! Implementing this specialized routing logic inside the `PeerSet` -- so that
//! it continues to abstract away "the rest of the network" into one endpoint --
//! is not a problem, as the `PeerSet` can simply maintain more information on
//! its peers and route requests appropriately. However, there is a problem with
//! maintaining accurate backpressure information, because the `Service` trait
//! requires that service readiness is independent of the data in the request.
//!
//! For this reason, in the future, this code will probably be refactored to
//! address this backpressure mismatch. One possibility is to refactor the code
//! so that one entity holds and maintains the peer set and metadata on the
//! peers, and each "backpressure category" of request is assigned to different
//! `Service` impls with specialized `poll_ready()` implementations. Another
//! less-elegant solution (which might be useful as an intermediate step for the
//! inventory case) is to provide a way to borrow a particular backing service,
//! say by address.
//!
//! [finagle]: https://twitter.github.io/finagle/guide/Clients.html#power-of-two-choices-p2c-least-loaded
//! [p2c]: http://www.eecs.harvard.edu/~michaelm/postscripts/handbook2001.pdf
//! [tower-balance]: https://github.com/tower-rs/tower/tree/master/tower/src/balance
//!
//! # Behavior During Network Upgrades
//!
//! [ZIP-201] specifies peer behavior during network upgrades:
//!
//! > With scheduled network upgrades, at the activation height, nodes on each consensus branch
//! > should disconnect from nodes on other consensus branches and only accept new incoming
//! > connections from nodes on the same consensus branch.
//!
//! Zebra handles this with the help of [`MinimumPeerVersion`], which determines the minimum peer
//! protocol version to accept based on the current best chain tip height. The minimum version is
//! therefore automatically increased when the block height reaches a network upgrade's activation
//! height. The helper type is then used to:
//!
//! - cancel handshakes to outdated peers, in `handshake::negotiate_version`
//! - cancel requests to and disconnect from peers that have become outdated, in
//!   [`PeerSet::push_unready`]
//! - disconnect from peers that have just responded and became outdated, in
//!   [`PeerSet::poll_unready`]
//! - disconnect from ready or busy peers that have become outdated, in
//!   [`PeerSet::disconnect_from_ineligible_peers`]
//!
//! ## Network Coalescence
//!
//! [ZIP-201] also specifies how Zcashd behaves [leading up to a activation
//! height][1]. Since Zcashd limits the number of connections to at most eight
//! peers, it will gradually migrate its connections to up-to-date peers as it
//! approaches the activation height.
//!
//! The motivation for this behavior is to avoid an abrupt partitioning the network, which can lead
//! to isolated peers and increases the chance of an eclipse attack on some peers of the network.
//!
//! Zebra does not gradually migrate its peers as it approaches an activation height. This is
//! because Zebra by default can connect to up to 75 peers, as can be seen in [`Config::default`].
//! Since this is a lot larger than the 8 peers Zcashd connects to, an eclipse attack becomes a lot
//! more costly to execute, and the probability of an abrupt network partition that isolates peers
//! is lower.
//!
//! Even if a Zebra node is manually configured to connect to a smaller number
//! of peers, the [`AddressBook`][2] is configured to hold a large number of
//! peer addresses ([`MAX_ADDRS_IN_ADDRESS_BOOK`][3]). Since the address book
//! prioritizes addresses it trusts (like those that it has successfully
//! connected to before), the node should be able to recover and rejoin the
//! network by itself, as long as the address book is populated with enough
//! entries.
//!
//! [1]: https://zips.z.cash/zip-0201#network-coalescence
//! [2]: crate::AddressBook
//! [3]: crate::constants::MAX_ADDRS_IN_ADDRESS_BOOK
//! [ZIP-201]: https://zips.z.cash/zip-0201

use std::{
    collections::{HashMap, HashSet, VecDeque},
    convert,
    fmt::Debug,
    marker::PhantomData,
    net::IpAddr,
    pin::Pin,
    sync::Arc,
    task::{Context, Poll, Waker},
    time::Instant,
};

use futures::{
    channel::{mpsc, oneshot},
    future::{FutureExt, TryFutureExt},
    prelude::*,
    stream::FuturesUnordered,
    task::{noop_waker, AtomicWaker},
};
use itertools::Itertools;
use num_integer::div_ceil;
use tokio::{
    sync::{broadcast, mpsc as tokio_mpsc, watch, Notify},
    task::JoinHandle,
};
use tower::{
    buffer::Buffer,
    discover::{Change, Discover},
    load::Load,
    util::BoxService,
    Service, ServiceExt,
};
use tracing_futures::Instrument;

use zebra_chain::{chain_tip::ChainTip, parameters::Network};

use crate::{
    address_book::AddressMetrics,
    connection_metrics::network_kind_label,
    constants::{
        FIND_BUSY_SERVING_PEER_WAIT_TIMEOUT, INVENTORY_BUSY_PEER_WAIT_TIMEOUT,
        MIN_PEER_SET_LOG_INTERVAL,
    },
    peer::{ConnectionInfo, LoadTrackedClient, MinimumPeerVersion},
    peer_set::{
        stall_tracker::FindResponseStallTracker,
        unready_service::{Error as UnreadyError, UnreadyService},
        InventoryChange, InventoryRegistry,
    },
    protocol::{
        external::{connection_limit_key, types::PeerServices, InventoryHash},
        internal::{Request, Response},
    },
    BanList, BoxError, Config, PeerError, PeerSocketAddr, SharedPeerError,
};

#[cfg(any(test, feature = "proptest-impl"))]
pub(crate) mod tests;

/// A signal sent by the [`PeerSet`] when it has no ready peers, and gets a request from Zebra.
///
/// In response to this signal, the crawler tries to open more peer connections.
#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
pub struct MorePeers;

/// A signal sent by the [`PeerSet`] to cancel a [`Client`][1]'s current request
/// or response.
///
/// When it receives this signal, the [`Client`][1] stops processing and exits.
///
/// [1]: crate::peer::Client
#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
pub struct CancelClientWork;

type ResponseFuture = Pin<Box<dyn Future<Output = Result<Response, BoxError>> + Send + 'static>>;

/// A block request that no ready peer could serve, waiting for a busy peer that might have the
/// block to become ready.
struct QueuedInvRequest {
    /// The request to route.
    request: Request,

    /// The requested inventory.
    hash: InventoryHash,

    /// The enqueue-time refusal deadline; downloaders enforce their own operation deadline.
    deadline: Option<tokio::time::Instant>,

    /// The original request's tracing context.
    span: tracing::Span,

    /// Receives the response future of the peer the request is routed to.
    ///
    /// Dropping the sender refuses the request.
    response_sender: oneshot::Sender<ResponseFuture>,
}

/// Bounds queued discovery requests; overflow falls back to any ready peer.
const MAX_QUEUED_FIND_REQUESTS: usize = 16;

/// A find request waiting for a serving peer, with a bounded serving-only preference.
struct QueuedFindRequest {
    /// The request to route.
    request: Request,

    /// When this request may fall back to any ready peer.
    deadline: tokio::time::Instant,

    /// Receives the selected peer's response future; cancellation abandons the queued request.
    response_sender: oneshot::Sender<ResponseFuture>,
}

/// Where the peer set routes an inventory request.
enum InvRoute<K> {
    /// Send the request to this ready peer.
    Peer(K),

    /// Queue the request until a busy peer that might have the inventory becomes ready.
    Wait,

    /// Refuse the request, because no connected peer might have the inventory.
    Refuse,
}

/// Returns the synthetic error for an inventory request that the peer set refuses.
///
/// # Security
///
/// Avoid routing requests to peers that are missing inventory.
/// If we kept trying doomed requests, peers that are missing our requested inventory
/// could take up a large amount of our bandwidth and retry limits.
fn not_found_registry_error(hash: InventoryHash) -> BoxError {
    SharedPeerError::from(PeerError::NotFoundRegistry(vec![hash])).into()
}

/// A waker for peer events, queue deadlines, and cancellations while requests are queued.
///
/// Wakes the task that polls the peer set, and notifies the task that runs
/// [`poll_peer_set_on_notify`]. The peer set is behind a [`tower::buffer::Buffer`], which only
/// polls it when it has a request, so queued requests otherwise wouldn't be routed when a
/// busy peer becomes ready or a find request's serving-only wait expires.
#[derive(Default)]
struct QueuedRequestWaker {
    /// Notified when requests are queued, peers change, waits expire, or callers cancel.
    notify: Notify,

    /// The waker of the task that polls the peer set.
    peer_set_waker: AtomicWaker,
}

impl std::task::Wake for QueuedRequestWaker {
    fn wake(self: Arc<Self>) {
        self.wake_by_ref();
    }

    fn wake_by_ref(self: &Arc<Self>) {
        self.notify.notify_one();
        self.peer_set_waker.wake();
    }
}

/// Polls `peer_set` with a [`Request::PollPeerSet`] each time `queued_request_waker` is notified,
/// so queued requests are routed when peers become ready or their serving-only wait expires.
///
/// `peer_set` must be the service that wraps the [`PeerSet`] that owns `queued_request_waker`.
/// See [`PeerSet::into_buffer`].
///
/// Only returns if the peer set fails.
async fn poll_peer_set_on_notify<S>(
    mut peer_set: S,
    queued_request_waker: Arc<QueuedRequestWaker>,
) -> Result<(), BoxError>
where
    S: Service<Request, Response = Response, Error = BoxError> + Send + Clone + 'static,
    S::Future: Send + 'static,
{
    loop {
        // Notifications while the peer set is being polled are combined into one poll, because
        // `Notify` stores at most one permit.
        //
        // This wait has no timeout, because the peer set only notifies while requests are queued.
        // The poll request can also wait a long time, while all peers are busy: then it is only
        // routed once a peer becomes ready, which is exactly when queued requests can be routed.
        queued_request_waker.notify.notified().await;
        peer_set.ready().await?.call(Request::PollPeerSet).await?;
    }
}

/// Classification of a `FindBlocks`/`FindHeaders` response, sent from a
/// response-wrapping future to [`PeerSet::poll_ready`] via an mpsc channel so
/// the stall tracker can be updated and the peer disconnected if needed.
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
enum StallOutcome {
    Stall,
    Clear,
}

type StallEvent = (PeerSocketAddr, Arc<ConnectionInfo>, StallOutcome);

type InvFailureEvent = (PeerSocketAddr, Arc<ConnectionInfo>, InventoryHash);

/// Empty replies are valid when the locator reaches the remote tip. Neither a local wall-clock
/// estimate nor an untrusted, frozen handshake height proves otherwise. Fair discovery routing
/// prevents fast empty replies from monopolizing requests.
fn classify_find_response<E>(result: &Result<Response, E>) -> Option<StallOutcome> {
    match result {
        Ok(Response::BlockHashes(_) | Response::BlockHeaders(_)) => Some(StallOutcome::Clear),
        Ok(_) => None,
        Err(_) => Some(StallOutcome::Stall),
    }
}

/// A [`tower::Service`] that abstractly represents "the rest of the network".
///
/// # Security
///
/// The `Discover::Key` must be the transient remote address of each peer. This
/// address may only be valid for the duration of a single connection. (For
/// example, inbound connections have an ephemeral remote port, and proxy
/// connections have an ephemeral local or proxy port.)
///
/// Otherwise, malicious peers could interfere with other peers' `PeerSet` state.
pub struct PeerSet<D, C>
where
    D: Discover<Key = PeerSocketAddr, Service = LoadTrackedClient> + Unpin,
    D::Error: Into<BoxError>,
    C: ChainTip,
{
    // Peer Tracking: New Peers
    //
    /// Provides new and deleted peer [`Change`]s to the peer set,
    /// via the [`Discover`] trait implementation.
    discover: D,

    /// A channel that asks the peer crawler task to connect to more peers.
    demand_signal: mpsc::Sender<MorePeers>,

    /// A watch channel receiver with a snapshot of the banned peer groups.
    bans_receiver: watch::Receiver<BanList>,

    /// Tracks peers returning failed `FindBlocks`/`FindHeaders` responses.
    /// Mutated only from [`Self::poll_ready`] via [`Self::stall_event_rx`].
    find_response_stalls: FindResponseStallTracker,

    /// Receives stall/clear events from tracked routing futures in
    /// [`Self::call_ready_peer`]. The channel keeps the tracker single-owner (no
    /// `Mutex`) and confines mutation to `poll_ready`, where the peer set can
    /// call [`Self::remove`] directly.
    stall_event_rx: tokio_mpsc::UnboundedReceiver<StallEvent>,

    /// Producer clones handed to each tracked request's response wrapper.
    stall_event_tx: tokio_mpsc::UnboundedSender<StallEvent>,

    /// Receives completed block request errors from response futures and records them in the
    /// [`InventoryRegistry`]. Caller cancellation is not evidence of a peer failure.
    ///
    /// Each routed block request sends at most one failure, so the channel is limited by the
    /// number of block requests in flight.
    inv_failure_rx: tokio_mpsc::UnboundedReceiver<InvFailureEvent>,

    /// Producer clones handed to each block request's response future.
    inv_failure_tx: tokio_mpsc::UnboundedSender<InvFailureEvent>,

    // Peer Tracking: Ready Peers
    //
    /// Connected peers that are ready to receive requests from Zebra,
    /// or send requests to Zebra.
    ready_services: HashMap<D::Key, D::Service>,

    // Request Routing
    //
    /// Connection identities in find-request order, preserved while peers are busy.
    ///
    /// New connections join at the back; disconnected connections are pruned in
    /// [`Self::poll_peers`]. The metadata allocation identifies a connection even if
    /// a later connection reuses its address.
    find_peer_queue: VecDeque<(D::Key, Arc<ConnectionInfo>)>,

    /// Find requests waiting for a serving peer, in request order.
    ///
    /// After [`FIND_BUSY_SERVING_PEER_WAIT_TIMEOUT`], any ready peer can receive the request.
    queued_find_requests: VecDeque<QueuedFindRequest>,

    /// Wakes the peer set at the next unexpired queued find request deadline.
    ///
    /// Expired requests wait for peer events rather than repeatedly waking a completed timer.
    queued_find_request_timer: Option<Pin<Box<tokio::time::Sleep>>>,

    /// Stores gossiped inventory hashes from connected peers.
    ///
    /// Used to route inventory requests to peers that are likely to have it.
    inventory_registry: InventoryRegistry,

    /// Stores requests that should be routed to peers once they are ready.
    queued_broadcast_all: Option<(
        Request,
        tokio::sync::mpsc::Sender<ResponseFuture>,
        HashSet<D::Key>,
    )>,

    /// Inbound peer IPs that must always receive block inventory broadcasts.
    block_gossip_peer_ips: HashSet<IpAddr>,

    /// The keys of connected peers that matched [`Self::block_gossip_peer_ips`]
    /// when they were inserted into the peer set.
    ///
    /// Stale keys of disconnected peers are pruned by
    /// [`Self::prune_disconnected_sidecar_keys`].
    zcashd_compat_peer_keys: HashSet<D::Key>,

    /// The keys of connected peers that advertised [`PeerServices::NODE_NETWORK`],
    /// so they can serve historic blocks.
    ///
    /// Busy peers' services are not otherwise accessible, but block and find routing need to
    /// know whether a busy peer can serve blocks. Stale keys of disconnected peers are pruned by
    /// [`Self::retain_connected_keys`].
    serving_peer_keys: HashSet<D::Key>,

    /// Block requests that are waiting for a busy peer that might have the block, in the order
    /// they were made.
    ///
    /// Each request is routed to the first of those peers that becomes ready, or refused once
    /// [`INVENTORY_BUSY_PEER_WAIT_TIMEOUT`] expires, or once no busy peer might have the block.
    queued_inv_requests: VecDeque<QueuedInvRequest>,

    /// Wakes the task that polls the peer set, and the task that runs
    /// [`poll_peer_set_on_notify`], while requests are queued.
    queued_request_waker: Arc<QueuedRequestWaker>,

    /// The most recent sidecar broadcast (a block advert or a pushed
    /// transaction) that has not been delivered to all connected zcashd-compat
    /// sidecar peers, and the sidecar peers that are still owed it.
    ///
    /// A sidecar can be busy with another request when a broadcast is routed.
    /// The request is queued here and delivered as soon as the sidecar is
    /// ready again, so configured sidecars never miss block gossip. A newer
    /// broadcast replaces an older undelivered one, even one of a different
    /// kind.
    queued_sidecar_broadcast: Option<(Request, HashSet<D::Key>)>,

    // Peer Tracking: Busy Peers
    //
    /// Connected peers that are handling a Zebra request,
    /// or Zebra is handling one of their requests.
    unready_services: FuturesUnordered<UnreadyService<D::Key, D::Service, Request>>,

    /// Channels used to cancel the request that an unready service is doing.
    cancel_handles: HashMap<D::Key, oneshot::Sender<CancelClientWork>>,

    // Peer Validation
    //
    /// An endpoint to see the minimum peer protocol version in real time.
    ///
    /// The minimum version depends on the block height, and [`MinimumPeerVersion`] listens for
    /// height changes and determines the correct minimum version.
    minimum_peer_version: MinimumPeerVersion<C>,

    /// The configured limit for inbound and outbound connections.
    ///
    /// The peer set panics if this size is exceeded.
    /// If that happens, our connection limit code has a bug.
    peerset_total_connection_limit: usize,

    // Background Tasks
    //
    /// Channel for passing ownership of tokio JoinHandles from PeerSet's background tasks
    ///
    /// The join handles passed into the PeerSet are used populate the `guards` member
    handle_rx: tokio::sync::oneshot::Receiver<Vec<JoinHandle<Result<(), BoxError>>>>,

    /// Unordered set of handles to background tasks associated with the `PeerSet`
    ///
    /// These guards are checked for errors as part of `poll_ready` which lets
    /// the `PeerSet` propagate errors from background tasks back to the user
    guards: futures::stream::FuturesUnordered<JoinHandle<Result<(), BoxError>>>,

    // Metrics and Logging
    //
    /// Address book metrics watch channel.
    ///
    /// Used for logging diagnostics.
    address_metrics: watch::Receiver<AddressMetrics>,

    /// The last time we logged a message about the peer set size
    last_peer_log: Option<Instant>,

    /// The configured maximum number of peers that can be in the
    /// peer set per connection limit key, defaults to [`crate::constants::DEFAULT_MAX_CONNS_PER_IP`]
    max_conns_per_ip: usize,

    /// The network of this peer set.
    network: Network,
}

impl<D, C> Drop for PeerSet<D, C>
where
    D: Discover<Key = PeerSocketAddr, Service = LoadTrackedClient> + Unpin,
    D::Error: Into<BoxError>,
    C: ChainTip,
{
    fn drop(&mut self) {
        // We don't have access to the current task (if any), so we just drop everything we can.
        let waker = noop_waker();
        let mut cx = Context::from_waker(&waker);

        self.shut_down_tasks_and_channels(&mut cx);
    }
}

impl<D, C> PeerSet<D, C>
where
    D: Discover<Key = PeerSocketAddr, Service = LoadTrackedClient> + Unpin,
    D::Error: Into<BoxError>,
    C: ChainTip,
{
    #[allow(clippy::too_many_arguments)]
    /// Construct a peerset which uses `discover` to manage peer connections.
    ///
    /// Arguments:
    /// - `config`: configures the peer set connection limit;
    /// - `block_gossip_peer_ips`: inbound peer IPs that must always receive block inventory broadcasts.
    /// - `discover`: handles peer connects and disconnects;
    /// - `demand_signal`: requests more peers when all peers are busy (unready);
    /// - `handle_rx`: receives background task handles,
    ///   monitors them to make sure they're still running,
    ///   and shuts down all the tasks as soon as one task exits;
    /// - `inv_stream`: receives inventory changes from peers,
    ///   allowing the peer set to direct inventory requests;
    /// - `bans_receiver`: receives a map of banned IP addresses that should be dropped;
    /// - `address_book`: when peer set is busy, it logs address book diagnostics.
    /// - `minimum_peer_version`: endpoint to see the minimum peer protocol version in real time.
    /// - `max_conns_per_ip`: configured maximum number of peers that can be in the
    ///   peer set per connection limit key, defaults to the config value or to
    ///   [`crate::constants::DEFAULT_MAX_CONNS_PER_IP`].
    pub fn new(
        config: &Config,
        block_gossip_peer_ips: Vec<IpAddr>,
        discover: D,
        demand_signal: mpsc::Sender<MorePeers>,
        handle_rx: tokio::sync::oneshot::Receiver<Vec<JoinHandle<Result<(), BoxError>>>>,
        inv_stream: broadcast::Receiver<InventoryChange>,
        bans_receiver: watch::Receiver<BanList>,
        address_metrics: watch::Receiver<AddressMetrics>,
        minimum_peer_version: MinimumPeerVersion<C>,
        max_conns_per_ip: Option<usize>,
    ) -> Self {
        let (stall_event_tx, stall_event_rx) = tokio_mpsc::unbounded_channel();
        let (inv_failure_tx, inv_failure_rx) = tokio_mpsc::unbounded_channel();
        Self {
            // New peers
            discover,
            demand_signal,
            // Banned peers
            bans_receiver,

            // Stall tracking
            find_response_stalls: FindResponseStallTracker::new(),
            stall_event_rx,
            stall_event_tx,

            // Failed block request tracking
            inv_failure_rx,
            inv_failure_tx,

            // Ready peers
            ready_services: HashMap::new(),
            // Request Routing
            find_peer_queue: VecDeque::new(),
            queued_find_requests: VecDeque::new(),
            queued_find_request_timer: None,
            inventory_registry: InventoryRegistry::new(inv_stream),
            queued_broadcast_all: None,
            block_gossip_peer_ips: block_gossip_peer_ips.into_iter().collect(),
            zcashd_compat_peer_keys: HashSet::new(),
            serving_peer_keys: HashSet::new(),
            queued_inv_requests: VecDeque::new(),
            queued_request_waker: Arc::default(),
            queued_sidecar_broadcast: None,

            // Busy peers
            unready_services: FuturesUnordered::new(),
            cancel_handles: HashMap::new(),

            // Peer validation
            minimum_peer_version,
            peerset_total_connection_limit: config.peerset_total_connection_limit(),

            // Background tasks
            handle_rx,
            guards: futures::stream::FuturesUnordered::new(),

            // Metrics
            last_peer_log: None,
            address_metrics,

            max_conns_per_ip: max_conns_per_ip.unwrap_or(config.max_connections_per_ip),

            network: config.network.clone(),
        }
    }

    /// Returns a waker for peer events, which also notifies [`poll_peer_set_on_notify`] while
    /// requests are queued.
    fn peer_event_waker(&self, cx: &Context<'_>) -> Waker {
        if self.queued_inv_requests.is_empty() && self.queued_find_requests.is_empty() {
            return cx.waker().clone();
        }

        self.queued_request_waker
            .peer_set_waker
            .register(cx.waker());
        Waker::from(self.queued_request_waker.clone())
    }

    /// Check background task handles to make sure they're still running.
    ///
    /// Never returns `Ok`.
    ///
    /// If any background task exits, shuts down all other background tasks,
    /// and returns an error. Otherwise, returns `Pending`, and registers a wakeup for
    /// receiving the background tasks, or the background tasks exiting.
    fn poll_background_errors(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), BoxError>> {
        futures::ready!(self.receive_tasks_if_needed(cx))?;

        // Return Pending if all background tasks are still running.
        match futures::ready!(Pin::new(&mut self.guards).poll_next(cx)) {
            Some(res) => {
                info!(
                    background_tasks = %self.guards.len(),
                    "a peer set background task exited, shutting down other peer set tasks"
                );

                self.shut_down_tasks_and_channels(cx);

                // Flatten the join result and inner result, and return any errors.
                res.map_err(Into::into)
                    // TODO: replace with Result::flatten when it stabilises (#70142)
                    .and_then(convert::identity)?;

                // Turn Ok() task exits into errors.
                Poll::Ready(Err("a peer set background task exited".into()))
            }

            None => {
                self.shut_down_tasks_and_channels(cx);
                Poll::Ready(Err("all peer set background tasks have exited".into()))
            }
        }
    }

    /// Receive background tasks, if they've been sent on the channel, but not consumed yet.
    ///
    /// Returns a result representing the current task state, or `Poll::Pending` if the background
    /// tasks should be polled again to check their state.
    fn receive_tasks_if_needed(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), BoxError>> {
        if self.guards.is_empty() {
            // Return Pending if the tasks have not been sent yet.
            let handles = futures::ready!(Pin::new(&mut self.handle_rx).poll(cx));

            match handles {
                // The tasks have been sent, but not consumed yet.
                Ok(handles) => {
                    // Currently, the peer set treats an empty background task set as an error.
                    //
                    // TODO: refactor `handle_rx` and `guards` into an enum
                    //       for the background task state: Waiting/Running/Shutdown.
                    assert!(
                        !handles.is_empty(),
                        "the peer set requires at least one background task"
                    );

                    self.guards.extend(handles);

                    Poll::Ready(Ok(()))
                }

                // The sender was dropped without sending the tasks.
                Err(_) => Poll::Ready(Err(
                    "sender did not send peer background tasks before it was dropped".into(),
                )),
            }
        } else {
            Poll::Ready(Ok(()))
        }
    }

    /// Shut down:
    /// - services by dropping the service lists
    /// - background tasks via their join handles or cancel handles
    /// - channels by closing the channel
    fn shut_down_tasks_and_channels(&mut self, cx: &mut Context<'_>) {
        // Drop services and cancel their background tasks.
        self.ready_services = HashMap::new();

        for (_peer_key, handle) in self.cancel_handles.drain() {
            let _ = handle.send(CancelClientWork);
        }
        self.unready_services = FuturesUnordered::new();

        // Refuse queued requests, because there are no peers left to route them to.
        self.queued_inv_requests.clear();
        self.queued_find_requests.clear();
        self.queued_find_request_timer = None;

        // Close the MorePeers channel for all senders,
        // so we don't add more peers to a shut down peer set.
        self.demand_signal.close_channel();

        // Shut down background tasks, ignoring pending polls.
        self.handle_rx.close();
        let _ = self.receive_tasks_if_needed(cx);
        for guard in self.guards.iter() {
            guard.abort();
        }
    }

    /// Checks for newly ready, disconnects from outdated peers, and polls ready peer errors.
    fn poll_peers(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), BoxError>> {
        // Busy connections must remain eligible too: neither a ban nor an upgrade waits for
        // their request channel to regain capacity.
        self.disconnect_from_ineligible_peers();

        // Check for newly ready peers, including newly added peers (which are added as unready).
        // So it needs to run after `poll_discover()`. Registers a wakeup if there are any unready
        // peers.
        //
        // Request-channel backpressure is bounded independently of the wire response timeout.
        let _poll_pending_or_ready: Poll<Option<()>> = self.poll_unready(cx)?;

        // Check for failures in ready peers, removing newly errored or disconnected peers.
        // So it needs to run after `poll_unready()`.
        let ready_peers = self.poll_ready_peer_errors(cx);
        self.find_peer_queue.retain(|(key, _)| {
            let connected =
                self.ready_services.contains_key(key) || self.cancel_handles.contains_key(key);
            if !connected {
                self.find_response_stalls.clear(*key);
            }
            connected
        });
        ready_peers.map(Ok)
    }

    /// Check busy peer services for request completion or errors.
    ///
    /// Move newly ready services to the ready list if they are for peers with supported protocol
    /// versions, otherwise they are dropped. Also drop failed services.
    ///
    /// Never returns an error.
    ///
    /// Returns `Ok(Some(())` if at least one peer became ready, `Poll::Pending` if there are
    /// unready peers, but none became ready, and `Ok(None)` if the unready peers were empty.
    ///
    /// If there are any remaining unready peers, registers a wakeup for the next time one becomes
    /// ready. If there are no unready peers, doesn't register any wakeups. (Since wakeups come
    /// from peers, there needs to be at least one peer to register a wakeup.)
    fn poll_unready(&mut self, cx: &mut Context<'_>) -> Poll<Result<Option<()>, BoxError>> {
        let mut result = Poll::Pending;

        // # Correctness
        //
        // `poll_next()` must always be called, because `self.unready_services` could have been
        // empty before the call to `self.poll_ready()`.
        //
        // > When new futures are added, `poll_next` must be called in order to begin receiving
        // > wake-ups for new futures.
        //
        // <https://docs.rs/futures/latest/futures/stream/futures_unordered/struct.FuturesUnordered.html>
        //
        // Returns Pending if we've finished processing the unready service changes,
        // but there are still some unready services.
        loop {
            // No ready peers left, but there are some unready peers pending.
            let Poll::Ready(ready_peer) = Pin::new(&mut self.unready_services).poll_next(cx) else {
                break;
            };

            match ready_peer {
                // No unready peers in the list.
                None => {
                    // If we've finished processing the unready service changes, and there are no
                    // unready services left, it doesn't make sense to return Pending, because
                    // their stream is terminated. But when we add more unready peers and call
                    // `poll_next()`, its termination status will be reset, and it will receive
                    // wakeups again.
                    if result.is_pending() {
                        result = Poll::Ready(Ok(None));
                    }

                    break;
                }

                // Unready -> Ready
                Some(Ok((key, svc))) => {
                    trace!(?key, "service became ready");

                    if self.bans_receiver.borrow().is_banned(key.ip()) {
                        warn!(?key, "service is banned, dropping service");
                        std::mem::drop(svc);
                        let cancel = self.cancel_handles.remove(&key);
                        debug_assert!(
                            cancel.is_some(),
                            "missing cancel handle for banned unready peer"
                        );
                        continue;
                    }

                    self.push_ready(true, key, svc);

                    // Return Ok if at least one peer became ready.
                    result = Poll::Ready(Ok(Some(())));
                }

                // Unready -> Canceled
                Some(Err((key, UnreadyError::Canceled))) => {
                    // A service be canceled because we've connected to the same service twice.
                    // In that case, there is a cancel handle for the peer address,
                    // but it belongs to the service for the newer connection.
                    trace!(
                        ?key,
                        duplicate_connection = self.cancel_handles.contains_key(&key),
                        "service was canceled, dropping service"
                    );
                }
                Some(Err((key, UnreadyError::CancelHandleDropped(_)))) => {
                    // Similarly, services with dropped cancel handes can have duplicates.
                    trace!(
                        ?key,
                        duplicate_connection = self.cancel_handles.contains_key(&key),
                        "cancel handle was dropped, dropping service"
                    );
                }
                Some(Err((key, UnreadyError::TimedOut))) => {
                    debug!(?key, "service readiness timed out, dropping service");
                    let cancel = self.cancel_handles.remove(&key);
                    assert!(
                        cancel.is_some(),
                        "timed-out unready service has a cancel handle"
                    );
                }

                // Unready -> Errored
                Some(Err((key, UnreadyError::Inner(error)))) => {
                    debug!(%error, "service failed while unready, dropping service");

                    let cancel = self.cancel_handles.remove(&key);
                    assert!(cancel.is_some(), "missing cancel handle");
                }
            }
        }

        result
    }

    /// Checks previously ready peer services for errors.
    ///
    /// The only way these peer `Client`s can become unready is when we send them a request,
    /// because the peer set has exclusive access to send requests to each peer. (If an inbound
    /// request is in progress, it will be handled, then our request will be sent by the connection
    /// task.)
    ///
    /// Returns `Poll::Ready` if there are some ready peers, and `Poll::Pending` if there are no
    /// ready peers. Registers a wakeup if any peer has failed due to a disconnection, hang, or protocol error.
    ///
    /// # Panics
    ///
    /// If any peers somehow became unready without being sent a request. This indicates a bug in the peer set, where requests
    /// are sent to peers without putting them in `unready_peers`.
    fn poll_ready_peer_errors(&mut self, cx: &mut Context<'_>) -> Poll<()> {
        let mut previous = HashMap::new();
        std::mem::swap(&mut previous, &mut self.ready_services);

        // TODO: consider only checking some peers each poll (for performance reasons),
        //       but make sure we eventually check all of them.
        for (key, mut svc) in previous.drain() {
            let Poll::Ready(peer_readiness) = Pin::new(&mut svc).poll_ready(cx) else {
                unreachable!(
                    "unexpected unready peer: peers must be put into the unready_peers list \
                     after sending them a request"
                );
            };

            match peer_readiness {
                // Still ready, add it back to the list.
                Ok(()) => {
                    if self.bans_receiver.borrow().is_banned(key.ip()) {
                        debug!(?key, "service ip is banned, dropping service");
                        std::mem::drop(svc);
                        continue;
                    }

                    self.push_ready(false, key, svc)
                }

                // Ready -> Errored
                Err(error) => {
                    debug!(%error, "service failed while ready, dropping service");

                    // Ready services can just be dropped, they don't need any cleanup.
                    std::mem::drop(svc);
                }
            }
        }

        if self.ready_services.is_empty() {
            Poll::Pending
        } else {
            Poll::Ready(())
        }
    }

    /// Returns the number of peer connections Zebra already has with the
    /// provided IPv4 address, or in the same IPv6 `/64` subnet.
    ///
    /// Peer set admission uses this count to enforce
    /// [`Config::max_connections_per_ip`](crate::config::Config).
    ///
    /// # Performance
    ///
    /// This method is `O(connected peers)`, so it should not be called from a loop
    /// that is already iterating through the peer set.
    fn num_peers_with_ip(&self, ip: IpAddr) -> usize {
        let limit_key = connection_limit_key(ip);
        self.ready_services
            .keys()
            .chain(self.cancel_handles.keys())
            .filter(|addr| connection_limit_key(addr.ip()) == limit_key)
            .count()
    }

    /// Returns `true` if Zebra is already connected to the IP and port in `addr`.
    fn has_peer_with_addr(&self, addr: PeerSocketAddr) -> bool {
        self.ready_services.contains_key(&addr) || self.cancel_handles.contains_key(&addr)
    }

    /// Processes the entire list of newly inserted or removed services.
    ///
    /// Puts inserted services in the unready list.
    /// Drops removed services, after cancelling any pending requests.
    ///
    /// If the peer connector channel is closed, returns an error.
    ///
    /// Otherwise, returns `Ok` if it discovered at least one peer, or `Poll::Pending` if it didn't
    /// discover any peers. Always registers a wakeup for new peers, even when it returns `Ok`.
    fn poll_discover(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), BoxError>> {
        // Return pending if there are no peers in the list.
        let mut result = Poll::Pending;

        loop {
            // If we've emptied the list, finish looping, otherwise process the new peer.
            let Poll::Ready(discovered) = Pin::new(&mut self.discover).poll_discover(cx) else {
                break;
            };

            // If the change channel has a permanent error, return that error.
            let change = discovered
                .ok_or("discovery stream closed")?
                .map_err(Into::into)?;

            // Otherwise we have successfully processed a peer.
            result = Poll::Ready(Ok(()));

            // Process each change.
            match change {
                Change::Remove(key) => {
                    trace!(?key, "got Change::Remove from Discover");
                    self.remove(&key);
                }
                Change::Insert(key, svc) => {
                    // We add peers as unready, so that we:
                    // - always do the same checks on every ready peer, and
                    // - check for any errors that happened right after the handshake
                    trace!(?key, "got Change::Insert from Discover");

                    // # Security
                    //
                    // Drop the new peer if we are already connected to it.
                    // Preferring old connections avoids connection thrashing.
                    if self.has_peer_with_addr(key) {
                        std::mem::drop(svc);
                        continue;
                    }

                    // Classify sidecars before the per-IP cap: a trusted sidecar
                    // reconnecting from a new ephemeral port must not be blocked
                    // by its own not-yet-swept dead connection still counting
                    // against `max_conns_per_ip` (which defaults to 1), or the
                    // wallet would silently stop following the chain.
                    let is_sidecar = self.is_zcashd_compat_peer(&svc);

                    // # Security
                    //
                    // drop the new peer if there are already `max_conns_per_ip` peers with
                    // the same IPv4 address, or in the same IPv6 `/64` subnet. Sidecars are
                    // exempt: they are trusted, and the listener already caps their inbound
                    // slots.
                    if !is_sidecar && self.num_peers_with_ip(key.ip()) >= self.max_conns_per_ip {
                        std::mem::drop(svc);
                        continue;
                    }

                    if is_sidecar {
                        self.zcashd_compat_peer_keys.insert(key);
                    }

                    if svc.remote_services().contains(PeerServices::NODE_NETWORK) {
                        self.serving_peer_keys.insert(key);
                    } else {
                        self.serving_peer_keys.remove(&key);
                    }
                    self.inventory_registry.clear_peer_failures(key);

                    self.find_peer_queue
                        .push_back((key, svc.connection_info().clone()));
                    self.push_unready(key, svc);
                }
            }
        }

        result
    }

    /// Disconnects banned and outdated connections, whether ready or backpressured.
    fn disconnect_from_ineligible_peers(&mut self) {
        let minimum_version = self.minimum_peer_version.current();
        let bans = self.bans_receiver.borrow();
        self.find_peer_queue.retain(|(key, info)| {
            !bans.is_banned(key.ip()) && info.remote.version >= minimum_version
        });
        self.ready_services.retain(|key, service| {
            let eligible = !bans.is_banned(key.ip()) && service.remote_version() >= minimum_version;
            if !eligible {
                self.find_response_stalls.clear(*key);
            }
            eligible
        });
        for pending in Pin::new(&self.unready_services).iter_pin_ref() {
            let pending = pending.get_ref();
            let key = pending
                .key
                .as_ref()
                .expect("unready services retain their key until completion");
            // Canceled futures can outlive their connection and share a replacement's address.
            if !self
                .cancel_handles
                .get(key)
                .is_some_and(|cancel| cancel.is_connected_to(&pending.cancel))
            {
                continue;
            }
            let service = pending
                .service
                .as_ref()
                .expect("unready services retain their client until completion");
            if bans.is_banned(key.ip()) || service.remote_version() < minimum_version {
                self.find_response_stalls.clear(*key);
                if let Some(cancel) = self.cancel_handles.remove(key) {
                    let _ = cancel.send(CancelClientWork);
                }
            }
        }
    }

    /// Takes a ready service by key.
    fn take_ready_service(&mut self, key: &D::Key) -> Option<D::Service> {
        if let Some(svc) = self.ready_services.remove(key) {
            assert!(
                !self.cancel_handles.contains_key(key),
                "cancel handles are only used for unready service work"
            );

            Some(svc)
        } else {
            None
        }
    }

    /// Whether an event belongs to the currently connected incarnation of an address.
    fn is_current_connection(&self, addr: PeerSocketAddr, info: &Arc<ConnectionInfo>) -> bool {
        self.find_peer_queue
            .iter()
            .any(|(key, current)| *key == addr && Arc::ptr_eq(current, info))
    }

    /// Drains pending stall/clear events from tracked routing futures and
    /// disconnects peers that have exceeded the stall threshold. The peer's
    /// TCP connection is closed when its service is dropped; address book and
    /// ban list are untouched, so the peer is free to reconnect.
    fn drain_stall_events(&mut self, cx: &mut Context<'_>) {
        while let Poll::Ready(Some((addr, connection_info, outcome))) =
            self.stall_event_rx.poll_recv(cx)
        {
            // A delayed response from an old connection must not affect its replacement.
            if !self.is_current_connection(addr, &connection_info) {
                continue;
            }
            match outcome {
                StallOutcome::Stall => {
                    if self.find_response_stalls.record_stall(addr) {
                        info!(
                            ?addr,
                            "dropping stalled peer: exceeded FindBlocks/FindHeaders stall threshold",
                        );
                        self.remove(&addr);
                    }
                }
                StallOutcome::Clear => self.find_response_stalls.clear(addr),
            }
        }
    }

    /// Records block requests that peers failed to answer in the inventory registry.
    ///
    /// Peers declare their own services, and request timeouts don't mark inventory as missing. So
    /// otherwise a peer that falsely claims to serve blocks, and withholds them, could get every
    /// retry for a block.
    fn drain_inv_failures(&mut self, cx: &mut Context<'_>) {
        while let Poll::Ready(Some((peer, connection_info, hash))) =
            self.inv_failure_rx.poll_recv(cx)
        {
            if self.is_current_connection(peer, &connection_info) {
                self.inventory_registry.register_failed(hash, peer);
            }
        }
    }

    /// Remove the service corresponding to `key` from the peer set.
    ///
    /// Drops the service, cancelling any pending request or response to that peer.
    /// If the peer does not exist, does nothing.
    fn remove(&mut self, key: &D::Key) {
        self.find_response_stalls.clear(*key);
        self.find_peer_queue.retain(|(addr, _)| addr != key);
        self.zcashd_compat_peer_keys.remove(key);
        self.serving_peer_keys.remove(key);
        if let Some((_, remaining_sidecars)) = self.queued_sidecar_broadcast.as_mut() {
            remaining_sidecars.remove(key);
            if remaining_sidecars.is_empty() {
                self.queued_sidecar_broadcast = None;
            }
        }

        if let Some(ready_service) = self.take_ready_service(key) {
            // A ready service has no work to cancel, so just drop it.
            std::mem::drop(ready_service);
        } else if let Some(handle) = self.cancel_handles.remove(key) {
            // Cancel the work, implicitly dropping the cancel handle.
            // The service future returns a `Canceled` error,
            // making `poll_unready` drop the service.
            let _ = handle.send(CancelClientWork);
        }
    }

    /// Adds a ready service to the ready list if it's for a peer with a supported version.
    /// If `was_unready` is true, also removes the peer's cancel handle.
    ///
    /// If the service is for a connection to an outdated peer, the service is dropped.
    fn push_ready(&mut self, was_unready: bool, key: D::Key, svc: D::Service) {
        let cancel = self.cancel_handles.remove(&key);
        assert_eq!(
            cancel.is_some(),
            was_unready,
            "missing or unexpected cancel handle"
        );

        if svc.remote_version() >= self.minimum_peer_version.current() {
            self.ready_services.insert(key, svc);
        } else {
            std::mem::drop(svc);
        }
    }

    /// Adds a busy service to the unready list if it's for a peer with a supported version,
    /// and adds a cancel handle for the service's current request.
    ///
    /// If the service is for a connection to an outdated peer, the request is cancelled and the
    /// service is dropped.
    fn push_unready(&mut self, key: D::Key, svc: D::Service) {
        let peer_version = svc.remote_version();
        let (tx, rx) = oneshot::channel();

        self.unready_services.push(UnreadyService {
            key: Some(key),
            service: Some(svc),
            cancel: rx,
            readiness_timeout: tokio::time::sleep(crate::constants::REQUEST_TIMEOUT),
            _req: PhantomData,
        });

        if peer_version >= self.minimum_peer_version.current() {
            self.cancel_handles.insert(key, tx);
        } else {
            self.find_response_stalls.clear(key);
            // Reject delayed response events before the next peer-set poll.
            self.find_peer_queue.retain(|(addr, _)| *addr != key);
            // Cancel any request made to the service because it is using an outdated protocol
            // version.
            let _ = tx.send(CancelClientWork);
        }
    }

    /// Performs P2C on `self.ready_services` to randomly select a less-loaded ready service.
    fn select_ready_p2c_peer(&self) -> Option<D::Key> {
        self.select_p2c_peer_from_list(&self.ready_services.keys().copied().collect())
    }

    /// Rotates find requests across ready connections, restricting them to serving peers until
    /// the queued wait expires, unless no serving peers are connected.
    fn select_ready_find_peer(&mut self, allow_non_serving: bool) -> Option<D::Key> {
        if self.ready_services.is_empty() {
            return None;
        }

        let serving_only = !allow_non_serving && !self.serving_peer_keys.is_empty();
        let (index, (key, _)) = self
            .find_peer_queue
            .iter()
            .enumerate()
            .find(|(_, (key, _))| {
                self.ready_services.contains_key(key)
                    && (!serving_only || self.serving_peer_keys.contains(key))
            })?;
        let key = *key;
        self.find_peer_queue.rotate_left(index + 1);
        Some(key)
    }

    /// Performs P2C on `ready_service_list` to randomly select a less-loaded ready service.
    #[allow(clippy::unwrap_in_result)]
    fn select_p2c_peer_from_list(&self, ready_service_list: &HashSet<D::Key>) -> Option<D::Key> {
        match ready_service_list.len() {
            0 => None,
            1 => Some(
                *ready_service_list
                    .iter()
                    .next()
                    .expect("just checked there is one service"),
            ),
            len => {
                // Choose 2 random peers, then return the least loaded of those 2 peers.
                let (a, b) = {
                    let idxs = rand::seq::index::sample(&mut rand::rng(), len, 2);
                    let a = idxs.index(0);
                    let b = idxs.index(1);

                    let a = *ready_service_list
                        .iter()
                        .nth(a)
                        .expect("sample returns valid indexes");
                    let b = *ready_service_list
                        .iter()
                        .nth(b)
                        .expect("sample returns valid indexes");

                    (a, b)
                };

                let a_load = self.query_load(&a).expect("supplied services are ready");
                let b_load = self.query_load(&b).expect("supplied services are ready");

                let selected = if a_load <= b_load { a } else { b };

                trace!(
                    a.key = ?a,
                    a.load = ?a_load,
                    b.key = ?b,
                    b.load = ?b_load,
                    selected = ?selected,
                    ?len,
                    "selected service by p2c"
                );

                Some(selected)
            }
        }
    }

    /// Randomly chooses `max_peers` ready services, ignoring service load.
    ///
    /// The chosen peers are unique, but their order is not fully random.
    fn select_random_ready_peers(&self, max_peers: usize) -> Vec<D::Key> {
        use rand::seq::IteratorRandom;

        self.ready_services
            .keys()
            .copied()
            .sample(&mut rand::rng(), max_peers)
    }

    /// Randomly chooses ready peers for a sidecar broadcast, always including
    /// configured zcashd compat sidecar peers.
    fn select_sidecar_broadcast_peers(&self, max_peers: usize) -> Vec<D::Key> {
        use rand::seq::IteratorRandom;

        let mut selected_peers: Vec<_> = self
            .zcashd_compat_peer_keys
            .iter()
            .filter(|key| self.ready_services.contains_key(*key))
            .copied()
            .collect();

        selected_peers.extend(
            self.ready_services
                .keys()
                .filter(|key| !self.zcashd_compat_peer_keys.contains(key))
                .copied()
                .sample(&mut rand::rng(), max_peers),
        );

        selected_peers
    }

    /// Returns true if `service` is a configured zcashd sidecar peer.
    ///
    /// Only used to classify peers once, when they are inserted into the peer
    /// set; every later check uses the O(1) [`Self::zcashd_compat_peer_keys`]
    /// set instead.
    fn is_zcashd_compat_peer(&self, service: &D::Service) -> bool {
        self.block_gossip_peer_ips
            .iter()
            .any(|ip| service.is_inbound_direct_from_ip(ip))
    }

    /// Forgets sidecar keys whose peer has disconnected.
    ///
    /// Sidecars reconnect from new ephemeral ports, and services are dropped on many
    /// paths (bans, version downgrades, cancelled requests), so this runs every
    /// poll cycle to keep the set from accumulating stale keys — otherwise a
    /// reused port could inherit a stale sidecar's stall-tracking exemption.
    fn prune_disconnected_sidecar_keys(&mut self) {
        Self::retain_connected_keys(
            &mut self.zcashd_compat_peer_keys,
            &self.ready_services,
            &self.cancel_handles,
        );
    }

    /// Removes the keys of disconnected peers from `keys`.
    ///
    /// A connected peer is either ready or unready with a registered cancel
    /// handle; a key in neither map belongs to a dropped connection.
    fn retain_connected_keys(
        keys: &mut HashSet<D::Key>,
        ready_services: &HashMap<D::Key, D::Service>,
        cancel_handles: &HashMap<D::Key, oneshot::Sender<CancelClientWork>>,
    ) {
        if keys.is_empty() {
            return;
        }

        keys.retain(|key| ready_services.contains_key(key) || cancel_handles.contains_key(key));
    }

    /// Accesses a ready endpoint by `key` and returns its current load.
    ///
    /// Returns `None` if the service is not in the ready service list.
    fn query_load(&self, key: &D::Key) -> Option<<D::Service as Load>::Metric> {
        let svc = self.ready_services.get(key);
        svc.map(|svc| svc.load())
    }

    /// Routes find requests fairly and other requests using P2C load-balancing.
    fn route_request(&mut self, req: Request) -> <Self as tower::Service<Request>>::Future {
        let is_find_request = matches!(
            &req,
            Request::FindBlocks { .. } | Request::FindHeaders { .. }
        );
        let peer_key = if is_find_request {
            // Fast empty responses must not gain a latency-based routing advantage.
            self.select_ready_find_peer(false)
        } else {
            self.select_ready_p2c_peer()
        };

        if let Some(peer_key) = peer_key {
            if is_find_request && !self.serving_peer_keys.contains(&peer_key) {
                metrics::counter!(
                    "zcash.net.peer_set.find_fallback",
                    "reason" => "no_serving_peer",
                    "network" => network_kind_label(&self.network)
                )
                .increment(1);
            }
            return self.call_ready_peer(peer_key, req);
        }

        if is_find_request {
            self.queued_find_requests
                .retain(|queued| !queued.response_sender.is_canceled());
            if self.queued_find_requests.len() < MAX_QUEUED_FIND_REQUESTS {
                return self.queue_find_request(req);
            }
            if let Some(peer_key) = self.select_ready_find_peer(true) {
                metrics::counter!(
                    "zcash.net.peer_set.find_fallback",
                    "reason" => "queue_full",
                    "network" => network_kind_label(&self.network)
                )
                .increment(1);
                return self.call_ready_peer(peer_key, req);
            }
        }

        async move {
            // Let other tasks run, so a retry request might get different ready peers.
            tokio::task::yield_now().await;
            Err(SharedPeerError::from(PeerError::NoReadyPeers))
        }
        .map_err(Into::into)
        .boxed()
    }

    /// Tries to route a request to a ready peer that advertised that inventory,
    /// falling back to a ready peer that isn't missing the inventory.
    ///
    /// If no ready peer can be used, but a busy peer might have the block, queues the request
    /// until one of those busy peers becomes ready. See [`Self::route_queued_inv_requests`].
    ///
    /// Otherwise, returns a synthetic [`NotFoundRegistry`](PeerError::NotFoundRegistry) error.
    ///
    /// See [`Self::select_inv_route`] for details.
    fn route_inv(
        &mut self,
        req: Request,
        hash: InventoryHash,
    ) -> <Self as tower::Service<Request>>::Future {
        match self.select_inv_route(&req, hash) {
            InvRoute::Peer(peer) => self.call_inv_peer(peer, req, hash),
            InvRoute::Wait => self.queue_inv_request(req, hash),
            InvRoute::Refuse => {
                tracing::debug!(
                    ?hash,
                    "no peer that might have the inventory can take the request, failing request"
                );
                metrics::counter!(
                    "zcash.net.peer_set.inventory_refusal",
                    "kind" => "instant",
                    "network" => network_kind_label(&self.network)
                )
                .increment(1);

                async move {
                    // Let other tasks run, so a retry request might get different ready peers.
                    tokio::task::yield_now().await;

                    Err(not_found_registry_error(hash))
                }
                .boxed()
            }
        }
    }

    /// Returns where to route an inventory request for `hash`.
    ///
    /// Prefers a ready peer that advertised the inventory, then a ready peer that isn't missing it.
    /// For blocks, that fallback prefers peers that advertised [`PeerServices::NODE_NETWORK`].
    /// Non-serving peers usually answer `notfound` for historic blocks, so they only get a block
    /// if no connected serving peer might have it, or if some peer advertised it (so it is recent).
    ///
    /// Only block requests wait for busy peers: transaction downloads don't retry, so waiting
    /// would only slow them down.
    ///
    /// Peers that recently failed to return a block are skipped for that block, unless no other
    /// peer might have it.
    ///
    /// Uses P2C to choose the least loaded peer in each list.
    fn select_inv_route(&self, request: &Request, hash: InventoryHash) -> InvRoute<D::Key> {
        // # Security
        //
        // Peers that recently failed to return this block lose any preference for it, and are only
        // asked again as a last resort. Otherwise, a peer that falsely claims to serve blocks and
        // withholds them could get every retry.
        let failed_peer_list: HashSet<PeerSocketAddr> = self
            .inventory_registry
            .failed_peers(hash)
            .copied()
            .collect();
        let has_failed = |addr: &PeerSocketAddr| failed_peer_list.contains(addr);
        let advertiser = match request {
            Request::BlockDownload { advertiser, .. } => advertiser.filter(|addr| {
                // Registry advertisements already exclude missing peers. Apply the same rule
                // to carried provenance without allocating a fallback set on the fast path.
                !self
                    .inventory_registry
                    .missing_peers(hash)
                    .any(|missing| missing == addr)
            }),
            _ => None,
        };

        let advertising_peers: Vec<PeerSocketAddr> = self
            .inventory_registry
            .advertising_peers(hash)
            .copied()
            .chain(advertiser)
            .collect();
        let advertising_peer_list = advertising_peers
            .iter()
            .filter(|addr| self.ready_services.contains_key(addr) && !has_failed(addr))
            .copied()
            .collect();

        // # Security
        //
        // Choose a random, less-loaded peer with the inventory.
        //
        // If we chose the first peer in HashMap order,
        // peers would be able to influence our choice by switching addresses.
        // But we need the choice to be random,
        // so that a peer can't provide all our inventory responses.
        if let Some(peer) = self.select_p2c_peer_from_list(&advertising_peer_list) {
            tracing::trace!(?hash, ?peer, "routing to a peer which advertised inventory");
            return InvRoute::Peer(peer);
        }

        let missing_peer_list: HashSet<PeerSocketAddr> = self
            .inventory_registry
            .missing_peers(hash)
            .copied()
            .collect();

        // Only block routing depends on peer services: non-serving peers can still serve
        // mempool transactions.
        let is_block = matches!(hash, InventoryHash::Block(_));
        let ready_peers_with_services = |serving: bool| -> HashSet<PeerSocketAddr> {
            self.ready_services
                .keys()
                .filter(|addr| {
                    !missing_peer_list.contains(addr)
                        && !has_failed(addr)
                        && (!is_block || self.serving_peer_keys.contains(addr) == serving)
                })
                .copied()
                .collect()
        };
        let preferred_peer_list = ready_peers_with_services(true);

        // Security: choose a random, less-loaded peer that might have the inventory.
        if let Some(peer) = self.select_p2c_peer_from_list(&preferred_peer_list) {
            tracing::trace!(?hash, ?peer, "routing to a peer that might have inventory");
            return InvRoute::Peer(peer);
        }

        if !is_block {
            return InvRoute::Refuse;
        }

        // Peers only advertise recent blocks, which non-serving peers can usually serve. So any
        // busy peer might have an advertised block, but only busy serving peers might have a
        // historic block.
        //
        // Downloaders carry their known advertiser separately from this lossy registry.
        //
        // A disconnected or failed advertiser must not enable speculative non-serving fallback.
        let is_advertised = advertising_peers
            .iter()
            .any(|addr| self.has_peer_with_addr(*addr) && !has_failed(addr));
        let busy_peer_might_have = |failed: bool| {
            self.cancel_handles.keys().any(|key| {
                !missing_peer_list.contains(key)
                    && has_failed(key) == failed
                    && (failed || is_advertised || self.serving_peer_keys.contains(key))
            })
        };
        let other_busy_peer_might_have = busy_peer_might_have(false);

        // Waiting for a busy serving peer is better than asking a non-serving peer, which would
        // most likely answer `notfound` for a historic block, and get marked as missing it.
        if is_advertised || !other_busy_peer_might_have {
            if let Some(peer) = self.select_p2c_peer_from_list(&ready_peers_with_services(false)) {
                tracing::trace!(?hash, ?peer, "routing to a non-serving peer");
                return InvRoute::Peer(peer);
            }
        }

        if other_busy_peer_might_have {
            return InvRoute::Wait;
        }

        // As a last resort, ask a peer that failed to return this block again, or wait for it.
        let failed_ready_peer_list = self
            .ready_services
            .keys()
            .filter(|addr| has_failed(addr) && !missing_peer_list.contains(addr))
            .copied()
            .collect();
        if let Some(peer) = self.select_p2c_peer_from_list(&failed_ready_peer_list) {
            tracing::trace!(
                ?hash,
                ?peer,
                "routing to a peer that failed to return the block"
            );
            return InvRoute::Peer(peer);
        }

        if busy_peer_might_have(true) {
            InvRoute::Wait
        } else {
            InvRoute::Refuse
        }
    }

    /// Sends the inventory request `req` for `hash` to the ready peer `key`, and returns its
    /// response future.
    ///
    /// If a block request fails, reports the failure to the peer set.
    fn call_inv_peer(
        &mut self,
        key: D::Key,
        req: Request,
        hash: InventoryHash,
    ) -> <Self as tower::Service<Request>>::Future {
        // Transaction requests don't wait for busy peers, so there is no preference to lose.
        if !matches!(hash, InventoryHash::Block(_)) {
            return self.call_ready_peer(key, req);
        }

        let connection_info = self
            .ready_services
            .get(&key)
            .expect("inventory routing selected a ready connection")
            .connection_info()
            .clone();
        let fut = self.call_ready_peer(key, req);

        let failure_tx = self.inv_failure_tx.clone();
        async move {
            let result = fut.await;
            if let Err(error) = &result {
                let missing = error
                    .downcast_ref::<SharedPeerError>()
                    .is_some_and(|error| error.inner_debug().contains("NotFoundResponse"));
                if !missing {
                    // Only a completed peer error, including its response timeout, is failure
                    // evidence. Dropping a caller, losing hedge or sync reset remains neutral.
                    let _ = failure_tx.send((key, connection_info, hash));
                }
            }
            result
        }
        .boxed()
    }

    /// Sends `req` to the ready peer `key`, and returns its response future.
    fn call_ready_peer(
        &mut self,
        key: D::Key,
        req: Request,
    ) -> <Self as tower::Service<Request>>::Future {
        let is_find_request = matches!(
            &req,
            Request::FindBlocks { .. } | Request::FindHeaders { .. }
        );
        tracing::trace!(peer_key = ?key, is_find_request, "routing request to peer");
        let mut svc = self
            .take_ready_service(&key)
            .expect("callers select `key` from the ready services just before calling");

        // Configured sidecars are trusted downstream consumers, not upstream sync sources.
        let track_stalls = is_find_request && !self.zcashd_compat_peer_keys.contains(&key);
        let fut = svc.call(req);
        let response = if track_stalls {
            let connection_info = svc.connection_info().clone();
            let stall_tx = self.stall_event_tx.clone();
            async move {
                // Cancellation is not evidence of failure: a caught-up peer can
                // legitimately send no getblocks reply before the caller times out.
                let result = fut.await;
                if let Some(outcome) = classify_find_response(&result) {
                    let _ = stall_tx.send((key, connection_info, outcome));
                }
                result.map_err(Into::into)
            }
            .boxed()
        } else {
            fut.map_err(Into::into).boxed()
        };
        self.push_unready(key, svc);
        response
    }

    /// Waits for a serving peer, falling back to any ready peer once the bounded wait expires.
    fn queue_find_request(&mut self, request: Request) -> ResponseFuture {
        metrics::counter!(
            "zcash.net.peer_set.find_queued",
            "network" => network_kind_label(&self.network)
        )
        .increment(1);
        let (response_sender, response_receiver) = oneshot::channel();
        self.queued_find_requests.push_back(QueuedFindRequest {
            request,
            deadline: tokio::time::Instant::now() + FIND_BUSY_SERVING_PEER_WAIT_TIMEOUT,
            response_sender,
        });

        // Register peer, timer, and cancellation wakeups even if the response is never polled.
        self.queued_request_waker.notify.notify_one();

        async move {
            let response_fut = response_receiver
                .await
                .map_err(|_| SharedPeerError::from(PeerError::NoReadyPeers))?;
            response_fut.await
        }
        .boxed()
    }

    /// Routes queued discovery before inventory, and registers deadline and cancellation wakeups.
    ///
    /// This also runs with no ready peers, so abandoned requests are removed and expired
    /// serving-only waits stop arming the timer.
    fn route_queued_find_requests(&mut self, cx: &mut Context<'_>) {
        if self.queued_find_requests.is_empty() {
            return;
        }

        let now = tokio::time::Instant::now();
        let mut next_deadline = None;
        for _ in 0..self.queued_find_requests.len() {
            let mut queued = self
                .queued_find_requests
                .pop_front()
                .expect("the loop visits only requests that were already queued");
            if queued.response_sender.poll_canceled(cx).is_ready() {
                continue;
            }

            let expired = queued.deadline <= now;
            if let Some(peer) = self.select_ready_find_peer(expired) {
                if !self.serving_peer_keys.contains(&peer) {
                    let reason = if self.serving_peer_keys.is_empty() {
                        "no_serving_peer"
                    } else {
                        "expired"
                    };
                    metrics::counter!(
                        "zcash.net.peer_set.find_fallback",
                        "reason" => reason,
                        "network" => network_kind_label(&self.network)
                    )
                    .increment(1);
                }
                let response_fut = self.call_ready_peer(peer, queued.request);
                // A concurrent cancellation drops the response future, cancelling peer work.
                let _ = queued.response_sender.send(response_fut);
                // Register the newly unready service even if an unrelated peer remains ready.
                // Otherwise an idle Buffer can leave later queued downloads without a wakeup.
                cx.waker().wake_by_ref();
            } else {
                if !expired {
                    // Requests enter this FIFO with monotonically increasing deadlines.
                    next_deadline.get_or_insert(queued.deadline);
                }
                self.queued_find_requests.push_back(queued);
            }
        }

        if let Some(deadline) = next_deadline {
            let timer = self
                .queued_find_request_timer
                .get_or_insert_with(|| Box::pin(tokio::time::sleep_until(deadline)));
            if timer.deadline() != deadline {
                timer.as_mut().reset(deadline);
            }
            if timer.as_mut().poll(cx).is_ready() {
                // The deadline might have passed while routing; retry with a fresh `now`.
                cx.waker().wake_by_ref();
            }
        } else {
            self.queued_find_request_timer = None;
        }
    }

    /// Queues an inventory request until a busy peer that might have `hash` becomes ready.
    ///
    /// The returned future resolves to that peer's response, or to a synthetic
    /// [`NotFoundRegistry`](PeerError::NotFoundRegistry) error if the request is refused.
    fn queue_inv_request(
        &mut self,
        request: Request,
        hash: InventoryHash,
    ) -> <Self as tower::Service<Request>>::Future {
        tracing::debug!(
            ?hash,
            "no ready peer can serve the inventory, waiting for a busy peer"
        );
        let network = network_kind_label(&self.network);
        metrics::counter!("zcash.net.peer_set.inventory_queued", "network" => network).increment(1);

        let deadline = (!matches!(request, Request::BlockDownload { .. }))
            .then(|| tokio::time::Instant::now() + INVENTORY_BUSY_PEER_WAIT_TIMEOUT);

        let (response_sender, response_receiver) = oneshot::channel();
        self.queued_inv_requests.push_back(QueuedInvRequest {
            request,
            hash,
            deadline,
            span: tracing::Span::current(),
            response_sender,
        });

        // Poll the peer set again, so peer events wake `poll_peer_set_on_notify()` while this
        // request is queued.
        self.queued_request_waker.notify.notify_one();

        async move {
            let response = if let Some(deadline) = deadline {
                tokio::time::timeout_at(deadline, response_receiver)
                    .await
                    .ok()
                    .and_then(Result::ok)
            } else {
                // Capacity waiting is not a peer attempt. The downloader bounds the complete
                // operation, including readiness, queueing, retries, and the response.
                response_receiver.await.ok()
            };
            match response {
                Some(response_fut) => response_fut.await,
                None => {
                    tracing::debug!(
                        ?hash,
                        "no busy peer that might have the inventory became ready, failing request"
                    );
                    metrics::counter!(
                        "zcash.net.peer_set.inventory_refusal",
                        "kind" => "waited",
                        "network" => network
                    )
                    .increment(1);

                    Err(not_found_registry_error(hash))
                }
            }
        }
        .boxed()
    }

    /// Routes queued inventory requests to ready peers that might have their inventory, in the
    /// order they were queued. Refuses queued requests once no busy peer might have their
    /// inventory.
    ///
    /// Queued requests are only routed when the peer set is polled. While requests are queued,
    /// peer events notify [`poll_peer_set_on_notify`], which polls the peer set.
    ///
    /// Reconsiders candidates on every poll, including advertisements, failures and removals.
    fn route_queued_inv_requests(&mut self, cx: &mut Context<'_>) {
        // Register caller cancellation too, so an expired operation releases its queue entry
        // even when no connection or inventory event follows it.
        self.queued_inv_requests.retain_mut(|queued| {
            queued
                .deadline
                .is_none_or(|deadline| deadline > tokio::time::Instant::now())
                && queued.response_sender.poll_canceled(cx).is_pending()
        });

        let mut routed_any = false;
        for _ in 0..self.queued_inv_requests.len() {
            let queued = self
                .queued_inv_requests
                .pop_front()
                .expect("each original queue entry is visited once");

            match self.select_inv_route(&queued.request, queued.hash) {
                InvRoute::Peer(peer) => {
                    let _span = queued.span.enter();
                    let response_fut = self.call_inv_peer(peer, queued.request, queued.hash);
                    routed_any = true;
                    // If the receiver was dropped since the check above, dropping the response
                    // future cancels the peer's request.
                    let _ = queued.response_sender.send(response_fut);
                }
                InvRoute::Wait => self.queued_inv_requests.push_back(queued),
                // Dropping the sender refuses the request.
                InvRoute::Refuse => {}
            }
        }

        // Poll the newly busy peer's readiness future even if another peer stays ready.
        if routed_any && !self.queued_inv_requests.is_empty() {
            self.queued_request_waker.notify.notify_one();
        }
    }

    /// Routes the same request to up to `max_peers` ready peers, ignoring return values.
    ///
    /// `max_peers` must be at least one, and at most the number of ready peers.
    fn route_multiple(
        &mut self,
        req: Request,
        max_peers: usize,
    ) -> <Self as tower::Service<Request>>::Future {
        assert!(
            max_peers > 0,
            "requests must be routed to at least one peer"
        );
        assert!(
            max_peers <= self.ready_services.len(),
            "requests can only be routed to ready peers"
        );

        let selected_peers = self.select_random_ready_peers(max_peers);
        self.send_multiple(req, selected_peers)
    }

    /// Sends the same request to the provided ready peers, ignoring return values.
    ///
    /// # Security
    ///
    /// Callers should choose peers randomly, ignoring load.
    /// This avoids favouring malicious peers, because peers can influence their own load.
    ///
    /// The order of peers isn't completely random,
    /// but peer request order is not security-sensitive.
    fn send_multiple(
        &mut self,
        req: Request,
        peers: Vec<D::Key>,
    ) -> <Self as tower::Service<Request>>::Future {
        let futs = FuturesUnordered::new();
        for key in peers {
            let mut svc = self
                .take_ready_service(&key)
                .expect("selected peers are ready");
            futs.push(svc.call(req.clone()).map_err(|_| ()));
            self.push_unready(key, svc);
        }

        async move {
            let results = futs.collect::<Vec<Result<_, _>>>().await;
            tracing::debug!(
                ok.len = results.iter().filter(|r| r.is_ok()).count(),
                err.len = results.iter().filter(|r| r.is_err()).count(),
                "sent peer request to multiple peers"
            );
            Ok(Response::Nil)
        }
        .boxed()
    }

    /// Broadcasts the same request to lots of ready peers, ignoring return values.
    fn route_broadcast(&mut self, req: Request) -> <Self as tower::Service<Request>>::Future {
        // Broadcasts ignore the response
        self.route_multiple(req, self.number_of_peers_to_broadcast())
    }

    /// Broadcasts a request to sampled peers and all configured sidecars.
    ///
    /// Used for requests that sidecars must always receive: block adverts and
    /// pushed transactions. Connected sidecars that are busy with another
    /// request are owed it: the request is queued and delivered by
    /// [`Self::send_queued_sidecar_broadcast`] as soon as they are ready again.
    fn route_sidecar_broadcast(
        &mut self,
        req: Request,
    ) -> <Self as tower::Service<Request>>::Future {
        self.prune_disconnected_sidecar_keys();

        let selected_peers =
            self.select_sidecar_broadcast_peers(self.number_of_peers_to_broadcast());

        // A newer broadcast supersedes any older undelivered one, even one of
        // a different kind. Sidecars only need the latest block advert to stay
        // live: they fetch any blocks in between over the same connection.
        let busy_sidecars: HashSet<D::Key> = self
            .zcashd_compat_peer_keys
            .iter()
            .filter(|key| !self.ready_services.contains_key(*key))
            .copied()
            .collect();
        self.queued_sidecar_broadcast =
            (!busy_sidecars.is_empty()).then(|| (req.clone(), busy_sidecars));

        self.send_multiple(req, selected_peers)
    }

    /// Delivers the queued sidecar broadcast to any owed sidecar peers that
    /// have become ready. See [`Self::route_sidecar_broadcast`].
    fn send_queued_sidecar_broadcast(&mut self) {
        let Some((req, mut remaining_sidecars)) = self.queued_sidecar_broadcast.take() else {
            return;
        };

        // Like `broadcast_all_queued`, don't deliver to peers that were banned
        // while the request was queued.
        let bans = self.bans_receiver.borrow().clone();
        remaining_sidecars.retain(|key| !bans.is_banned(key.ip()));

        let ready_sidecars: Vec<D::Key> = remaining_sidecars
            .iter()
            .filter(|key| self.ready_services.contains_key(*key))
            .copied()
            .collect();
        for key in ready_sidecars {
            remaining_sidecars.remove(&key);

            let mut svc = self
                .take_ready_service(&key)
                .expect("sidecars are ready because they were filtered from ready_services above");
            let req_fut = svc.call(req.clone());
            self.push_unready(key, svc);

            // Detach the response future: the connection cancels requests whose
            // response channel is dropped, and there is no caller left to drive
            // this delivery.
            tokio::spawn(req_fut.map(|_| ()));
        }

        // Drop sidecars that disconnected while the request was queued.
        Self::retain_connected_keys(
            &mut remaining_sidecars,
            &self.ready_services,
            &self.cancel_handles,
        );

        if !remaining_sidecars.is_empty() {
            self.queued_sidecar_broadcast = Some((req, remaining_sidecars));
        }
    }

    /// Broadcasts the same request to all ready peers, ignoring return values.
    fn broadcast_all(&mut self, req: Request) -> <Self as tower::Service<Request>>::Future {
        let ready_peers = self.ready_services.keys().copied().collect();
        let send_multiple_fut = self.send_multiple(req.clone(), ready_peers);
        let Some(mut queued_broadcast_fut_receiver) = self.queue_broadcast_all_unready(&req) else {
            return send_multiple_fut;
        };

        async move {
            let _ = send_multiple_fut.await?;
            while queued_broadcast_fut_receiver.recv().await.is_some() {}
            Ok(Response::Nil)
        }
        .boxed()
    }

    /// If there are unready peers, queues a request to be broadcasted to them and
    /// returns a channel receiver for callers to await the broadcast_all() futures, or
    /// returns None if there are no unready peers.
    fn queue_broadcast_all_unready(
        &mut self,
        req: &Request,
    ) -> Option<tokio::sync::mpsc::Receiver<ResponseFuture>> {
        if !self.cancel_handles.is_empty() {
            /// How many broadcast all futures to send to the channel until the peer set should wait for the channel consumer
            /// to read a message before continuing to send the queued broadcast request to peers that were originally unready.
            const QUEUED_BROADCAST_FUTS_CHANNEL_SIZE: usize = 3;

            let (sender, receiver) = tokio::sync::mpsc::channel(QUEUED_BROADCAST_FUTS_CHANNEL_SIZE);
            let unready_peers: HashSet<_> = self.cancel_handles.keys().cloned().collect();
            let queued = (req.clone(), sender, unready_peers);

            // Drop the existing queued broadcast all request, if any.
            self.queued_broadcast_all = Some(queued);

            Some(receiver)
        } else {
            None
        }
    }

    /// Broadcasts the same requests to all ready peers which were unready when
    /// [`PeerSet::broadcast_all()`] was last called, ignoring return values.
    fn broadcast_all_queued(&mut self) {
        let Some((req, sender, mut remaining_peers)) = self.queued_broadcast_all.take() else {
            return;
        };

        let bans = self.bans_receiver.borrow().clone();
        remaining_peers.retain(|addr| !bans.is_banned(addr.ip()));

        let Ok(reserved_send_slot) = sender.try_reserve() else {
            self.queued_broadcast_all = Some((req, sender, remaining_peers));
            return;
        };

        let peers: Vec<_> = self
            .ready_services
            .keys()
            .filter(|ready_peer| remaining_peers.remove(ready_peer))
            .copied()
            .collect();

        reserved_send_slot.send(self.send_multiple(req.clone(), peers).boxed());

        if !remaining_peers.is_empty() {
            self.queued_broadcast_all = Some((req, sender, remaining_peers));
        }
    }

    /// Given a number of ready peers calculate to how many of them Zebra will
    /// actually send the request to. Return this number.
    pub(crate) fn number_of_peers_to_broadcast(&self) -> usize {
        if self.network.is_regtest() {
            // In regtest, we broadcast to all peers, so that we can test the
            // peer set with a small number of peers.
            self.ready_services.len()
        } else {
            // We are currently sending broadcast messages to a third of the total peers.
            const PEER_FRACTION_TO_BROADCAST: usize = 3;

            // Round up, so that if we have one ready peer, it gets the request.
            div_ceil(self.ready_services.len(), PEER_FRACTION_TO_BROADCAST)
        }
    }

    /// Returns the list of addresses in the peer set.
    fn peer_set_addresses(&self) -> Vec<PeerSocketAddr> {
        self.ready_services
            .keys()
            .chain(self.cancel_handles.keys())
            .cloned()
            .collect()
    }

    /// Logs the peer set size, and any potential connectivity issues.
    fn log_peer_set_size(&mut self) {
        let ready_services_len = self.ready_services.len();
        let unready_services_len = self.unready_services.len();
        trace!(ready_peers = ?ready_services_len, unready_peers = ?unready_services_len);

        let now = Instant::now();

        // These logs are designed to be human-readable in a terminal, at the
        // default Zebra log level. If you need to know the peer set size for
        // every request, use the trace-level logs, or the metrics exporter.
        if let Some(last_peer_log) = self.last_peer_log {
            // Avoid duplicate peer set logs
            if now.duration_since(last_peer_log) < MIN_PEER_SET_LOG_INTERVAL {
                return;
            }
        } else {
            // Suppress initial logs until the peer set has started up.
            // There can be multiple initial requests before the first peer is
            // ready.
            self.last_peer_log = Some(now);
            return;
        }

        self.last_peer_log = Some(now);

        // Log potential duplicate connections.
        let peers = self.peer_set_addresses();

        // Check for duplicates by address and port: these are unexpected and represent a bug.
        let duplicates: Vec<PeerSocketAddr> = peers.iter().duplicates().cloned().collect();

        let mut peer_counts = peers.iter().counts();
        peer_counts.retain(|peer, _count| duplicates.contains(peer));

        if !peer_counts.is_empty() {
            let duplicate_connections: usize = peer_counts.values().sum();

            warn!(
                ?duplicate_connections,
                duplicated_peers = ?peer_counts.len(),
                peers = ?peers.len(),
                "duplicate peer connections in peer set"
            );
        }

        // Check for duplicates by address: these can happen if there are multiple nodes
        // behind a NAT or on a single server.
        let peers: Vec<IpAddr> = peers.iter().map(|addr| addr.ip()).collect();
        let duplicates: Vec<IpAddr> = peers.iter().duplicates().cloned().collect();

        let mut peer_counts = peers.iter().counts();
        peer_counts.retain(|peer, _count| duplicates.contains(peer));

        if !peer_counts.is_empty() {
            let duplicate_connections: usize = peer_counts.values().sum();

            info!(
                ?duplicate_connections,
                duplicated_peers = ?peer_counts.len(),
                peers = ?peers.len(),
                "duplicate IP addresses in peer set"
            );
        }

        // Only log connectivity warnings if all our peers are busy (or there are no peers).
        if ready_services_len > 0 {
            return;
        }

        let address_metrics = *self.address_metrics.borrow();
        if unready_services_len == 0 {
            warn!(
                ?address_metrics,
                "network request with no peer connections. Hint: check your network connection"
            );
        } else {
            info!(?address_metrics, "network request with no ready peers: finding more peers, waiting for {} peers to answer requests",
                  unready_services_len);
        }
    }

    /// Updates the peer set metrics.
    ///
    /// # Panics
    ///
    /// If the peer set size exceeds the connection limit.
    fn update_metrics(&self) {
        let num_ready = self.ready_services.len();
        let num_unready = self.unready_services.len();
        let num_peers = num_ready + num_unready;
        let network = network_kind_label(&self.network);
        metrics::gauge!("pool.num_ready", "network" => network).set(num_ready as f64);
        metrics::gauge!("pool.num_unready", "network" => network).set(num_unready as f64);
        metrics::gauge!("zcash.net.peers", "network" => network).set(num_peers as f64);

        // Security: make sure we haven't exceeded the connection limit
        if num_peers > self.peerset_total_connection_limit {
            let address_metrics = *self.address_metrics.borrow();
            panic!(
                "unexpectedly exceeded configured peer set connection limit: \n\
                 peers: {num_peers:?}, ready: {num_ready:?}, unready: {num_unready:?}, \n\
                 address_metrics: {address_metrics:?}",
            );
        }
    }
}

impl<D, C> PeerSet<D, C>
where
    D: Discover<Key = PeerSocketAddr, Service = LoadTrackedClient> + Unpin + Send + 'static,
    D::Error: Into<BoxError>,
    C: ChainTip + Send + 'static,
{
    /// Buffers the peer set and retains shutdown ownership only in external service handles.
    ///
    /// The worker and self-poller are aborted when the final external handle is dropped.
    pub(crate) fn into_buffer(
        self,
        bound: usize,
    ) -> (crate::NetworkService, JoinHandle<Result<(), BoxError>>) {
        let queued_request_waker = self.queued_request_waker.clone();
        let (peer_set, worker) = Buffer::pair(BoxService::new(self), bound);
        let worker = tokio::spawn(worker.in_current_span());
        let poll_task = tokio::spawn(
            poll_peer_set_on_notify(peer_set.clone(), queued_request_waker).in_current_span(),
        );

        let network = crate::NetworkService::new(
            peer_set,
            vec![worker.abort_handle(), poll_task.abort_handle()],
        );
        (network, poll_task)
    }
}

impl<D, C> Service<Request> for PeerSet<D, C>
where
    D: Discover<Key = PeerSocketAddr, Service = LoadTrackedClient> + Unpin,
    D::Error: Into<BoxError>,
    C: ChainTip,
{
    type Response = Response;
    type Error = BoxError;
    type Future =
        Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        // Update service and peer statuses.
        //
        // # Correctness
        //
        // All of the futures that receive a context from this method can wake the peer set buffer
        // task. If there are no ready peers, and no new peers, network requests will pause until:
        // - an unready peer becomes ready, or
        // - a new peer arrives.

        // Peer events, find deadlines, and cancellations also notify the queued request task,
        // so the peer set gets polled even when the buffer has no other requests.
        let peer_event_waker = self.peer_event_waker(cx);
        let mut peer_cx = Context::from_waker(&peer_event_waker);
        // Events can change queued candidates even without a readiness transition. Register
        // their wakeups with the notification task as well as the Buffer worker.
        self.drain_stall_events(&mut peer_cx);

        // Check for new peers, and register a task wakeup when the next new peers arrive. New peers
        // can be infrequent if our connection slots are full, or we're connected to all
        // available/useful peers.
        let _poll_pending_or_ready: Poll<()> = self.poll_discover(&mut peer_cx)?;

        // These tasks don't provide new peers or newly ready peers.
        let _poll_pending: Poll<()> = self.poll_background_errors(cx)?;
        let _poll_pending_or_ready: Poll<()> =
            self.inventory_registry.poll_inventory(&mut peer_cx)?;
        // Register failures after rotation so they survive a full interval before aging.
        self.drain_inv_failures(&mut peer_cx);

        let ready_peers = self.poll_peers(&mut peer_cx)?;

        self.prune_disconnected_sidecar_keys();
        Self::retain_connected_keys(
            &mut self.serving_peer_keys,
            &self.ready_services,
            &self.cancel_handles,
        );
        // Discovery gets newly ready serving peers before queued block downloads can take them.
        // Even without ready peers, poll deadlines and cancellations to keep the queue live.
        self.route_queued_find_requests(&mut peer_cx);

        // These metrics should run last, to report the most up-to-date information.
        self.log_peer_set_size();
        self.update_metrics();

        if ready_peers.is_pending() {
            // # Correctness
            //
            // If the channel is full, drop the demand signal rather than waiting. If we waited
            // here, the crawler could deadlock sending a request to fetch more peers, because it
            // also empties the channel.
            trace!("no ready services, sending demand signal");
            let _ = self.demand_signal.try_send(MorePeers);

            // # Correctness
            //
            // The current task must be scheduled for wakeup every time we return `Poll::Pending`.
            //
            // As long as there are unready or new peers, this task will run, because:
            // - `poll_discover` schedules this task for wakeup when new peers arrive.
            // - if there are unready peers, `poll_unready` or `poll_ready_peers` schedule this
            //   task for wakeup when peer services become ready.
            //
            // To avoid peers blocking on a full peer status/error channel:
            // - `poll_background_errors` schedules this task for wakeup when the peer status
            //   update task exits.
            return Poll::Pending;
        }

        // Waiting block requests still take priority over broadcasts.
        self.route_queued_inv_requests(&mut peer_cx);
        self.broadcast_all_queued();
        self.send_queued_sidecar_broadcast();

        if self.ready_services.is_empty() {
            if self.poll_peers(&mut peer_cx)?.is_pending() {
                return Poll::Pending;
            }
            // A peer can recover capacity while the first queued request is dispatched.
            // Do not expose that capacity to the Buffer's held request before queued work.
            self.route_queued_find_requests(&mut peer_cx);
            self.route_queued_inv_requests(&mut peer_cx);
            if self.ready_services.is_empty() {
                // Newly inserted readiness futures need a first poll even with an idle Buffer.
                cx.waker().wake_by_ref();
                Poll::Pending
            } else {
                Poll::Ready(Ok(()))
            }
        } else {
            Poll::Ready(Ok(()))
        }
    }

    fn call(&mut self, req: Request) -> Self::Future {
        let fut = match req {
            Request::BlockDownload { hash, .. } => self.route_inv(req, hash.into()),
            // Only do inventory-aware routing on individual items.
            Request::BlocksByHash(ref hashes) if hashes.len() == 1 => {
                let hash = InventoryHash::from(*hashes.iter().next().unwrap());
                self.route_inv(req, hash)
            }
            Request::TransactionsById(ref hashes) if hashes.len() == 1 => {
                let hash = InventoryHash::from(*hashes.iter().next().unwrap());
                self.route_inv(req, hash)
            }

            // Broadcast advertisements to lots of peers
            Request::AdvertiseTransactionIds(_, _) => self.route_broadcast(req),
            Request::AdvertiseBlockToAll(_) => self.broadcast_all(req),

            // Broadcasts that must always reach the configured zcashd-compat sidecar peers
            Request::AdvertiseBlock(_, _) | Request::PushTransaction(_, _) => {
                self.route_sidecar_broadcast(req)
            }

            // Queued requests were already routed by `poll_ready()`.
            Request::PollPeerSet => async { Ok(Response::Nil) }.boxed(),

            // Find requests rotate across serving peers with a bounded wait; others use P2C.
            _ => self.route_request(req),
        };
        self.update_metrics();

        fut
    }
}