protocols/identify: Allow at most one inbound identify push stream (#2694)

An identify push contains the whole identify information of a remote
peer. Upgrading multiple inbound identify push streams is useless.
Instead older streams are dropped in favor of newer streams.

protocols/identify/src/handler.rs

@@ -22,6 +22,7 @@ use crate::protocol::{
     IdentifyInfo, IdentifyProtocol, IdentifyPushProtocol, InboundPush, OutboundPush,
     ReplySubstream, UpgradeError,
 };
+use futures::future::BoxFuture;
 use futures::prelude::*;
 use futures_timer::Delay;
 use libp2p_core::either::{EitherError, EitherOutput};
@@ -30,6 +31,7 @@ use libp2p_swarm::{
     ConnectionHandler, ConnectionHandlerEvent, ConnectionHandlerUpgrErr, KeepAlive,
     NegotiatedSubstream, SubstreamProtocol,
 };
+use log::warn;
 use smallvec::SmallVec;
 use std::{io, pin::Pin, task::Context, task::Poll, time::Duration};
 
@@ -39,6 +41,7 @@ use std::{io, pin::Pin, task::Context, task::Poll, time::Duration};
 /// at least one identification request to be answered by the remote before
 /// permitting the underlying connection to be closed.
 pub struct IdentifyHandler {
+    inbound_identify_push: Option<BoxFuture<'static, Result<IdentifyInfo, UpgradeError>>>,
     /// Pending events to yield.
     events: SmallVec<
         [ConnectionHandlerEvent<
@@ -80,6 +83,7 @@ impl IdentifyHandler {
     /// Creates a new `IdentifyHandler`.
     pub fn new(initial_delay: Duration, interval: Duration) -> Self {
         IdentifyHandler {
+            inbound_identify_push: Default::default(),
             events: SmallVec::new(),
             trigger_next_identify: Delay::new(initial_delay),
             keep_alive: KeepAlive::Yes,
@@ -113,9 +117,14 @@ impl ConnectionHandler for IdentifyHandler {
             EitherOutput::First(substream) => self.events.push(ConnectionHandlerEvent::Custom(
                 IdentifyHandlerEvent::Identify(substream),
             )),
-            EitherOutput::Second(info) => self.events.push(ConnectionHandlerEvent::Custom(
-                IdentifyHandlerEvent::Identified(info),
-            )),
+            EitherOutput::Second(fut) => {
+                if self.inbound_identify_push.replace(fut).is_some() {
+                    warn!(
+                        "New inbound identify push stream while still upgrading previous one. \
+                        Replacing previous with new.",
+                    );
+                }
+            }
         }
     }
 
@@ -189,14 +198,30 @@ impl ConnectionHandler for IdentifyHandler {
 
         // Poll the future that fires when we need to identify the node again.
         match Future::poll(Pin::new(&mut self.trigger_next_identify), cx) {
-            Poll::Pending => Poll::Pending,
+            Poll::Pending => {}
             Poll::Ready(()) => {
                 self.trigger_next_identify.reset(self.interval);
                 let ev = ConnectionHandlerEvent::OutboundSubstreamRequest {
                     protocol: SubstreamProtocol::new(EitherUpgrade::A(IdentifyProtocol), ()),
                 };
-                Poll::Ready(ev)
+                return Poll::Ready(ev);
             }
         }
+
+        if let Some(Poll::Ready(res)) = self
+            .inbound_identify_push
+            .as_mut()
+            .map(|f| f.poll_unpin(cx))
+        {
+            self.inbound_identify_push.take();
+
+            if let Ok(info) = res {
+                return Poll::Ready(ConnectionHandlerEvent::Custom(
+                    IdentifyHandlerEvent::Identified(info),
+                ));
+            }
+        }
+
+        Poll::Pending
     }
 }