Struct Connection

pub struct Connection<S, Tx>
where
    Tx: Sink<Message, Error = SerializationError> + Unpin,
{
    pub connection_info: Arc<ConnectionInfo>,
    pub(super) state: State,
    pub(super) request_timer: Option<Pin<Box<Sleep>>>,
    pub(super) cached_addrs: Vec<MetaAddr>,
    pub(super) svc: S,
    pub(super) client_rx: ClientRequestReceiver,
    pub(super) error_slot: ErrorSlot,
    pub(super) peer_tx: PeerTx<Tx>,
    pub(super) connection_tracker: ConnectionTracker,
    pub(super) metrics_label: String,
    pub(super) last_metrics_state: Option<Cow<'static, str>>,
    last_overload_time: Option<Instant>,
}

The channels, services, and associated state for a peer connection.

Fields

connection_info: Arc<ConnectionInfo>

The metadata for the connected peer service.

This field is used for debugging.

state: State

The state of this connection’s current request or response.

request_timer: Option<Pin<Box<Sleep>>>

A timeout for a client request. This is stored separately from State so that we can move the future out of it independently of other state handling.

cached_addrs: Vec<MetaAddr>

Unused peers from recent addr or addrv2 messages from this peer. Also holds the initial addresses sent in version messages, or guessed from the remote IP.

When peers send solicited or unsolicited peer advertisements, Zebra puts them in this cache.

When Zebra’s components request peers, some cached peers are randomly selected, consumed, and returned as a modified response. This works around zcashd’s address response rate-limit.

The cache size is limited to avoid denial of service attacks.

svc: S

The inbound service, used to answer requests from this connection’s peer.

client_rx: ClientRequestReceiver

A channel for requests that Zebra’s internal services want to send to remote peers.

This channel accepts Requests, and produces InProgressClientRequests.

error_slot: ErrorSlot

A slot for an error shared between the Connection and the Client that uses it.

None unless the connection or client have errored.

peer_tx: PeerTx<Tx>

A channel for sending Zcash messages to the connected peer.

This channel accepts Messages.

The corresponding peer message receiver is passed to Connection::run.

connection_tracker: ConnectionTracker

A connection tracker that reduces the open connection count when dropped. Used to limit the number of open connections in Zebra.

This field does nothing until it is dropped.

Security

If this connection tracker or Connections are leaked, the number of active connections will appear higher than it actually is. If enough connections leak, Zebra will stop making new connections.

metrics_label: String

The metrics label for this peer. Usually the remote IP and port.

last_metrics_state: Option<Cow<'static, str>>

The state for this peer, when the metrics were last updated.

last_overload_time: Option<Instant>

The time of the last overload error response from the inbound service to a request from this connection, or None if this connection hasn’t yet received an overload error.

Implementations

impl<S, Tx> Connection<S, Tx>

where Tx: Sink<Message, Error = SerializationError> + Unpin,

pub(crate) fn new( inbound_service: S, client_rx: Receiver<ClientRequest>, error_slot: ErrorSlot, peer_tx: Tx, connection_tracker: ConnectionTracker, connection_info: Arc<ConnectionInfo>, initial_cached_addrs: Vec<MetaAddr>, ) -> Self

Return a new connection from its channels, services, shared state, and metadata.

impl<S, Tx> Connection<S, Tx>

where S: Service<Request, Response = Response, Error = BoxError>, S::Error: Into<BoxError>, Tx: Sink<Message, Error = SerializationError> + Unpin,

pub async fn run<Rx>(self, peer_rx: Rx)

where Rx: Stream<Item = Result<Message, SerializationError>> + Unpin,

Consume this Connection to form a spawnable future containing its event loop.

peer_rx is a channel for receiving Zcash Messages from the connected peer. The corresponding peer message receiver is [Connection.peer_tx].

async fn fail_with(&mut self, error: impl Into<SharedPeerError>)

Fail this connection, log the failure, and shut it down. See Self::shutdown_async() for details.

Use Self::shutdown_async() to avoid logging the failure, and Self::shutdown() from non-async code.

async fn handle_client_request(&mut self, req: InProgressClientRequest)

Handle an internal client request, possibly generating outgoing messages to the remote peer.

NOTE: the caller should use .instrument(msg.span) to instrument the function.

async fn handle_message_as_request(&mut self, msg: Message) -> Option<Message>

Handle msg as a request from a peer to this Zebra instance.

If the message is not handled, it is returned.

async fn drive_peer_request(&mut self, req: Request)

Given a req originating from the peer, drive it to completion and send any appropriate messages to the remote peer. If an error occurs while processing the request (e.g., the service is shedding load), then we call fail_with to terminate the entire peer connection, shrinking the number of connected peers.

async fn handle_inbound_overload( &mut self, req: Request, now: Instant, error: PeerError, )

Handle inbound service overload and timeout error responses by randomly terminating some connections.

Security

When the inbound service is overloaded with requests, Zebra needs to drop some connections, to reduce the load on the application. But dropping every connection that receives an Overloaded error from the inbound service could cause Zebra to drop too many peer connections, and stop itself downloading blocks or transactions.

Malicious or misbehaving peers can also overload the inbound service, and make Zebra drop its connections to other peers.

So instead, Zebra drops some overloaded connections at random. If a connection has recently overloaded the inbound service, it is more likely to be dropped. This makes it harder for a single peer (or multiple peers) to perform a denial of service attack.

The inbound connection rate-limit also makes it hard for multiple peers to perform this attack, because each inbound connection can only send one inbound request before its probability of being disconnected increases.

impl<S, Tx> Connection<S, Tx>

where Tx: Sink<Message, Error = SerializationError> + Unpin,

fn update_state_metrics(&mut self, extra_state_info: impl Into<Option<String>>)

Update the connection state metrics for this connection, using extra_state_info as additional state information.

fn erase_state_metrics(&mut self)

Erase the connection state metrics for this connection.

async fn shutdown_async(&mut self, error: impl Into<SharedPeerError>)

Marks the peer as having failed with error, and performs connection cleanup, including async channel closes.

If the connection has errored already, re-use the original error. Otherwise, fail the connection with error.

fn shutdown(&mut self, error: impl Into<SharedPeerError>)

Marks the peer as having failed with error, and performs connection cleanup. See Self::shutdown_async() for details.

Call Self::shutdown_async() in async code, because it can shut down more channels.

Trait Implementations

impl<S, Tx> Debug for Connection<S, Tx>

where Tx: Sink<Message, Error = SerializationError> + Unpin,

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<S, Tx> Drop for Connection<S, Tx>

where Tx: Sink<Message, Error = SerializationError> + Unpin,

fn drop(&mut self)

Executes the destructor for this type.