/__w/smoldot/smoldot/repo/full-node/src/network_service.rs

Source (jump to first uncovered line)
// Smoldot
// Copyright (C) 2019-2022  Parity Technologies (UK) Ltd.
// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0

// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.

// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.

// You should have received a copy of the GNU General Public License
// along with this program.  If not, see <http://www.gnu.org/licenses/>.

//! Background network service.
//!
//! The [`NetworkService`] manages background tasks dedicated to connecting to other nodes.
//! Importantly, its design is oriented towards the particular use case of the full node.
//!
//! The [`NetworkService`] spawns one background task (using the [`Config::tasks_executor`]) for
//! each active TCP socket, plus one for each TCP listening socket. Messages are exchanged between
//! the service and these background tasks.

// TODO: doc
// TODO: re-review this once finished

use crate::{database_thread, jaeger_service, LogCallback, LogLevel};

use core::{cmp, future::Future, mem, pin::Pin, task::Poll, time::Duration};
use futures_channel::oneshot;
use futures_lite::FutureExt as _;
use futures_util::stream::{self, SelectAll};
use hashbrown::HashMap;
use smol::{
    channel, future,
    lock::Mutex,
    net::TcpStream,
    stream::{Stream, StreamExt as _},
};
use smoldot::{
    database::full_sqlite,
    header,
    informant::{BytesDisplay, HashDisplay},
    libp2p::{
        connection,
        multiaddr::{self, Multiaddr, Protocol},
        peer_id::{self, PeerId},
    },
    network::{basic_peering_strategy, codec, service},
};
use std::{
    io,
    net::{IpAddr, SocketAddr},
    sync::Arc,
    time::Instant,
    vec,
};

pub use smoldot::network::service::ChainId;

mod tasks;

/// Configuration for a [`NetworkService`].
pub struct Config {
    /// Closure that spawns background tasks.
    pub tasks_executor: Box<dyn FnMut(Pin<Box<dyn Future<Output = ()> + Send>>) + Send>,

    /// Function called in order to notify of something.
    pub log_callback: Arc<dyn LogCallback + Send + Sync>,

    /// Number of event receivers returned by [`NetworkService::new`].
    pub num_events_receivers: usize,

    /// Addresses to listen for incoming connections.
    pub listen_addresses: Vec<Multiaddr>,

    /// List of block chains to be connected to.
    pub chains: Vec<ChainConfig>,

    /// Value sent back for the agent version when receiving an identification request.
    pub identify_agent_version: String,

    /// Key used for the encryption layer.
    /// This is a Noise static key, according to the Noise specification.
    /// Signed using the actual libp2p key.
    pub noise_key: connection::NoiseKey,

    /// Service to use to report traces.
    pub jaeger_service: Arc<jaeger_service::JaegerService>,
}

/// Configuration for one chain.
pub struct ChainConfig {
    /// Name of the chain to use for logging purposes.
    pub log_name: String,

    /// List of node identities and addresses that are known to belong to the chain's peer-to-pee
    /// network.
    pub bootstrap_nodes: Vec<(PeerId, Multiaddr)>,

    /// Database to use to read blocks from when answering requests.
    pub database: Arc<database_thread::DatabaseThread>,

    /// Hash of the genesis block of the chain. Sent to other nodes in order to determine whether
    /// the chains match.
    pub genesis_block_hash: [u8; 32],

    /// Number and hash of the current best block. Can later be updated with // TODO: which function?
    pub best_block: (u64, [u8; 32]),

    /// Optional identifier to insert into the networking protocol names. Used to differentiate
    /// between chains with the same genesis hash.
    pub fork_id: Option<String>,

    /// Number of bytes of the block number in the networking protocol.
    pub block_number_bytes: usize,

    /// Must be `Some` if and only if the chain uses the GrandPa networking protocol. Contains the
    /// number of the finalized block at the time of the initialization.
    pub grandpa_protocol_finalized_block_height: Option<u64>,

    /// Maximum number of peers that have slots attributed to them.
    pub max_slots: usize,

    /// Maximum number of peers that have gossip links open but without having slots attributed
    /// to them.
    pub max_in_peers: usize,
}

/// Event generated by the events reporters returned by [`NetworkService::new`].
#[derive(Debug, Clone)]
pub enum Event {
    Connected {
        chain_id: ChainId,
        peer_id: PeerId,
        best_block_number: u64,
        best_block_hash: [u8; 32],
    },
    Disconnected {
        chain_id: ChainId,
        peer_id: PeerId,
    },
    BlockAnnounce {
        chain_id: ChainId,
        peer_id: PeerId,
        scale_encoded_header: Vec<u8>,
        is_best: bool,
    },
    GrandpaNeighborPacket {
        chain_id: ChainId,
        peer_id: PeerId,
        finalized_block_height: u64,
    },
}

pub struct NetworkService {
    /// Identity of the local node.
    local_peer_id: PeerId,

    /// Service to use to report traces.
    // TODO: unused
    _jaeger_service: Arc<jaeger_service::JaegerService>,

    /// Channel to send messages to the background task.
    to_background_tx: Mutex<channel::Sender<ToBackground>>,

    /// Name of all the chains that have been registered, for logging purposes.
    chain_names: hashbrown::HashMap<ChainId, String, fnv::FnvBuildHasher>,
}

enum ToBackground {
    ForegroundDisconnectAndBan {
        peer_id: PeerId,
        chain_id: ChainId,
        severity: BanSeverity,
        reason: &'static str,
    },
    ForegroundAnnounceBlock {
        target: PeerId,
        chain_id: ChainId,
        scale_encoded_header: Vec<u8>,
        is_best: bool,
        result_tx: oneshot::Sender<Result<(), service::QueueNotificationError>>,
    },
    ForegroundSetLocalBestBlock {
        chain_id: ChainId,
        best_hash: [u8; 32],
        best_number: u64,
    },
    ForegroundBlocksRequest {
        target: PeerId,
        chain_id: ChainId,
        config: codec::BlocksRequestConfig,
        result_tx: oneshot::Sender<Result<Vec<codec::BlockData>, BlocksRequestError>>,
    },
    ForegroundWarpSyncRequest {
        target: PeerId,
        chain_id: ChainId,
        begin_hash: [u8; 32],
        result_tx:
            oneshot::Sender<Result<service::EncodedGrandpaWarpSyncResponse, WarpSyncRequestError>>,
    },
    ForegroundStorageProofRequest {
        target: PeerId,
        chain_id: ChainId,
        config: codec::StorageProofRequestConfig<vec::IntoIter<Vec<u8>>>,
        result_tx: oneshot::Sender<Result<service::EncodedMerkleProof, ()>>,
    },
    ForegroundCallProofRequest {
        target: PeerId, // TODO: takes by value because of futures longevity issue
        chain_id: ChainId,
        config: codec::CallProofRequestConfig<'static, vec::IntoIter<Vec<u8>>>,
        result_tx: oneshot::Sender<Result<service::EncodedMerkleProof, ()>>,
    },
    ForegroundGetNumConnections {
        result_tx: oneshot::Sender<usize>,
    },
    ForegroundGetNumPeers {
        chain_id: ChainId,
        result_tx: oneshot::Sender<usize>,
    },
    ForegroundGetNumTotalPeers {
        result_tx: oneshot::Sender<usize>,
    },
}

struct Inner {
    /// Value provided through [`Config::identify_agent_version`].
    identify_agent_version: String,

    /// Sending events through the public API.
    ///
    /// Contains either senders, or a `Future` that is currently sending an event and will yield
    /// the senders back once it is finished.
    event_senders: either::Either<
        Vec<channel::Sender<Event>>,
        Pin<Box<dyn Future<Output = Vec<channel::Sender<Event>>> + Send>>,
    >,

    /// Event about to be sent on the senders of [`Inner::event_senders`].
    event_pending_send: Option<Event>,

    /// Identity of the local node.
    noise_key: service::NoiseKey,

    /// Identity of the local node. Can be derived from [`Inner::noise_key`].
    local_peer_id: PeerId,

    /// Service to use to report traces.
    jaeger_service: Arc<jaeger_service::JaegerService>,

    /// Data structure holding the entire state of the networking.
    network:
        service::ChainNetwork<Chain, channel::Sender<service::CoordinatorToConnection>, Instant>,

    /// Data structure holding the addresses and assigned slots.
    peering_strategy: basic_peering_strategy::BasicPeeringStrategy<ChainId, Instant>,

    /// Current number of outgoing connection attempts.
    ///
    /// This counter is used to limit the number of simultaneous connection attempts, as some
    /// ISPs/cloud providers don't like seeing too many dialing connections at the same time.
    num_pending_out_attempts: usize,

    /// Stream of incoming connections.
    incoming_connections: SelectAll<Pin<Box<dyn Stream<Item = (TcpStream, SocketAddr)> + Send>>>,

    /// See [`Config::tasks_executor`].
    tasks_executor: Box<dyn FnMut(Pin<Box<dyn Future<Output = ()> + Send>>) + Send>,

    /// See [`Config::log_callback`].
    log_callback: Arc<dyn LogCallback + Send + Sync>,

    /// Channel for the frontend to send messages to the background task.
    to_background_rx: Pin<Box<channel::Receiver<ToBackground>>>,

    /// Channel where connections send messages destined to the coordinator.
    from_connections_rx: Pin<
        Box<
            channel::Receiver<(
                service::ConnectionId,
                Option<service::ConnectionToCoordinator>,
            )>,
        >,
    >,

    /// Sending side of [`Inner::from_connections_rx`].
    from_connections_tx: channel::Sender<(
        service::ConnectionId,
        Option<service::ConnectionToCoordinator>,
    )>,

    /// List of all block requests that have been started but not finished yet.
    blocks_requests: HashMap<
        service::SubstreamId,
        oneshot::Sender<Result<Vec<codec::BlockData>, BlocksRequestError>>,
        fnv::FnvBuildHasher,
    >,

    /// List of all warp sync requests that have been started but not finished yet.
    warp_sync_requests: HashMap<
        service::SubstreamId,
        oneshot::Sender<Result<service::EncodedGrandpaWarpSyncResponse, WarpSyncRequestError>>,
        fnv::FnvBuildHasher,
    >,

    /// List of all storage requests that have been started but not finished yet.
    storage_requests: HashMap<
        service::SubstreamId,
        oneshot::Sender<Result<service::EncodedMerkleProof, ()>>,
        fnv::FnvBuildHasher,
    >,

    /// List of all call proof requests that have been started but not finished yet.
    call_proof_requests: HashMap<
        service::SubstreamId,
        oneshot::Sender<Result<service::EncodedMerkleProof, ()>>,
        fnv::FnvBuildHasher,
    >,

    /// When to start the next discovery process.
    next_discovery: smol::Timer,

    /// Time between [`Inner::next_discovery`] and the follow-up discovery.
    next_discovery_period: Duration,
}

/// Extra information of a chain.
struct Chain {
    /// Name of the chain to use for logging purposes.
    log_name: String,

    /// How to access data to answer requests from the remotes.
    database: Arc<database_thread::DatabaseThread>,

    /// Maximum number of peers that have slots attributed to them.
    max_slots: usize,

    /// Maximum number of peers that have gossip links open but without having slots attributed
    /// to them.
    max_in_peers: usize,
}

/// Severity of a ban. See [`NetworkService::ban_and_disconnect`].
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
pub enum BanSeverity {
    Low,
    High,
}

impl NetworkService {
    /// Initializes the network service with the given configuration.
    pub async fn new(
        config: Config,
    ) -> Result<
        (
            Arc<Self>,
            Vec<ChainId>,
            Vec<Pin<Box<dyn Stream<Item = Event> + Send>>>,
        ),
        InitError,
    > {
        let (event_senders, event_receivers): (Vec<_>, Vec<_>) = (0..config.num_events_receivers)
            .map(|_| channel::bounded(16))
            .unzip();

        let mut network = service::ChainNetwork::new(service::Config {
            chains_capacity: config.chains.len(),
            connections_capacity: 100, // TODO: ?
            handshake_timeout: Duration::from_secs(8),
            randomness_seed: rand::random(),
        });

        let mut peering_strategy =
            basic_peering_strategy::BasicPeeringStrategy::new(basic_peering_strategy::Config {
                randomness_seed: rand::random(),
                peers_capacity: 200, // TODO: ?
                chains_capacity: config.chains.len(),
            });

        let mut chain_names =
            hashbrown::HashMap::with_capacity_and_hasher(config.chains.len(), Default::default());

        for chain21 in config.chains {
            let chain_id = network
                .add_chain(service::ChainConfig {
                    fork_id: chain.fork_id.clone(),
                    block_number_bytes: chain.block_number_bytes,
                    best_hash: chain.best_block.1,
                    best_number: chain.best_block.0,
                    genesis_hash: chain.genesis_block_hash,
                    role: codec::Role::Full,
                    grandpa_protocol_config: chain.grandpa_protocol_finalized_block_height.map(
                        // TODO: dummy values
                        |commit_finalized_height| service::GrandpaState {
                            commit_finalized_height,
                            round_number: 1,
                            set_id: 0,
                        },
                    ),
                    allow_inbound_block_requests: true,
                    user_data: Chain {
                        log_name: chain.log_name.clone(),
                        database: chain.database,
                        max_in_peers: chain.max_in_peers,
                        max_slots: chain.max_slots,
                    },
                })
                .unwrap(); // TODO: don't unwrap?

            for (peer_id, addr0) in chain.bootstrap_nodes {
                // Note that we must call this function before `insert_address`, as documented
                // in `basic_peering_strategy`.
                peering_strategy.insert_chain_peer(chain_id, peer_id.clone(), usize::MAX);
                peering_strategy.insert_address(&peer_id, addr.into_bytes(), usize::MAX);
            }

            chain_names.insert(chain_id, chain.log_name);
        }

        let (to_background_tx, to_background_rx) = channel::bounded(16);
        let (from_connections_tx, from_connections_rx) = channel::bounded(64);

        let local_peer_id =
            peer_id::PublicKey::Ed25519(*config.noise_key.libp2p_public_ed25519_key())
                .into_peer_id();

        // For each listening address in the configuration, create a background task dedicated to
        // listening on that address.
        let mut incoming_connections = SelectAll::new();
        for listen_address0 in config.listen_addresses {
            // Try to parse the requested address and create the corresponding listening socket.
            let tcp_listener: smol::net::TcpListener = {
                let addr = {
                    let mut iter = listen_address.iter();
                    let proto1 = iter.next();
                    let proto2 = iter.next();
                    let proto3 = iter.next();
                    match (proto1, proto2, proto3) {
                        (Some(Protocol::Ip4(ip)), Some(Protocol::Tcp(port)), None) => {
                            Some(SocketAddr::from((ip, port)))
                        }
                        (Some(Protocol::Ip6(ip)), Some(Protocol::Tcp(port)), None) => {
                            Some(SocketAddr::from((ip, port)))
                        }
                        _ => None,
                    }
                };

                if let Some(addr) = addr {
                    match smol::net::TcpListener::bind(addr).await {
                        Ok(l) => l,
                        Err(err) => {
                            return Err(InitError::ListenerIo(listen_address, err));
                        }
                    }
                } else {
                    // TODO: support WebSocket server
                    return Err(InitError::BadListenMultiaddr(listen_address));
                }
            };

            // Add a task dedicated to this listener.
            let log_callback = config.log_callback.clone();
            incoming_connections.push(Box::pin(stream::unfold(tcp_listener, move |tcp_listener| {
                let log_callback = log_callback.clone();
                async move {
                    loop {
                        match tcp_listener.accept().await {
                            Ok((socket, socket_addr)) => {
                                break Some(((socket, socket_addr), tcp_listener))
                            }
                            Err(error) => {
                                // Errors here can happen if the accept failed, for example
                                // if no file descriptor is available.
                                // A wait is added in order to avoid having a busy-loop
                                // failing to accept connections.
                                log_callback.log(
                                    LogLevel::Warn,
                                    format!("tcp-accept-error; error={}", error),
                                );
                                smol::Timer::after(Duration::from_secs(2)).await;
                            }
                        }
                    }
                }
            })) as Pin<Box<_>>);
        }

        // Initialize the inner network service.
        run(Inner {
            local_peer_id: local_peer_id.clone(),
            identify_agent_version: config.identify_agent_version,
            event_senders: either::Left(event_senders),
            event_pending_send: None,
            num_pending_out_attempts: 0,
            to_background_rx: Box::pin(to_background_rx),
            from_connections_rx: Box::pin(from_connections_rx),
            from_connections_tx,
            tasks_executor: config.tasks_executor,
            log_callback: config.log_callback,
            network,
            noise_key: config.noise_key,
            peering_strategy,
            blocks_requests: hashbrown::HashMap::with_capacity_and_hasher(
                50, // TODO: ?
                Default::default(),
            ),
            warp_sync_requests: hashbrown::HashMap::with_capacity_and_hasher(
                2, // TODO: ?
                Default::default(),
            ),
            storage_requests: hashbrown::HashMap::with_capacity_and_hasher(
                5, // TODO: ?
                Default::default(),
            ),
            call_proof_requests: hashbrown::HashMap::with_capacity_and_hasher(
                5, // TODO: ?
                Default::default(),
            ),
            jaeger_service: config.jaeger_service.clone(),
            next_discovery: smol::Timer::after(Duration::from_secs(1)),
            next_discovery_period: Duration::from_secs(1),
            incoming_connections,
        });

        // Build the final network service.
        let network_service = Arc::new(NetworkService {
            local_peer_id,
            chain_names,
            _jaeger_service: config.jaeger_service,
            to_background_tx: Mutex::new(to_background_tx),
        });

        // Adjust the receivers to keep the `network_service` alive.
        // TODO: no, hacky
        let receivers = event_receivers
            .into_iter()
            .map(|rx| {
                let mut network_service = Some(network_service.clone());
                rx.chain(smol::stream::poll_fn(move |_| {
                    drop(network_service.take());
                    Poll::Ready(None)
                }))
                .boxed()
            })
            .collect();

        let chain_ids = network_service.chain_names.keys().cloned().collect();
        Ok((network_service, chain_ids, receivers))
    }

    /// Returns the peer ID of the local node.
    pub fn local_peer_id(&self) -> &PeerId {
        &self.local_peer_id
    }

    /// Returns the number of connections, both handshaking or established, both incoming and
    /// outgoing.
    pub async fn num_connections(&self) -> usize {
        let (result_tx, result_rx) = oneshot::channel();

        let _ = self
            .to_background_tx
            .lock()
            .await
            .send(ToBackground::ForegroundGetNumConnections { result_tx })
            .await;

        result_rx.await.unwrap()
    }

    /// Returns the number of peers we have a substream with,.
    pub async fn num_peers(&self, chain_id: ChainId) -> usize {
        let (result_tx, result_rx) = oneshot::channel();

        let _ = self
            .to_background_tx
            .lock()
            .await
            .send(ToBackground::ForegroundGetNumPeers {
                chain_id,
                result_tx,
            })
            .await;

        result_rx.await.unwrap()
    }

    /// Returns the number of peers we have a substream with, all chains added together.
    pub async fn num_total_peers(&self) -> usize {
        let (result_tx, result_rx) = oneshot::channel();

        let _ = self
            .to_background_tx
            .lock()
            .await
            .send(ToBackground::ForegroundGetNumTotalPeers { result_tx })
            .await;

        result_rx.await.unwrap()
    }

    pub async fn set_local_best_block(
        &self,
        chain_id: ChainId,
        best_hash: [u8; 32],
        best_number: u64,
    ) {
        let _ = self
            .to_background_tx
            .lock()
            .await
            .send(ToBackground::ForegroundSetLocalBestBlock {
                chain_id,
                best_hash,
                best_number,
            })
            .await;
    }

    /// Starts asynchronously disconnecting the given peer. A [`Event::Disconnected`] will later be
    /// generated. Prevents a new gossip link with the same peer from being reopened for a
    /// little while.
    ///
    /// `reason` is a human-readable string printed in the logs.
    ///
    /// Due to race conditions, it is possible to reconnect to the peer soon after, in case the
    /// reconnection was already happening as the call to this function is still being processed.
    /// If that happens another [`Event::Disconnected`] will be delivered afterwards. In other
    /// words, this function guarantees that we will be disconnected in the future rather than
    /// guarantees that we will disconnect.
    pub async fn ban_and_disconnect(
        &self,
        peer_id: PeerId,
        chain_id: ChainId,
        severity: BanSeverity,
        reason: &'static str,
    ) {
        let _ = self
            .to_background_tx
            .lock()
            .await
            .send(ToBackground::ForegroundDisconnectAndBan {
                peer_id,
                chain_id,
                severity,
                reason,
            })
            .await;
    }

    pub async fn send_block_announce(
        self: Arc<Self>,
        target: PeerId,
        chain_id: ChainId,
        scale_encoded_header: Vec<u8>,
        is_best: bool,
    ) -> Result<(), service::QueueNotificationError> {
        let (result_tx, result_rx) = oneshot::channel();

        let _ = self
            .to_background_tx
            .lock()
            .await
            .send(ToBackground::ForegroundAnnounceBlock {
                target,
                chain_id,
                scale_encoded_header,
                is_best,
                result_tx,
            })
            .await;

        result_rx.await.unwrap()
    }

    /// Sends a blocks request to the given peer.
    // TODO: more docs
    // TODO: proper error type
    pub async fn blocks_request(
        self: Arc<Self>,
        target: PeerId, // TODO: by value?
        chain_id: ChainId,
        config: codec::BlocksRequestConfig,
    ) -> Result<Vec<codec::BlockData>, BlocksRequestError> {
        let (result_tx, result_rx) = oneshot::channel();

        let _ = self
            .to_background_tx
            .lock()
            .await
            .send(ToBackground::ForegroundBlocksRequest {
                target: target.clone(),
                chain_id,
                config,
                result_tx,
            })
            .await;

        result_rx.await.unwrap()
    }

    /// Sends a warp sync request to the given peer.
    // TODO: more docs
    // TODO: proper error type
    pub async fn warp_sync_request(
        self: Arc<Self>,
        target: PeerId, // TODO: by value?
        chain_id: ChainId,
        begin_hash: [u8; 32],
    ) -> Result<service::EncodedGrandpaWarpSyncResponse, WarpSyncRequestError> {
        let (result_tx, result_rx) = oneshot::channel();

        let _ = self
            .to_background_tx
            .lock()
            .await
            .send(ToBackground::ForegroundWarpSyncRequest {
                target: target.clone(),
                chain_id,
                begin_hash,
                result_tx,
            })
            .await;

        result_rx.await.unwrap()
    }

    /// Sends a storage proof request to the given peer.
    // TODO: more docs
    // TODO: proper error type
    pub async fn storage_request(
        self: Arc<Self>,
        target: PeerId, // TODO: by value?
        chain_id: ChainId,
        config: codec::StorageProofRequestConfig<impl Iterator<Item = impl AsRef<[u8]> + Clone>>,
    ) -> Result<service::EncodedMerkleProof, ()> {
        // TODO: logs and jaeger integration
        let (result_tx, result_rx) = oneshot::channel();

        let _ = self
            .to_background_tx
            .lock()
            .await
            .send(ToBackground::ForegroundStorageProofRequest {
                target: target.clone(),
                chain_id,
                config: codec::StorageProofRequestConfig {
                    block_hash: config.block_hash,
                    keys: config
                        .keys
                        .map(|key| key.as_ref().to_vec()) // TODO: to_vec() overhead
                        .collect::<Vec<_>>()
                        .into_iter(),
                },
                result_tx,
            })
            .await;

        result_rx.await.unwrap()
    }

    /// Sends a call proof request to the given peer.
    // TODO: more docs
    // TODO: proper error type
    pub async fn call_proof_request(
        self: Arc<Self>,
        target: PeerId, // TODO: by value?
        chain_id: ChainId,
        config: codec::CallProofRequestConfig<'_, impl Iterator<Item = impl AsRef<[u8]>>>,
    ) -> Result<service::EncodedMerkleProof, ()> {
        // TODO: logs and jaeger integration
        let (result_tx, result_rx) = oneshot::channel();

        let _ = self
            .to_background_tx
            .lock()
            .await
            .send(ToBackground::ForegroundCallProofRequest {
                target: target.clone(),
                chain_id,
                config: codec::CallProofRequestConfig {
                    block_hash: config.block_hash,
                    method: config.method.into_owned().into(),
                    parameter_vectored: config
                        .parameter_vectored
                        .map(|v| v.as_ref().to_vec()) // TODO: to_vec() overhead
                        .collect::<Vec<_>>()
                        .into_iter(),
                },
                result_tx,
            })
            .await;

        result_rx.await.unwrap()
    }
}

/// Error when initializing the network service.
#[derive(Debug, derive_more::Display)]
pub enum InitError {
    /// I/O error when initializing a listener.
    #[display(fmt = "I/O error when creating listener for {_0}: {_1}")]
    ListenerIo(Multiaddr, io::Error),
    /// A listening address passed through the configuration isn't valid.
    #[display(fmt = "A listening address passed through the configuration isn't valid: {_0}")]
    BadListenMultiaddr(Multiaddr),
}

/// Error returned by [`NetworkService::blocks_request`].
#[derive(Debug, derive_more::Display)]
pub enum BlocksRequestError {
    /// No established connection with the target.
    NoConnection,
    /// Error during the request.
    #[display(fmt = "{_0}")]
    Request(service::BlocksRequestError),
}

/// Error returned by [`NetworkService::warp_sync_request`].
#[derive(Debug, derive_more::Display)]
pub enum WarpSyncRequestError {
    /// No established connection with the target.
    NoConnection,
    /// Error during the request.
    #[display(fmt = "{_0}")]
    Request(service::GrandpaWarpSyncRequestError),
}

fn run(mut inner: Inner) {
    // This function is a small hack because I didn't find a better way to store the executor
    // within `Inner` while at the same time spawning the `Inner` using said executor.
    let mut actual_executor = mem::replace(&mut inner.tasks_executor, Box::new(|_| unreachable!()0));
    let (tx, rx) = oneshot::channel();
    actual_executor(Box::pin(async move {
        let actual_executor = rx.await0.unwrap();
        inner.tasks_executor = actual_executor;
        background_task(inner).await;
    }));
    tx.send(actual_executor).unwrap_or_else(|_| panic!()0);
}

async fn background_task(mut inner: Inner) {
    loop {
        enum WakeUpReason {
            IncomingConnection {
                socket: TcpStream,
                socket_addr: SocketAddr,
            },
            NetworkEvent(service::Event<channel::Sender<service::CoordinatorToConnection>>),
            Message(ToBackground),
            ForegroundClosed,
            FromConnectionTask {
                connection_id: service::ConnectionId,
                // TODO: this Option is weird
                message: Option<service::ConnectionToCoordinator>,
            },
            EventSendersReady,
            CanAssignSlot(PeerId, ChainId),
            CanStartConnect(PeerId),
            CanOpenGossip(PeerId, ChainId),
            StartKademliaDiscoveries,
            MessageToConnection {
                connection_id: service::ConnectionId,
                message: service::CoordinatorToConnection,
            },
        }

        let wake_up_reason = 126async {
            inner
                .to_background_rx
                .next()
                .await
                .map_or(WakeUpReason::ForegroundClosed, WakeUpReason::Message)
        }
        .or({
            let event_senders_ready = matches!0(inner.event_senders, either::Left(_));
            let event_pending_send = &inner.event_pending_send;
            let network = &mut inner.network;
            let peering_strategy = &mut inner.peering_strategy;
            let num_pending_out_attempts = &inner.num_pending_out_attempts;
            async move {
                if let Some(event0) = (event_senders_ready && event_pending_send.is_none())
                    .then(|| network.next_event())
                    .flatten()
                {
                    WakeUpReason::NetworkEvent(event)
                } else if let Some((connection_id, message0)) = network.pull_message_to_connection()
                {
                    WakeUpReason::MessageToConnection {
                        connection_id,
                        message,
                    }
                } else if let Some((peer_id, chain_id0)) = network
                    .connected_unopened_gossip_desired()
                    .next()
                    .map(|(peer_id, chain_id, _)| (peer_id.clone(), chain_id)0)
                {
                    WakeUpReason::CanOpenGossip(peer_id, chain_id)
                } else if let Some(peer_id0) = (*num_pending_out_attempts < 16)
                    .then(|| network.unconnected_desired().next().cloned())
                    .flatten()
                {
                    WakeUpReason::CanStartConnect(peer_id)
                } else {
                    'search: loop {
                        let mut earlier_unban = None;

                        for chain_id in network.chains().collect::<Vec<_>>() {
                            if network.gossip_desired_num(
                                chain_id,
                                service::GossipKind::ConsensusTransactions,
                            ) >= network[chain_id].max_slots
                            {
                                continue;
                            }

                            match peering_strategy.pick_assignable_peer(&chain_id, &Instant::now())
                            {
                                basic_peering_strategy::AssignablePeer::Assignable(peer_id) => {
                                    break 'search WakeUpReason::CanAssignSlot(
                                        peer_id.clone(),
                                        chain_id,
                                    )
                                }
                                basic_peering_strategy::AssignablePeer::AllPeersBanned {
                                    next_unban,
                                } => {
                                    if earlier_unban.as_ref().map_or(true, |b| b > next_unban) {
                                        earlier_unban = Some(next_unban.clone());
                                    }
                                }
                                basic_peering_strategy::AssignablePeer::NoPeer => continue,
                            }
                        }

                        if let Some(earlier_unban0) = earlier_unban {
                            smol::Timer::at(earlier_unban).await;
                        } else {
                            future::pending::<()>().await100;
                        }
                    }
                }
            }
        })
        .or(async {
            if let either::Right(sending0) = &mut inner.event_senders {
                let event_senders = sending.await;
                inner.event_senders = either::Left(event_senders);
                WakeUpReason::EventSendersReady
            } else if inner.event_pending_send.is_some() {
                WakeUpReason::EventSendersReady
            } else {
                future::pending().await100
            }
        })
        .or(async {
            (&mut inner.next_discovery).await100;
            inner.next_discovery = smol::Timer::after(inner.next_discovery_period);
            inner.next_discovery_period =
                cmp::min(inner.next_discovery_period * 2, Duration::from_secs(120));
            WakeUpReason::StartKademliaDiscoveries
        })
        .or(async {143
            let (connection_id, message0) = inner.from_connections_rx.next().await0.unwrap();
            WakeUpReason::FromConnectionTask {
                connection_id,
                message,
            }
        })
        .or(async {143
            let Some((socket, socket_addr0)) = inner.incoming_connections.next().await0 else {
                future::pending().await0
            };
            WakeUpReason::IncomingConnection {
                socket,
                socket_addr,
            }
        })
        .await;

        match wake_up_reason {
            WakeUpReason::MessageToConnection {
                connection_id,
                message,
            } => {
                // Note that it is critical for the sending to not take too long here, in order to
                // not block the process of the network service.
                // In particular, if sending the message to the connection is blocked due to
                // sending a message on the connection-to-coordinator channel, this will result
                // in a deadlock.
                // For this reason, the connection task is always ready to immediately accept a
                // message on the coordinator-to-connection channel.
                inner.network[connection_id].send(message).await.unwrap();
            }

            WakeUpReason::FromConnectionTask {
                connection_id,
                message,
            } => {
                if let Some(message) = message {
                    inner
                        .network
                        .inject_connection_message(connection_id, message);
                }
            }

            WakeUpReason::IncomingConnection {
                socket,
                socket_addr,
            } => {
                // The Nagle algorithm, implemented in the kernel, consists in buffering the
                // data to be sent out and waiting a bit before actually sending it out, in
                // order to potentially merge multiple writes in a row into one packet. In
                // the implementation below, it is guaranteed that the buffer in `WithBuffers`
                // is filled with as much data as possible before the operating system gets
                // involved. As such, we disable the Nagle algorithm, in order to avoid adding
                // an artificial delay to all sends.
                let _ = socket.set_nodelay(true);

                let multiaddr = [
                    match socket_addr.ip() {
                        IpAddr::V4(ip) => Protocol::<&[u8]>::Ip4(ip.octets()),
                        IpAddr::V6(ip) => Protocol::Ip6(ip.octets()),
                    },
                    Protocol::Tcp(socket_addr.port()),
                ]
                .into_iter()
                .collect::<Multiaddr>();

                inner.log_callback.log(
                    LogLevel::Debug,
                    format!("incoming-connection; multiaddr={}", multiaddr),
                );

                let (tx, rx) = channel::bounded(16); // TODO: ?!

                let (connection_id, connection_task) = inner.network.add_single_stream_connection(
                    Instant::now(),
                    service::SingleStreamHandshakeKind::MultistreamSelectNoiseYamux {
                        is_initiator: false,
                        noise_key: &inner.noise_key,
                    },
                    multiaddr.clone().into_bytes(),
                    None,
                    tx,
                );

                (inner.tasks_executor)(Box::pin(tasks::connection_task(
                    inner.log_callback.clone(),
                    multiaddr.to_string(),
                    async move { Ok(socket) },
                    connection_id,
                    connection_task,
                    rx,
                    inner.from_connections_tx.clone(),
                )));
            }

            WakeUpReason::StartKademliaDiscoveries => {
                for chain_id in inner.network.chains().collect::<Vec<_>>() {
                    let random_peer_id =
                        PeerId::from_public_key(&peer_id::PublicKey::Ed25519(rand::random()));

                    // TODO: select target closest to the random peer instead
                    let target = inner
                        .network
                        .gossip_connected_peers(
                            chain_id,
                            service::GossipKind::ConsensusTransactions,
                        )
                        .next()
                        .cloned();

                    if let Some(target0) = target {
                        match inner.network.start_kademlia_find_node_request(
                            &target,
                            chain_id,
                            &random_peer_id,
                            Duration::from_secs(20),
                        ) {
                            Ok(_) => {}
                            Err(service::StartRequestError::NoConnection) => unreachable!(),
                        };
                    } else {
                        // TODO: log message
                    }
                }
            }

            WakeUpReason::ForegroundClosed => {
                // TODO: do a clean shutdown of all the connections
                return;
            }

            WakeUpReason::Message(ToBackground::ForegroundDisconnectAndBan {
                peer_id,
                chain_id,
                severity,
                reason,
            }) => {
                // Note that peer doesn't necessarily have an out slot.
                inner.peering_strategy.unassign_slot_and_ban(
                    &chain_id,
                    &peer_id,
                    Instant::now()
                        + Duration::from_secs(match severity {
                            BanSeverity::Low => 10,
                            BanSeverity::High => 40,
                        }),
                );
                if inner.network.gossip_remove_desired(
                    chain_id,
                    &peer_id,
                    service::GossipKind::ConsensusTransactions,
                ) {
                    inner.log_callback.log(
                        LogLevel::Debug,
                        format!(
                            "slot-unassigned; peer_id={}; chain={}; reason=user-ban; user-reason={}",
                            peer_id, inner.network[chain_id].log_name, reason
                        ),
                    );
                }

                if inner.network.gossip_is_connected(
                    chain_id,
                    &peer_id,
                    service::GossipKind::ConsensusTransactions,
                ) {
                    let _close_result = inner.network.gossip_close(
                        chain_id,
                        &peer_id,
                        service::GossipKind::ConsensusTransactions,
                    );
                    debug_assert!(_close_result.is_ok());

                    inner.log_callback.log(
                        LogLevel::Debug,
                        format!(
                            "chain-disconnected; peer_id={}; chain={}",
                            peer_id, inner.network[chain_id].log_name
                        ),
                    );

                    debug_assert!(inner.event_pending_send.is_none());
                    inner.event_pending_send = Some(Event::Disconnected { chain_id, peer_id });
                }
            }

            WakeUpReason::Message(ToBackground::ForegroundAnnounceBlock {
                target,
                chain_id,
                scale_encoded_header,
                is_best,
                result_tx,
            }) => {
                let _ = result_tx.send(inner.network.gossip_send_block_announce(
                    &target,
                    chain_id,
                    &scale_encoded_header,
                    is_best,
                ));
            }
            WakeUpReason::Message(ToBackground::ForegroundSetLocalBestBlock {
                chain_id,
                best_hash,
                best_number,
            }) => {
                inner
                    .network
                    .set_chain_local_best_block(chain_id, best_hash, best_number);
            }
            WakeUpReason::Message(ToBackground::ForegroundBlocksRequest {
                target,
                chain_id,
                config,
                result_tx,
            }) => {
                inner.log_callback.log(
                    LogLevel::Debug,
                    format!(
                        "blocks-request-start; peer_id={}; chain={}; start={}; desired_count={}; direction={}",
                        target,
                        inner.network[chain_id].log_name,
                        match &config.start {
                            codec::BlocksRequestConfigStart::Hash(h) => either::Left(HashDisplay(h)),
                            codec::BlocksRequestConfigStart::Number(n) => either::Right(n),
                        },
                        config.desired_count,
                        match config.direction {
                            codec::BlocksRequestDirection::Ascending => "ascending",
                            codec::BlocksRequestDirection::Descending => "descending",
                        },
                    ),
                );

                match inner.network.start_blocks_request(
                    &target,
                    chain_id,
                    config,
                    Duration::from_secs(12),
                ) {
                    Ok(request_id) => {
                        // TODO: somehow cancel the request if the `rx` is dropped?
                        inner.blocks_requests.insert(request_id, result_tx);
                    }
                    Err(service::StartRequestError::NoConnection) => {
                        inner.log_callback.log(
                            LogLevel::Debug,
                            format!(
                                "blocks-request-ended; peer_id={}; chain={}; outcome=failure; error=no-connection",
                                target,
                                inner.network[chain_id].log_name,
                            ),
                        );
                        let _ = result_tx.send(Err(BlocksRequestError::NoConnection));
                    }
                }
            }
            WakeUpReason::Message(ToBackground::ForegroundWarpSyncRequest {
                target,
                chain_id,
                begin_hash,
                result_tx,
            }) => {
                inner.log_callback.log(
                    LogLevel::Debug,
                    format!(
                        "warp-sync-request-start; peer_id={}; chain={}; begin-hash={}",
                        target,
                        inner.network[chain_id].log_name,
                        HashDisplay(&begin_hash)
                    ),
                );

                match inner.network.start_grandpa_warp_sync_request(
                    &target,
                    chain_id,
                    begin_hash,
                    Duration::from_secs(12),
                ) {
                    Ok(request_id) => {
                        // TODO: somehow cancel the request if the `rx` is dropped?
                        inner.warp_sync_requests.insert(request_id, result_tx);
                    }
                    Err(service::StartRequestError::NoConnection) => {
                        inner.log_callback.log(
                            LogLevel::Debug,
                            format!(
                                "warp-sync-request-ended; peer_id={}; chain={}; outcome=failure; error=no-connection",
                                target,
                                inner.network[chain_id].log_name,
                            ),
                        );
                        let _ = result_tx.send(Err(WarpSyncRequestError::NoConnection));
                    }
                }
            }
            WakeUpReason::Message(ToBackground::ForegroundStorageProofRequest {
                target,
                chain_id,
                config,
                result_tx,
            }) => {
                inner.log_callback.log(
                    LogLevel::Debug,
                    format!(
                        "storage-request-start; peer_id={}; chain={}; block-hash={}; num-keys={}",
                        target,
                        inner.network[chain_id].log_name,
                        HashDisplay(&config.block_hash),
                        config.keys.len()
                    ),
                );

                match inner.network.start_storage_proof_request(
                    &target,
                    chain_id,
                    config,
                    Duration::from_secs(12),
                ) {
                    Ok(request_id) => {
                        // TODO: somehow cancel the request if the `rx` is dropped?
                        inner.storage_requests.insert(request_id, result_tx);
                    }
                    Err(service::StartRequestMaybeTooLargeError::NoConnection) => {
                        inner.log_callback.log(
                            LogLevel::Debug,
                            format!(
                                "storage-request-ended; peer_id={}; chain={}; outcome=failure; error=no-connection",
                                target,
                                inner.network[chain_id].log_name,
                            ),
                        );
                        let _ = result_tx.send(Err(()));
                    }
                    Err(service::StartRequestMaybeTooLargeError::RequestTooLarge) => {
                        inner.log_callback.log(
                            LogLevel::Debug,
                            format!(
                                "storage-request-ended; peer_id={}; chain={}; outcome=failure; error=request-too-large",
                                target,
                                inner.network[chain_id].log_name,
                            ),
                        );
                        let _ = result_tx.send(Err(()));
                    }
                }
            }
            WakeUpReason::Message(ToBackground::ForegroundCallProofRequest {
                target,
                chain_id,
                config,
                result_tx,
            }) => {
                inner.log_callback.log(
                    LogLevel::Debug,
                    format!(
                        "call-proof-request-start; peer_id={}; chain={}; block-hash={}; function={}",
                        target,
                        inner.network[chain_id].log_name,
                        HashDisplay(&config.block_hash),
                        config.method
                    ),
                );

                match inner.network.start_call_proof_request(
                    &target,
                    chain_id,
                    config,
                    Duration::from_secs(12),
                ) {
                    Ok(request_id) => {
                        // TODO: somehow cancel the request if the `rx` is dropped?
                        inner.call_proof_requests.insert(request_id, result_tx);
                    }
                    Err(service::StartRequestMaybeTooLargeError::NoConnection) => {
                        inner.log_callback.log(
                            LogLevel::Debug,
                            format!(
                                "call-proof-request-ended; peer_id={}; chain={}; outcome=failure; error=no-connection",
                                target,
                                inner.network[chain_id].log_name,
                            ),
                        );
                        let _ = result_tx.send(Err(()));
                    }
                    Err(service::StartRequestMaybeTooLargeError::RequestTooLarge) => {
                        inner.log_callback.log(
                            LogLevel::Debug,
                            format!(
                                "call-proof-request-ended; peer_id={}; chain={}; outcome=failure; error=request-too-large",
                                target,
                                inner.network[chain_id].log_name,
                            ),
                        );
                        let _ = result_tx.send(Err(()));
                    }
                }
            }
            WakeUpReason::Message(ToBackground::ForegroundGetNumConnections { result_tx }) => {
                let _ = result_tx.send(inner.network.num_connections());
            }
            WakeUpReason::Message(ToBackground::ForegroundGetNumPeers {
                chain_id,
                result_tx,
            }) => {
                // TODO: optimize?
                let _ = result_tx.send(
                    inner
                        .network
                        .gossip_connected_peers(
                            chain_id,
                            service::GossipKind::ConsensusTransactions,
                        )
                        .count(),
                );
            }
            WakeUpReason::Message(ToBackground::ForegroundGetNumTotalPeers { result_tx }) => {
                // TODO: optimize?
                let total = inner
                    .network
                    .chains()
                    .map(|chain_id| {
                        inner
                            .network
                            .gossip_connected_peers(
                                chain_id,
                                service::GossipKind::ConsensusTransactions,
                            )
                            .count()
                    })
                    .sum();
                let _ = result_tx.send(total);
            }

            WakeUpReason::EventSendersReady => {
                // Dispatch the pending event, if any, to the various senders.

                // We made sure that the senders were ready before generating an event.
                let either::Left(event_senders) = &mut inner.event_senders else {
                    unreachable!()
                };

                if let Some(event_to_dispatch) = inner.event_pending_send.take() {
                    let mut event_senders = mem::take(event_senders);
                    inner.event_senders = either::Right(Box::pin(async move {
                        // Elements in `event_senders` are removed one by one and inserted
                        // back if the channel is still open.
                        for index in (0..event_senders.len()).rev() {
                            let event_sender = event_senders.swap_remove(index);
                            if event_sender.send(event_to_dispatch.clone()).await.is_err() {
                                continue;
                            }

                            event_senders.push(event_sender);
                        }
                        event_senders
                    }));
                }
            }

            WakeUpReason::NetworkEvent(service::Event::HandshakeFinished {
                id,
                expected_peer_id,
                peer_id,
                ..
            }) => {
                inner.num_pending_out_attempts -= 1;

                let remote_addr =
                    Multiaddr::from_bytes(inner.network.connection_remote_addr(id).to_owned())
                        .unwrap(); // TODO: review this unwrap
                if let Some(expected_peer_id) = expected_peer_id.as_ref().filter(|p| **p != peer_id)
                {
                    inner
                    .log_callback
                    .log(LogLevel::Debug, format!("connected-peer-id-mismatch; expected_peer_id={}; actual_peer_id={}; address={}", expected_peer_id, peer_id, remote_addr));

                    let _was_in = inner
                        .peering_strategy
                        .decrease_address_connections_and_remove_if_zero(
                            expected_peer_id,
                            remote_addr.as_ref(),
                        );
                    debug_assert!(_was_in.is_ok());
                    if let basic_peering_strategy::InsertAddressConnectionsResult::Inserted {
                        address_removed: Some(addr_rm),
                    } = inner.peering_strategy.increase_address_connections(
                        &peer_id,
                        remote_addr.into_bytes().to_owned(),
                        10, // TODO: constant
                    ) {
                        let addr_rm = Multiaddr::from_bytes(addr_rm).unwrap();
                        inner.log_callback.log(
                            LogLevel::Debug,
                            format!("address-purged; peer_id={}; address={}", peer_id, addr_rm),
                        );
                    }
                } else {
                    inner
                        .log_callback
                        .log(LogLevel::Debug, format!("connected; peer_id={}", peer_id));
                }
            }

            WakeUpReason::NetworkEvent(service::Event::PreHandshakeDisconnected {
                expected_peer_id: Some(_),
                ..
            })
            | WakeUpReason::NetworkEvent(service::Event::Disconnected { .. }) => {
                let (address, peer_id, handshake_finished) = match wake_up_reason {
                    WakeUpReason::NetworkEvent(service::Event::PreHandshakeDisconnected {
                        address,
                        expected_peer_id: Some(peer_id),
                        ..
                    }) => (address, peer_id, false),
                    WakeUpReason::NetworkEvent(service::Event::Disconnected {
                        address,
                        peer_id,
                        ..
                    }) => (address, peer_id, true),
                    _ => unreachable!(),
                };

                if !handshake_finished {
                    inner.num_pending_out_attempts -= 1;
                }

                inner
                    .peering_strategy
                    .decrease_address_connections(&peer_id, &address)
                    .unwrap();
                let address = Multiaddr::from_bytes(&address).unwrap();
                inner.log_callback.log(
                    LogLevel::Debug,
                    format!(
                        "disconnected; handshake-finished={}; peer_id={}; address={}",
                        handshake_finished, peer_id, address
                    ),
                );

                // Ban the peer in order to avoid trying over and over again the same address(es).
                // Even if the handshake was finished, it is possible that the peer simply shuts
                // down connections immediately after it has been opened, hence the ban.
                // Due to race conditions and peerid mismatches, it is possible that there is
                // another existing connection or connection attempt with that same peer. However,
                // it is not possible to be sure that we will reach 0 connections or connection
                // attempts, and thus we ban the peer every time.
                let ban_duration = Duration::from_secs(5);
                inner.network.gossip_remove_desired_all(
                    &peer_id,
                    service::GossipKind::ConsensusTransactions,
                );
                for (&chain_id, what_happened) in inner
                    .peering_strategy
                    .unassign_slots_and_ban(&peer_id, Instant::now() + ban_duration)
                {
                    if matches!(
                        what_happened,
                        basic_peering_strategy::UnassignSlotsAndBan::Banned { had_slot: true }
                    ) {
                        inner.log_callback.log(
                            LogLevel::Debug,
                            format!(
                                "slot-unassigned; peer_id={}; chain={}; reason=disconnected",
                                peer_id, inner.network[chain_id].log_name
                            ),
                        );
                    }
                }
            }

            WakeUpReason::NetworkEvent(service::Event::PreHandshakeDisconnected {
                expected_peer_id: None,
                address,
                ..
            }) => {
                inner.log_callback.log(
                    LogLevel::Debug,
                    format!(
                        "disconnected; handshake-finished=false; address={}",
                        Multiaddr::from_bytes(&address).unwrap()
                    ),
                );
            }

            WakeUpReason::NetworkEvent(service::Event::PingOutSuccess {
                id,
                peer_id,
                ping_time,
            }) => {
                let remote_addr =
                    Multiaddr::from_bytes(inner.network.connection_remote_addr(id).to_owned())
                        .unwrap(); // TODO: review this unwrap
                inner.log_callback.log(
                    LogLevel::Debug,
                    format!("ping; peer_id={peer_id}; remote_addr={remote_addr}); ping-time={ping_time:?}"),
                );
            }

            WakeUpReason::NetworkEvent(service::Event::BlockAnnounce {
                chain_id,
                peer_id,
                announce,
            }) => {
                let decoded = announce.decode();
                let header_hash =
                    header::hash_from_scale_encoded_header(decoded.scale_encoded_header);
                match header::decode(
                    decoded.scale_encoded_header,
                    inner.network.block_number_bytes(chain_id),
                ) {
                    Ok(decoded_header) => {
                        let mut _jaeger_span = inner.jaeger_service.block_announce_receive_span(
                            &inner.local_peer_id,
                            &peer_id,
                            decoded_header.number,
                            &decoded_header.hash(inner.network.block_number_bytes(chain_id)),
                        );

                        inner.log_callback.log(LogLevel::Debug, format!(
                            "block-announce; peer_id={}; chain={}; hash={}; number={}; is_best={:?}",
                            peer_id, inner.network[chain_id].log_name, HashDisplay(&header_hash), decoded_header.number, decoded.is_best
                        ));

                        debug_assert!(inner.event_pending_send.is_none());
                        inner.event_pending_send = Some(Event::BlockAnnounce {
                            chain_id,
                            peer_id,
                            is_best: decoded.is_best,
                            scale_encoded_header: decoded.scale_encoded_header.to_owned(), // TODO: somewhat wasteful to copy here, could pass the entire announce
                        });
                    }
                    Err(error) => {
                        inner.log_callback.log(LogLevel::Warn, format!(
                            "block-announce-bad-header; peer_id={}; chain={}; hash={}; is_best={:?}; error={}",
                            peer_id, inner.network[chain_id].log_name, HashDisplay(&header_hash), decoded.is_best, error
                        ));

                        if inner.network.gossip_remove_desired(
                            chain_id,
                            &peer_id,
                            service::GossipKind::ConsensusTransactions,
                        ) {
                            inner.peering_strategy.unassign_slot_and_ban(
                                &chain_id,
                                &peer_id,
                                Instant::now() + Duration::from_secs(10),
                            );
                            inner.log_callback.log(
                                LogLevel::Debug,
                                format!(
                                    "slot-unassigned; peer_id={}; chain={}; reason=bad-block-announce",
                                    peer_id, inner.network[chain_id].log_name
                                ),
                            );
                        }
                        let _ = inner.network.gossip_close(
                            chain_id,
                            &peer_id,
                            service::GossipKind::ConsensusTransactions,
                        ); // TODO: what is the return value?
                    }
                }
            }
            WakeUpReason::NetworkEvent(service::Event::GossipConnected {
                peer_id,
                chain_id,
                best_number,
                best_hash,
                ..
            }) => {
                inner.log_callback.log(
                    LogLevel::Debug,
                    format!(
                        "chain-connected; peer_id={}; chain={}; best_number={}; best_hash={}",
                        peer_id,
                        inner.network[chain_id].log_name,
                        best_number,
                        HashDisplay(&best_hash),
                    ),
                );
                debug_assert!(inner.event_pending_send.is_none());
                inner.event_pending_send = Some(Event::Connected {
                    peer_id,
                    chain_id,
                    best_block_number: best_number,
                    best_block_hash: best_hash,
                });
            }
            WakeUpReason::NetworkEvent(service::Event::GossipDisconnected {
                peer_id,
                chain_id,
                ..
            }) => {
                inner.log_callback.log(
                    LogLevel::Debug,
                    format!(
                        "chain-disconnected; peer_id={}; chain={}",
                        peer_id, inner.network[chain_id].log_name
                    ),
                );

                // Note that peer doesn't necessarily have an out slot, as this event
                // might happen as a result of an inbound gossip connection.
                inner.peering_strategy.unassign_slot_and_ban(
                    &chain_id,
                    &peer_id,
                    Instant::now() + Duration::from_secs(10),
                );
                if inner.network.gossip_remove_desired(
                    chain_id,
                    &peer_id,
                    service::GossipKind::ConsensusTransactions,
                ) {
                    inner.log_callback.log(
                        LogLevel::Debug,
                        format!(
                            "slot-unassigned; peer_id={}; chain={}; reason=gossip-disconnected",
                            peer_id, inner.network[chain_id].log_name
                        ),
                    );
                }

                debug_assert!(inner.event_pending_send.is_none());
                inner.event_pending_send = Some(Event::Disconnected { chain_id, peer_id });
            }
            WakeUpReason::NetworkEvent(service::Event::GossipOpenFailed {
                chain_id,
                peer_id,
                error,
                ..
            }) => {
                inner.log_callback.log(
                    LogLevel::Debug,
                    format!(
                        "chain-connect-attempt-failed; peer_id={}; chain={}; error={}",
                        peer_id, inner.network[chain_id].log_name, error
                    ),
                );

                // Note that peer doesn't necessarily have an out slot, as this event
                // might happen as a result of an inbound gossip connection.
                if inner.network.gossip_remove_desired(
                    chain_id,
                    &peer_id,
                    service::GossipKind::ConsensusTransactions,
                ) {
                    inner.log_callback.log(
                        LogLevel::Debug,
                        format!(
                            "slot-unassigned; peer_id={}; chain={}; reason=gossip-open-failed",
                            peer_id, inner.network[chain_id].log_name
                        ),
                    );
                }

                if let service::GossipConnectError::GenesisMismatch { .. } = error {
                    inner
                        .peering_strategy
                        .unassign_slot_and_remove_chain_peer(&chain_id, &peer_id);
                } else {
                    inner.peering_strategy.unassign_slot_and_ban(
                        &chain_id,
                        &peer_id,
                        Instant::now() + Duration::from_secs(15),
                    );
                }
            }
            WakeUpReason::NetworkEvent(service::Event::GossipInDesired {
                chain_id,
                peer_id,
                kind: service::GossipKind::ConsensusTransactions,
            }) => {
                // TODO: log this
                // The networking state machine guarantees that `GossipInDesired`
                // can't happen if we are already opening an out slot, which we do
                // immediately.
                // TODO: add debug_assert! ^
                if inner
                    .network
                    .opened_gossip_undesired_by_chain(chain_id)
                    .count()
                    < inner.network[chain_id].max_in_peers
                {
                    inner
                        .network
                        .gossip_open(
                            chain_id,
                            &peer_id,
                            service::GossipKind::ConsensusTransactions,
                        )
                        .unwrap();
                } else {
                    inner
                        .network
                        .gossip_close(
                            chain_id,
                            &peer_id,
                            service::GossipKind::ConsensusTransactions,
                        )
                        .unwrap();
                }
            }
            WakeUpReason::NetworkEvent(service::Event::GossipInDesiredCancel { .. }) => {
                // All `GossipInDesired` are immediately accepted or rejected, meaning
                // that this event can't happen.
                unreachable!()
            }
            WakeUpReason::NetworkEvent(service::Event::RequestResult {
                substream_id,
                peer_id,
                chain_id,
                response: service::RequestResult::Blocks(response),
            }) => {
                match &response {
                    Ok(success) => {
                        inner.log_callback.log(
                            LogLevel::Debug,
                            format!(
                                "blocks-request-ended; outcome=success; peer_id={peer_id}; chain={}; response-blocks={}",
                                inner.network[chain_id].log_name,
                                success.len()
                            ),
                        );
                    }
                    Err(err) => {
                        inner.log_callback.log(
                            LogLevel::Debug,
                            format!("blocks-request-ended; outcome=failure; peer_id={peer_id}; chain={}; error={}",
                            inner.network[chain_id].log_name, err),
                        );
                    }
                }

                let _ = inner
                    .blocks_requests
                    .remove(&substream_id)
                    .unwrap()
                    .send(response.map_err(BlocksRequestError::Request));
            }
            WakeUpReason::NetworkEvent(service::Event::RequestResult {
                substream_id,
                peer_id,
                chain_id,
                response: service::RequestResult::GrandpaWarpSync(response),
            }) => {
                match &response {
                    Ok(success) => {
                        let decoded = success.decode();
                        inner.log_callback.log(
                            LogLevel::Debug,
                            format!(
                                "warp-sync-request-ended; outcome=success; peer_id={peer_id}; chain={}; num-fragments={}; is-finished={:?}",
                                inner.network[chain_id].log_name,
                                decoded.fragments.len(), decoded.is_finished,
                            ),
                        );
                    }
                    Err(err) => {
                        inner.log_callback.log(
                            LogLevel::Debug,
                            format!("warp-sync-request-ended; outcome=failure; peer_id={peer_id}; chain={}; error={}",
                            inner.network[chain_id].log_name, err),
                        );
                    }
                }

                let _ = inner
                    .warp_sync_requests
                    .remove(&substream_id)
                    .unwrap()
                    .send(response.map_err(WarpSyncRequestError::Request));
            }
            WakeUpReason::NetworkEvent(service::Event::RequestResult {
                substream_id,
                peer_id,
                chain_id,
                response: service::RequestResult::StorageProof(response),
            }) => {
                match &response {
                    Ok(success) => {
                        inner.log_callback.log(
                            LogLevel::Debug,
                            format!(
                                "storage-request-ended; outcome=success; peer_id={peer_id}; chain={}; proof-size={}",
                                inner.network[chain_id].log_name,
                                BytesDisplay(u64::try_from(success.decode().len()).unwrap()),
                            ),
                        );
                    }
                    Err(err) => {
                        inner.log_callback.log(
                            LogLevel::Debug,
                            format!(
                                "storage-request-ended; outcome=failure; peer_id={peer_id}; chain={}; error={}",
                                inner.network[chain_id].log_name, err
                            ),
                        );
                    }
                }

                let _ = inner
                    .storage_requests
                    .remove(&substream_id)
                    .unwrap()
                    .send(response.map_err(|_| ()));
            }
            WakeUpReason::NetworkEvent(service::Event::RequestResult {
                substream_id,
                peer_id,
                chain_id,
                response: service::RequestResult::CallProof(response),
            }) => {
                match &response {
                    Ok(success) => {
                        inner.log_callback.log(
                            LogLevel::Debug,
                            format!(
                                "call-proof-request-ended; outcome=success; peer_id={peer_id}; chain={}; proof-size={}",
                                inner.network[chain_id].log_name,
                                BytesDisplay(u64::try_from(success.decode().len()).unwrap()),
                            ),
                        );
                    }
                    Err(err) => {
                        inner.log_callback.log(
                            LogLevel::Debug,
                            format!(
                                "call-proof-request-ended; outcome=failure; peer_id={peer_id}; chain={}; error={}",
                                inner.network[chain_id].log_name,
                                err
                            ),
                        );
                    }
                }

                let _ = inner
                    .call_proof_requests
                    .remove(&substream_id)
                    .unwrap()
                    .send(response.map_err(|_| ()));
            }
            WakeUpReason::NetworkEvent(service::Event::RequestResult {
                peer_id: kademlia_request_target,
                chain_id,
                response: service::RequestResult::KademliaFindNode(Ok(nodes)),
                ..
            }) => {
                for (peer_id, addrs) in nodes {
                    let mut valid_addrs = Vec::with_capacity(addrs.len());
                    for addr in addrs {
                        match Multiaddr::from_bytes(addr) {
                            Ok(a) => valid_addrs.push(a),
                            Err((error, addr)) => {
                                inner.log_callback.log(
                                    LogLevel::Debug,
                                    format!(
                                        "discovery-invalid-address; error={error}, addr={}, discovered_from={kademlia_request_target}",
                                        hex::encode(&addr)
                                    ),
                                );
                                continue;
                            }
                        }
                    }

                    if !valid_addrs.is_empty() {
                        // Note that we must call this function before `insert_address`,
                        // as documented in `basic_peering_strategy`.
                        if let basic_peering_strategy::InsertChainPeerResult::Inserted {
                            peer_removed: Some(peer_removed),
                        } = inner.peering_strategy.insert_chain_peer(
                            chain_id,
                            peer_id.clone(),
                            100, // TODO: constant
                        ) {
                            inner.log_callback.log(
                                LogLevel::Debug,
                                format!(
                                    "peer-forgotten; peer_id={}; chain={}",
                                    peer_removed, inner.network[chain_id].log_name
                                ),
                            );
                        }
                    }

                    for addr in valid_addrs {
                        inner.log_callback.log(
                            LogLevel::Debug,
                            format!(
                                "discovered; chain={}; peer_id={peer_id}; address={addr}; discovered_from={kademlia_request_target}",
                                inner.network[chain_id].log_name
                            ),
                        );

                        match inner
                            .peering_strategy
                             .insert_address(&peer_id, addr.into_bytes(), 10) // TODO: constant
                            {
                                basic_peering_strategy::InsertAddressResult::Inserted { address_removed: Some(addr_rm) } => {
                                    let addr_rm = Multiaddr::from_bytes(addr_rm).unwrap();
                                    inner
                                        .log_callback
                                        .log(LogLevel::Debug, format!("address-purged; peer_id={}; address={}", peer_id, addr_rm));
                                }
                                basic_peering_strategy::InsertAddressResult::UnknownPeer => unreachable!(),
                                _ => {}
                            }
                    }
                }
            }
            WakeUpReason::NetworkEvent(service::Event::RequestResult {
                peer_id,
                chain_id,
                response: service::RequestResult::KademliaFindNode(Err(error)),
                ..
            }) => {
                inner.log_callback.log(
                    LogLevel::Debug,
                    format!(
                        "discovery-error; chain={}; peer_id={peer_id}; error={error}",
                        inner.network[chain_id].log_name
                    ),
                );
            }
            WakeUpReason::NetworkEvent(service::Event::RequestResult { .. }) => {
                // We never start a request of any other kind.
                unreachable!()
            }
            WakeUpReason::NetworkEvent(service::Event::RequestInCancel { .. }) => {
                // Requests are answered immediately, and thus cancelling events can't happen.
                unreachable!()
            }
            WakeUpReason::NetworkEvent(service::Event::IdentifyRequestIn {
                peer_id,
                substream_id,
            }) => {
                inner.log_callback.log(
                    LogLevel::Debug,
                    format!("identify-request; peer_id={}", peer_id),
                );
                inner
                    .network
                    .respond_identify(substream_id, &inner.identify_agent_version);
            }
            WakeUpReason::NetworkEvent(service::Event::BlocksRequestIn {
                peer_id,
                chain_id,
                config,
                substream_id,
            }) => {
                inner.log_callback.log(
                    LogLevel::Debug,
                    format!(
                        "incoming-blocks-request; peer_id={}; chain={}",
                        peer_id, inner.network[chain_id].log_name
                    ),
                );
                let mut _jaeger_span = inner.jaeger_service.incoming_block_request_span(
                    &inner.local_peer_id,
                    &peer_id,
                    config.desired_count.get(),
                    if let (1, codec::BlocksRequestConfigStart::Hash(block_hash)) =
                        (config.desired_count.get(), &config.start)
                    {
                        Some(block_hash)
                    } else {
                        None
                    },
                );

                // TODO: is it a good idea to await here while the lock is held and freezing the entire networking background task?
                let response = blocks_request_response(
                    &inner.network[chain_id].database,
                    inner.network.block_number_bytes(chain_id),
                    config,
                )
                .await;
                inner.network.respond_blocks(
                    substream_id,
                    match response {
                        Ok(b) => Some(b),
                        Err(error) => {
                            inner.log_callback.log(
                                LogLevel::Warn,
                                format!("incoming-blocks-request-error; error={}", error),
                            );
                            None
                        }
                    },
                );
            }
            WakeUpReason::NetworkEvent(service::Event::GrandpaNeighborPacket {
                chain_id,
                peer_id,
                state,
            }) => {
                inner.log_callback.log(LogLevel::Debug, format!(
                    "grandpa-neighbor-packet; peer_id={}; chain={}; round_number={}; set_id={}; commit_finalized_height={}",
                    peer_id,
                    inner.network[chain_id].log_name,
                    state.round_number,
                    state.set_id,
                    state.commit_finalized_height,
                ));

                debug_assert!(inner.event_pending_send.is_none());
                inner.event_pending_send = Some(Event::GrandpaNeighborPacket {
                    chain_id,
                    peer_id,
                    finalized_block_height: state.commit_finalized_height,
                });
            }
            WakeUpReason::NetworkEvent(service::Event::GrandpaCommitMessage {
                chain_id,
                peer_id,
                message,
            }) => {
                inner.log_callback.log(
                    LogLevel::Debug,
                    format!(
                        "grandpa-commit-message; peer_id={}; chain={}; target_hash={}",
                        peer_id,
                        inner.network[chain_id].log_name,
                        HashDisplay(message.decode().target_hash),
                    ),
                );
            }
            WakeUpReason::NetworkEvent(service::Event::ProtocolError { peer_id, error }) => {
                inner.log_callback.log(
                    LogLevel::Warn,
                    format!("protocol-error; peer_id={}; error={}", peer_id, error),
                );
                inner
                    .peering_strategy
                    .unassign_slots_and_ban(&peer_id, Instant::now() + Duration::from_secs(5));
                // TODO: log chain names?
                inner.log_callback.log(
                    LogLevel::Debug,
                    format!(
                        "all-slots-unassigned; reason=no-address; peer_id={}",
                        peer_id
                    ),
                );
            }

            WakeUpReason::CanAssignSlot(peer_id, chain_id) => {
                inner.peering_strategy.assign_slot(&chain_id, &peer_id);

                inner.log_callback.log(
                    LogLevel::Debug,
                    format!(
                        "slot-assigned; peer_id={}; chain={}",
                        peer_id, inner.network[chain_id].log_name
                    ),
                );

                inner.network.gossip_insert_desired(
                    chain_id,
                    peer_id,
                    service::GossipKind::ConsensusTransactions,
                );
            }

            WakeUpReason::CanStartConnect(peer_id) => {
                inner.num_pending_out_attempts += 1;

                let Some(multiaddr) = inner
                    .peering_strategy
                    .pick_address_and_add_connection(&peer_id)
                else {
                    // There is no address for that peer in the address book.
                    inner.network.gossip_remove_desired_all(
                        &peer_id,
                        service::GossipKind::ConsensusTransactions,
                    );
                    for (chain_id, what_happened) in inner
                        .peering_strategy
                        .unassign_slots_and_ban(&peer_id, Instant::now() + Duration::from_secs(10))
                    {
                        if matches!(
                            what_happened,
                            basic_peering_strategy::UnassignSlotsAndBan::Banned { had_slot: true }
                        ) {
                            inner.log_callback.log(
                                LogLevel::Debug,
                                format!(
                                    "slot-unassigned; peer_id={}; chain={}; reason=no-address",
                                    peer_id, inner.network[*chain_id].log_name
                                ),
                            );
                        }
                    }
                    continue;
                };

                let multiaddr = match multiaddr::Multiaddr::from_bytes(multiaddr.to_owned()) {
                    Ok(a) => a,
                    Err((multiaddr::FromBytesError, multiaddr)) => {
                        // Address is in an invalid format.
                        inner.log_callback.log(
                            LogLevel::Debug,
                            format!(
                                "invalid-address; peer_id={}; address={:?}",
                                peer_id, multiaddr
                            ),
                        );
                        let _was_in = inner
                            .peering_strategy
                            .decrease_address_connections_and_remove_if_zero(&peer_id, &multiaddr);
                        debug_assert!(_was_in.is_ok());
                        continue;
                    }
                };

                // Convert the `multiaddr` (typically of the form `/ip4/a.b.c.d/tcp/d`) into
                // a `Future<dyn Output = Result<TcpStream, ...>>`.
                let socket = match tasks::multiaddr_to_socket(&multiaddr) {
                    Ok(socket) => socket,
                    Err(_) => {
                        // Address is in an invalid format or isn't supported.
                        inner.log_callback.log(
                            LogLevel::Debug,
                            format!(
                                "invalid-address; peer_id={}; address={}",
                                peer_id, multiaddr
                            ),
                        );
                        let _was_in = inner
                            .peering_strategy
                            .decrease_address_connections_and_remove_if_zero(
                                &peer_id,
                                multiaddr.as_ref(),
                            );
                        debug_assert!(_was_in.is_ok());
                        continue;
                    }
                };

                inner.log_callback.log(
                    LogLevel::Debug,
                    format!("start-connecting; peer_id={peer_id}; address={multiaddr}"),
                );

                let (tx, rx) = channel::bounded(16); // TODO: ?!

                let (connection_id, connection_task) = inner.network.add_single_stream_connection(
                    Instant::now(),
                    service::SingleStreamHandshakeKind::MultistreamSelectNoiseYamux {
                        is_initiator: true,
                        noise_key: &inner.noise_key,
                    },
                    multiaddr.clone().into_bytes(),
                    Some(peer_id.clone()),
                    tx,
                );

                // Handle the connection in a separate task.
                (inner.tasks_executor)(Box::pin(tasks::connection_task(
                    inner.log_callback.clone(),
                    multiaddr.to_string(),
                    socket,
                    connection_id,
                    connection_task,
                    rx,
                    inner.from_connections_tx.clone(),
                )));
            }

            WakeUpReason::CanOpenGossip(peer_id, chain_id) => {
                inner
                    .network
                    .gossip_open(
                        chain_id,
                        &peer_id,
                        service::GossipKind::ConsensusTransactions,
                    )
                    .unwrap();

                inner.log_callback.log(
                    LogLevel::Debug,
                    format!(
                        "gossip-open; peer_id={}; chain={}",
                        peer_id, &inner.network[chain_id].log_name
                    ),
                );
            }
        }
    }
}

/// Builds the response to a block request by reading from the given database.
async fn blocks_request_response(
    database: &database_thread::DatabaseThread,
    block_number_bytes: usize,
    config: codec::BlocksRequestConfig,
) -> Result<Vec<codec::BlockData>, full_sqlite::CorruptedError> {
    database
        .with_database(move |database| {
            let num_blocks = cmp::min(
                usize::try_from(config.desired_count.get()).unwrap_or(usize::MAX),
                128,
            );

            let mut output = Vec::with_capacity(num_blocks);
            let mut next_block = config.start;

            loop {
                if output.len() >= num_blocks {
                    break;
                }

                let hash = match next_block {
                    codec::BlocksRequestConfigStart::Hash(hash) => hash,
                    codec::BlocksRequestConfigStart::Number(number) => {
                        // TODO: naive block selection ; should choose the best chain instead
                        match database.block_hash_by_number(number)?.next() {
                            Some(h) => h,
                            None => break,
                        }
                    }
                };

                let header = match database.block_scale_encoded_header(&hash)? {
                    Some(h) => h,
                    None => break,
                };

                next_block = {
                    let decoded = header::decode(&header, block_number_bytes).unwrap();
                    match config.direction {
                        codec::BlocksRequestDirection::Ascending => {
                            // TODO: right now, since we don't necessarily pick the best chain in `block_hash_by_number`, it is possible that the next block doesn't have the current block as parent
                            codec::BlocksRequestConfigStart::Number(decoded.number + 1)
                        }
                        codec::BlocksRequestDirection::Descending => {
                            codec::BlocksRequestConfigStart::Hash(*decoded.parent_hash)
                        }
                    }
                };

                output.push(codec::BlockData {
                    hash,
                    header: if config.fields.header {
                        Some(header)
                    } else {
                        None
                    },
                    body: if config.fields.body {
                        Some(match database.block_extrinsics(&hash)? {
                            Some(body) => body.collect(),
                            None => break,
                        })
                    } else {
                        None
                    },
                    justifications: if config.fields.justifications {
                        // TODO: justifications aren't saved in database at the moment
                        Some(Vec::new())
                    } else {
                        None
                    },
                });
            }

            Ok(output)
        })
        .await
}

Coverage Report

Created: 2024-05-16 12:16