Merge pull request #89 from chainbound/nico/chore/maintenance

merklefruit · web-flow · commit 069120be12e0 · 2025-10-03T09:41:54.000+02:00
chore: update crates, workspace edition 2024, MSRV, fmt
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -11,14 +11,28 @@ resolver = "2"
 
 [workspace.package]
 version = "0.1.3"
-edition = "2021"
-rust-version = "1.75"
+edition = "2024"
+rust-version = "1.85"
 license = "MIT"
 description = "A flexible and lightweight messaging library for distributed systems"
 authors = ["Jonas Bostoen", "Nicolas Racchi"]
 homepage = "https://github.com/chainbound/msg-rs"
 repository = "https://github.com/chainbound/msg-rs"
-keywords = ["messaging", "distributed", "systems", "networking", "quic", "quinn", "tokio", "async", "simulation", "pnet", "udp", "tcp", "socket"]
+keywords = [
+    "messaging",
+    "distributed",
+    "systems",
+    "networking",
+    "quic",
+    "quinn",
+    "tokio",
+    "async",
+    "simulation",
+    "pnet",
+    "udp",
+    "tcp",
+    "socket",
+]
 
 [workspace.dependencies]
 msg-wire = { path = "./msg-wire" }
@@ -70,7 +84,10 @@ codegen-units = 1
 opt-level = 3
 
 [profile.debug-maxperf]
-inherits = "maxperf"
+inherits = "release"
+lto = true
+codegen-units = 1
+opt-level = 3
 debug = true
 
 [workspace.metadata.docs.rs]
diff --git a/README.md b/README.md
@@ -52,7 +52,7 @@ The 📖 [MSG-RS Book][book] contains detailed information on how to use the lib
 
 ## MSRV
 
-The minimum supported Rust version is 1.75.
+The minimum supported Rust version is 1.85 (in accordance with Edition 2024 crates).
 
 ## Contributions & Bug Reports
 
diff --git a/msg-common/src/channel.rs b/msg-common/src/channel.rs
@@ -5,9 +5,8 @@ use std::{
 
 use futures::{Sink, SinkExt, Stream};
 use tokio::sync::mpsc::{
-    self,
+    self, Receiver,
     error::{TryRecvError, TrySendError},
-    Receiver,
 };
 use tokio_util::sync::{PollSendError, PollSender};
 
diff --git a/msg-common/src/lib.rs b/msg-common/src/lib.rs
@@ -9,7 +9,7 @@ use std::time::SystemTime;
 use futures::future::BoxFuture;
 
 mod channel;
-pub use channel::{channel, Channel};
+pub use channel::{Channel, channel};
 
 mod task;
 pub use task::JoinMap;
diff --git a/msg-common/src/task.rs b/msg-common/src/task.rs
@@ -1,7 +1,7 @@
-use futures::{future::poll_fn, Future};
+use futures::{Future, future::poll_fn};
 use std::{
     collections::HashSet,
-    task::{ready, Context, Poll},
+    task::{Context, Poll, ready},
 };
 use tokio::task::{JoinError, JoinSet};
 
diff --git a/msg-socket/src/pub/driver.rs b/msg-socket/src/pub/driver.rs
@@ -5,13 +5,13 @@ use std::{
     task::{Context, Poll},
 };
 
-use futures::{stream::FuturesUnordered, Future, SinkExt, StreamExt};
+use futures::{Future, SinkExt, StreamExt, stream::FuturesUnordered};
 use tokio::{sync::broadcast, task::JoinSet};
 use tokio_util::codec::Framed;
 use tracing::{debug, error, info, warn};
 
 use super::{
-    session::SubscriberSession, trie::PrefixTrie, PubError, PubMessage, PubOptions, SocketState,
+    PubError, PubMessage, PubOptions, SocketState, session::SubscriberSession, trie::PrefixTrie,
 };
 use crate::{AuthResult, Authenticator};
 use msg_transport::{Address, PeerAddress, Transport};
diff --git a/msg-socket/src/pub/mod.rs b/msg-socket/src/pub/mod.rs
@@ -319,7 +319,9 @@ mod tests {
         sub2.subscribe("HELLO".to_string()).await.unwrap();
         tokio::time::sleep(Duration::from_millis(100)).await;
 
-        let original_msg = Bytes::from("WOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOORLD");
+        let original_msg = Bytes::from(
+            "WOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOORLD",
+        );
 
         pub_socket.publish("HELLO".to_string(), original_msg.clone()).await.unwrap();
 
diff --git a/msg-socket/src/pub/session.rs b/msg-socket/src/pub/session.rs
@@ -11,7 +11,7 @@ use tokio_stream::wrappers::BroadcastStream;
 use tokio_util::codec::Framed;
 use tracing::{debug, error, trace, warn};
 
-use super::{trie::PrefixTrie, PubMessage, SocketState};
+use super::{PubMessage, SocketState, trie::PrefixTrie};
 use msg_wire::pubsub;
 
 /// A subscriber session. This struct represents a single subscriber session, which is a
@@ -112,7 +112,7 @@ enum ControlMsg<'a> {
 /// Converts the message to a control message. If the message is not a control message,
 /// the session is closed.
 #[inline]
-fn msg_to_control(msg: &pubsub::Message) -> ControlMsg {
+fn msg_to_control(msg: &pubsub::Message) -> ControlMsg<'_> {
     if msg.payload_size() == 0 {
         if msg.topic().starts_with(b"MSG.SUB.") {
             let topic = msg.topic().strip_prefix(b"MSG.SUB.").unwrap();
diff --git a/msg-socket/src/pub/socket.rs b/msg-socket/src/pub/socket.rs
@@ -3,13 +3,13 @@ use std::{net::SocketAddr, path::PathBuf, sync::Arc};
 use bytes::Bytes;
 use futures::stream::FuturesUnordered;
 use tokio::{
-    net::{lookup_host, ToSocketAddrs},
+    net::{ToSocketAddrs, lookup_host},
     sync::broadcast,
     task::JoinSet,
 };
 use tracing::{debug, trace, warn};
 
-use super::{driver::PubDriver, stats::PubStats, PubError, PubMessage, PubOptions, SocketState};
+use super::{PubError, PubMessage, PubOptions, SocketState, driver::PubDriver, stats::PubStats};
 use crate::Authenticator;
 
 use msg_transport::{Address, Transport};
diff --git a/msg-socket/src/rep/driver.rs b/msg-socket/src/rep/driver.rs
@@ -7,7 +7,7 @@ use std::{
 };
 
 use bytes::Bytes;
-use futures::{stream::FuturesUnordered, Future, FutureExt, SinkExt, Stream, StreamExt};
+use futures::{Future, FutureExt, SinkExt, Stream, StreamExt, stream::FuturesUnordered};
 use tokio::{
     io::{AsyncRead, AsyncWrite},
     sync::{mpsc, oneshot},
@@ -17,12 +17,12 @@ use tokio_stream::{StreamMap, StreamNotifyClose};
 use tokio_util::codec::Framed;
 use tracing::{debug, error, info, trace, warn};
 
-use crate::{rep::SocketState, AuthResult, Authenticator, RepOptions, Request};
+use crate::{AuthResult, Authenticator, RepOptions, Request, rep::SocketState};
 
 use msg_transport::{Address, PeerAddress, Transport};
 use msg_wire::{
     auth,
-    compression::{try_decompress_payload, Compressor},
+    compression::{Compressor, try_decompress_payload},
     reqrep,
 };
 
diff --git a/msg-socket/src/rep/mod.rs b/msg-socket/src/rep/mod.rs
@@ -98,7 +98,7 @@ mod tests {
     use rand::Rng;
     use tracing::{debug, info};
 
-    use crate::{req::ReqSocket, Authenticator, ReqOptions};
+    use crate::{Authenticator, ReqOptions, req::ReqSocket};
 
     use super::*;
 
diff --git a/msg-socket/src/rep/socket.rs b/msg-socket/src/rep/socket.rs
@@ -6,18 +6,18 @@ use std::{
     task::{Context, Poll},
 };
 
-use futures::{stream::FuturesUnordered, Stream};
+use futures::{Stream, stream::FuturesUnordered};
 use tokio::{
-    net::{lookup_host, ToSocketAddrs},
+    net::{ToSocketAddrs, lookup_host},
     sync::mpsc,
     task::JoinSet,
 };
 use tokio_stream::StreamMap;
 use tracing::{debug, warn};
 
 use crate::{
-    rep::{driver::RepDriver, RepError, SocketState},
-    Authenticator, RepOptions, Request, DEFAULT_BUFFER_SIZE,
+    Authenticator, DEFAULT_BUFFER_SIZE, RepOptions, Request,
+    rep::{RepError, SocketState, driver::RepDriver},
 };
 
 use msg_transport::{Address, Transport};
diff --git a/msg-socket/src/req/driver.rs b/msg-socket/src/req/driver.rs
@@ -3,7 +3,7 @@ use std::{
     io,
     pin::Pin,
     sync::Arc,
-    task::{ready, Context, Poll},
+    task::{Context, Poll, ready},
     time::{Duration, Instant},
 };
 
@@ -18,12 +18,12 @@ use tokio_util::codec::Framed;
 use tracing::{debug, error, trace};
 
 use super::{Command, ReqError, ReqOptions};
-use crate::{req::SocketState, ConnectionState, ExponentialBackoff};
+use crate::{ConnectionState, ExponentialBackoff, req::SocketState};
 
 use msg_transport::{Address, Transport};
 use msg_wire::{
     auth,
-    compression::{try_decompress_payload, Compressor},
+    compression::{Compressor, try_decompress_payload},
     reqrep,
 };
 
diff --git a/msg-socket/src/req/socket.rs b/msg-socket/src/req/socket.rs
@@ -2,17 +2,17 @@ use bytes::Bytes;
 use rustc_hash::FxHashMap;
 use std::{marker::PhantomData, net::SocketAddr, path::PathBuf, sync::Arc, time::Duration};
 use tokio::{
-    net::{lookup_host, ToSocketAddrs},
+    net::{ToSocketAddrs, lookup_host},
     sync::{mpsc, oneshot},
 };
 
 use msg_transport::{Address, Transport};
 use msg_wire::compression::Compressor;
 
-use super::{Command, ReqError, ReqOptions, DEFAULT_BUFFER_SIZE};
+use super::{Command, DEFAULT_BUFFER_SIZE, ReqError, ReqOptions};
 use crate::{
-    req::{driver::ReqDriver, stats::ReqStats, SocketState},
     ConnectionState, ExponentialBackoff, ReqMessage,
+    req::{SocketState, driver::ReqDriver, stats::ReqStats},
 };
 
 /// The request socket.
diff --git a/msg-socket/src/sub/driver.rs b/msg-socket/src/sub/driver.rs
@@ -13,13 +13,13 @@ use tokio_util::codec::Framed;
 use tracing::{debug, error, info, warn};
 
 use super::{
+    Command, PubMessage, SocketState, SubOptions,
     session::{PublisherSession, SessionCommand},
     stream::{PublisherStream, TopicMessage},
-    Command, PubMessage, SocketState, SubOptions,
 };
 use crate::{ConnectionState, ExponentialBackoff};
 
-use msg_common::{channel, Channel, JoinMap};
+use msg_common::{Channel, JoinMap, channel};
 use msg_transport::{Address, Transport};
 use msg_wire::{auth, compression::try_decompress_payload, pubsub};
 
@@ -370,7 +370,9 @@ where
                                     );
                                 }
                                 Err(TrySendError::Closed(_)) => {
-                                    error!("SubSocket frontend channel closed unexpectedly while driver is active.");
+                                    error!(
+                                        "SubSocket frontend channel closed unexpectedly while driver is active."
+                                    );
                                     // Consider shutting down or marking as inactive?
                                     // For now, just log. No counter increment.
                                 }
@@ -401,7 +403,10 @@ where
                                 debug!(backoff = ?duration, "Not retrying connection to {:?} as there is already a connection task", addr);
                             }
                         } else {
-                            error!("Exceeded maximum number of retries for {:?}, terminating connection", addr);
+                            error!(
+                                "Exceeded maximum number of retries for {:?}, terminating connection",
+                                addr
+                            );
                             to_terminate.push(addr.clone());
                         }
                     }
@@ -424,10 +429,6 @@ where
             self.publishers.remove(&addr);
         }
 
-        if progress {
-            Poll::Ready(())
-        } else {
-            Poll::Pending
-        }
+        if progress { Poll::Ready(()) } else { Poll::Pending }
     }
 }
diff --git a/msg-socket/src/sub/mod.rs b/msg-socket/src/sub/mod.rs
@@ -179,7 +179,7 @@ mod tests {
         net::TcpListener,
     };
     use tokio_stream::StreamExt;
-    use tracing::{info, info_span, Instrument};
+    use tracing::{Instrument, info, info_span};
 
     use super::*;
 
diff --git a/msg-socket/src/sub/session.rs b/msg-socket/src/sub/session.rs
@@ -10,7 +10,7 @@ use futures::{Future, StreamExt};
 use tokio::io::{AsyncRead, AsyncWrite};
 use tracing::{debug, error, warn};
 
-use msg_common::{unix_micros, Channel};
+use msg_common::{Channel, unix_micros};
 use msg_transport::Address;
 use msg_wire::pubsub;
 
diff --git a/msg-socket/src/sub/socket.rs b/msg-socket/src/sub/socket.rs
@@ -10,7 +10,7 @@ use std::{
 use futures::Stream;
 use rustc_hash::FxHashMap;
 use tokio::{
-    net::{lookup_host, ToSocketAddrs},
+    net::{ToSocketAddrs, lookup_host},
     sync::mpsc,
 };
 
@@ -24,12 +24,12 @@ use super::{
     // REMOVED: Old/removed stats structs
     // Command, PubMessage, SocketState, SocketStats, SocketWideStats, SubDriver, SubError,
     Command,
+    DEFAULT_BUFFER_SIZE,
     PubMessage,
     SocketState,
     SubDriver,
     SubError,
     SubOptions,
-    DEFAULT_BUFFER_SIZE,
 };
 
 /// A subscriber socket. This socket implements [`Stream`] and yields incoming [`PubMessage`]s.
diff --git a/msg-socket/src/sub/stats.rs b/msg-socket/src/sub/stats.rs
@@ -1,8 +1,8 @@
 use std::{
     collections::HashMap,
     sync::{
-        atomic::{AtomicU64, AtomicUsize, Ordering},
         Arc,
+        atomic::{AtomicU64, AtomicUsize, Ordering},
     },
 };
 
diff --git a/msg-socket/src/sub/stream.rs b/msg-socket/src/sub/stream.rs
@@ -1,6 +1,6 @@
 use std::{
     pin::Pin,
-    task::{ready, Context, Poll},
+    task::{Context, Poll, ready},
 };
 
 use bytes::Bytes;
diff --git a/msg-socket/tests/it/pubsub.rs b/msg-socket/tests/it/pubsub.rs
@@ -8,7 +8,7 @@ use tokio_stream::StreamExt;
 use tracing::info;
 
 use msg_socket::{PubSocket, SubSocket};
-use msg_transport::{quic::Quic, tcp::Tcp, Address, Transport};
+use msg_transport::{Address, Transport, quic::Quic, tcp::Tcp};
 
 const TOPIC: &str = "test";
 
diff --git a/msg-transport/src/quic/config.rs b/msg-transport/src/quic/config.rs
@@ -1,6 +1,6 @@
 use std::{sync::Arc, time::Duration};
 
-use quinn::{congestion::ControllerFactory, IdleTimeout};
+use quinn::{IdleTimeout, congestion::ControllerFactory};
 
 use super::tls::{self_signed_certificate, unsafe_client_config};
 
diff --git a/msg-transport/src/quic/mod.rs b/msg-transport/src/quic/mod.rs
@@ -3,7 +3,7 @@ use std::{
     net::{SocketAddr, UdpSocket},
     pin::Pin,
     sync::Arc,
-    task::{ready, Poll},
+    task::{Poll, ready},
 };
 
 use futures::future::BoxFuture;
@@ -198,7 +198,9 @@ impl Transport<SocketAddr> for Quic {
                             endpoint.accept().await.ok_or(Error::ClosedEndpoint);
 
                         if tx.send(connection_result).await.is_err() {
-                            error!("Failed to notify new incoming connection, channel closed. Shutting down task.");
+                            error!(
+                                "Failed to notify new incoming connection, channel closed. Shutting down task."
+                            );
                             break;
                         };
                     }
diff --git a/msg-wire/src/compression/gzip.rs b/msg-wire/src/compression/gzip.rs
@@ -1,5 +1,5 @@
 use bytes::Bytes;
-use flate2::{read::GzDecoder, write::GzEncoder, Compression};
+use flate2::{Compression, read::GzDecoder, write::GzEncoder};
 use std::io::{self, Read, Write};
 
 use super::{CompressionType, Compressor, Decompressor};
diff --git a/msg/benches/pubsub.rs b/msg/benches/pubsub.rs
diff --git a/msg/benches/reqrep.rs b/msg/benches/reqrep.rs
diff --git a/msg/examples/durable.rs b/msg/examples/durable.rs
diff --git a/msg/examples/ipc.rs b/msg/examples/ipc.rs
diff --git a/msg/examples/pubsub.rs b/msg/examples/pubsub.rs
diff --git a/msg/examples/pubsub_auth.rs b/msg/examples/pubsub_auth.rs
diff --git a/msg/examples/pubsub_compression.rs b/msg/examples/pubsub_compression.rs
diff --git a/msg/examples/quic_vs_tcp.rs b/msg/examples/quic_vs_tcp.rs
diff --git a/msg/examples/reqrep.rs b/msg/examples/reqrep.rs
diff --git a/msg/examples/reqrep_auth.rs b/msg/examples/reqrep_auth.rs
diff --git a/msg/examples/reqrep_compression.rs b/msg/examples/reqrep_compression.rs
