ping: Fix memory leak of unremoved pending_opens (#274)

The purpose of the `pending_opens` field is to double check outbound
substream opens.
This was used to ensure that the substream ID was indeed opened to a
given peer ID.
However, this is not needed considering the `identify` implementation.

Further, the `pending_opens` was leaking `(SubstramId, PeerId)` tuples
in cases where the
substream opening would later fail. In other words, the implementation
did not remove the tuples on the `TransportEvent::SubstreamOpenFailure`
event.

Part of endeavors to fix memory leaks:
paritytech/polkadot-sdk#6149

### Testing Done
- custom patched litep2p to dump the internal state of identify protocol
running in kusama

The code is similar to the identify protocol. However, this leak was
more subtle and not of the magnitude of the `identify` protocol since
substream open failures are not that frequent:
 - #273
 
 cc @paritytech/networking

Signed-off-by: Alexandru Vasile <[email protected]>

src/protocol/libp2p/ping/mod.rs

@@ -32,7 +32,7 @@ use futures::{future::BoxFuture, stream::FuturesUnordered, StreamExt};
 use tokio::sync::mpsc::Sender;
 
 use std::{
-    collections::{HashMap, HashSet},
+    collections::HashSet,
     time::{Duration, Instant},
 };
 
@@ -72,9 +72,6 @@ pub(crate) struct Ping {
     /// Connected peers.
     peers: HashSet<PeerId>,
 
-    /// Pending outbound substreams.
-    pending_opens: HashMap<SubstreamId, PeerId>,
-
     /// Pending outbound substreams.
     pending_outbound: FuturesUnordered<BoxFuture<'static, crate::Result<(PeerId, Duration)>>>,
 
@@ -89,7 +86,6 @@ impl Ping {
             service,
             tx: config.tx_event,
             peers: HashSet::new(),
-            pending_opens: HashMap::new(),
             pending_outbound: FuturesUnordered::new(),
             pending_inbound: FuturesUnordered::new(),
             _max_failures: config.max_failures,
@@ -100,8 +96,7 @@ impl Ping {
     fn on_connection_established(&mut self, peer: PeerId) -> crate::Result<()> {
         tracing::trace!(target: LOG_TARGET, ?peer, "connection established");
 
-        let substream_id = self.service.open_substream(peer)?;
-        self.pending_opens.insert(substream_id, peer);
+        self.service.open_substream(peer)?;
         self.peers.insert(peer);
 
         Ok(())
@@ -191,19 +186,7 @@ impl Ping {
                             self.on_inbound_substream(peer, substream);
                         }
                         Direction::Outbound(substream_id) => {
-                            match self.pending_opens.remove(&substream_id) {
-                                Some(stored_peer) => {
-                                    debug_assert!(peer == stored_peer);
-                                    self.on_outbound_substream(peer, substream_id, substream);
-                                }
-                                None => {
-                                    tracing::warn!(
-                                        target: LOG_TARGET,
-                                        ?substream_id,
-                                        "outbound ping substream ID does not exist",
-                                    );
-                                }
-                            }
+                            self.on_outbound_substream(peer, substream_id, substream);
                         }
                     },
                     Some(_) => {}