feat: deprecate 'KeepAlive::Until' globally

protocols/dcutr/src/handler/relayed.rs

@@ -24,7 +24,6 @@ use crate::protocol;
 use either::Either;
 use futures::future;
 use futures::future::{BoxFuture, FutureExt};
-use instant::Instant;
 use libp2p_core::multiaddr::Multiaddr;
 use libp2p_core::upgrade::{DeniedUpgrade, NegotiationError, UpgradeError};
 use libp2p_core::ConnectedPoint;
@@ -144,6 +143,7 @@ pub struct Handler {
     inbound_connect:
         Option<BoxFuture<'static, Result<Vec<Multiaddr>, protocol::inbound::UpgradeError>>>,
     keep_alive: KeepAlive,
+    timeout: futures_timer::Delay,
 }
 
 impl Handler {
@@ -153,7 +153,8 @@ impl Handler {
             pending_error: Default::default(),
             queued_events: Default::default(),
             inbound_connect: Default::default(),
-            keep_alive: KeepAlive::Until(Instant::now() + Duration::from_secs(30)),
+            keep_alive: KeepAlive::Yes,
+            timeout: futures_timer::Delay::new(Duration::from_secs(30)),
         }
     }
 
@@ -392,6 +393,10 @@ impl ConnectionHandler for Handler {
             }
         }
 
+        if self.timeout.poll_unpin(cx).is_ready() {
+            self.keep_alive = KeepAlive::No // Close the connection 30 seconds after initiated.
+        }
+
         Poll::Pending
     }
 

protocols/gossipsub/Cargo.toml

@@ -19,6 +19,7 @@ bytes = "1.4"
 byteorder = "1.3.4"
 fnv = "1.0.7"
 futures = "0.3.28"
+futures-timer = "3.0.2"
 rand = "0.8"
 asynchronous-codec = "0.6"
 unsigned-varint = { version = "0.7.0", features = ["asynchronous_codec"] }

protocols/gossipsub/src/handler.rs

@@ -122,6 +122,10 @@ pub struct EnabledHandler {
     /// The amount of time we keep an idle connection alive.
     idle_timeout: Duration,
 
+    keep_alive: KeepAlive,
+
+    timeout: Option<futures_timer::Delay>,
+
     /// Keeps track of whether this connection is for a peer in the mesh. This is used to make
     /// decisions about the keep alive state for this connection.
     in_mesh: bool,
@@ -177,6 +181,8 @@ impl Handler {
             peer_kind_sent: false,
             last_io_activity: Instant::now(),
             idle_timeout,
+            keep_alive: KeepAlive::Yes,
+            timeout: None,
             in_mesh: false,
         })
     }
@@ -388,6 +394,29 @@ impl EnabledHandler {
             }
         }
 
+        match self.outbound_substream {
+            Some(
+                OutboundSubstreamState::PendingSend(_, _) | OutboundSubstreamState::PendingFlush(_),
+            ) => self.timeout = None, // Cancel the timer when there's pending activity
+            _ => {
+                if !self.in_mesh {
+                    // Criteria met to set a timer if none present.
+                    self.timeout.get_or_insert(futures_timer::Delay::new(
+                        (self.last_io_activity + self.idle_timeout).duration_since(Instant::now()),
+                    ));
+                } else {
+                    // In mesh, cancel the timer.
+                    self.timeout = None
+                }
+            }
+        }
+        // Note that cancellation happens before checking the timeout,
+        // which may cause the connection not be droped after timeout reached.
+        if let Some(timer) = &mut self.timeout {
+            if timer.poll_unpin(cx).is_ready() {
+                self.keep_alive = KeepAlive::No
+            }
+        }
         Poll::Pending
     }
 }
@@ -431,21 +460,7 @@ impl ConnectionHandler for Handler {
 
     fn connection_keep_alive(&self) -> KeepAlive {
         match self {
-            Handler::Enabled(handler) => {
-                if handler.in_mesh {
-                    return KeepAlive::Yes;
-                }
-
-                if let Some(
-                    OutboundSubstreamState::PendingSend(_, _)
-                    | OutboundSubstreamState::PendingFlush(_),
-                ) = handler.outbound_substream
-                {
-                    return KeepAlive::Yes;
-                }
-
-                KeepAlive::Until(handler.last_io_activity + handler.idle_timeout)
-            }
+            Handler::Enabled(handler) => handler.keep_alive,
             Handler::Disabled(_) => KeepAlive::No,
         }
     }

protocols/kad/src/handler_priv.rs

@@ -26,7 +26,6 @@ use crate::record_priv::{self, Record};
 use either::Either;
 use futures::prelude::*;
 use futures::stream::SelectAll;
-use instant::Instant;
 use libp2p_core::{upgrade, ConnectedPoint};
 use libp2p_identity::PeerId;
 use libp2p_swarm::handler::{
@@ -75,6 +74,8 @@ pub struct KademliaHandler<TUserData> {
     /// Until when to keep the connection alive.
     keep_alive: KeepAlive,
 
+    timeout: Option<futures_timer::Delay>,
+
     /// The connected endpoint of the connection that the handler
     /// is associated with.
     endpoint: ConnectedPoint,
@@ -488,8 +489,7 @@ where
         endpoint: ConnectedPoint,
         remote_peer_id: PeerId,
     ) -> Self {
-        let keep_alive = KeepAlive::Until(Instant::now() + config.idle_timeout);
-
+        let timeout = Some(futures_timer::Delay::new(config.idle_timeout));
         KademliaHandler {
             config,
             endpoint,
@@ -499,7 +499,8 @@ where
             outbound_substreams: Default::default(),
             num_requested_outbound_streams: 0,
             pending_messages: Default::default(),
-            keep_alive,
+            keep_alive: KeepAlive::Yes,
+            timeout,
             protocol_status: ProtocolStatus::Unconfirmed,
         }
     }
@@ -746,14 +747,22 @@ where
         }
 
         let no_streams = self.outbound_substreams.is_empty() && self.inbound_substreams.is_empty();
-        self.keep_alive = match (no_streams, self.keep_alive) {
-            // No open streams. Preserve the existing idle timeout.
-            (true, k @ KeepAlive::Until(_)) => k,
-            // No open streams. Set idle timeout.
-            (true, _) => KeepAlive::Until(Instant::now() + self.config.idle_timeout),
-            // Keep alive for open streams.
-            (false, _) => KeepAlive::Yes,
-        };
+        if no_streams {
+            // Criteria met to set a timer if none present.
+            self.timeout
+                .get_or_insert(futures_timer::Delay::new(self.config.idle_timeout));
+        } else {
+            // Cancel the timer.
+            // Note that cancellation happens before checking the timeout,
+            // which may cause the connection not be droped after timeout reached.
+            self.timeout.take();
+        }
+
+        if let Some(timer) = &mut self.timeout {
+            if timer.poll_unpin(cx).is_ready() {
+                self.keep_alive = KeepAlive::No
+            }
+        }
 
         Poll::Pending
     }

protocols/perf/Cargo.toml

@@ -16,6 +16,7 @@ async-std = { version = "1.9.0", features = ["attributes"] }
 clap = { version = "4.2.7", features = ["derive"] }
 env_logger = "0.10.0"
 futures = "0.3.28"
+futures-timer = "3.0.2"
 instant = "0.1.11"
 libp2p-core = { workspace = true }
 libp2p-dns = { workspace = true, features = ["async-std"] }

protocols/perf/src/client/handler.rs

@@ -21,7 +21,7 @@
 use std::{
     collections::VecDeque,
     task::{Context, Poll},
-    time::{Duration, Instant},
+    time::Duration,
 };
 
 use futures::{future::BoxFuture, stream::FuturesUnordered, FutureExt, StreamExt, TryFutureExt};
@@ -66,6 +66,8 @@ pub struct Handler {
     outbound: FuturesUnordered<BoxFuture<'static, Result<Event, std::io::Error>>>,
 
     keep_alive: KeepAlive,
+
+    timeout: Option<futures_timer::Delay>,
 }
 
 impl Handler {
@@ -75,6 +77,7 @@ impl Handler {
             requested_streams: Default::default(),
             outbound: Default::default(),
             keep_alive: KeepAlive::Yes,
+            timeout: None,
         }
     }
 }
@@ -192,15 +195,20 @@ impl ConnectionHandler for Handler {
         }
 
         if self.outbound.is_empty() {
-            match self.keep_alive {
-                KeepAlive::Yes => {
-                    self.keep_alive = KeepAlive::Until(Instant::now() + Duration::from_secs(10));
-                }
-                KeepAlive::Until(_) => {}
-                KeepAlive::No => panic!("Handler never sets KeepAlive::No."),
-            }
+            // Criteria met to set a timer if none present.
+            self.timeout
+                .get_or_insert(futures_timer::Delay::new(Duration::from_secs(10)));
         } else {
-            self.keep_alive = KeepAlive::Yes
+            // Cancel the timer.
+            // Note that cancellation happens before checking the timeout,
+            // which may cause the connection not be droped after 10 seconds.
+            self.timeout.take();
+        }
+
+        if let Some(timer) = &mut self.timeout {
+            if timer.poll_unpin(cx).is_ready() {
+                self.keep_alive = KeepAlive::No
+            }
         }
 
         Poll::Pending

protocols/perf/src/server/handler.rs

@@ -20,7 +20,7 @@
 
 use std::{
     task::{Context, Poll},
-    time::{Duration, Instant},
+    time::Duration,
 };
 
 use futures::{future::BoxFuture, stream::FuturesUnordered, FutureExt, StreamExt};
@@ -46,13 +46,15 @@ pub struct Event {
 pub struct Handler {
     inbound: FuturesUnordered<BoxFuture<'static, Result<RunStats, std::io::Error>>>,
     keep_alive: KeepAlive,
+    timeout: Option<futures_timer::Delay>,
 }
 
 impl Handler {
     pub fn new() -> Self {
         Self {
             inbound: Default::default(),
             keep_alive: KeepAlive::Yes,
+            timeout: None,
         }
     }
 }
@@ -143,15 +145,20 @@ impl ConnectionHandler for Handler {
         }
 
         if self.inbound.is_empty() {
-            match self.keep_alive {
-                KeepAlive::Yes => {
-                    self.keep_alive = KeepAlive::Until(Instant::now() + Duration::from_secs(10));
-                }
-                KeepAlive::Until(_) => {}
-                KeepAlive::No => panic!("Handler never sets KeepAlive::No."),
-            }
+            // Criteria met to set a timer if none present.
+            self.timeout
+                .get_or_insert(futures_timer::Delay::new(Duration::from_secs(10)));
         } else {
-            self.keep_alive = KeepAlive::Yes
+            // Cancel the timer.
+            // Note that cancellation happens before checking the timeout,
+            // which may cause the connection not be droped after 10 seconds.
+            self.timeout.take();
+        }
+
+        if let Some(timer) = &mut self.timeout {
+            if timer.poll_unpin(cx).is_ready() {
+                self.keep_alive = KeepAlive::No
+            }
         }
 
         Poll::Pending

protocols/relay/src/behaviour/handler.rs

@@ -29,7 +29,6 @@ use futures::future::{BoxFuture, FutureExt, TryFutureExt};
 use futures::io::AsyncWriteExt;
 use futures::stream::{FuturesUnordered, StreamExt};
 use futures_timer::Delay;
-use instant::Instant;
 use libp2p_core::{upgrade, ConnectedPoint, Multiaddr};
 use libp2p_identity::PeerId;
 use libp2p_swarm::handler::{
@@ -363,6 +362,8 @@ pub struct Handler {
     /// Until when to keep the connection alive.
     keep_alive: KeepAlive,
 
+    timeout: Option<futures_timer::Delay>,
+
     /// Future handling inbound reservation request.
     reservation_request_future: Option<ReservationRequestFuture>,
     /// Timeout for the currently active reservation.
@@ -404,6 +405,7 @@ impl Handler {
             circuits: Default::default(),
             active_reservation: Default::default(),
             keep_alive: KeepAlive::Yes,
+            timeout: None,
         }
     }
 
@@ -914,15 +916,18 @@ impl ConnectionHandler for Handler {
             && self.circuits.is_empty()
             && self.active_reservation.is_none()
         {
-            match self.keep_alive {
-                KeepAlive::Yes => {
-                    self.keep_alive = KeepAlive::Until(Instant::now() + Duration::from_secs(10));
-                }
-                KeepAlive::Until(_) => {}
-                KeepAlive::No => panic!("Handler never sets KeepAlive::No."),
-            }
+            // Criteria met to set a timer.
+            self.timeout
+                .get_or_insert(futures_timer::Delay::new(Duration::from_secs(10)));
         } else {
-            self.keep_alive = KeepAlive::Yes;
+            // Cancel the timer.
+            self.timeout.take();
+        }
+
+        if let Some(timer) = &mut self.timeout {
+            if timer.poll_unpin(cx).is_ready() {
+                self.keep_alive = KeepAlive::No
+            }
         }
 
         Poll::Pending