chore: Rust to 1.89 and edition 2024 (#2659)

lib/runtime/src/pipeline.rs

@@ -31,8 +31,8 @@ pub use network::egress::push_router::{PushRouter, RouterMode};
 pub mod registry;
 
 pub use crate::engine::{
-    self as engine, async_trait, AsyncEngine, AsyncEngineContext, AsyncEngineContextProvider, Data,
-    DataStream, Engine, EngineStream, EngineUnary, ResponseStream,
+    self as engine, AsyncEngine, AsyncEngineContext, AsyncEngineContextProvider, Data, DataStream,
+    Engine, EngineStream, EngineUnary, ResponseStream, async_trait,
 };
 pub use anyhow::Error;
 pub use context::Context;

lib/runtime/src/pipeline/context.rs

@@ -316,7 +316,7 @@ impl<T: Send + Sync + 'static> From<Context<T>> for StreamContext {
 
 // TODO - refactor here - this came from the dynamo.llm-async-engine crate
 
-use tokio::sync::watch::{channel, Receiver, Sender};
+use tokio::sync::watch::{Receiver, Sender, channel};
 
 #[derive(Debug, Eq, PartialEq)]
 enum State {

lib/runtime/src/pipeline/error.rs

@@ -17,7 +17,7 @@
 //
 use async_nats::error::Error as NatsError;
 
-pub use anyhow::{anyhow, anyhow as error, bail, ensure, Context, Error, Result};
+pub use anyhow::{Context, Error, Result, anyhow, anyhow as error, bail, ensure};
 
 pub trait PipelineErrorExt {
     /// Downcast the [`Error`] to a [`PipelineError`]

lib/runtime/src/pipeline/network.rs

@@ -33,8 +33,8 @@ use super::{AsyncEngine, AsyncEngineContext, AsyncEngineContextProvider, Respons
 use serde::{Deserialize, Serialize};
 
 use super::{
-    context, AsyncTransportEngine, Context, Data, Error, ManyOut, PipelineError, PipelineIO,
-    SegmentSource, ServiceBackend, ServiceEngine, SingleIn, Source,
+    AsyncTransportEngine, Context, Data, Error, ManyOut, PipelineError, PipelineIO, SegmentSource,
+    ServiceBackend, ServiceEngine, SingleIn, Source, context,
 };
 use ingress::push_handler::WorkHandlerMetrics;
 

lib/runtime/src/pipeline/network/codec/two_part.rs

@@ -70,10 +70,10 @@ impl Decoder for TwoPartCodec {
         let total_len = 24 + header_len + body_len;
 
         // Check if total_len exceeds max_message_size
-        if let Some(max_size) = self.max_message_size {
-            if total_len > max_size {
-                return Err(TwoPartCodecError::MessageTooLarge(total_len, max_size));
-            }
+        if let Some(max_size) = self.max_message_size
+            && total_len > max_size
+        {
+            return Err(TwoPartCodecError::MessageTooLarge(total_len, max_size));
         }
 
         // Check if enough data is available
@@ -124,10 +124,10 @@ impl Encoder<TwoPartMessage> for TwoPartCodec {
         let total_len = 24 + header_len + body_len; // 24 bytes for lengths and checksum
 
         // Check if total_len exceeds max_message_size
-        if let Some(max_size) = self.max_message_size {
-            if total_len > max_size {
-                return Err(TwoPartCodecError::MessageTooLarge(total_len, max_size));
-            }
+        if let Some(max_size) = self.max_message_size
+            && total_len > max_size
+        {
+            return Err(TwoPartCodecError::MessageTooLarge(total_len, max_size));
         }
 
         dst.put_u64(header_len as u64);

lib/runtime/src/pipeline/network/egress/addressed_router.rs

@@ -18,10 +18,10 @@ use async_nats::{HeaderMap, HeaderValue};
 use tracing as log;
 
 use super::*;
-use crate::logging::get_distributed_tracing_context;
 use crate::logging::DistributedTraceContext;
-use crate::{protocols::maybe_error::MaybeError, Result};
-use tokio_stream::{wrappers::ReceiverStream, StreamExt, StreamNotifyClose};
+use crate::logging::get_distributed_tracing_context;
+use crate::{Result, protocols::maybe_error::MaybeError};
+use tokio_stream::{StreamExt, StreamNotifyClose, wrappers::ReceiverStream};
 use tracing::Instrument;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]

lib/runtime/src/pipeline/network/egress/push_router.rs

@@ -7,8 +7,8 @@ use crate::{
     component::{Client, Endpoint, InstanceSource},
     engine::{AsyncEngine, Data},
     pipeline::{
-        error::{PipelineError, PipelineErrorExt},
         AddressedPushRouter, AddressedRequest, Error, ManyOut, SingleIn,
+        error::{PipelineError, PipelineErrorExt},
     },
     protocols::maybe_error::MaybeError,
     traits::DistributedRuntimeProvider,
@@ -23,8 +23,8 @@ use std::{
     future::Future,
     marker::PhantomData,
     sync::{
-        atomic::{AtomicU64, Ordering},
         Arc,
+        atomic::{AtomicU64, Ordering},
     },
 };
 use tokio_stream::StreamExt;
@@ -242,10 +242,10 @@ where
                 Ok(ResponseStream::new(Box::pin(stream), engine_ctx))
             }
             Err(err) => {
-                if let Some(req_err) = err.downcast_ref::<NatsRequestError>() {
-                    if matches!(req_err.kind(), NatsNoResponders) {
-                        self.client.report_instance_down(instance_id);
-                    }
+                if let Some(req_err) = err.downcast_ref::<NatsRequestError>()
+                    && matches!(req_err.kind(), NatsNoResponders)
+                {
+                    self.client.report_instance_down(instance_id);
                 }
                 Err(err)
             }

lib/runtime/src/pipeline/network/ingress/push_endpoint.rs

@@ -4,10 +4,10 @@
 use std::sync::atomic::{AtomicU64, Ordering};
 
 use super::*;
+use crate::SystemHealth;
 use crate::config::HealthStatus;
 use crate::logging::TraceParent;
 use crate::protocols::LeaseId;
-use crate::SystemHealth;
 use anyhow::Result;
 use async_nats::service::endpoint::Endpoint;
 use derive_builder::Builder;
@@ -77,7 +77,10 @@ impl PushEndpoint {
             if let Some(req) = req {
                 let response = "".to_string();
                 if let Err(e) = req.respond(Ok(response.into())).await {
-                    tracing::warn!("Failed to respond to request; this may indicate the request has shutdown: {:?}", e);
+                    tracing::warn!(
+                        "Failed to respond to request; this may indicate the request has shutdown: {:?}",
+                        e
+                    );
                 }
 
                 let ingress = self.service_handler.clone();

lib/runtime/src/pipeline/network/ingress/push_handler.rs

@@ -19,8 +19,8 @@ use prometheus::{Histogram, IntCounter, IntCounterVec, IntGauge};
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
 use std::time::Instant;
-use tracing::info_span;
 use tracing::Instrument;
+use tracing::info_span;
 
 /// Metrics configuration for profiling work handlers
 #[derive(Clone, Debug)]
@@ -175,9 +175,9 @@ where
                                 .with_label_values(&["deserialization"])
                                 .inc();
                         }
-                        return Err(PipelineError::DeserializationError(
-                            format!("Failed deserializing to RequestControlMessage. err={err}, json_str={json_str}"),
-                        ));
+                        return Err(PipelineError::DeserializationError(format!(
+                            "Failed deserializing to RequestControlMessage. err={err}, json_str={json_str}"
+                        )));
                     }
                 };
                 let request: T = serde_json::from_slice(&data)?;
@@ -189,7 +189,9 @@ where
                         .with_label_values(&["invalid_message"])
                         .inc();
                 }
-                return Err(PipelineError::Generic(String::from("Unexpected message from work queue; unable extract a TwoPartMessage with a header and data")));
+                return Err(PipelineError::Generic(String::from(
+                    "Unexpected message from work queue; unable extract a TwoPartMessage with a header and data",
+                )));
             }
         };
 

lib/runtime/src/pipeline/network/tcp.rs

@@ -37,8 +37,8 @@ use serde::{Deserialize, Serialize};
 
 #[allow(unused_imports)]
 use super::{
-    codec::TwoPartCodec, ConnectionInfo, PendingConnections, RegisteredStream, ResponseService,
-    StreamOptions, StreamReceiver, StreamSender, StreamType,
+    ConnectionInfo, PendingConnections, RegisteredStream, ResponseService, StreamOptions,
+    StreamReceiver, StreamSender, StreamType, codec::TwoPartCodec,
 };
 
 const TCP_TRANSPORT: &str = "tcp_server";

lib/runtime/src/pipeline/network/tcp/client.rs

@@ -27,11 +27,11 @@ use tokio_util::codec::{FramedRead, FramedWrite};
 use super::{CallHomeHandshake, ControlMessage, TcpStreamConnectionInfo};
 use crate::engine::AsyncEngineContext;
 use crate::pipeline::network::{
+    ConnectionInfo, ResponseStreamPrologue, StreamSender,
     codec::{TwoPartCodec, TwoPartMessage},
     tcp::StreamType,
-    ConnectionInfo, ResponseStreamPrologue, StreamSender,
 };
-use crate::{error, ErrorContext, Result}; // Import SinkExt to use the `send` method
+use crate::{ErrorContext, Result, error}; // Import SinkExt to use the `send` method
 
 #[allow(dead_code)]
 pub struct TcpClient {

lib/runtime/src/pipeline/network/tcp/server.rs

@@ -26,7 +26,7 @@ use tokio::sync::Mutex;
 use bytes::Bytes;
 use derive_builder::Builder;
 use futures::{SinkExt, StreamExt};
-use local_ip_address::{list_afinet_netifas, local_ip, local_ipv6, Error};
+use local_ip_address::{Error, list_afinet_netifas, local_ip, local_ipv6};
 use serde::{Deserialize, Serialize};
 use tokio::{
     io::AsyncWriteExt,
@@ -41,14 +41,14 @@ use super::{
 };
 use crate::engine::AsyncEngineContext;
 use crate::pipeline::{
+    PipelineError,
     network::{
+        ResponseService, ResponseStreamPrologue,
         codec::{TwoPartMessage, TwoPartMessageType},
         tcp::StreamType,
-        ResponseService, ResponseStreamPrologue,
     },
-    PipelineError,
 };
-use crate::{error, ErrorContext, Result};
+use crate::{ErrorContext, Result, error};
 
 #[allow(dead_code)]
 type ResponseType = TwoPartMessage;
@@ -461,7 +461,9 @@ async fn tcp_listener(
             }))
             .is_err()
         {
-            return Err(error!("The requester of the stream has been dropped before the connection was established"));
+            return Err(error!(
+                "The requester of the stream has been dropped before the connection was established"
+            ));
         }
 
         let (control_tx, control_rx) = mpsc::channel::<ControlMessage>(1);
@@ -539,13 +541,12 @@ async fn tcp_listener(
                                 }
                             }
 
-                            if !data.is_empty() {
-                                if let Err(err) = response_tx.send(data).await {
+                            if !data.is_empty()
+                                && let Err(err) = response_tx.send(data).await {
                                     tracing::debug!("forwarding body/data message to response channel failed: {}", err);
                                     control_tx.send(ControlMessage::Kill).await.expect("the control channel should not be closed");
                                     break;
                                 };
-                            }
                         }
                         Some(Err(_)) => {
                             // TODO(#171) - address fatal errors

lib/runtime/src/pipeline/nodes/sinks.rs

@@ -14,7 +14,7 @@
 // limitations under the License.
 
 use super::{
-    async_trait, private::Token, Arc, Edge, OnceLock, PipelineError, Service, Sink, Source,
+    Arc, Edge, OnceLock, PipelineError, Service, Sink, Source, async_trait, private::Token,
 };
 use crate::pipeline::{PipelineIO, ServiceEngine};
 

lib/runtime/src/pipeline/nodes/sources/base.rs

@@ -83,7 +83,7 @@ impl<In: PipelineIO + Sync, Out: PipelineIO> AsyncEngine<In, Out, Error> for Fro
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::pipeline::{error::PipelineErrorExt, ManyOut, SingleIn};
+    use crate::pipeline::{ManyOut, SingleIn, error::PipelineErrorExt};
 
     #[tokio::test]
     async fn test_frontend_no_edge() {

lib/runtime/src/protocols/annotated.rs

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use super::*;
-use crate::{error, Result};
+use crate::{Result, error};
 use maybe_error::MaybeError;
 
 pub trait AnnotationsProvider {
@@ -68,13 +68,13 @@ impl<R> Annotated<R> {
     /// Convert to a [`Result<Self, String>`]
     /// If [`Self::event`] is "error", return an error message(s) held by [`Self::comment`]
     pub fn ok(self) -> Result<Self, String> {
-        if let Some(event) = &self.event {
-            if event == "error" {
-                return Err(self
-                    .comment
-                    .unwrap_or(vec!["unknown error".to_string()])
-                    .join(", "));
-            }
+        if let Some(event) = &self.event
+            && event == "error"
+        {
+            return Err(self
+                .comment
+                .unwrap_or(vec!["unknown error".to_string()])
+                .join(", "));
         }
         Ok(self)
     }
@@ -125,10 +125,11 @@ impl<R> Annotated<R> {
         match self.data {
             Some(data) => Ok(Some(data)),
             None => match self.event {
-                Some(event) if event == "error" => Err(error!(self
-                    .comment
-                    .unwrap_or(vec!["unknown error".to_string()])
-                    .join(", ")))?,
+                Some(event) if event == "error" => Err(error!(
+                    self.comment
+                        .unwrap_or(vec!["unknown error".to_string()])
+                        .join(", ")
+                ))?,
                 _ => Ok(None),
             },
         }
@@ -145,10 +146,10 @@ where
 
     fn err(&self) -> Option<anyhow::Error> {
         if self.is_error() {
-            if let Some(comment) = &self.comment {
-                if !comment.is_empty() {
-                    return Some(anyhow::Error::msg(comment.join("; ")));
-                }
+            if let Some(comment) = &self.comment
+                && !comment.is_empty()
+            {
+                return Some(anyhow::Error::msg(comment.join("; ")));
             }
             Some(anyhow::Error::msg("unknown error"))
         } else {

lib/runtime/src/runtime.rs

@@ -25,7 +25,7 @@
 //! Notes: We will need to do an evaluation on what is fully public, what is pub(crate) and what is
 //! private; however, for now we are exposing most objects as fully public while the API is maturing.
 
-use super::{error, Result, Runtime, RuntimeType};
+use super::{Result, Runtime, RuntimeType, error};
 use crate::config::{self, RuntimeConfig};
 
 use futures::Future;

lib/runtime/src/service.rs

@@ -20,13 +20,13 @@
 // component's "service state"
 
 use crate::{
+    DistributedRuntime, Result,
     component::Component,
     error,
-    metrics::{prometheus_names, prometheus_names::nats_service, MetricsRegistry},
+    metrics::{MetricsRegistry, prometheus_names, prometheus_names::nats_service},
     traits::*,
     transports::nats,
     utils::stream,
-    DistributedRuntime, Result,
 };
 
 use async_nats::Message;
@@ -35,7 +35,7 @@ use bytes::Bytes;
 use derive_getters::Dissolve;
 use futures::stream::{StreamExt, TryStreamExt};
 use prometheus;
-use serde::{de::DeserializeOwned, Deserialize, Serialize};
+use serde::{Deserialize, Serialize, de::DeserializeOwned};
 use std::time::Duration;
 
 pub struct ServiceClient {

lib/runtime/src/slug.rs

@@ -43,11 +43,7 @@ impl Slug {
             .chars()
             .map(|c| {
                 let is_valid = c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-' || c == '_';
-                if is_valid {
-                    c
-                } else {
-                    REPLACEMENT_CHAR
-                }
+                if is_valid { c } else { REPLACEMENT_CHAR }
             })
             .collect::<String>();
         Slug::new(out)
@@ -61,11 +57,7 @@ impl Slug {
             .chars()
             .map(|c| {
                 let is_valid = c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_';
-                if is_valid {
-                    c
-                } else {
-                    REPLACEMENT_CHAR
-                }
+                if is_valid { c } else { REPLACEMENT_CHAR }
             })
             .collect::<String>();
         let hash = blake3::hash(s.as_bytes()).to_string();

lib/runtime/src/storage/key_value_store.rs

@@ -22,8 +22,8 @@ use std::pin::Pin;
 use std::sync::Arc;
 use std::time::Duration;
 
-use crate::slug::Slug;
 use crate::CancellationToken;
+use crate::slug::Slug;
 use async_trait::async_trait;
 use futures::StreamExt;
 use serde::{Deserialize, Serialize};
@@ -278,7 +278,7 @@ mod tests {
     use std::sync::Arc;
 
     use super::*;
-    use futures::{pin_mut, StreamExt};
+    use futures::{StreamExt, pin_mut};
 
     const BUCKET_NAME: &str = "mdc";
 

lib/runtime/src/storage/key_value_store/etcd.rs

@@ -186,11 +186,10 @@ impl EtcdBucket {
         // Key already existed, get its version
         if let Some(etcd_client::TxnOpResponse::Get(get_resp)) =
             result.op_responses().into_iter().next()
+            && let Some(kv) = get_resp.kvs().first()
         {
-            if let Some(kv) = get_resp.kvs().first() {
-                let version = kv.version() as u64;
-                return Ok(StorageOutcome::Exists(version));
-            }
+            let version = kv.version() as u64;
+            return Ok(StorageOutcome::Exists(version));
         }
         // Shouldn't happen, but handle edge case
         Err(StorageError::EtcdError(
@@ -259,7 +258,7 @@ fn make_key(bucket_name: &str, key: &str) -> String {
 #[cfg(test)]
 mod concurrent_create_tests {
     use super::*;
-    use crate::{distributed::DistributedConfig, DistributedRuntime, Runtime};
+    use crate::{DistributedRuntime, Runtime, distributed::DistributedConfig};
     use std::sync::Arc;
     use tokio::sync::Barrier;