protocol: Ensure transport manager knows about closed connections (#515)

This PR improves the connection management robustness by ensuring that
the closed connections are reported.

If a registered protocol could not receive the
`InnerTransportEvent::ConnectionClosed` the function would fail to
inform the transport manager. This would cause litep2p to think that the
connection is still active, while the async task of the raw socket would
exit immediately after the function call.

The PR makes sure are not returning on the first encountered error and
switch to a best-effort approach.

---------

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

Author: lexnv

src/protocol/protocol_set.rs

@@ -396,31 +396,53 @@ impl ProtocolSet {
     ) -> crate::Result<()> {
         let mut futures = self
             .protocols
-            .values()
-            .map(|sender| async move {
+            .iter()
+            .map(|(protocol, sender)| async move {
                 sender
                     .tx
                     .send(InnerTransportEvent::ConnectionClosed {
                         peer,
                         connection: connection_id,
                     })
                     .await
+                    .inspect_err(|err| {
+                        tracing::debug!(
+                            target: LOG_TARGET,
+                            %protocol,
+                            ?peer,
+                            ?connection_id,
+                            ?err,
+                            "failed to report connection closed to protocol",
+                        );
+                    })
             })
             .collect::<FuturesUnordered<_>>();
 
+        // Capture the first error that occurs while reporting to protocols.
+        let mut protocol_error = None;
         while !futures.is_empty() {
-            if let Some(Err(error)) = futures.next().await {
-                return Err(error.into());
+            if let Some(Err(err)) = futures.next().await {
+                if protocol_error.is_none() {
+                    protocol_error = Some(err.into());
+                }
             }
         }
 
+        // Ensure the manager receives the connection closed event. Otherwise, the
+        // manager will think the connection is still open, while the underlying
+        // protocols and raw connection are closed.
         self.mgr_tx
             .send(TransportManagerEvent::ConnectionClosed {
                 peer,
                 connection: connection_id,
             })
-            .await
-            .map_err(From::from)
+            .await?;
+
+        // If any protocol report failed, return that error now
+        match protocol_error {
+            Some(e) => Err(e),
+            None => Ok(()),
+        }
     }
 }