chore: Rust 1.86 Edition 2024 upgrade (#2581)

Signed-off-by: Sreekanth <[email protected]>

Author: BulkBeing

Dockerfile

@@ -22,7 +22,7 @@ RUN chmod +x /bin/entrypoint
 ####################################################################################################
 # Rust binary
 ####################################################################################################
-FROM lukemathwalker/cargo-chef:latest-rust-1.84 AS chef
+FROM lukemathwalker/cargo-chef:latest-rust-1.86 AS chef
 ARG TARGETPLATFORM
 WORKDIR /numaflow
 RUN apt-get update && apt-get install -y protobuf-compiler

rust/.rustfmt.toml

@@ -1 +1 @@
-edition = "2021"
+edition = "2024"

rust/backoff/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "backoff"
 version = "0.1.0"
-edition = "2021"
+edition = "2024"
 
 [lints]
 workspace = true

rust/backoff/src/retry.rs

@@ -4,7 +4,7 @@ use std::task::{Context, Poll};
 use std::time::Duration;
 
 use pin_project::pin_project;
-use tokio::time::{sleep_until, Instant, Sleep};
+use tokio::time::{Instant, Sleep, sleep_until};
 
 use crate::{Condition, Operation};
 
@@ -144,8 +144,8 @@ where
 #[cfg(test)]
 mod tests {
     use std::future;
-    use std::sync::atomic::{AtomicUsize, Ordering};
     use std::sync::Arc;
+    use std::sync::atomic::{AtomicUsize, Ordering};
 
     use super::*;
     use crate::strategy::fixed;

rust/extns/numaflow-jetstream/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "numaflow-jetstream"
 version = "0.1.0"
-edition = "2021"
+edition = "2024"
 
 [dependencies]
 backoff.workspace = true

rust/extns/numaflow-jetstream/src/lib.rs

@@ -3,11 +3,11 @@ use std::{collections::HashMap, time::Duration};
 
 use async_nats::jetstream::{AckKind, Message as JetstreamMessage};
 use async_nats::{
+    ConnectOptions,
     jetstream::consumer::{
-        pull::{Config, Stream},
         Consumer, PullConsumer,
+        pull::{Config, Stream},
     },
-    ConnectOptions,
 };
 use backoff::retry::Retry;
 use backoff::strategy::fixed;
@@ -185,7 +185,9 @@ impl JetstreamActor {
         tls_config: TlsConfig,
     ) -> Result<ConnectOptions> {
         if tls_config.insecure_skip_verify {
-            tracing::warn!("'insecureSkipVerify' is set to true, certificate validation will not be performed when connecting to NATS server");
+            tracing::warn!(
+                "'insecureSkipVerify' is set to true, certificate validation will not be performed when connecting to NATS server"
+            );
             let tls_client_config = rustls::ClientConfig::builder()
                 .dangerous()
                 .with_custom_certificate_verifier(Arc::new(NoVerifier))
@@ -447,7 +449,7 @@ impl MessageProcessingTracker {
         let nack_retry_interval =
             fixed::Interval::from_millis(ACK_RETRY_INTERVAL).take(ACK_RETRY_ATTEMPTS);
 
-        let ack_msg = || async {
+        let ack_msg = async || {
             if let Err(err) = msg.ack().await {
                 tracing::error!(?err, "Failed to Ack message");
                 return Err(format!("Acknowledging Jetstream message: {:?}", err));
@@ -457,14 +459,14 @@ impl MessageProcessingTracker {
 
         let ack_with_retry = Retry::retry(ack_retry_interval, ack_msg, |_: &String| true);
 
-        let ack_in_progress = || async {
+        let ack_in_progress = async || {
             let ack_result = msg.ack_with(AckKind::Progress).await;
             if let Err(e) = ack_result {
                 tracing::error!(?e, "Failed to send InProgress Ack to Jetstream for message");
             }
         };
 
-        let nack_msg = || async {
+        let nack_msg = async || {
             let ack_result = msg.ack_with(AckKind::Nak(None)).await;
             if let Err(e) = ack_result {
                 tracing::error!(?e, "Failed to send InProgress Ack to Jetstream for message");
@@ -660,8 +662,8 @@ XdvExDsAdjbkBG7ynn9pmMgIJg==
         assert!(result.is_ok());
     }
 
-    use async_nats::jetstream::stream::Config as StreamConfig;
     use async_nats::jetstream::Context;
+    use async_nats::jetstream::stream::Config as StreamConfig;
     use tokio::time::Duration;
 
     async fn setup_jetstream() -> (Context, String) {

rust/extns/numaflow-pulsar/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "numaflow-pulsar"
 version = "0.1.0"
-edition = "2021"
+edition = "2024"
 
 [lints]
 workspace = true

rust/extns/numaflow-pulsar/src/source.rs

@@ -4,7 +4,7 @@ use std::{collections::HashMap, time::Duration};
 use bytes::Bytes;
 use chrono::{DateTime, Utc};
 use pulsar::Authentication;
-use pulsar::{proto::MessageIdData, Consumer, ConsumerOptions, Pulsar, SubType, TokioExecutor};
+use pulsar::{Consumer, ConsumerOptions, Pulsar, SubType, TokioExecutor, proto::MessageIdData};
 use tokio::time::Instant;
 use tokio::{
     sync::{mpsc, oneshot},

rust/extns/numaflow-sqs/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "numaflow-sqs"
 version = "0.1.0"
-edition = "2021"
+edition = "2024"
 
 [lints]
 workspace = true

rust/extns/numaflow-sqs/src/lib.rs

@@ -37,7 +37,7 @@ impl From<String> for Error {
 #[cfg(test)]
 mod tests {
     use aws_config::BehaviorVersion;
-    use aws_smithy_mocks_experimental::{mock, MockResponseInterceptor, RuleMode};
+    use aws_smithy_mocks_experimental::{MockResponseInterceptor, RuleMode, mock};
     use aws_smithy_types::error::ErrorMetadata;
 
     use super::*;
@@ -66,9 +66,11 @@ mod tests {
 
         let converted_error = Error::Sqs(err.into());
         assert!(matches!(converted_error, Error::Sqs(_)));
-        assert!(converted_error
-            .to_string()
-            .contains("Failed with SQS error"));
+        assert!(
+            converted_error
+                .to_string()
+                .contains("Failed with SQS error")
+        );
     }
 
     #[test]