smoldot 1.1.0

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

//! Collection of libp2p connections.
//!
//! The [`Network`] struct in this module is a collection of libp2p connections. In the
//! documentation below, it is also called the *coordinator*.
//!
//! When a connection is inserted in the collection with [`Network::insert_single_stream`] or
//! [`Network::insert_multi_stream`], two objects are returned: an identifier for this new
//! connection assigned by the collection, and a [`SingleStreamConnectionTask`] or
//! [`MultiStreamConnectionTask`].
//!
//! All the [`SingleStreamConnectionTask`]s/[`MultiStreamConnectionTask`] created by the
//! [`Network`] communicate with that [`Network`] by passing messages. Passing the messages has
//! to be done explicitly by the API user. It is the responsibility of the API user to pull
//! messages from the coordinator (i.e. the [`Network`]) and push them onto the
//! [`SingleStreamConnectionTask`] or [`MultiStreamConnectionTask`] and vice-versa.
//!
//! # Usage
//!
//! - Whenever a new connection is established, use  [`Network::insert_single_stream`] or
//! [`Network::insert_multi_stream`] to allocate a connection in the collection.
//! - When a connection has received data or is ready to send more data, use
//! [`SingleStreamConnectionTask::read_write`], [`SingleStreamConnectionTask::reset`],
//! [`MultiStreamConnectionTask::substream_read_write`], [`MultiStreamConnectionTask::reset`],
//! [`MultiStreamConnectionTask::add_substream`], and/or
//! [`MultiStreamConnectionTask::desired_outbound_substreams`] to synchronize the state of the
//! task with the actual state of the connection.
//! - Pull messages from the [`SingleStreamConnectionTask`]s and [`MultiStreamConnectionTask`]s
//! and inject them into the [`Network`], and vice versa.
//! - In parallel, continuously call [`Network::next_event`] to process the events generated by
//! the calls to [`Network::inject_connection_message`].
//!

use crate::libp2p::connection::noise;

use super::connection::{established, single_stream_handshake};
use alloc::{
    collections::{BTreeMap, BTreeSet, VecDeque},
    string::String,
    sync::Arc,
    vec::Vec,
};
use core::{
    hash::Hash,
    marker::PhantomData,
    ops::{self, Add, Sub},
    time::Duration,
};
use rand_chacha::{
    ChaCha20Rng,
    rand_core::{RngCore as _, SeedableRng as _},
};

pub use super::peer_id::PeerId;
pub use super::read_write::ReadWrite;
pub use established::{InboundError, InboundTy, SubstreamFate};
pub use single_stream_handshake::HandshakeError;

pub use multi_stream::MultiStreamConnectionTask;
pub use single_stream::SingleStreamConnectionTask;

mod multi_stream;
mod single_stream;

/// What kind of handshake to perform on the newly-added connection.
pub enum SingleStreamHandshakeKind<'a> {
    /// Use the multistream-select protocol to negotiate the Noise encryption, then use the
    /// multistream-select protocol to negotiate the Yamux multiplexing.
    MultistreamSelectNoiseYamux {
        /// Must be `true` if the connection has been initiated locally, or `false` if it has been
        /// initiated by the remote.
        is_initiator: bool,
        /// Local secret key to use for the handshake.
        noise_key: &'a noise::NoiseKey,
    },
}

/// What kind of handshake to perform on the newly-added connection.
pub enum MultiStreamHandshakeKind<'a> {
    /// The connection is a WebRTC connection.
    ///
    /// See <https://github.com/libp2p/specs/pull/412> for details.
    ///
    /// The reading and writing side of substreams must never be closed. Substreams can only be
    /// abruptly destroyed by either side.
    WebRtc {
        /// Must be `true` if the connection has been initiated locally, or `false` if it has been
        /// initiated by the remote.
        is_initiator: bool,
        /// Local secret key to use for the handshake.
        noise_key: &'a noise::NoiseKey,
        /// Multihash encoding of the TLS certificate used by the local node at the DTLS layer.
        local_tls_certificate_multihash: Vec<u8>,
        /// Multihash encoding of the TLS certificate used by the remote node at the DTLS layer.
        remote_tls_certificate_multihash: Vec<u8>,
    },
}

/// Configuration for a [`Network`].
pub struct Config {
    /// Seed for the randomness within the networking state machine.
    pub randomness_seed: [u8; 32],

    /// Number of connections containers should initially allocate for.
    pub capacity: usize,

    /// Maximum number of substreams that each remote can have simultaneously opened.
    ///
    /// > **Note**: This limit is necessary in order to avoid DoS attacks where a remote opens too
    /// >           many substreams.
    pub max_inbound_substreams: usize,

    /// Maximum size in bytes of the protocols supported by the local node. Any protocol larger
    /// than that requested by the remote is automatically refused. Necessary in order to avoid
    /// situations where the remote sends an infinitely-sized protocol name.
    pub max_protocol_name_len: usize,

    /// Amount of time after which a connection handshake is considered to have taken too long
    /// and must be aborted.
    pub handshake_timeout: Duration,

    /// Name of the ping protocol on the network.
    pub ping_protocol: String,
}

/// Identifier of a connection spawned by the [`Network`].
//
// Identifiers are never reused.
#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct ConnectionId(u64);

impl ConnectionId {
    /// Value that compares inferior or equal to any possible [`ConnectionId`].
    pub const MIN: Self = ConnectionId(u64::MIN);
    /// Value that compares superior or equal to any possible [`ConnectionId`].
    pub const MAX: Self = ConnectionId(u64::MAX);
}

/// Identifier of a request, or an inbound substream, or an outbound substream.
//
// Identifiers are never reused.
#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct SubstreamId(u64);

impl SubstreamId {
    /// Value that compares inferior or equal to any possible [`SubstreamId`].
    pub const MIN: Self = SubstreamId(u64::MIN);
    /// Value that compares superior or equal to any possible [`SubstreamId`].
    pub const MAX: Self = SubstreamId(u64::MAX);
}

/// Data structure containing the list of all connections, pending or not, and their latest known
/// state. See also [the module-level documentation](..).
pub struct Network<TConn, TNow> {
    /// Messages waiting to be sent to connection tasks.
    messages_to_connections: VecDeque<(ConnectionId, CoordinatorToConnectionInner)>,

    /// Messages received from connection tasks. Processed in [`Network::next_event`].
    pending_incoming_messages: VecDeque<(ConnectionId, ConnectionToCoordinatorInner)>,

    /// Id to assign to the next connection.
    next_connection_id: ConnectionId,

    /// Id to assign to the next substream, such as the next request or next notifications
    /// substream.
    ///
    /// All substreams share the same pool of ids across all connections.
    next_substream_id: SubstreamId,

    /// List of all connections in the data structure.
    connections: hashbrown::HashMap<ConnectionId, Connection<TConn>, fnv::FnvBuildHasher>,

    /// If `Some`, the given connection is in the process of shutting down. Calling
    /// [`Network::next_event`] will cancel all ongoing requests and notification substreams
    /// that concern this connection before processing any other incoming message.
    shutting_down_connection: Option<ConnectionId>,

    /// List of connections for which a [`ConnectionToCoordinatorInner::ShutdownFinished`] has
    /// been received and a [`CoordinatorToConnectionInner::ShutdownFinishedAck`] has been sent.
    /// We can now remove these connections and generate a [`Event::Shutdown`].
    shutdown_finished_connections: VecDeque<ConnectionId>,

    /// List of all outgoing notification substreams that we have opened. Can be either pending
    /// (waiting for the connection task to say whether it has been accepted or not) or fully
    /// open.
    outgoing_notification_substreams:
        hashbrown::HashMap<SubstreamId, (ConnectionId, SubstreamState), fnv::FnvBuildHasher>,

    /// Always contains the same entries as [`Network::outgoing_notification_substreams`] but
    /// ordered differently.
    // TODO: group with the other similar BTreeSets?
    outgoing_notification_substreams_by_connection: BTreeSet<(ConnectionId, SubstreamId)>,

    /// List of all requests that have been started locally.
    outgoing_requests: BTreeSet<(ConnectionId, SubstreamId)>,

    /// List of inbound negotiated substreams that connections have received and haven't been
    /// answered by the API user yet.
    ingoing_negotiated_substreams: hashbrown::HashMap<
        SubstreamId,
        (ConnectionId, established::SubstreamId, bool),
        fnv::FnvBuildHasher,
    >,

    /// Always contains the same entries as [`Network::ingoing_negotiated_substreams`] but ordered
    /// differently.
    // TODO: group with the other similar BTreeSets?
    ingoing_negotiated_substreams_by_connection:
        BTreeMap<(ConnectionId, established::SubstreamId), SubstreamId>,

    /// List in incoming notification substreams that connections have received. Can be either
    /// pending (waiting to be accepted/refused) or fully opened.
    ///
    /// The substream ID of the substream is allocated by the connection task, and thus we need
    /// to keep a mapping of inner `<->` substream IDs.
    ingoing_notification_substreams: hashbrown::HashMap<
        SubstreamId,
        (ConnectionId, SubstreamState, established::SubstreamId),
        fnv::FnvBuildHasher,
    >,

    /// Always contains the same entries as [`Network::ingoing_notification_substreams`] but
    /// ordered differently.
    // TODO: group with the other similar BTreeSets?
    ingoing_notification_substreams_by_connection:
        BTreeMap<(ConnectionId, established::SubstreamId), SubstreamId>,

    /// List of incoming Bitswap substreams with corresponding connections.
    /// All substreams here are fully open as there is no negotiation phase in Bitswap.
    ///
    /// The substream ID of the substream is allocated by the connection task, and thus we need
    /// to keep a mapping of inner `<->` substream IDs.
    ingoing_bitswap_substreams: hashbrown::HashMap<
        SubstreamId,
        (ConnectionId, established::SubstreamId),
        fnv::FnvBuildHasher,
    >,

    /// Always contains the same entries as [`Network::ingoing_bitswap_substreams`] but
    /// ordered differently.
    // TODO: group with the other similar BTreeSets?
    ingoing_bitswap_substreams_by_connection:
        BTreeMap<(ConnectionId, established::SubstreamId), SubstreamId>,

    /// List of all outgoing Bitswap substreams that we have opened. Can be either pending (waiting
    /// for the connection task to say whether it has been accepted or not) or fully open.
    outgoing_bitswap_substreams:
        hashbrown::HashMap<SubstreamId, (ConnectionId, SubstreamState), fnv::FnvBuildHasher>,

    /// Always contains the same entries as [`Network::outgoing_bitswap_substreams`] but
    /// ordered differently.
    // TODO: group with the other similar BTreeSets?
    outgoing_bitswap_substreams_by_connection: BTreeSet<(ConnectionId, SubstreamId)>,

    /// List of requests that connections have received and haven't been answered by the API user
    /// yet.
    ingoing_requests: hashbrown::HashMap<
        SubstreamId,
        (ConnectionId, established::SubstreamId),
        fnv::FnvBuildHasher,
    >,

    /// Always contains the same entries as [`Network::ingoing_requests`] but ordered differently.
    // TODO: group with the other similar BTreeSets?
    ingoing_requests_by_connection: BTreeSet<(ConnectionId, SubstreamId)>,

    /// Generator for randomness seeds given to the established connections.
    randomness_seeds: ChaCha20Rng,

    /// See [`Config::max_inbound_substreams`].
    max_inbound_substreams: usize,

    /// See [`Config::max_protocol_name_len`].
    max_protocol_name_len: usize,

    /// See [`Config::handshake_timeout`].
    handshake_timeout: Duration,

    /// See [`Config::ping_protocol`].
    ping_protocol: Arc<str>,

    // Phantom data to keep the `TNow` type pinned.
    // TODO: considering removing
    now_pin: PhantomData<fn() -> TNow>,
}

struct Connection<TConn> {
    state: InnerConnectionState,

    user_data: TConn,
}

enum InnerConnectionState {
    /// The connection is still in its handshaking state.
    Handshaking,
    /// The connection is fully established.
    Established,
    /// The connection is in the process of shutting down.
    ShuttingDown {
        /// `true` if the state before the shutdown was [`InnerConnectionState::Established`].
        was_established: bool,

        /// `true` if [`Network::start_shutdown`] has been called on this connection.
        ///
        /// Even if the remote starts the shutdown at the same time, from an API perspective if
        /// this flag is `true` it will be considered as if the API user had initiated the
        /// shutdown.
        api_initiated: bool,
    },
}

/// See [`Network::outgoing_notification_substreams`] and
/// [`Network::ingoing_notification_substreams`].
///
/// > **Note**: There is no `Closed` variant, as this corresponds to a lack of entry in the map.
#[derive(Debug, Copy, Clone, PartialEq, Eq, Ord, PartialOrd, Hash)]
enum SubstreamState {
    /// Substream hasn't been accepted or refused yet.
    Pending,
    Open,
    /// Substream is in the process of being closed. Only relevant for inbound substreams.
    RequestedClosing,
}

impl<TConn, TNow> Network<TConn, TNow>
where
    TNow: Clone + Add<Duration, Output = TNow> + Sub<TNow, Output = Duration> + Ord,
{
    /// Initializes a new network data structure.
    pub fn new(config: Config) -> Self {
        // The initial capacities given to the containers below are more or less arbitrary, the
        // objective being to avoid relocating the containers.
        Network {
            messages_to_connections: VecDeque::with_capacity(config.capacity * 2),
            pending_incoming_messages: VecDeque::with_capacity(config.capacity * 2),
            next_substream_id: SubstreamId(0),
            handshake_timeout: config.handshake_timeout,
            next_connection_id: ConnectionId(0),
            connections: hashbrown::HashMap::with_capacity_and_hasher(
                config.capacity,
                Default::default(),
            ),
            shutting_down_connection: None,
            shutdown_finished_connections: VecDeque::with_capacity(config.capacity),
            outgoing_requests: BTreeSet::new(),
            ingoing_requests: hashbrown::HashMap::with_capacity_and_hasher(
                4 * config.capacity,
                Default::default(),
            ),
            ingoing_requests_by_connection: BTreeSet::new(),
            outgoing_notification_substreams: hashbrown::HashMap::with_capacity_and_hasher(
                4 * config.capacity,
                Default::default(),
            ),
            outgoing_notification_substreams_by_connection: BTreeSet::new(),
            ingoing_notification_substreams: hashbrown::HashMap::with_capacity_and_hasher(
                4 * config.capacity,
                Default::default(),
            ),
            ingoing_notification_substreams_by_connection: BTreeMap::new(),
            ingoing_negotiated_substreams: hashbrown::HashMap::with_capacity_and_hasher(
                4 * config.capacity,
                Default::default(),
            ),
            ingoing_negotiated_substreams_by_connection: BTreeMap::new(),
            outgoing_bitswap_substreams: hashbrown::HashMap::with_capacity_and_hasher(
                config.capacity,
                Default::default(),
            ),
            outgoing_bitswap_substreams_by_connection: BTreeSet::new(),
            ingoing_bitswap_substreams: hashbrown::HashMap::with_capacity_and_hasher(
                config.capacity,
                Default::default(),
            ),
            ingoing_bitswap_substreams_by_connection: BTreeMap::new(),
            randomness_seeds: ChaCha20Rng::from_seed(config.randomness_seed),
            max_inbound_substreams: config.max_inbound_substreams,
            max_protocol_name_len: config.max_protocol_name_len,
            ping_protocol: config.ping_protocol.into(),
            now_pin: PhantomData,
        }
    }

    /// Adds a new single-stream connection to the collection.
    ///
    /// Must be passed the moment (as a `TNow`) when the connection process has been started, in
    /// order to determine when the handshake timeout expires.
    pub fn insert_single_stream(
        &mut self,
        when_connection_start: TNow,
        handshake_kind: SingleStreamHandshakeKind,
        substreams_capacity: usize,
        user_data: TConn,
    ) -> (ConnectionId, SingleStreamConnectionTask<TNow>) {
        let connection_id = self.next_connection_id;
        self.next_connection_id.0 += 1;

        // We only support one kind of handshake at the moment. Make sure (at compile time) that
        // the value provided as parameter is indeed the one expected.
        let SingleStreamHandshakeKind::MultistreamSelectNoiseYamux {
            is_initiator,
            noise_key,
        } = handshake_kind;

        let connection_task = SingleStreamConnectionTask::new(single_stream::Config {
            randomness_seed: {
                let mut seed = [0; 32];
                self.randomness_seeds.fill_bytes(&mut seed);
                seed
            },
            handshake: {
                let mut ephemeral_secret_key = zeroize::Zeroizing::new([0; 32]);
                self.randomness_seeds.fill_bytes(&mut *ephemeral_secret_key);
                single_stream_handshake::HealthyHandshake::noise_yamux(
                    noise_key,
                    &ephemeral_secret_key,
                    is_initiator,
                )
            },
            handshake_timeout: when_connection_start + self.handshake_timeout,
            max_inbound_substreams: self.max_inbound_substreams,
            substreams_capacity,
            max_protocol_name_len: self.max_protocol_name_len,
            ping_protocol: self.ping_protocol.clone(),
        });

        let _previous_value = self.connections.insert(
            connection_id,
            Connection {
                state: InnerConnectionState::Handshaking,
                user_data,
            },
        );
        debug_assert!(_previous_value.is_none());

        (connection_id, connection_task)
    }

    /// Adds a new multi-stream connection to the collection.
    ///
    /// Must be passed the moment (as a `TNow`) when the connection process has been started, in
    /// order to determine when the handshake timeout expires.
    pub fn insert_multi_stream<TSubId>(
        &mut self,
        when_connection_start: TNow,
        handshake_kind: MultiStreamHandshakeKind,
        substreams_capacity: usize,
        user_data: TConn,
    ) -> (ConnectionId, MultiStreamConnectionTask<TNow, TSubId>)
    where
        TSubId: Clone + PartialEq + Eq + Hash,
    {
        let connection_id = self.next_connection_id;
        self.next_connection_id.0 += 1;

        // In the WebRTC handshake, the Noise prologue must be set to `"libp2p-webrtc-noise:"`
        // followed with the multihash-encoded fingerprints of the initiator's certificate
        // and the receiver's certificate.
        // See <https://github.com/libp2p/specs/pull/412>.
        let (noise_key, noise_prologue, local_is_noise_initiator) = {
            let MultiStreamHandshakeKind::WebRtc {
                noise_key,
                is_initiator,
                local_tls_certificate_multihash,
                remote_tls_certificate_multihash,
            } = handshake_kind;
            const PREFIX: &[u8] = b"libp2p-webrtc-noise:";
            let mut out = Vec::with_capacity(
                PREFIX.len()
                    + local_tls_certificate_multihash.len()
                    + remote_tls_certificate_multihash.len(),
            );
            out.extend_from_slice(PREFIX);
            if is_initiator {
                out.extend_from_slice(&local_tls_certificate_multihash);
                out.extend_from_slice(&remote_tls_certificate_multihash);
            } else {
                out.extend_from_slice(&remote_tls_certificate_multihash);
                out.extend_from_slice(&local_tls_certificate_multihash);
            }

            // In the WebRTC libp2p protocol, the initiator of the connection is *not* the
            // initiator of the Noise handshake. Instead, it's the "server" that initiates the
            // Noise handshake. This saves a round-trip.
            (noise_key, out, !is_initiator)
        };

        let handshake = {
            let mut noise_ephemeral_key = zeroize::Zeroizing::new([0; 32]);
            self.randomness_seeds.fill_bytes(&mut *noise_ephemeral_key);
            noise::HandshakeInProgress::new(noise::Config {
                key: noise_key,
                is_initiator: local_is_noise_initiator,
                prologue: &noise_prologue,
                ephemeral_secret_key: &noise_ephemeral_key,
            })
        };

        let connection_task = MultiStreamConnectionTask::new(
            {
                let mut seed = [0; 32];
                self.randomness_seeds.fill_bytes(&mut seed);
                seed
            },
            when_connection_start,
            handshake,
            self.max_inbound_substreams,
            substreams_capacity,
            self.max_protocol_name_len,
            self.ping_protocol.clone(),
        );

        let _previous_value = self.connections.insert(
            connection_id,
            Connection {
                state: InnerConnectionState::Handshaking,
                user_data,
            },
        );
        debug_assert!(_previous_value.is_none());

        (connection_id, connection_task)
    }

    /// Switches the connection to a state where it will shut down soon.
    ///
    /// Calling this function does **not** generate a [`Event::StartShutdown`] event for this
    /// connection. The event is implied.
    ///
    /// It is no longer possible to start requests, open notifications substreams, or send
    /// notifications. No new incoming requests or notification substreams will be reported. The
    /// incoming notifications that were sent by the remote before the shutdown started will still
    /// be reported.
    ///
    /// It is possible to call this method on connection that are still in their handshaking
    /// phase.
    ///
    /// This function generates a message destined to the connection. Use
    /// [`Network::pull_message_to_connection`] to process these messages after it has returned.
    ///
    /// # Panic
    ///
    /// Panics if the connection is already shutting down, either because
    /// [`Network::start_shutdown`] was called or a [`Event::StartShutdown`] event was yielded.
    ///
    #[track_caller]
    pub fn start_shutdown(&mut self, connection_id: ConnectionId) {
        let connection = match self.connections.get_mut(&connection_id) {
            Some(c) => c,
            None => panic!(),
        };

        let is_established = match connection.state {
            InnerConnectionState::Handshaking => false,
            InnerConnectionState::Established => true,
            InnerConnectionState::ShuttingDown {
                api_initiated: true,
                ..
            } => panic!("start_shutdown called twice on same connection"), // Forbidden.
            InnerConnectionState::ShuttingDown {
                api_initiated: false,
                ..
            } => panic!("start_shutdown called after StartShutdown event"), // Forbidden.
        };

        connection.state = InnerConnectionState::ShuttingDown {
            was_established: is_established,
            api_initiated: true,
        };

        self.messages_to_connections
            .push_back((connection_id, CoordinatorToConnectionInner::StartShutdown));
    }

    /// Returns true if the collection doesn't contain any connection.
    pub fn is_empty(&self) -> bool {
        self.connections.is_empty()
    }

    /// Returns the number of connections in the collection.
    pub fn len(&self) -> usize {
        self.connections.len()
    }

    /// Returns the state of the given connection.
    ///
    /// # Panic
    ///
    /// Panics if the identifier is invalid or corresponds to a connection that has already
    /// entirely shut down.
    ///
    pub fn connection_state(&self, connection_id: ConnectionId) -> ConnectionState {
        let connection = self.connections.get(&connection_id).unwrap();
        match connection.state {
            InnerConnectionState::Handshaking => ConnectionState {
                established: false,
                shutting_down: false,
            },
            InnerConnectionState::Established => ConnectionState {
                established: true,
                shutting_down: false,
            },
            InnerConnectionState::ShuttingDown {
                was_established, ..
            } => ConnectionState {
                established: was_established,
                shutting_down: true,
            },
        }
    }

    /// Modifies the value that was initially passed through [`Config::max_protocol_name_len`].
    ///
    /// The new value only applies to substreams opened after this function has been called.
    pub fn set_max_protocol_name_len(&mut self, new_max_length: usize) {
        if self.max_protocol_name_len == new_max_length {
            return;
        }

        self.max_protocol_name_len = new_max_length;

        // Send a message to all connections to update this value.
        self.messages_to_connections.reserve(self.connections.len());
        for connection_id in self.connections.keys() {
            self.messages_to_connections.push_back((
                *connection_id,
                CoordinatorToConnectionInner::SetMaxProtocolNameLen { new_max_length },
            ));
        }
    }

    /// Call after an [`Event::InboundNegotiated`] has been emitted in order to accept the protocol
    /// name and indicate the type of the protocol.
    ///
    /// # Panic
    ///
    /// Panics if the substream is not in the correct state.
    ///
    pub fn accept_inbound(&mut self, substream_id: SubstreamId, ty: InboundTy) {
        let (connection_id, inner_substream_id, already_accepted) =
            match self.ingoing_negotiated_substreams.get_mut(&substream_id) {
                Some(s) => s,
                None => panic!(),
            };
        assert!(!*already_accepted);

        self.messages_to_connections.push_back((
            *connection_id,
            CoordinatorToConnectionInner::AcceptInbound {
                substream_id: *inner_substream_id,
                inbound_ty: ty,
            },
        ));

        *already_accepted = true;
    }

    /// Call after an [`Event::InboundNegotiated`] has been emitted in order to reject the
    /// protocol name as not supported.
    ///
    /// # Panic
    ///
    /// Panics if the substream is not in the correct state.
    ///
    pub fn reject_inbound(&mut self, substream_id: SubstreamId) {
        let (connection_id, inner_substream_id, already_accepted) =
            match self.ingoing_negotiated_substreams.remove(&substream_id) {
                Some(s) => s,
                None => panic!(),
            };
        let _was_in = self
            .ingoing_negotiated_substreams_by_connection
            .remove(&(connection_id, inner_substream_id));
        debug_assert!(_was_in.is_some());
        assert!(!already_accepted);

        self.messages_to_connections.push_back((
            connection_id,
            CoordinatorToConnectionInner::RejectInbound {
                substream_id: inner_substream_id,
            },
        ));
    }

    /// Sends a request to the given peer.
    ///
    /// A [`Event::Response`] event will later be generated containing the result of the request.
    /// This event is generated even if the connection the request was sent on has been closed.
    ///
    /// It is invalid to start a request on a connection before a [`Event::HandshakeFinished`]
    /// or after a [`Event::StartShutdown`] has been generated, or after
    /// [`Network::start_shutdown`] has been called.
    ///
    /// Returns a newly-allocated identifier for this substream.
    ///
    /// This function generates a message destined to the connection. Use
    /// [`Network::pull_message_to_connection`] to process these messages after it has returned.
    ///
    /// # Requests
    ///
    /// A request consists in:
    ///
    /// - Opening a substream on an established connection with the target.
    /// - Negotiating the requested protocol (`protocol_index`) on this substream using the
    ///   *multistream-select* protocol.
    /// - Sending the request (`request_data` parameter), prefixed with its length.
    /// - Waiting for the response (prefixed with its length), which is then returned.
    ///
    /// An error happens if the connection closes while the request is in progress, if the remote
    /// doesn't support the given protocol, if the request or response doesn't respect the protocol
    /// limits, or if the remote takes too much time to answer.
    ///
    /// The timeout is the time between the moment the substream is opened and the moment the
    /// response is sent back. It starts ticking only after the request starts being sent. If the
    /// emitter doesn't send the request or if the receiver doesn't answer during this time
    /// window, the request is considered failed.
    ///
    /// # Panic
    ///
    /// Panics if the [`ConnectionId`] is invalid or is a connection that hasn't finished its
    /// handshake or is shutting down.
    ///
    #[track_caller]
    pub fn start_request(
        &mut self,
        target: ConnectionId,
        protocol_name: String,
        request_data: Option<Vec<u8>>,
        timeout: Duration,
        max_response_size: usize,
    ) -> SubstreamId {
        let connection = match self.connections.get(&target) {
            Some(c) => c,
            None => panic!(),
        };
        assert!(matches!(
            connection.state,
            InnerConnectionState::Established
        ));

        let substream_id = self.next_substream_id;
        self.next_substream_id.0 += 1;

        let _was_inserted = self.outgoing_requests.insert((target, substream_id));
        debug_assert!(_was_inserted);

        self.messages_to_connections.push_back((
            target,
            CoordinatorToConnectionInner::StartRequest {
                protocol_name,
                request_data,
                timeout,
                max_response_size,
                substream_id,
            },
        ));

        substream_id
    }

    /// Start opening a notifications substream.
    ///
    /// It is invalid to open a notifications substream on a connection before a
    /// [`Event::HandshakeFinished`] or after a [`Event::StartShutdown`] has been generated, or
    /// after [`Network::start_shutdown`] has been called.
    ///
    /// Returns a newly-allocated identifier for this substream.
    ///
    /// This function generates a message destined to the connection. Use
    /// [`Network::pull_message_to_connection`] to process these messages after it has returned.
    ///
    /// # Panic
    ///
    /// Panics if the [`ConnectionId`] is invalid or is a connection that hasn't finished its
    /// handshake or is shutting down.
    ///
    #[track_caller]
    pub fn open_out_notifications(
        &mut self,
        connection_id: ConnectionId,
        protocol_name: String,
        handshake_timeout: Duration,
        handshake: impl Into<Vec<u8>>,
        max_handshake_size: usize,
    ) -> SubstreamId {
        let connection = match self.connections.get(&connection_id) {
            Some(c) => c,
            None => panic!(),
        };
        assert!(matches!(
            connection.state,
            InnerConnectionState::Established
        ));

        let substream_id = self.next_substream_id;
        self.next_substream_id.0 += 1;

        let _prev_value = self
            .outgoing_notification_substreams
            .insert(substream_id, (connection_id, SubstreamState::Pending));
        debug_assert!(_prev_value.is_none());
        let _was_inserted = self
            .outgoing_notification_substreams_by_connection
            .insert((connection_id, substream_id));
        debug_assert!(_was_inserted);

        self.messages_to_connections.push_back((
            connection_id,
            CoordinatorToConnectionInner::OpenOutNotifications {
                protocol_name,
                handshake: handshake.into(),
                handshake_timeout,
                max_handshake_size,
                substream_id,
            },
        ));

        substream_id
    }

    /// Start closing a previously-open notifications substream, or cancels opening a
    /// notifications substream.
    ///
    /// All the notifications that have been queued remain queued. The substream actually closes
    /// only once the queue is empty.
    ///
    /// Calling this method does *not* emit any event. The [`SubstreamId`] is considered invalid
    /// after this function returns.
    ///
    /// This function generates a message destined to the connection. Use
    /// [`Network::pull_message_to_connection`] to process these messages after it has returned.
    ///
    /// # Panic
    ///
    /// Panics if [`SubstreamId`] doesn't correspond to an outbound notifications substream.
    ///
    #[track_caller]
    pub fn close_out_notifications(&mut self, substream_id: SubstreamId) {
        // Both `Pending` and `Open` states are accepted.
        let (connection_id, _state) =
            match self.outgoing_notification_substreams.remove(&substream_id) {
                Some(s) => s,
                None => panic!(),
            };
        let _was_in = self
            .outgoing_notification_substreams_by_connection
            .remove(&(connection_id, substream_id));
        debug_assert!(_was_in);

        self.messages_to_connections.push_back((
            connection_id,
            CoordinatorToConnectionInner::CloseOutNotifications { substream_id },
        ));
    }

    /// Adds a notification to the queue of notifications to send to the given peer.
    ///
    /// It is invalid to call this on a [`SubstreamId`] before a successful
    /// [`Event::NotificationsOutResult`] has been yielded.
    ///
    /// Each substream maintains a queue of notifications to be sent to the remote. This method
    /// attempts to push a notification to this queue.
    ///
    /// An error is also returned if the queue exceeds a certain size in bytes, for two reasons:
    ///
    /// - Since the content of the queue is transferred at a limited rate, each notification
    /// pushed at the end of the queue will take more time than the previous one to reach the
    /// destination. Once the queue reaches a certain size, the time it would take for
    /// newly-pushed notifications to reach the destination would start being unreasonably large.
    ///
    /// - If the remote deliberately applies back-pressure on the substream, it is undesirable to
    /// increase the memory usage of the local node.
    ///
    /// Similarly, the queue being full is a normal situation and notification protocols should
    /// be designed in such a way that discarding notifications shouldn't have a too negative
    /// impact.
    ///
    /// Regardless of the success of this function, no guarantee exists about the successful
    /// delivery of notifications.
    ///
    /// This function generates a message destined to the connection. Use
    /// [`Network::pull_message_to_connection`] to process these messages after it has returned.
    ///
    /// # Panics
    ///
    /// Panics if [`SubstreamId`] is not a fully open outbound notifications substream.
    ///
    #[track_caller]
    pub fn queue_notification(
        &mut self,
        substream_id: SubstreamId,
        notification: impl Into<Vec<u8>>,
    ) -> Result<(), QueueNotificationError> {
        let (connection_id, state) = match self.outgoing_notification_substreams.get(&substream_id)
        {
            Some(s) => s,
            None => panic!(),
        };
        assert!(matches!(state, SubstreamState::Open));

        //  TODO: add some back-pressure system and return a `QueueNotificationError` if full

        self.messages_to_connections.push_back((
            *connection_id,
            CoordinatorToConnectionInner::QueueNotification {
                substream_id,
                notification: notification.into(),
            },
        ));

        Ok(())
    }

    /// Accepts a request for an inbound notifications substream reported by an
    /// [`Event::NotificationsInOpen`].
    ///
    /// If a [`Event::NotificationsInClose`] event is yielded, then this function must not be
    /// called and will panic.
    ///
    /// This function generates a message destined to the connection. Use
    /// [`Network::pull_message_to_connection`] to process these messages after it has returned.
    ///
    /// # Panic
    ///
    /// Panics if the [`SubstreamId`] doesn't correspond to an inbound notifications substream.
    ///
    #[track_caller]
    pub fn accept_in_notifications(
        &mut self,
        substream_id: SubstreamId,
        handshake: Vec<u8>,
        max_notification_size: usize,
    ) {
        let (connection_id, state, inner_substream_id) =
            match self.ingoing_notification_substreams.get_mut(&substream_id) {
                Some(s) => s,
                None => panic!(),
            };
        assert!(matches!(state, SubstreamState::Pending));

        self.messages_to_connections.push_back((
            *connection_id,
            CoordinatorToConnectionInner::AcceptInNotifications {
                substream_id: *inner_substream_id,
                handshake,
                max_notification_size,
            },
        ));

        *state = SubstreamState::Open;
    }

    /// Rejects a request for an inbound notifications substream reported by an
    /// [`Event::NotificationsInOpen`].
    ///
    /// If a [`Event::NotificationsInClose`] event is yielded, then this function must not be
    /// called and will panic.
    ///
    /// The [`SubstreamId`] is considered no longer valid after this function returns.
    ///
    /// This function generates a message destined to the connection. Use
    /// [`Network::pull_message_to_connection`] to process these messages after it has returned.
    ///
    /// # Panic
    ///
    /// Panics if the [`SubstreamId`] doesn't correspond to an inbound notifications substream.
    ///
    #[track_caller]
    pub fn reject_in_notifications(&mut self, substream_id: SubstreamId) {
        if let Some((connection_id, SubstreamState::Pending, inner_substream_id)) =
            self.ingoing_notification_substreams.remove(&substream_id)
        {
            let _was_in = self
                .ingoing_notification_substreams_by_connection
                .remove(&(connection_id, inner_substream_id));
            debug_assert_eq!(_was_in, Some(substream_id));

            self.messages_to_connections.push_back((
                connection_id,
                CoordinatorToConnectionInner::RejectInNotifications {
                    substream_id: inner_substream_id,
                },
            ));
        } else {
            // Note that, if this is reached, the pending substream is not inserted back
            // in the state machine, meaning that `self` is now in an inconsistent state.
            // But considering that we panic, this state mismatch isn't actually observable.
            panic!()
        }
    }

    /// Start the closing of an inbound notifications substream that was previously accepted with
    /// [`Network::accept_in_notifications`]
    ///
    /// Calling this function will later generate a [`Event::NotificationsInClose`] event once the
    /// substream is effectively closed.
    /// This function gracefully asks the remote to close the substream. The remote has the
    /// duration indicated with `timeout` to effectively close the substream. In the meanwhile,
    /// notifications can still be received.
    ///
    /// This function generates a message destined to the connection. Use
    /// [`Network::pull_message_to_connection`] to process these messages after it has returned.
    ///
    /// # Panic
    ///
    /// Panics if the [`SubstreamId`] doesn't correspond to an accepted inbound notifications
    /// substream.
    ///
    #[track_caller]
    pub fn start_close_in_notifications(&mut self, substream_id: SubstreamId, timeout: Duration) {
        let (connection_id, state, inner_substream_id) =
            match self.ingoing_notification_substreams.get_mut(&substream_id) {
                Some(s) => s,
                None => panic!(),
            };
        assert!(matches!(state, SubstreamState::Open));

        self.messages_to_connections.push_back((
            *connection_id,
            CoordinatorToConnectionInner::CloseInNotifications {
                substream_id: *inner_substream_id,
                timeout,
            },
        ));

        *state = SubstreamState::RequestedClosing;
    }

    /// Start opening a Bitswap substream.
    ///
    /// It is invalid to open a Bitswap substream on a connection before a
    /// [`Event::HandshakeFinished`] or after a [`Event::StartShutdown`] has been generated, or
    /// after [`Network::start_shutdown`] has been called.
    ///
    /// Returns a newly-allocated identifier for this substream.
    ///
    /// This function generates a message destined to the connection. Use
    /// [`Network::pull_message_to_connection`] to process these messages after it has returned.
    ///
    /// # Panic
    ///
    /// Panics if the [`ConnectionId`] is invalid or is a connection that hasn't finished its
    /// handshake or is shutting down.
    ///
    #[track_caller]
    pub fn open_out_bitswap(&mut self, connection_id: ConnectionId) -> SubstreamId {
        let connection = match self.connections.get(&connection_id) {
            Some(c) => c,
            None => panic!(),
        };
        assert!(matches!(
            connection.state,
            InnerConnectionState::Established
        ));

        let substream_id = self.next_substream_id;
        self.next_substream_id.0 += 1;

        let _prev_value = self
            .outgoing_bitswap_substreams
            .insert(substream_id, (connection_id, SubstreamState::Pending));
        debug_assert!(_prev_value.is_none());
        let _was_inserted = self
            .outgoing_bitswap_substreams_by_connection
            .insert((connection_id, substream_id));
        debug_assert!(_was_inserted);

        self.messages_to_connections.push_back((
            connection_id,
            CoordinatorToConnectionInner::OpenOutBitswap { substream_id },
        ));

        substream_id
    }

    /// Start closing a previously-open outbound Bitswap substream, or cancels opening an
    /// outbound Bitswap substream.
    ///
    /// All the messages that have been queued remain queued. The substream actually closes
    /// only once the queue is empty.
    ///
    /// Calling this method does *not* emit any event. The [`SubstreamId`] is considered invalid
    /// after this function returns.
    ///
    /// This function generates a message destined to the connection. Use
    /// [`Network::pull_message_to_connection`] to process these messages after it has returned.
    ///
    /// # Panic
    ///
    /// Panics if [`SubstreamId`] doesn't correspond to an outbound Bitswap substream.
    ///
    #[track_caller]
    pub fn close_out_bitswap(&mut self, substream_id: SubstreamId) {
        // Both `Pending` and `Open` states are accepted.
        let (connection_id, _state) = match self.outgoing_bitswap_substreams.remove(&substream_id) {
            Some(s) => s,
            None => panic!(),
        };
        let _was_in = self
            .outgoing_bitswap_substreams_by_connection
            .remove(&(connection_id, substream_id));
        debug_assert!(_was_in);

        self.messages_to_connections.push_back((
            connection_id,
            CoordinatorToConnectionInner::CloseOutBitswap { substream_id },
        ));
    }

    /// Close inbound Bitswap substream. Because Bitswap protocol doesn't specify a way to signal
    /// remote about a desire to close an inbound substream, this in fact resets the substream.
    ///
    /// Unlike Notifications, Bitswap in substreams do not have `SubstreamState::RequestedClosing`,
    /// and are instantly removed. This affects the race condition resolution between the
    /// coordinator and connection task for Bitswap in substreams.
    ///
    /// This is used to limit the number of inbound Bitswap substreams remote has opened with us.
    ///
    /// This function generates a message destined to the connection. Use
    /// [`Network::pull_message_to_connection`] to process these messages after it has returned.
    ///
    /// # Panic
    ///
    /// Panics if [`SubstreamId`] doesn't correspond to an inbound Bitswap substream.
    ///
    #[track_caller]
    pub fn close_in_bitswap(&mut self, substream_id: SubstreamId) {
        let (connection_id, inner_substream_id) =
            match self.ingoing_bitswap_substreams.remove(&substream_id) {
                Some(s) => s,
                None => panic!(),
            };
        let _was_in = self
            .ingoing_bitswap_substreams_by_connection
            .remove(&(connection_id, inner_substream_id))
            .is_some();
        debug_assert!(_was_in);

        self.messages_to_connections.push_back((
            connection_id,
            CoordinatorToConnectionInner::CloseInBitswap {
                substream_id: inner_substream_id,
            },
        ));
    }

    /// Adds a Bitswap message to the queue of messages to send to the given peer.
    ///
    /// It is invalid to call this on a [`SubstreamId`] before a successful
    /// [`Event::BitswapOutOpenResult`] has been yielded.
    ///
    /// Each substream maintains a queue of messages to be sent to the remote. This method
    /// attempts to push a message to this queue.
    ///
    /// An error is returned if the queue exceeds a certain size in bytes to handle the remote
    /// back-pressure and limited transfer rate.
    ///
    /// Regardless of the success of this function, no guarantee exists about the successful
    /// delivery of the message.
    ///
    /// This function generates a message destined to the connection. Use
    /// [`Network::pull_message_to_connection`] to process these messages after it has returned.
    ///
    /// # Panics
    ///
    /// Panics if [`SubstreamId`] is not a fully open outbound Bitswap substream.
    ///
    #[track_caller]
    pub fn queue_bitswap_message(
        &mut self,
        substream_id: SubstreamId,
        message: impl Into<Vec<u8>>,
    ) -> Result<(), QueueBitswapMessageError> {
        let (connection_id, state) = match self.outgoing_bitswap_substreams.get(&substream_id) {
            Some(s) => s,
            None => panic!(),
        };
        assert!(matches!(state, SubstreamState::Open));

        // TODO: add some back-pressure system and return a `QueueBitswapMessageError` if full

        self.messages_to_connections.push_back((
            *connection_id,
            CoordinatorToConnectionInner::QueueBitswapMessage {
                substream_id,
                message: message.into(),
            },
        ));

        Ok(())
    }

    /// Responds to an incoming request. Must be called in response to a [`Event::RequestIn`].
    ///
    /// If the substream was in the meanwhile yielded in an [`Event::RequestInCancel`], then this
    /// function must not be called and will panic.
    ///
    /// The [`SubstreamId`] is considered no longer valid after this function returns.
    ///
    /// This function generates a message destined to the connection. Use
    /// [`Network::pull_message_to_connection`] to process these messages after it has returned.
    ///
    /// # Panic
    ///
    /// Panics if the [`SubstreamId`] doesn't correspond to an active incoming request.
    ///
    #[track_caller]
    pub fn respond_in_request(&mut self, substream_id: SubstreamId, response: Result<Vec<u8>, ()>) {
        let (connection_id, inner_substream_id) = match self.ingoing_requests.remove(&substream_id)
        {
            Some(s) => s,
            None => panic!(),
        };

        self.ingoing_requests_by_connection
            .remove(&(connection_id, substream_id));

        self.messages_to_connections.push_back((
            connection_id,
            CoordinatorToConnectionInner::AnswerRequest {
                substream_id: inner_substream_id,
                response,
            },
        ));
    }

    /// Pulls a message that must be sent to a connection.
    ///
    /// The message must be passed to [`SingleStreamConnectionTask::inject_coordinator_message`]
    /// or [`MultiStreamConnectionTask::inject_coordinator_message`] in the appropriate connection.
    ///
    /// This function guarantees that the [`ConnectionId`] always refers to a connection that
    /// is still alive, in the sense that [`SingleStreamConnectionTask::inject_coordinator_message`]
    /// or [`MultiStreamConnectionTask::inject_coordinator_message`] has never returned `None`
    /// and that no [`Event::Shutdown`] has been generated for this connection.
    pub fn pull_message_to_connection(
        &mut self,
    ) -> Option<(ConnectionId, CoordinatorToConnection)> {
        let message = self
            .messages_to_connections
            .pop_front()
            .map(|(id, inner)| (id, CoordinatorToConnection { inner }))?;

        if let CoordinatorToConnectionInner::ShutdownFinishedAck = message.1.inner {
            self.shutdown_finished_connections.push_back(message.0);
        }

        Some(message)
    }

    /// Injects into the state machine a message generated by
    /// [`SingleStreamConnectionTask::pull_message_to_coordinator`] or
    /// [`MultiStreamConnectionTask::pull_message_to_coordinator`].
    ///
    /// This message is queued and is later processed in [`Network::next_event`]. This means that
    /// it is [`Network::next_event`] and not [`Network::inject_connection_message`] that updates
    /// the internals of the state machine according to the content of the message. For example,
    /// if a [`SingleStreamConnectionTask`] sends a message to the coordinator indicating that a
    /// notifications substream has been closed, the coordinator will still believe that it is
    /// open until [`Network::next_event`] processes this message and at the same time returns a
    /// corresponding [`Event`]. Processing messages directly in
    /// [`Network::inject_connection_message`] would introduce "race conditions" where the API
    /// user can't be sure in which state a connection or a substream is.
    pub fn inject_connection_message(
        &mut self,
        connection_id: ConnectionId,
        message: ConnectionToCoordinator,
    ) {
        assert!(self.connections.contains_key(&connection_id));

        // TODO: add a limit for a back-pressure-like system?
        self.pending_incoming_messages
            .push_back((connection_id, message.inner));
    }

    /// Returns the next event produced by the service. Returns `None` if no event is available.
    ///
    /// Call this function in a loop after having injected messages using
    /// [`Network::inject_connection_message`].
    pub fn next_event(&mut self) -> Option<Event<TConn>> {
        loop {
            if let Some(connection_id) = self.shutdown_finished_connections.pop_front() {
                let connection = self.connections.remove(&connection_id).unwrap();
                let was_established = match &connection.state {
                    InnerConnectionState::ShuttingDown {
                        was_established, ..
                    } => *was_established,
                    _ => unreachable!(),
                };

                return Some(Event::Shutdown {
                    id: connection_id,
                    was_established,
                    user_data: connection.user_data,
                });
            }

            // When a connection starts its shutdown, its id is put in `shutting_down_connection`.
            // When that happens, we go through the local state and clean up all requests and
            // notification substreams that are in progress/open and return the cancellations
            // as events.
            //
            // `shutting_down_connection` is set back to `None` only if it turns out that there
            // is no request or notification substream in progress/open anymore.
            if let Some(shutting_down_connection) = self.shutting_down_connection {
                // Outgoing notification substreams to close.
                if let Some((_, substream_id)) = self
                    .outgoing_notification_substreams_by_connection
                    .range(
                        (shutting_down_connection, SubstreamId::MIN)
                            ..=(shutting_down_connection, SubstreamId::MAX),
                    )
                    .cloned()
                    .next()
                {
                    self.outgoing_notification_substreams_by_connection
                        .remove(&(shutting_down_connection, substream_id));
                    let (_, state) = self
                        .outgoing_notification_substreams
                        .remove(&substream_id)
                        .unwrap();
                    return Some(match state {
                        SubstreamState::Open => Event::NotificationsOutReset { substream_id },
                        SubstreamState::Pending => Event::NotificationsOutResult {
                            substream_id,
                            result: Err(NotificationsOutErr::ConnectionShutdown),
                        },
                        SubstreamState::RequestedClosing => unreachable!(), // Never set for outgoing notification substreams.
                    });
                }

                // Ingoing notification substreams to close.
                if let Some((key, substream_id)) = self
                    .ingoing_notification_substreams_by_connection
                    .range(
                        (shutting_down_connection, established::SubstreamId::MIN)
                            ..=(shutting_down_connection, established::SubstreamId::MAX),
                    )
                    .map(|(k, v)| (*k, *v))
                    .next()
                {
                    let (_, state, _) = self
                        .ingoing_notification_substreams
                        .remove(&substream_id)
                        .unwrap();
                    self.ingoing_notification_substreams_by_connection
                        .remove(&key)
                        .unwrap();

                    return Some(match state {
                        SubstreamState::Open | SubstreamState::RequestedClosing => {
                            Event::NotificationsInClose {
                                substream_id,
                                outcome: Err(NotificationsInClosedErr::ConnectionShutdown),
                            }
                        }
                        SubstreamState::Pending => {
                            Event::NotificationsInOpenCancel { substream_id }
                        }
                    });
                }

                // Outgoing Bitswap substreams to close.
                if let Some((_, substream_id)) = self
                    .outgoing_bitswap_substreams_by_connection
                    .range(
                        (shutting_down_connection, SubstreamId::MIN)
                            ..=(shutting_down_connection, SubstreamId::MAX),
                    )
                    .cloned()
                    .next()
                {
                    self.outgoing_bitswap_substreams_by_connection
                        .remove(&(shutting_down_connection, substream_id));
                    let (_, state) = self
                        .outgoing_bitswap_substreams
                        .remove(&substream_id)
                        .unwrap();
                    return Some(match state {
                        SubstreamState::Open => Event::BitswapOutClose {
                            substream_id,
                            error: BitswapOutClosedErr::ConnectionShutdown,
                        },
                        SubstreamState::Pending => Event::BitswapOutOpenResult {
                            substream_id,
                            result: Err(BitswapOutOpenErr::ConnectionShutdown),
                        },
                        SubstreamState::RequestedClosing => unreachable!(), // Never set for outgoing Bitswap substreams.
                    });
                }

                // Ingoing Bitswap substreams to close.
                if let Some((key, substream_id)) = self
                    .ingoing_bitswap_substreams_by_connection
                    .range(
                        (shutting_down_connection, established::SubstreamId::MIN)
                            ..=(shutting_down_connection, established::SubstreamId::MAX),
                    )
                    .map(|(k, v)| (*k, *v))
                    .next()
                {
                    self.ingoing_bitswap_substreams.remove(&substream_id);
                    self.ingoing_bitswap_substreams_by_connection.remove(&key);

                    return Some(Event::BitswapInClose {
                        substream_id,
                        outcome: Err(BitswapInClosedErr::ConnectionShutdown),
                    });
                }

                // Find outgoing requests to cancel.
                if let Some((_, substream_id)) = self
                    .outgoing_requests
                    .range(
                        (shutting_down_connection, SubstreamId::MIN)
                            ..=(shutting_down_connection, SubstreamId::MAX),
                    )
                    .next()
                {
                    let substream_id = *substream_id;
                    self.outgoing_requests
                        .remove(&(shutting_down_connection, substream_id));

                    return Some(Event::Response {
                        substream_id,
                        response: Err(RequestError::ConnectionShutdown),
                    });
                }

                // Find ingoing requests to cancel.
                if let Some((_, substream_id)) = self
                    .ingoing_requests_by_connection
                    .range(
                        (shutting_down_connection, SubstreamId::MIN)
                            ..=(shutting_down_connection, SubstreamId::MAX),
                    )
                    .next()
                {
                    let substream_id = *substream_id;

                    let _was_in = self.ingoing_requests.remove(&substream_id);
                    debug_assert!(_was_in.is_some());
                    let _was_in = self
                        .ingoing_requests_by_connection
                        .remove(&(shutting_down_connection, substream_id));
                    debug_assert!(_was_in);

                    return Some(Event::RequestInCancel { substream_id });
                }

                // Find ingoing negotiated substreams to cancel.
                if let Some((key, substream_id)) = self
                    .ingoing_negotiated_substreams_by_connection
                    .range(
                        (shutting_down_connection, established::SubstreamId::MIN)
                            ..=(shutting_down_connection, established::SubstreamId::MAX),
                    )
                    .map(|(k, v)| (*k, *v))
                    .next()
                {
                    let Some((_, _, was_accepted)) =
                        self.ingoing_negotiated_substreams.remove(&substream_id)
                    else {
                        unreachable!()
                    };
                    let _was_in = self
                        .ingoing_negotiated_substreams_by_connection
                        .remove(&key);
                    debug_assert!(_was_in.is_some());

                    if was_accepted {
                        return Some(Event::InboundAcceptedCancel { substream_id });
                    } else {
                        return Some(Event::InboundNegotiatedCancel { substream_id });
                    }
                }

                // If this is reached, this connection has no more request or notifications
                // substream that is still in progress or open. The connection is successfully
                // shut down.
                self.shutting_down_connection = None;
            }

            // Now actually process messages.
            let (connection_id, message) = self.pending_incoming_messages.pop_front()?;
            let connection = &mut self.connections.get_mut(&connection_id).unwrap();

            break Some(match message {
                ConnectionToCoordinatorInner::StartShutdown(reason) => {
                    // The `ConnectionToCoordinator` message contains a shutdown reason if
                    // and only if it sends `StartShutdown` as a response to a shutdown
                    // initiated by the remote. If the shutdown was initiated locally
                    // (`api_initiated` is `true`), then it can contain `None`, but it can also
                    // contain `Some` in case the shutdown was initiated by the remote at the same
                    // time as it was initiated locally.

                    let report_event = match &mut connection.state {
                        InnerConnectionState::ShuttingDown {
                            api_initiated: true,
                            ..
                        } => false,
                        InnerConnectionState::ShuttingDown {
                            api_initiated: false,
                            ..
                        } => unreachable!(),
                        st @ InnerConnectionState::Handshaking => {
                            *st = InnerConnectionState::ShuttingDown {
                                api_initiated: false,
                                was_established: false,
                            };
                            true
                        }
                        st @ InnerConnectionState::Established => {
                            *st = InnerConnectionState::ShuttingDown {
                                api_initiated: false,
                                was_established: true,
                            };
                            true
                        }
                    };

                    // Control flow can't reach here if `shutting_down_connection` is ̀`Some`.
                    debug_assert!(self.shutting_down_connection.is_none());
                    self.shutting_down_connection = Some(connection_id);

                    if !report_event {
                        // No `StartShutdown` event is generated if the API user has started
                        // the shutdown themselves. In that case, `StartShutdown` is merely a
                        // confirmation.
                        continue;
                    } else {
                        Event::StartShutdown {
                            id: connection_id,
                            reason: reason.unwrap(), // See comment above.
                        }
                    }
                }
                ConnectionToCoordinatorInner::ShutdownFinished => {
                    self.messages_to_connections.push_back((
                        connection_id,
                        CoordinatorToConnectionInner::ShutdownFinishedAck,
                    ));
                    continue;
                }
                ConnectionToCoordinatorInner::HandshakeFinished(peer_id) => {
                    debug_assert_eq!(
                        self.ingoing_notification_substreams_by_connection
                            .range(
                                (connection_id, established::SubstreamId::MIN)
                                    ..=(connection_id, established::SubstreamId::MAX)
                            )
                            .count(),
                        0
                    );
                    debug_assert_eq!(
                        self.outgoing_notification_substreams_by_connection
                            .range(
                                (connection_id, SubstreamId::MIN)
                                    ..=(connection_id, SubstreamId::MAX)
                            )
                            .count(),
                        0
                    );
                    debug_assert_eq!(
                        self.ingoing_requests_by_connection
                            .range(
                                (connection_id, SubstreamId::MIN)
                                    ..=(connection_id, SubstreamId::MAX)
                            )
                            .count(),
                        0
                    );

                    match &mut connection.state {
                        InnerConnectionState::ShuttingDown {
                            was_established,
                            api_initiated,
                        } => {
                            debug_assert!(!*was_established);
                            debug_assert!(*api_initiated);
                            continue;
                        }
                        st @ InnerConnectionState::Handshaking => {
                            *st = InnerConnectionState::Established
                        }
                        InnerConnectionState::Established => unreachable!(),
                    }

                    Event::HandshakeFinished {
                        id: connection_id,
                        peer_id,
                    }
                }
                ConnectionToCoordinatorInner::InboundError(error) => {
                    // Ignore events if a shutdown has been initiated by the coordinator.
                    if let InnerConnectionState::ShuttingDown { api_initiated, .. } =
                        connection.state
                    {
                        debug_assert!(api_initiated);
                        continue;
                    }

                    Event::InboundError {
                        id: connection_id,
                        error,
                    }
                }
                ConnectionToCoordinatorInner::InboundNegotiated {
                    id: connection_substream_id,
                    protocol_name,
                } => {
                    // Ignore events if a shutdown has been initiated by the coordinator.
                    if let InnerConnectionState::ShuttingDown { api_initiated, .. } =
                        connection.state
                    {
                        debug_assert!(api_initiated);
                        continue;
                    }

                    let substream_id = self.next_substream_id;
                    self.next_substream_id.0 += 1;

                    self.ingoing_negotiated_substreams.insert(
                        substream_id,
                        (connection_id, connection_substream_id, false),
                    );
                    self.ingoing_negotiated_substreams_by_connection
                        .insert((connection_id, connection_substream_id), substream_id);

                    Event::InboundNegotiated {
                        id: connection_id,
                        substream_id,
                        protocol_name,
                    }
                }
                ConnectionToCoordinatorInner::InboundAcceptedCancel {
                    id: connection_substream_id,
                } => {
                    // Ignore events if a shutdown has been initiated by the coordinator.
                    if let InnerConnectionState::ShuttingDown { api_initiated, .. } =
                        connection.state
                    {
                        debug_assert!(api_initiated);
                        continue;
                    }

                    let substream_id = self
                        .ingoing_negotiated_substreams_by_connection
                        .remove(&(connection_id, connection_substream_id))
                        .unwrap_or_else(|| unreachable!());
                    let _was_in = self.ingoing_negotiated_substreams.remove(&substream_id);
                    debug_assert!(_was_in.is_some());

                    Event::InboundAcceptedCancel { substream_id }
                }
                ConnectionToCoordinatorInner::RequestIn {
                    id: connection_substream_id,
                    request,
                } => {
                    // Ignore events if a shutdown has been initiated by the coordinator.
                    if let InnerConnectionState::ShuttingDown { api_initiated, .. } =
                        connection.state
                    {
                        debug_assert!(api_initiated);
                        continue;
                    }

                    let substream_id = self
                        .ingoing_negotiated_substreams_by_connection
                        .remove(&(connection_id, connection_substream_id))
                        .unwrap_or_else(|| unreachable!());
                    let _was_in = self.ingoing_negotiated_substreams.remove(&substream_id);
                    debug_assert!(_was_in.is_some());

                    self.ingoing_requests
                        .insert(substream_id, (connection_id, connection_substream_id));
                    self.ingoing_requests_by_connection
                        .insert((connection_id, substream_id));

                    Event::RequestIn {
                        substream_id,
                        request_payload: request,
                    }
                }
                ConnectionToCoordinatorInner::Response {
                    id: substream_id,
                    response,
                    ..
                } => {
                    // Ignore events if a shutdown has been initiated by the coordinator.
                    if let InnerConnectionState::ShuttingDown { api_initiated, .. } =
                        connection.state
                    {
                        debug_assert!(api_initiated);
                        continue;
                    }

                    let _was_in = self
                        .outgoing_requests
                        .remove(&(connection_id, substream_id));
                    debug_assert!(_was_in);

                    Event::Response {
                        substream_id,
                        response: response.map_err(RequestError::Substream),
                    }
                }
                ConnectionToCoordinatorInner::NotificationsInOpen {
                    id: inner_substream_id,
                    handshake,
                } => {
                    // Ignore events if a shutdown has been initiated by the coordinator.
                    if let InnerConnectionState::ShuttingDown { api_initiated, .. } =
                        connection.state
                    {
                        debug_assert!(api_initiated);
                        continue;
                    }

                    let substream_id = self
                        .ingoing_negotiated_substreams_by_connection
                        .remove(&(connection_id, inner_substream_id))
                        .unwrap_or_else(|| unreachable!());
                    let _was_in = self.ingoing_negotiated_substreams.remove(&substream_id);
                    debug_assert!(_was_in.is_some());

                    self.ingoing_notification_substreams.insert(
                        substream_id,
                        (connection_id, SubstreamState::Pending, inner_substream_id),
                    );
                    self.ingoing_notification_substreams_by_connection
                        .insert((connection_id, inner_substream_id), substream_id);

                    Event::NotificationsInOpen {
                        substream_id,
                        remote_handshake: handshake,
                    }
                }
                ConnectionToCoordinatorInner::NotificationsInOpenCancel {
                    id: inner_substream_id,
                    ..
                } => {
                    // Ignore events if a shutdown has been initiated by the coordinator.
                    if let InnerConnectionState::ShuttingDown { api_initiated, .. } =
                        connection.state
                    {
                        debug_assert!(api_initiated);
                        continue;
                    }

                    // The event might concern a substream that we have already accepted or
                    // refused. In that situation, either reinterpret the event as
                    // "NotificationsInClose" or discard it.
                    if let Some(substream_id) = self
                        .ingoing_notification_substreams_by_connection
                        .remove(&(connection_id, inner_substream_id))
                    {
                        let (_, state, _) = self
                            .ingoing_notification_substreams
                            .remove(&substream_id)
                            .unwrap();
                        match state {
                            SubstreamState::Open | SubstreamState::RequestedClosing => {
                                Event::NotificationsInClose {
                                    substream_id,
                                    outcome: Err(NotificationsInClosedErr::Substream(
                                        established::NotificationsInClosedErr::SubstreamReset,
                                    )),
                                }
                            }
                            SubstreamState::Pending => {
                                Event::NotificationsInOpenCancel { substream_id }
                            }
                        }
                    } else {
                        // Substream was refused. As documented, we must confirm the reception of
                        // the event by sending back a rejection.
                        self.messages_to_connections.push_back((
                            connection_id,
                            CoordinatorToConnectionInner::RejectInNotifications {
                                substream_id: inner_substream_id,
                            },
                        ));
                        continue;
                    }
                }
                ConnectionToCoordinatorInner::NotificationIn {
                    id: inner_substream_id,
                    notification,
                } => {
                    // Ignore events if a shutdown has been initiated by the coordinator.
                    if let InnerConnectionState::ShuttingDown { api_initiated, .. } =
                        connection.state
                    {
                        debug_assert!(api_initiated);
                        continue;
                    }

                    let substream_id = *self
                        .ingoing_notification_substreams_by_connection
                        .get(&(connection_id, inner_substream_id))
                        .unwrap();

                    Event::NotificationsIn {
                        substream_id,
                        notification,
                    }
                }
                ConnectionToCoordinatorInner::NotificationsInClose {
                    id: inner_substream_id,
                    outcome,
                    ..
                } => {
                    // Ignore events if a shutdown has been initiated by the coordinator.
                    if let InnerConnectionState::ShuttingDown { api_initiated, .. } =
                        connection.state
                    {
                        debug_assert!(api_initiated);
                        continue;
                    }

                    let substream_id = self
                        .ingoing_notification_substreams_by_connection
                        .remove(&(connection_id, inner_substream_id))
                        .unwrap();
                    let (_, state, _) = self
                        .ingoing_notification_substreams
                        .remove(&substream_id)
                        .unwrap();
                    debug_assert!(matches!(
                        state,
                        SubstreamState::Open | SubstreamState::RequestedClosing
                    ));

                    if let SubstreamState::Open = state {
                        // As documented, we must confirm the reception of the event by sending
                        // back a rejection, provided that no such event has been sent beforehand.
                        self.messages_to_connections.push_back((
                            connection_id,
                            CoordinatorToConnectionInner::CloseInNotifications {
                                substream_id: inner_substream_id,
                                timeout: Duration::new(0, 0),
                            },
                        ));
                    }

                    Event::NotificationsInClose {
                        substream_id,
                        outcome: outcome.map_err(NotificationsInClosedErr::Substream),
                    }
                }
                ConnectionToCoordinatorInner::NotificationsOutResult {
                    id: substream_id,
                    result,
                } => {
                    // Ignore events if a shutdown has been initiated by the coordinator.
                    if let InnerConnectionState::ShuttingDown { api_initiated, .. } =
                        connection.state
                    {
                        debug_assert!(api_initiated);
                        continue;
                    }

                    let mut entry = match self.outgoing_notification_substreams.entry(substream_id)
                    {
                        hashbrown::hash_map::Entry::Occupied(e) => e,
                        hashbrown::hash_map::Entry::Vacant(_) => {
                            // This can be reached if the API user closed the substream while it
                            // was being open.
                            continue;
                        }
                    };

                    debug_assert!(matches!(entry.get_mut().1, SubstreamState::Pending));

                    if result.is_ok() {
                        entry.insert((connection_id, SubstreamState::Open));
                    } else {
                        entry.remove();

                        let _was_removed = self
                            .outgoing_notification_substreams_by_connection
                            .remove(&(connection_id, substream_id));
                        debug_assert!(_was_removed);
                    }

                    Event::NotificationsOutResult {
                        substream_id,
                        result,
                    }
                }
                ConnectionToCoordinatorInner::NotificationsOutCloseDemanded {
                    id: substream_id,
                    ..
                } => {
                    // Ignore events if a shutdown has been initiated by the coordinator.
                    if let InnerConnectionState::ShuttingDown { api_initiated, .. } =
                        connection.state
                    {
                        debug_assert!(api_initiated);
                        continue;
                    }

                    match self.outgoing_notification_substreams.get(&substream_id) {
                        Some((_connection_id, _substream_state)) => {
                            debug_assert_eq!(*_connection_id, connection_id);
                            debug_assert!(matches!(_substream_state, SubstreamState::Open));
                        }
                        None => {
                            // The substream might already have been destroyed if the user closed
                            // the substream while this message was pending in the queue.
                            continue;
                        }
                    }

                    Event::NotificationsOutCloseDemanded { substream_id }
                }
                ConnectionToCoordinatorInner::NotificationsOutReset { id: substream_id } => {
                    // Ignore events if a shutdown has been initiated by the coordinator.
                    if let InnerConnectionState::ShuttingDown { api_initiated, .. } =
                        connection.state
                    {
                        debug_assert!(api_initiated);
                        continue;
                    }

                    match self.outgoing_notification_substreams.remove(&substream_id) {
                        Some((_connection_id, _substream_state)) => {
                            debug_assert_eq!(_connection_id, connection_id);
                            debug_assert!(matches!(_substream_state, SubstreamState::Open));
                        }
                        None => {
                            // The substream might already have been destroyed if the user closed
                            // the substream while this message was pending in the queue.
                            continue;
                        }
                    }

                    let _was_removed = self
                        .outgoing_notification_substreams_by_connection
                        .remove(&(connection_id, substream_id));
                    debug_assert!(_was_removed);

                    Event::NotificationsOutReset { substream_id }
                }
                ConnectionToCoordinatorInner::BitswapInOpen {
                    id: inner_substream_id,
                } => {
                    // Ignore events if a shutdown has been initiated by the coordinator.
                    if let InnerConnectionState::ShuttingDown { api_initiated, .. } =
                        connection.state
                    {
                        debug_assert!(api_initiated);
                        continue;
                    }

                    let substream_id = self
                        .ingoing_negotiated_substreams_by_connection
                        .remove(&(connection_id, inner_substream_id))
                        .unwrap();
                    let _was_in = self.ingoing_negotiated_substreams.remove(&substream_id);
                    debug_assert!(_was_in.is_some());

                    self.ingoing_bitswap_substreams
                        .insert(substream_id, (connection_id, inner_substream_id));
                    self.ingoing_bitswap_substreams_by_connection
                        .insert((connection_id, inner_substream_id), substream_id);

                    Event::BitswapInOpen { substream_id }
                }
                ConnectionToCoordinatorInner::BitswapIn {
                    id: inner_substream_id,
                    message,
                } => {
                    // Ignore events if a shutdown has been initiated by the coordinator.
                    if let InnerConnectionState::ShuttingDown { api_initiated, .. } =
                        connection.state
                    {
                        debug_assert!(api_initiated);
                        continue;
                    }

                    let substream_id = match self
                        .ingoing_bitswap_substreams_by_connection
                        .get(&(connection_id, inner_substream_id))
                    {
                        Some(&id) => id,
                        None => {
                            // It is possible that the coordinator has already removed this
                            // substream via `close_in_bitswap()` while a `BitswapIn` message
                            // was in flight. Silently discard the message.
                            continue;
                        }
                    };

                    Event::BitswapIn {
                        substream_id,
                        message,
                    }
                }
                ConnectionToCoordinatorInner::BitswapInClose {
                    id: inner_substream_id,
                    outcome,
                } => {
                    // Ignore events if a shutdown has been initiated by the coordinator.
                    if let InnerConnectionState::ShuttingDown { api_initiated, .. } =
                        connection.state
                    {
                        debug_assert!(api_initiated);
                        continue;
                    }

                    match self
                        .ingoing_bitswap_substreams_by_connection
                        .remove(&(connection_id, inner_substream_id))
                    {
                        Some(substream_id) => {
                            self.ingoing_bitswap_substreams
                                .remove(&substream_id)
                                .unwrap();

                            // As documented, we must acknowledge the reception of the
                            // event by sending back a `CloseInBitswap`, provided that
                            // no such message has been sent beforehand (i.e.,
                            // `close_in_bitswap()` hasn't been called).
                            self.messages_to_connections.push_back((
                                connection_id,
                                CoordinatorToConnectionInner::CloseInBitswap {
                                    substream_id: inner_substream_id,
                                },
                            ));

                            Event::BitswapInClose {
                                substream_id,
                                outcome: outcome.map_err(BitswapInClosedErr::Substream),
                            }
                        }
                        None => {
                            // The coordinator has already removed this substream via
                            // `close_in_bitswap()`. The `CloseInBitswap` message sent
                            // by `close_in_bitswap()` will serve as the acknowledgment
                            // consumed by the connection task's
                            // `bitswap_in_close_acknowledgments`.
                            continue;
                        }
                    }
                }
                ConnectionToCoordinatorInner::BitswapOutOpenResult {
                    id: substream_id,
                    result,
                } => {
                    // Ignore events if a shutdown has been initiated by the coordinator.
                    if let InnerConnectionState::ShuttingDown { api_initiated, .. } =
                        connection.state
                    {
                        debug_assert!(api_initiated);
                        continue;
                    }

                    let mut entry = match self.outgoing_bitswap_substreams.entry(substream_id) {
                        hashbrown::hash_map::Entry::Occupied(e) => e,
                        hashbrown::hash_map::Entry::Vacant(_) => {
                            // This can be reached if the API user closed the substream while it
                            // was being open.
                            continue;
                        }
                    };

                    debug_assert!(matches!(entry.get_mut().1, SubstreamState::Pending));

                    if result.is_ok() {
                        entry.insert((connection_id, SubstreamState::Open));
                    } else {
                        entry.remove();

                        let _was_removed = self
                            .outgoing_bitswap_substreams_by_connection
                            .remove(&(connection_id, substream_id));
                        debug_assert!(_was_removed);
                    }

                    Event::BitswapOutOpenResult {
                        substream_id,
                        result,
                    }
                }
                ConnectionToCoordinatorInner::BitswapOutClose {
                    id: substream_id,
                    error,
                } => {
                    // Ignore events if a shutdown has been initiated by the coordinator.
                    if let InnerConnectionState::ShuttingDown { api_initiated, .. } =
                        connection.state
                    {
                        debug_assert!(api_initiated);
                        continue;
                    }

                    if self
                        .outgoing_bitswap_substreams
                        .remove(&substream_id)
                        .is_none()
                    {
                        // This can happen if the API user closed the substream at the same
                        // time as it was closed by remote.
                        continue;
                    }

                    let _was_removed = self
                        .outgoing_bitswap_substreams_by_connection
                        .remove(&(connection_id, substream_id));
                    debug_assert!(_was_removed);

                    Event::BitswapOutClose {
                        substream_id,
                        error: BitswapOutClosedErr::Substream(error),
                    }
                }
                ConnectionToCoordinatorInner::PingOutSuccess { ping_time } => {
                    // Ignore events if a shutdown has been initiated by the coordinator.
                    if let InnerConnectionState::ShuttingDown { api_initiated, .. } =
                        connection.state
                    {
                        debug_assert!(api_initiated);
                        continue;
                    }

                    Event::PingOutSuccess {
                        id: connection_id,
                        ping_time,
                    }
                }
                ConnectionToCoordinatorInner::PingOutFailed => {
                    // Ignore events if a shutdown has been initiated by the coordinator.
                    if let InnerConnectionState::ShuttingDown { api_initiated, .. } =
                        connection.state
                    {
                        debug_assert!(api_initiated);
                        continue;
                    }

                    Event::PingOutFailed { id: connection_id }
                }
            });
        }
    }
}

impl<TConn, TNow> ops::Index<ConnectionId> for Network<TConn, TNow> {
    type Output = TConn;
    fn index(&self, id: ConnectionId) -> &TConn {
        &self.connections.get(&id).unwrap().user_data
    }
}

impl<TConn, TNow> ops::IndexMut<ConnectionId> for Network<TConn, TNow> {
    fn index_mut(&mut self, id: ConnectionId) -> &mut TConn {
        &mut self.connections.get_mut(&id).unwrap().user_data
    }
}

/// See [`Network::connection_state`].
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
pub struct ConnectionState {
    /// If `true`, the connection has finished its handshaking phase.
    pub established: bool,

    /// If `true`, the connection is shutting down.
    pub shutting_down: bool,
}

/// Message from a connection task destined to the coordinator.
pub struct ConnectionToCoordinator {
    inner: ConnectionToCoordinatorInner,
}

enum ConnectionToCoordinatorInner {
    HandshakeFinished(PeerId),

    /// See the corresponding event in [`established::Event`].
    InboundError(established::InboundError),

    /// See the corresponding event in [`established::Event`].
    InboundNegotiated {
        id: established::SubstreamId,
        protocol_name: String,
    },

    /// See the corresponding event in [`established::Event`].
    InboundAcceptedCancel {
        id: established::SubstreamId,
    },

    /// See the corresponding event in [`established::Event`].
    RequestIn {
        id: established::SubstreamId,
        request: Vec<u8>,
    },

    /// See the corresponding event in [`established::Event`].
    Response {
        response: Result<Vec<u8>, established::RequestError>,
        id: SubstreamId,
    },

    /// See the corresponding event in [`established::Event`].
    NotificationsInOpen {
        id: established::SubstreamId,
        handshake: Vec<u8>,
    },
    /// See the corresponding event in [`established::Event`].
    ///
    /// The coordinator should be aware that, due to the asynchronous nature of communications, it
    /// might receive this event after having sent a
    /// [`CoordinatorToConnectionInner::AcceptInNotifications`] or
    /// [`CoordinatorToConnectionInner::RejectInNotifications`]. In that situation, the coordinator
    /// should either reinterpret the message as a `NotificationsInClose` (if it had accepted it)
    /// or ignore it (if it had rejected it).
    ///
    /// The connection should be aware that, due to the asynchronous nature of communications, it
    /// might later receive an [`CoordinatorToConnectionInner::AcceptInNotifications`] or
    /// [`CoordinatorToConnectionInner::RejectInNotifications`] concerning this substream. In that
    /// situation, the connection should ignore this message.
    ///
    /// Because substream IDs can be reused, this introduces an ambiguity in the following sequence
    /// of events: send `NotificationsInOpen`, send `NotificationsInOpenCancel`, send
    /// `NotificationsInOpen`, receive `AcceptInNotifications`. Does the `AcceptInNotifications`
    /// refer to the first `NotificationsInOpen` or to the second?
    /// In order to solve this problem, the coordinator must always send back a
    /// [`CoordinatorToConnectionInner::RejectInNotifications`] in order to acknowledge a
    /// `NotificationsInOpenCancel`.
    NotificationsInOpenCancel {
        id: established::SubstreamId,
    },
    /// See the corresponding event in [`established::Event`].
    NotificationIn {
        id: established::SubstreamId,
        notification: Vec<u8>,
    },
    /// See the corresponding event in [`established::Event`].
    ///
    /// In order to avoid race conditions, this must always be acknowledged by sending back a
    /// [`CoordinatorToConnectionInner::CloseInNotifications`] message if no such message was
    /// sent in the past.
    NotificationsInClose {
        id: established::SubstreamId,
        outcome: Result<(), established::NotificationsInClosedErr>,
    },
    /// See the corresponding event in [`established::Event`].
    NotificationsOutResult {
        id: SubstreamId,
        result: Result<Vec<u8>, NotificationsOutErr>,
    },
    /// See the corresponding event in [`established::Event`].
    NotificationsOutCloseDemanded {
        id: SubstreamId,
    },
    /// See the corresponding event in [`established::Event`].
    NotificationsOutReset {
        id: SubstreamId,
    },

    /// Remote has opened an inbound Bitswap substream. We can use this event to close other
    /// Bitswap substreams the remote has opened before.
    BitswapInOpen {
        /// Inner substream ID.
        id: established::SubstreamId,
    },
    /// Remote has sent a Bitswap message.
    BitswapIn {
        /// Inner substream ID.
        id: established::SubstreamId,
        /// Message sent by the remote.
        message: Vec<u8>,
    },
    /// Remote has closed the inbound Bitswap substream with us, or the substream error occurred.
    ///
    /// In order to avoid race conditions, this must always be acknowledged by sending back a
    /// [`CoordinatorToConnectionInner::CloseInBitswap`] message if no such message was
    /// sent in the past (i.e., if `close_in_bitswap()` hasn't already been called).
    BitswapInClose {
        /// Inner substream ID.
        id: established::SubstreamId,
        /// `Ok(())` if the substream was gracefully closed, an `Err()` otherwise.
        outcome: Result<(), established::BitswapInClosedErr>,
    },
    /// The result of opening an outbound Bitswap substream.
    BitswapOutOpenResult {
        /// Substream ID.
        id: SubstreamId,
        /// `Ok()` if the substream was successfully opened, `Err()` otherwise.
        result: Result<(), BitswapOutOpenErr>,
    },
    /// The outbound Bitswap substream was closed or reset by remote.
    BitswapOutClose {
        /// Substream ID.
        id: SubstreamId,
        /// Bitswap spec doesn't contain a mechanism for closing outbound substreams by remote, so
        /// this is always an error.
        error: established::BitswapOutClosedErr,
    },

    /// See the corresponding event in [`established::Event`].
    PingOutSuccess {
        ping_time: Duration,
    },
    /// See the corresponding event in [`established::Event`].
    PingOutFailed,

    /// Sent either in response to [`ConnectionToCoordinatorInner::StartShutdown`] (in which case
    /// the content is `None`) or if the remote has initiated the shutdown (in which case the
    /// content is `Some`). After this, no more [`ConnectionToCoordinatorInner`] will be sent
    /// anymore except for [`ConnectionToCoordinatorInner::ShutdownFinished`].
    StartShutdown(Option<ShutdownCause>),

    /// Shutdown has now finished. Always sent after
    /// [`ConnectionToCoordinatorInner::StartShutdown`]. No message is sent by the connection
    /// task anymore after that.
    ///
    /// Must be confirmed with a [`CoordinatorToConnectionInner::ShutdownFinishedAck`].
    ShutdownFinished,
}

/// Message from the coordinator destined to a connection task.
pub struct CoordinatorToConnection {
    inner: CoordinatorToConnectionInner,
}

enum CoordinatorToConnectionInner {
    /// Connection task must terminate. This is always sent back after a
    /// [`ConnectionToCoordinatorInner::ShutdownFinished`].
    ///
    /// This final message is necessary in order to make sure that the coordinator doesn't
    /// generate messages destined to a connection that isn't alive anymore.
    ShutdownFinishedAck,

    /// Connection must start shutting down if it is not already the case.
    /// Before of concurrency, it is possible for this message to be sent/received *after* a
    /// [`ConnectionToCoordinatorInner::StartShutdown`] has been sent.
    StartShutdown,

    AcceptInbound {
        substream_id: established::SubstreamId,
        /// Configuration of the protocol.
        inbound_ty: InboundTy,
    },
    RejectInbound {
        substream_id: established::SubstreamId,
    },
    SetMaxProtocolNameLen {
        new_max_length: usize,
    },

    StartRequest {
        protocol_name: String,
        request_data: Option<Vec<u8>>,
        timeout: Duration,
        max_response_size: usize,
        /// Id of the substream assigned by the coordinator.
        /// This is **not** the same as the actual substream used in the connection.
        substream_id: SubstreamId,
    },
    OpenOutNotifications {
        /// Id of the substream assigned by the coordinator.
        /// This is **not** the same as the actual substream used in the connection.
        substream_id: SubstreamId,
        protocol_name: String,
        max_handshake_size: usize,
        handshake_timeout: Duration,
        handshake: Vec<u8>,
    },
    CloseOutNotifications {
        /// Id of the substream assigned by the coordinator.
        /// This is **not** the same as the actual substream used in the connection.
        substream_id: SubstreamId,
    },
    QueueNotification {
        /// Id of the substream assigned by the coordinator.
        /// This is **not** the same as the actual substream used in the connection.
        substream_id: SubstreamId,
        notification: Vec<u8>,
    },
    AcceptInNotifications {
        substream_id: established::SubstreamId,
        handshake: Vec<u8>,
        max_notification_size: usize,
    },
    RejectInNotifications {
        substream_id: established::SubstreamId,
    },
    CloseInNotifications {
        substream_id: established::SubstreamId,
        timeout: Duration,
    },

    /// Start opening outbound Bitswap substream.
    OpenOutBitswap {
        /// Id of the substream assigned by the coordinator.
        /// This is **not** the same as the actual substream used in the connection.
        substream_id: SubstreamId,
    },
    /// Start closing outbound Bitswap substream.
    CloseOutBitswap {
        /// Id of the substream assigned by the coordinator.
        /// This is **not** the same as the actual substream used in the connection.
        substream_id: SubstreamId,
    },
    /// Push Bitswap message to the queue of messages for this substream.
    QueueBitswapMessage {
        /// Id of the substream assigned by the coordinator.
        /// This is **not** the same as the actual substream used in the connection.
        substream_id: SubstreamId,
        message: Vec<u8>,
    },
    /// Close inbound Bitswap substream. Because Bitswap spec doesn't include a mechanism for
    /// notifying the remote about the close desire by closing the local sending end of the stream,
    /// this actually resets the substream.
    ///
    /// This message is used to limit the number of inbound Bitswap substreams per peer.
    CloseInBitswap {
        substream_id: established::SubstreamId,
    },

    /// Answer an incoming request.
    ///
    /// Since the API doesn't provide any feedback about whether responses have been successfully
    /// received by the remote, the response should simply be ignored in case the substream is
    /// obsolete. In any case, answering an obsolete request is not an API error because the remote
    /// might have canceled their request while the message containing the response was waiting
    /// in queue.
    AnswerRequest {
        substream_id: established::SubstreamId,
        response: Result<Vec<u8>, ()>,
    },
}

/// Event generated by [`Network::next_event`].
#[derive(Debug)]
pub enum Event<TConn> {
    /// Handshake of the given connection has completed.
    ///
    /// This event can only happen once per connection and only for single-stream connections.
    HandshakeFinished {
        /// Identifier of the connection whose handshake is finished.
        id: ConnectionId,
        /// Identity of the peer on the other side of the connection.
        peer_id: PeerId,
    },

    /// A transport-level connection (e.g. a TCP socket) is starting its shutdown.
    ///
    /// It is no longer possible to start requests, open notification substreams, or open
    /// notifications on this connection, and no new incoming requests or notification substreams
    /// will be reported as events.
    ///
    /// Further events will close all existing substreams (requests and notifications) one by one.
    /// Once all substreams have been closed, a [`Event::Shutdown`] is reported.
    ///
    /// Keep in mind that this event can also happen for connections that haven't finished their
    /// handshake.
    ///
    /// This event is **not** generated when [`Network::start_shutdown`] is called.
    StartShutdown {
        /// Identifier of the connection that is starting its shutdown.
        id: ConnectionId,
        /// Reason why the connection is starting its shutdown. Because this event is not generated
        /// when the shutdown is initiated locally, the reason is always cause by the remote.
        reason: ShutdownCause,
    },

    /// A transport-level connection (e.g. a TCP socket) has been shut down.
    ///
    /// This [`ConnectionId`] is no longer valid, and using it will result in panics.
    Shutdown {
        /// Identifier of the connection that has finished its shutdown.
        id: ConnectionId,
        /// `true` if the connection was in its established phase before the shutdown.
        was_established: bool,
        /// User data that was stored in the state machine for this connection.
        user_data: TConn,
    },

    /// Received an incoming substream, but this substream has produced an error.
    ///
    /// > **Note**: This event exists only for diagnostic purposes. No action is expected in
    /// >           return.
    InboundError {
        /// Identifier of the connection that has received the substream.
        id: ConnectionId,
        /// Error that happened.
        error: InboundError,
    },

    /// An inbound substream has requested to use a protocol. Call [`Network::accept_inbound`] or
    /// [`Network::reject_inbound`] in order to accept or reject this substream.
    InboundNegotiated {
        /// Identifier of the connection that has received the substream.
        id: ConnectionId,
        /// Identifier of the substream. Needs to be provided back when accepting or rejecting
        /// the substream.
        substream_id: SubstreamId,
        /// Name of the protocol requested by the remote.
        protocol_name: String,
    },

    /// An inbound substream that is waiting for a call to [`Network::accept_inbound`] or
    /// [`Network::reject_inbound`] has been abruptly closed.
    InboundNegotiatedCancel {
        /// Identifier of the substream.
        substream_id: SubstreamId,
    },

    /// An inbound substream that was previously accepted using [`Network::accept_inbound`] was
    /// closed by the remote or has generated an error.
    InboundAcceptedCancel {
        /// Identifier of the substream.
        substream_id: SubstreamId,
    },

    /// Outcome of a request started using [`Network::start_request`].
    ///
    /// *All* requests always lead to an outcome, even if the connection has been closed while the
    /// request was in progress.
    Response {
        /// Substream that was returned by [`Network::start_request`].
        substream_id: SubstreamId,
        /// If the request is successful, contains the response sent back by the remote. Otherwise,
        /// contains the reason why the request isn't successful.
        response: Result<Vec<u8>, RequestError>,
    },

    /// Received a request from a request-response protocol.
    RequestIn {
        /// Substream on which the request has been received. Must be passed back when providing
        /// the response.
        substream_id: SubstreamId,
        /// Payload that has been sent by the remote. Its interpretation is beyond the scope of
        /// this module.
        request_payload: Vec<u8>,
    },

    /// Request received earlier has been canceled by the remote.
    ///
    /// The [`SubstreamId`] is now invalid.
    RequestInCancel { substream_id: SubstreamId },

    /// Outcome of trying to open a substream with [`Network::open_out_notifications`].
    ///
    /// If `Ok`, it is now possible to send notifications on this substream.
    /// If `Err`, the substream no longer exists and the [`SubstreamId`] becomes invalid.
    NotificationsOutResult {
        substream_id: SubstreamId,
        /// If `Ok`, contains the handshake sent back by the remote. Its interpretation is out of
        /// scope of this module.
        result: Result<Vec<u8>, NotificationsOutErr>,
    },

    /// Remote has closed an outgoing notifications substream, meaning that it demands the closing
    /// of the substream. Use [`Network::close_out_notifications`] as soon as possible, which is
    /// typically after all outbound notifications that need to be queued have been queued.
    ///
    /// This event is only generated for notification substreams that are fully open.
    NotificationsOutCloseDemanded { substream_id: SubstreamId },

    /// A previously open outbound substream has been closed, by the remote or as a consequence of
    /// the connection shutting down.
    ///
    /// This event is only generated for notification substreams that are fully open.
    ///
    /// The substream no longer exists and the [`SubstreamId`] becomes invalid.
    NotificationsOutReset { substream_id: SubstreamId },

    /// The remote would like to open a notifications substream.
    ///
    /// The substream needs to be accepted or refused using [`Network::accept_in_notifications`]
    /// or [`Network::reject_in_notifications`].
    NotificationsInOpen {
        /// Newly-generated identifier for the substream on which the request has been received.
        /// Must be passed back when accepting or refusing the substream.
        substream_id: SubstreamId,
        /// Handshake that has been sent by the remote. Its interpretation is beyond the scope of
        /// this module.
        remote_handshake: Vec<u8>,
    },

    /// The remote has canceled the opening an incoming notifications substream.
    ///
    /// This can only happen before the notification substream has been accepted or refused.
    NotificationsInOpenCancel {
        /// Substream that has been closed. Guaranteed to match a substream that was earlier
        /// reported with a [`Event::NotificationsInOpen`].
        substream_id: SubstreamId,
    },

    /// Received a notification on a notifications substream of a connection.
    NotificationsIn {
        /// Substream on which the notification has been received. Guaranteed to be a substream
        /// that has been accepted with [`Network::accept_in_notifications`].
        substream_id: SubstreamId,
        /// Notification that the remote has sent. The meaning of this data is out of scope of
        /// this module.
        notification: Vec<u8>,
    },

    /// The remote has closed an incoming notifications substream.
    ///
    /// This can only happen after the notification substream has been accepted.
    NotificationsInClose {
        /// Substream that has been closed. Guaranteed to match a substream that was earlier
        /// reported with a [`Event::NotificationsInOpen`].
        substream_id: SubstreamId,
        /// Reason why the substream has been closed.
        outcome: Result<(), NotificationsInClosedErr>,
    },

    /// The remote has opened an inbound Bitswap substream.
    BitswapInOpen {
        /// Newly-generated identifier for the substream that has been opened.
        substream_id: SubstreamId,
    },
    /// The remote has sent a Bitswap message.
    BitswapIn {
        /// Substream on which the Bitswap message has been received.
        substream_id: SubstreamId,
        /// Message sent by the remote.
        message: Vec<u8>,
    },
    /// The remote has closed an incoming Bitswap substream or it has been closed due to an error.
    BitswapInClose {
        /// Substream that has been closed. Guaranteed to match a substream that was earlier
        /// reported with an [`Event::BitswapInOpen`].
        substream_id: SubstreamId,
        /// Reason why the substream has been closed.
        outcome: Result<(), BitswapInClosedErr>,
    },
    /// The result of opening an outbound Bitswap substream with [`Network::open_out_bitswap`].
    BitswapOutOpenResult {
        /// Substream identifier previously returned by [`Network::open_out_bitswap`].
        substream_id: SubstreamId,
        /// `Ok()` if the substream was successfully opened, `Err()` with the details otherwise.
        result: Result<(), BitswapOutOpenErr>,
    },
    /// The remote has closed/reset the outbound Bitswap substream or it was reset due to an error.
    BitswapOutClose {
        /// Substream that has been closed.
        substream_id: SubstreamId,
        /// An error that has caused the substream to close. Note that Bitswap spec doesn't specify
        /// a way of closing outbound substreams by the remote, so this is always an error.
        error: BitswapOutClosedErr,
    },

    /// An outgoing ping has succeeded. This event is generated automatically over time for each
    /// connection in the collection.
    PingOutSuccess {
        id: ConnectionId,
        /// Time between sending the ping and receiving the pong.
        ping_time: Duration,
    },
    /// An outgoing ping has failed. This event is generated automatically over time for each
    /// connection in the collection.
    PingOutFailed { id: ConnectionId },
}

/// Reason why a connection is shutting down. See [`Event::StartShutdown`].
#[derive(Debug, derive_more::Display, derive_more::Error)]
pub enum ShutdownCause {
    /// Shutdown was demanded by the remote and performed cleanly.
    CleanShutdown,
    /// Remote has abruptly reset the connection.
    RemoteReset,
    /// Error in the connection protocol of a fully established connection.
    ProtocolError(established::Error),
    /// Error in the protocol of the handshake.
    HandshakeError(HandshakeError),
    /// Handshake phase took too long.
    HandshakeTimeout,
}

#[derive(Debug, derive_more::Display, derive_more::Error, Clone)]
pub enum RequestError {
    /// Request has been canceled because the connection as a whole is being shut down.
    ConnectionShutdown,

    /// Error happened in the context of the substream.
    #[display("{_0}")]
    Substream(established::RequestError),
}

impl RequestError {
    /// Returns `true` if the error is caused by a faulty behavior by the remote. Returns `false`
    /// if the error can happen in normal situations.
    pub fn is_protocol_error(&self) -> bool {
        match self {
            RequestError::ConnectionShutdown => false,
            RequestError::Substream(err) => err.is_protocol_error(),
        }
    }
}

#[derive(Debug, derive_more::Display, derive_more::Error, Clone)]
pub enum NotificationsOutErr {
    /// Opening has been interrupted because the connection as a whole is being shut down.
    ConnectionShutdown,

    /// Error happened in the context of the substream.
    #[display("{_0}")]
    Substream(established::NotificationsOutErr),
}

#[derive(Debug, derive_more::Display, derive_more::Error, Clone)]
pub enum NotificationsInClosedErr {
    /// Substream has been closed because the connection as a whole is being shut down.
    ConnectionShutdown,

    /// Error happened in the context of the substream.
    #[display("{_0}")]
    Substream(established::NotificationsInClosedErr),
}

/// Error potentially returned by [`Network::queue_notification`].
#[derive(Debug, derive_more::Display, derive_more::Error)]
pub enum QueueNotificationError {
    /// Queue of notifications with that peer is full.
    QueueFull,
}

/// Error opening outbound Bitswap substream.
#[derive(Debug, derive_more::Display, derive_more::Error, Clone)]
pub enum BitswapOutOpenErr {
    /// Opening has been interrupted because the connection as a whole is being shut down.
    ConnectionShutdown,

    /// Error happened in the context of the substream.
    #[display("{_0}")]
    Substream(established::BitswapOutOpenErr),
}

/// Error that caused closing of outbound Bitswap substream.
#[derive(Debug, derive_more::Display, derive_more::Error, Clone)]
pub enum BitswapOutClosedErr {
    /// Substream has been closed because the connection as a whole is being shut down.
    ConnectionShutdown,

    /// Error happened in the context of the substream.
    #[display("{_0}")]
    Substream(established::BitswapOutClosedErr),
}

/// Error that led to closing of inbound Bitswap substream.
#[derive(Debug, derive_more::Display, derive_more::Error, Clone)]
pub enum BitswapInClosedErr {
    /// Substream has been closed because the connection as a whole is being shut down.
    ConnectionShutdown,

    /// Error happened in the context of the substream.
    #[display("{_0}")]
    Substream(established::BitswapInClosedErr),
}

/// Error potentially returned by [`Network::queue_bitswap_message`].
#[derive(Debug, derive_more::Display, derive_more::Error)]
pub enum QueueBitswapMessageError {
    /// Queue of bitswap messages with that peer is full.
    QueueFull,
}