refactor(relay): move stream-handling away from {In,Out}boundUpgrade

Cargo.toml

@@ -19,6 +19,7 @@ members = [
     "interop-tests",
     "misc/allow-block-list",
     "misc/connection-limits",
+    "misc/futures-bounded",
     "misc/keygen",
     "misc/memory-connection-limits",
     "misc/metrics",
@@ -69,6 +70,7 @@ resolver = "2"
 rust-version = "1.65.0"
 
 [workspace.dependencies]
+futures-bounded = { version = "0.1.0", path = "misc/futures-bounded" }
 libp2p = { version = "0.52.3", path = "libp2p" }
 libp2p-allow-block-list = { version = "0.2.0", path = "misc/allow-block-list" }
 libp2p-autonat = { version = "0.11.0", path = "protocols/autonat" }

misc/futures-bounded/CHANGELOG.md

@@ -0,0 +1,3 @@
+## 0.1.0 - unreleased
+
+Initial release.

misc/futures-bounded/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "futures-bounded"
+version = "0.1.0"
+edition = "2021"
+rust-version.workspace = true
+license = "MIT"
+repository = "https://github.com/libp2p/rust-libp2p"
+keywords = ["futures", "async", "backpressure"]
+categories = ["data-structures", "asynchronous"]
+description = "Utilities for bounding futures in size and time."
+publish = false # TEMP FIX until https://github.com/obi1kenobi/cargo-semver-checks-action/issues/53 is fixed.
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+futures-util = { version = "0.3.28" }
+futures-timer = "3.0.2"
+
+[dev-dependencies]
+tokio = { version = "1.29.1", features = ["macros", "rt"] }

misc/futures-bounded/src/lib.rs

@@ -0,0 +1,28 @@
+mod list;
+mod map;
+
+pub use list::FuturesList;
+pub use map::{FuturesMap, PushError};
+use std::fmt;
+use std::fmt::Formatter;
+use std::time::Duration;
+
+/// A future failed to complete within the given timeout.
+#[derive(Debug)]
+pub struct Timeout {
+    limit: Duration,
+}
+
+impl Timeout {
+    fn new(duration: Duration) -> Self {
+        Self { limit: duration }
+    }
+}
+
+impl fmt::Display for Timeout {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        write!(f, "future failed to complete within {:?}", self.limit)
+    }
+}
+
+impl std::error::Error for Timeout {}

misc/futures-bounded/src/list.rs

@@ -0,0 +1,58 @@
+use std::future::Future;
+use std::task::{ready, Context, Poll};
+use std::time::Duration;
+
+use futures_util::future::BoxFuture;
+
+use crate::{FuturesMap, PushError, Timeout};
+
+/// Represents a list of [Future]s.
+///
+/// Each future must finish within the specified time and the list never outgrows its capacity.
+pub struct FuturesList<O> {
+    id: i32,
+    inner: FuturesMap<i32, O>,
+}
+
+impl<O> FuturesList<O> {
+    pub fn new(timeout: Duration, capacity: usize) -> Self {
+        Self {
+            id: i32::MIN,
+            inner: FuturesMap::new(timeout, capacity),
+        }
+    }
+}
+
+impl<O> FuturesList<O> {
+    /// Push a future into the list.
+    ///
+    /// This method adds the given future to the list.
+    /// If the length of the list is equal to the capacity, this method returns a error that contains the passed future.
+    /// In that case, the future is not added to the set.
+    pub fn try_push<F>(&mut self, future: F) -> Result<(), BoxFuture<O>>
+    where
+        F: Future<Output = O> + Send + 'static,
+    {
+        (self.id, _) = self.id.overflowing_add(1);
+
+        match self.inner.try_push(self.id, future) {
+            Ok(()) => Ok(()),
+            Err(PushError::BeyondCapacity(w)) => Err(w),
+            Err(PushError::ReplacedFuture(_)) => unreachable!("we never reuse IDs"),
+        }
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.inner.is_empty()
+    }
+
+    pub fn poll_ready_unpin(&mut self, cx: &mut Context<'_>) -> Poll<()> {
+        self.inner.poll_ready_unpin(cx)
+    }
+
+    pub fn poll_unpin(&mut self, cx: &mut Context<'_>) -> Poll<Result<O, Timeout>> {
+        let (_, res) = ready!(self.inner.poll_unpin(cx));
+
+        Poll::Ready(res)
+    }
+}