Struct PeerSet 

pub struct PeerSet<D, C>
where
    D: Discover<Key = PeerSocketAddr, Service = LoadTrackedClient> + Unpin,
    D::Error: Into<BoxError>,
    C: ChainTip,
{

    discover: D,
    demand_signal: Sender<MorePeers>,
    bans_receiver: Receiver<BanList>,
    find_response_stalls: FindResponseStallTracker,
    stall_event_rx: UnboundedReceiver<(PeerSocketAddr, Arc<ConnectionInfo>, StallOutcome)>,
    stall_event_tx: UnboundedSender<(PeerSocketAddr, Arc<ConnectionInfo>, StallOutcome)>,
    inv_failure_rx: UnboundedReceiver<(PeerSocketAddr, Arc<ConnectionInfo>, InventoryHash)>,
    inv_failure_tx: UnboundedSender<(PeerSocketAddr, Arc<ConnectionInfo>, InventoryHash)>,
    ready_services: HashMap<D::Key, D::Service>,
    find_peer_queue: VecDeque<(D::Key, Arc<ConnectionInfo>)>,
    queued_find_requests: VecDeque<QueuedFindRequest>,
    queued_find_request_timer: Option<Pin<Box<Sleep>>>,
    inventory_registry: InventoryRegistry,
    queued_broadcast_all: Option<(Request, Sender<Pin<Box<dyn Future<Output = Result<Response, BoxError>> + Send + 'static>>>, HashSet<D::Key>)>,
    block_gossip_peer_ips: HashSet<IpAddr>,
    zcashd_compat_peer_keys: HashSet<D::Key>,
    serving_peer_keys: HashSet<D::Key>,
    queued_inv_requests: VecDeque<QueuedInvRequest>,
    queued_request_waker: Arc<QueuedRequestWaker>,
    queued_sidecar_broadcast: Option<(Request, HashSet<D::Key>)>,
    unready_services: FuturesUnordered<UnreadyService<D::Key, D::Service, Request>>,
    cancel_handles: HashMap<D::Key, Sender<CancelClientWork>>,
    minimum_peer_version: MinimumPeerVersion<C>,
    peerset_total_connection_limit: usize,
    handle_rx: Receiver<Vec<JoinHandle<Result<(), BoxError>>>>,
    guards: FuturesUnordered<JoinHandle<Result<(), BoxError>>>,
    address_metrics: Receiver<AddressMetrics>,
    last_peer_log: Option<Instant>,
    max_conns_per_ip: usize,
    network: Network,
}

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.

Fields

discover: D

Provides new and deleted peer [Change]s to the peer set, via the [Discover] trait implementation.

demand_signal: Sender<MorePeers>

A channel that asks the peer crawler task to connect to more peers.

bans_receiver: Receiver<BanList>

A watch channel receiver with a snapshot of the banned peer groups.

find_response_stalls: FindResponseStallTracker

Tracks peers returning failed FindBlocks/FindHeaders responses. Mutated only from Self::poll_ready via Self::stall_event_rx.

stall_event_rx: UnboundedReceiver<(PeerSocketAddr, Arc<ConnectionInfo>, StallOutcome)>

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_tx: UnboundedSender<(PeerSocketAddr, Arc<ConnectionInfo>, StallOutcome)>

Producer clones handed to each tracked request’s response wrapper.

inv_failure_rx: UnboundedReceiver<(PeerSocketAddr, Arc<ConnectionInfo>, InventoryHash)>

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_tx: UnboundedSender<(PeerSocketAddr, Arc<ConnectionInfo>, InventoryHash)>

Producer clones handed to each block request’s response future.

ready_services: HashMap<D::Key, D::Service>

Connected peers that are ready to receive requests from Zebra, or send requests to Zebra.

find_peer_queue: VecDeque<(D::Key, Arc<ConnectionInfo>)>

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.

queued_find_requests: VecDeque<QueuedFindRequest>

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_request_timer: Option<Pin<Box<Sleep>>>

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.

inventory_registry: InventoryRegistry

Stores gossiped inventory hashes from connected peers.

Used to route inventory requests to peers that are likely to have it.

queued_broadcast_all: Option<(Request, Sender<Pin<Box<dyn Future<Output = Result<Response, BoxError>> + Send + 'static>>>, HashSet<D::Key>)>

Stores requests that should be routed to peers once they are ready.

block_gossip_peer_ips: HashSet<IpAddr>

Inbound peer IPs that must always receive block inventory broadcasts.

zcashd_compat_peer_keys: HashSet<D::Key>

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.

serving_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.

queued_inv_requests: VecDeque<QueuedInvRequest>

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_request_waker: Arc<QueuedRequestWaker>

Wakes the task that polls the peer set, and the task that runs poll_peer_set_on_notify, while requests are queued.

queued_sidecar_broadcast: Option<(Request, HashSet<D::Key>)>

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.

unready_services: FuturesUnordered<UnreadyService<D::Key, D::Service, Request>>

Connected peers that are handling a Zebra request, or Zebra is handling one of their requests.

cancel_handles: HashMap<D::Key, Sender<CancelClientWork>>

Channels used to cancel the request that an unready service is doing.

minimum_peer_version: MinimumPeerVersion<C>

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.

peerset_total_connection_limit: usize

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.

handle_rx: Receiver<Vec<JoinHandle<Result<(), BoxError>>>>

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

guards: FuturesUnordered<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

address_metrics: Receiver<AddressMetrics>

Address book metrics watch channel.

Used for logging diagnostics.

last_peer_log: Option<Instant>

The last time we logged a message about the peer set size

max_conns_per_ip: usize

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

network: Network

The network of this peer set.

Implementations

impl<D, C> PeerSet<D, C>

where D: Discover<Key = PeerSocketAddr, Service = LoadTrackedClient> + Unpin, D::Error: Into<BoxError>, C: ChainTip,

pub fn new( config: &Config, block_gossip_peer_ips: Vec<IpAddr>, discover: D, demand_signal: Sender<MorePeers>, handle_rx: Receiver<Vec<JoinHandle<Result<(), BoxError>>>>, inv_stream: Receiver<InventoryResponse<(AtLeastOne<InventoryHash>, PeerSocketAddr), (AtLeastOne<InventoryHash>, PeerSocketAddr)>>, bans_receiver: Receiver<BanList>, address_metrics: Receiver<AddressMetrics>, minimum_peer_version: MinimumPeerVersion<C>, max_conns_per_ip: Option<usize>, ) -> Self

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.

fn peer_event_waker(&self, cx: &Context<'_>) -> Waker

Returns a waker for peer events, which also notifies poll_peer_set_on_notify while requests are queued.

fn poll_background_errors( &mut self, cx: &mut Context<'_>, ) -> Poll<Result<(), BoxError>>

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 receive_tasks_if_needed( &mut self, cx: &mut Context<'_>, ) -> Poll<Result<(), BoxError>>

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 shut_down_tasks_and_channels(&mut self, cx: &mut Context<'_>)

Shut down:

• services by dropping the service lists
• background tasks via their join handles or cancel handles
• channels by closing the channel

fn poll_peers(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), BoxError>>

Checks for newly ready, disconnects from outdated peers, and polls ready peer errors.

fn poll_unready( &mut self, cx: &mut Context<'_>, ) -> Poll<Result<Option<()>, BoxError>>

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_ready_peer_errors(&mut self, cx: &mut Context<'_>) -> Poll<()>

Checks previously ready peer services for errors.

The only way these peer Clients 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 num_peers_with_ip(&self, ip: IpAddr) -> usize

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.

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 has_peer_with_addr(&self, addr: PeerSocketAddr) -> bool

Returns true if Zebra is already connected to the IP and port in addr.

fn poll_discover(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), BoxError>>

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 disconnect_from_ineligible_peers(&mut self)

Disconnects banned and outdated connections, whether ready or backpressured.

fn take_ready_service(&mut self, key: &D::Key) -> Option<D::Service>

Takes a ready service by key.

fn is_current_connection( &self, addr: PeerSocketAddr, info: &Arc<ConnectionInfo>, ) -> bool

Whether an event belongs to the currently connected incarnation of an address.

fn drain_stall_events(&mut self, cx: &mut Context<'_>)

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_inv_failures(&mut self, cx: &mut Context<'_>)

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 remove(&mut self, key: &D::Key)

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 push_ready(&mut self, was_unready: bool, key: D::Key, svc: D::Service)

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_unready(&mut self, key: D::Key, svc: D::Service)

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 select_ready_p2c_peer(&self) -> Option<D::Key>

Performs P2C on self.ready_services to randomly select a less-loaded ready service.

fn select_ready_find_peer(&mut self, allow_non_serving: bool) -> Option<D::Key>

Rotates find requests across ready connections, restricting them to serving peers until the queued wait expires, unless no serving peers are connected.

fn select_p2c_peer_from_list( &self, ready_service_list: &HashSet<D::Key>, ) -> Option<D::Key>

Performs P2C on ready_service_list to randomly select a less-loaded ready service.

fn select_random_ready_peers(&self, max_peers: usize) -> Vec<D::Key>

Randomly chooses max_peers ready services, ignoring service load.

The chosen peers are unique, but their order is not fully random.

fn select_sidecar_broadcast_peers(&self, max_peers: usize) -> Vec<D::Key>

Randomly chooses ready peers for a sidecar broadcast, always including configured zcashd compat sidecar peers.

fn is_zcashd_compat_peer(&self, service: &D::Service) -> bool

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 prune_disconnected_sidecar_keys(&mut self)

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 retain_connected_keys( keys: &mut HashSet<D::Key>, ready_services: &HashMap<D::Key, D::Service>, cancel_handles: &HashMap<D::Key, Sender<CancelClientWork>>, )

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 query_load(&self, key: &D::Key) -> Option<<D::Service as Load>::Metric>

Accesses a ready endpoint by key and returns its current load.

Returns None if the service is not in the ready service list.

fn route_request(&mut self, req: Request) -> <Self as Service<Request>>::Future

Routes find requests fairly and other requests using P2C load-balancing.

fn route_inv( &mut self, req: Request, hash: InventoryHash, ) -> <Self as Service<Request>>::Future

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 error.

See Self::select_inv_route for details.

fn select_inv_route( &self, request: &Request, hash: InventoryHash, ) -> InvRoute<D::Key>

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 call_inv_peer( &mut self, key: D::Key, req: Request, hash: InventoryHash, ) -> <Self as Service<Request>>::Future

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_ready_peer( &mut self, key: D::Key, req: Request, ) -> <Self as Service<Request>>::Future

Sends req to the ready peer key, and returns its response future.

fn queue_find_request( &mut self, request: Request, ) -> Pin<Box<dyn Future<Output = Result<Response, BoxError>> + Send + 'static>>

Waits for a serving peer, falling back to any ready peer once the bounded wait expires.

fn route_queued_find_requests(&mut self, cx: &mut Context<'_>)

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 queue_inv_request( &mut self, request: Request, hash: InventoryHash, ) -> <Self as Service<Request>>::Future

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 error if the request is refused.

fn route_queued_inv_requests(&mut self, cx: &mut Context<'_>)

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_multiple( &mut self, req: Request, max_peers: usize, ) -> <Self as Service<Request>>::Future

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 send_multiple( &mut self, req: Request, peers: Vec<D::Key>, ) -> <Self as Service<Request>>::Future

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 route_broadcast( &mut self, req: Request, ) -> <Self as Service<Request>>::Future

Broadcasts the same request to lots of ready peers, ignoring return values.

fn route_sidecar_broadcast( &mut self, req: Request, ) -> <Self as Service<Request>>::Future

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 send_queued_sidecar_broadcast(&mut self)

Delivers the queued sidecar broadcast to any owed sidecar peers that have become ready. See Self::route_sidecar_broadcast.

fn broadcast_all(&mut self, req: Request) -> <Self as Service<Request>>::Future

Broadcasts the same request to all ready peers, ignoring return values.

fn queue_broadcast_all_unready( &mut self, req: &Request, ) -> Option<Receiver<Pin<Box<dyn Future<Output = Result<Response, BoxError>> + Send + 'static>>>>

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 broadcast_all_queued(&mut self)

Broadcasts the same requests to all ready peers which were unready when PeerSet::broadcast_all() was last called, ignoring return values.

pub(crate) fn number_of_peers_to_broadcast(&self) -> usize

Given a number of ready peers calculate to how many of them Zebra will actually send the request to. Return this number.

fn peer_set_addresses(&self) -> Vec<PeerSocketAddr>

Returns the list of addresses in the peer set.

fn log_peer_set_size(&mut self)

Logs the peer set size, and any potential connectivity issues.

fn update_metrics(&self)

Updates the peer set metrics.

Panics

If the peer set size exceeds the connection limit.

impl<D, C> PeerSet<D, C>

where D: Discover<Key = PeerSocketAddr, Service = LoadTrackedClient> + Unpin + Send + 'static, D::Error: Into<BoxError>, C: ChainTip + Send + 'static,

pub(crate) fn into_buffer( self, bound: usize, ) -> (NetworkService, JoinHandle<Result<(), BoxError>>)

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.

Trait Implementations

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)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.

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

Responses given by the service.

type Error = Box<dyn Error + Send + Sync>

Errors produced by the service.

type Future = Pin<Box<dyn Future<Output = Result<<PeerSet<D, C> as Service<Request>>::Response, <PeerSet<D, C> as Service<Request>>::Error>> + Send>>

The future response value.

fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>

Returns Poll::Ready(Ok(())) when the service is able to process requests.

fn call(&mut self, req: Request) -> Self::Future

Process the request and return the response asynchronously.

Layout

Note: Unable to compute type layout, possibly due to this type having generic parameters. Layout can only be computed for concrete, fully-instantiated types.