chore: add more docs, rm unused async_trait, fix workspace lint rules, clean up unnecessary trait bounds

Author: merklefruit

Cargo.toml

@@ -108,9 +108,10 @@ all = "warn"
 [workspace.lints.rust]
 missing_debug_implementations = "warn"
 missing_docs = "warn"
-rust-2018-idioms = { level = "deny", priority = -1 }
-unreachable-pub = "warn"
-unused-must-use = "deny"
+rust_2018_idioms = { level = "deny", priority = -1 }
+unreachable_pub = "warn"
+unused_must_use = "deny"
+rust_2024_incompatible_pat = "warn"
 
 [workspace.lints.clippy]
 # These are some of clippy's nursery (i.e., experimental) lints that we like.

msg-socket/src/connection/backoff.rs

@@ -1,12 +1,16 @@
 use futures::{FutureExt, Stream};
-use std::{pin::Pin, task::Poll, time::Duration};
+use std::{
+    pin::Pin,
+    task::{Context, Poll},
+    time::Duration,
+};
 use tokio::time::sleep;
 
 /// Helper trait alias for backoff streams.
 /// We define any stream that yields `Duration`s as a backoff
 pub trait Backoff: Stream<Item = Duration> + Unpin {}
 
-/// Blanket implementation of `Backoff` for any stream that yields `Duration`s.
+// Blanket implementation of `Backoff` for any stream that yields `Duration`s.
 impl<T> Backoff for T where T: Stream<Item = Duration> + Unpin {}
 
 /// A stream that yields exponentially increasing backoff durations.
@@ -23,6 +27,7 @@ pub struct ExponentialBackoff {
 }
 
 impl ExponentialBackoff {
+    /// Creates a new exponential backoff stream with the given initial duration and max retries.
     pub fn new(initial: Duration, max_retries: usize) -> Self {
         Self { retry_count: 0, max_retries, backoff: initial, timeout: None }
     }
@@ -38,38 +43,36 @@ impl Stream for ExponentialBackoff {
 
     /// Polls the exponential backoff stream. Returns `Poll::Ready` with the current backoff
     /// duration if the backoff timeout has elapsed, otherwise returns `Poll::Pending`.
-    fn poll_next(
-        self: Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> Poll<Option<Self::Item>> {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let this = self.get_mut();
 
         loop {
-            if let Some(ref mut timeout) = this.timeout {
-                if timeout.poll_unpin(cx).is_ready() {
-                    // Timeout has elapsed, so reset the timeout and double the backoff
-                    this.backoff *= 2;
-                    this.retry_count += 1;
+            let Some(ref mut timeout) = this.timeout else {
+                // Set the initial timeout
+                this.reset_timeout();
+                continue;
+            };
+
+            if timeout.poll_unpin(cx).is_ready() {
+                // Timeout has elapsed, so reset the timeout and double the backoff
+                this.backoff *= 2;
+                this.retry_count += 1;
 
-                    // Close the stream
-                    if this.retry_count >= this.max_retries {
-                        return Poll::Ready(None);
-                    }
+                // Close the stream
+                if this.retry_count >= this.max_retries {
+                    return Poll::Ready(None);
+                }
 
-                    this.reset_timeout();
+                this.reset_timeout();
 
-                    // Wake up the task to poll the timeout again
-                    cx.waker().wake_by_ref();
+                // Wake up the task to poll the timeout again
+                cx.waker().wake_by_ref();
 
-                    // Return the current backoff duration
-                    return Poll::Ready(Some(this.backoff));
-                } else {
-                    // Timeout has not elapsed, so return pending
-                    return Poll::Pending;
-                }
+                // Return the current backoff duration
+                return Poll::Ready(Some(this.backoff));
             } else {
-                // Set initial timeout
-                this.reset_timeout();
+                // Timeout has not elapsed, so return pending
+                return Poll::Pending;
             }
         }
     }

msg-socket/src/connection/state.rs

@@ -1,12 +1,9 @@
-use msg_transport::Address;
-
-use super::Backoff;
-
 /// Abstraction to represent the state of a connection.
 ///
 /// * `C` is the channel type, which is used to send and receive generic messages.
 /// * `B` is the backoff type, used to control the backoff state for inactive connections.
-pub enum ConnectionState<C, B, A: Address> {
+/// * `A` is the address type, used to represent the address of the connection.
+pub enum ConnectionState<C, B, A> {
     Active {
         /// Channel to control the underlying connection. This is used to send
         /// and receive any kind of message in any direction.
@@ -19,15 +16,13 @@ pub enum ConnectionState<C, B, A: Address> {
     },
 }
 
-impl<C, B: Backoff, A: Address> ConnectionState<C, B, A> {
+impl<C, B, A> ConnectionState<C, B, A> {
     /// Returns `true` if the connection is active.
-    #[allow(unused)]
     pub fn is_active(&self) -> bool {
         matches!(self, Self::Active { .. })
     }
 
     /// Returns `true` if the connection is inactive.
-    #[allow(unused)]
     pub fn is_inactive(&self) -> bool {
         matches!(self, Self::Inactive { .. })
     }

msg-socket/src/req/socket.rs

@@ -113,7 +113,6 @@ where
     /// Tries to connect to the target endpoint with the default options.
     /// A ReqSocket can only be connected to a single address.
     pub async fn try_connect(&mut self, endpoint: A) -> Result<(), ReqError> {
-        // Initialize communication channels
         let (to_driver, from_socket) = mpsc::channel(DEFAULT_BUFFER_SIZE);
 
         // TODO: Don't panic, return error

msg-transport/src/ipc/mod.rs

@@ -41,17 +41,20 @@ pub struct Ipc {
 }
 
 impl Ipc {
+    /// Creates a new `Ipc` transport.
     pub fn new(config: Config) -> Self {
         Self { config, listener: None, path: None }
     }
 }
 
+/// A stream representing an IPC connection.
 pub struct IpcStream {
     peer: PathBuf,
     stream: UnixStream,
 }
 
 impl IpcStream {
+    /// Connects to the given peer path.
     pub async fn connect(peer: PathBuf) -> io::Result<Self> {
         let stream = UnixStream::connect(&peer).await?;
         Ok(Self { peer, stream })

msg-transport/src/lib.rs

@@ -196,6 +196,7 @@ pub trait Transport<A: Address>: Send + Sync + Unpin + 'static {
     fn poll_accept(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Accept>;
 }
 
+/// Extension trait for transports that provides additional methods.
 pub trait TransportExt<A: Address>: Transport<A> {
     /// Async-friendly interface for accepting inbound connections.
     fn accept(&mut self) -> Acceptor<'_, Self, A>
@@ -226,6 +227,7 @@ where
     T: Transport<A>,
     A: Address,
 {
+    /// Creates a new `Acceptor` for the given transport.
     fn new(inner: &'a mut T) -> Self {
         Self { inner, pending: None, _marker: PhantomData }
     }

msg-transport/src/quic/config.rs

@@ -25,12 +25,20 @@ pub struct ConfigBuilder<C> {
     keep_alive_interval: Duration,
 }
 
+impl<C> Default for ConfigBuilder<C>
+where
+    C: ControllerFactory + Default + Send + Sync + 'static,
+{
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 impl<C> ConfigBuilder<C>
 where
     C: ControllerFactory + Default + Send + Sync + 'static,
 {
     /// Creates a new [`ConfigBuilder`] with sensible defaults.
-    #[allow(clippy::new_without_default)]
     pub fn new() -> Self {
         Self {
             cc: C::default(),
@@ -43,37 +51,39 @@ where
     }
 
     /// Sets the initial MTU.
-    pub fn initial_mtu(mut self, mtu: u16) -> Self {
+    pub fn with_initial_mtu(mut self, mtu: u16) -> Self {
         self.initial_mtu = mtu;
         self
     }
 
     /// Sets the maximum stream bandwidth in bytes per second.
-    pub fn max_stream_bandwidth(mut self, bandwidth: u32) -> Self {
+    pub fn with_max_stream_bandwidth(mut self, bandwidth: u32) -> Self {
         self.max_stream_bandwidth = bandwidth;
         self
     }
 
     /// Sets the expected round-trip time in milliseconds.
-    pub fn expected_rtt(mut self, rtt: u32) -> Self {
+    pub fn with_expected_rtt(mut self, rtt: u32) -> Self {
         self.expected_rtt = rtt;
         self
     }
 
     /// Sets the maximum idle timeout.
-    pub fn max_idle_timeout(mut self, timeout: Duration) -> Self {
+    ///
+    /// NOTE: this value must be less than 2^62 milliseconds.
+    pub fn with_max_idle_timeout(mut self, timeout: Duration) -> Self {
         self.max_idle_timeout = timeout;
         self
     }
 
     /// Sets the keep-alive interval.
-    pub fn keep_alive_interval(mut self, interval: Duration) -> Self {
+    pub fn with_keep_alive_interval(mut self, interval: Duration) -> Self {
         self.keep_alive_interval = interval;
         self
     }
 
     /// Sets the congestion controller.
-    pub fn congestion_controller(mut self, cc: C) -> Self {
+    pub fn with_congestion_controller(mut self, cc: C) -> Self {
         self.cc = cc;
         self
     }
@@ -85,11 +95,11 @@ where
         // Stream receive window
         let stream_rwnd = self.max_stream_bandwidth / 1000 * self.expected_rtt;
 
+        let timeout = IdleTimeout::try_from(self.max_idle_timeout).expect("Valid idle timeout");
+
         transport
             .keep_alive_interval(Some(self.keep_alive_interval))
-            .max_idle_timeout(Some(
-                IdleTimeout::try_from(self.max_idle_timeout).expect("Valid idle timeout"),
-            ))
+            .max_idle_timeout(Some(timeout))
             // Disable datagram support
             .datagram_receive_buffer_size(None)
             .datagram_send_buffer_size(0)

msg-transport/src/tcp/mod.rs

@@ -89,7 +89,6 @@ impl Transport<SocketAddr> for Tcp {
     }
 }
 
-#[async_trait::async_trait]
 impl TransportExt<SocketAddr> for Tcp {
     fn accept(&mut self) -> Acceptor<'_, Self, SocketAddr>
     where

msg-transport/src/tcp_tls/mod.rs

@@ -273,7 +273,6 @@ impl Transport<SocketAddr> for TcpTls {
     }
 }
 
-#[async_trait::async_trait]
 impl TransportExt<SocketAddr> for TcpTls {
     fn accept(&mut self) -> Acceptor<'_, Self, SocketAddr>
     where