transport_service: Improve connection stability by downgrading connections on substream inactivity

src/protocol/connection.rs

@@ -94,6 +94,15 @@ impl ConnectionHandle {
         }
     }
 
+    /// Try to upgrade the connection to active state.
+    pub fn try_open(&mut self) {
+        if let ConnectionType::Inactive(inactive) = &self.connection {
+            if let Some(active) = inactive.upgrade() {
+                self.connection = ConnectionType::Active(active);
+            }
+        }
+    }
+
     /// Attempt to acquire permit which will keep the connection open for indefinite time.
     pub fn try_get_permit(&self) -> Option<Permit> {
         match &self.connection {
@@ -120,6 +129,7 @@ impl ConnectionHandle {
             protocol: protocol.clone(),
             fallback_names,
             substream_id,
+            connection_id: self.connection_id.clone(),
             permit,
         })
         .map_err(|error| match error {
@@ -141,10 +151,15 @@ impl ConnectionHandle {
             TrySendError::Closed(_) => Error::ConnectionClosed,
         })
     }
+
+    /// Check if the connection is active.
+    pub fn is_active(&self) -> bool {
+        matches!(self.connection, ConnectionType::Active(_))
+    }
 }
 
 /// Type which allows the connection to be kept open.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct Permit {
     /// Active connection.
     _connection: Sender<ProtocolCommand>,

src/protocol/notification/tests/notification.rs

@@ -475,6 +475,7 @@ async fn remote_opens_multiple_inbound_substreams() {
             SubstreamId::from(0usize),
             Box::new(DummySubstream::new()),
         ),
+        connection_id: ConnectionId::from(0usize),
     })
     .await
     .unwrap();
@@ -511,6 +512,7 @@ async fn remote_opens_multiple_inbound_substreams() {
             SubstreamId::from(0usize),
             Box::new(substream),
         ),
+        connection_id: ConnectionId::from(0usize),
     })
     .await
     .unwrap();

src/protocol/protocol_set.rs

@@ -122,6 +122,9 @@ pub enum InnerTransportEvent {
         /// distinguish between different outbound substreams.
         direction: Direction,
 
+        /// Connection ID.
+        connection_id: ConnectionId,
+
         /// Substream.
         substream: Substream,
     },
@@ -149,6 +152,7 @@ impl From<InnerTransportEvent> for TransportEvent {
                 fallback,
                 direction,
                 substream,
+                ..
             } => TransportEvent::SubstreamOpened {
                 peer,
                 protocol,
@@ -164,7 +168,7 @@ impl From<InnerTransportEvent> for TransportEvent {
 }
 
 /// Events emitted by the installed protocols to transport.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub enum ProtocolCommand {
     /// Open substream.
     OpenSubstream {
@@ -192,6 +196,9 @@ pub enum ProtocolCommand {
         /// and associate incoming substreams with whatever logic it has.
         substream_id: SubstreamId,
 
+        /// Connection ID.
+        connection_id: ConnectionId,
+
         /// Connection permit.
         ///
         /// `Permit` allows the connection to be kept open while the permit is held and it is given
@@ -300,6 +307,7 @@ impl ProtocolSet {
             fallback,
             direction,
             substream,
+            connection_id: self.connection.connection_id().clone(),
         };
 
         protocol_context