chore(deps): Upgrade rust to 1.70.0 (#17585)

lib/vector-buffers/src/variants/disk_v2/tests/model/sequencer.rs

@@ -25,11 +25,9 @@ impl<T> TrackedFuture<T> {
     where
         F: Future<Output = T> + Send + 'static,
     {
-        let wrapped = async move { fut.await };
-
         Self {
             polled_once: false,
-            fut: spawn(wrapped.boxed()),
+            fut: spawn(fut.boxed()),
         }
     }
 

lib/vector-config/Cargo.toml

@@ -18,7 +18,6 @@ indexmap = { version = "1.9", default-features = false }
 inventory = { version = "0.3" }
 no-proxy = { version  = "0.3.1", default-features = false, features = ["serialize"] }
 num-traits = { version = "0.2.15", default-features = false }
-once_cell = { version = "1", default-features = false }
 serde = { version = "1.0", default-features = false }
 serde_json = { version = "1.0", default-features = false, features = ["std"] }
 serde_with = { version = "2.3.2", default-features = false, features = ["std"] }

lib/vector-config/src/schema/parser/query.rs

@@ -1,6 +1,5 @@
-use std::{fs::File, io::BufReader, path::Path};
+use std::{fs::File, io::BufReader, path::Path, sync::OnceLock};
 
-use once_cell::sync::OnceCell;
 use serde_json::Value;
 use snafu::Snafu;
 use vector_config_common::{
@@ -415,8 +414,8 @@ impl<'a> QueryableSchema for SimpleSchema<'a> {
 }
 
 fn schema_to_simple_schema(schema: &Schema) -> SimpleSchema<'_> {
-    static TRUE_SCHEMA_OBJECT: OnceCell<SchemaObject> = OnceCell::new();
-    static FALSE_SCHEMA_OBJECT: OnceCell<SchemaObject> = OnceCell::new();
+    static TRUE_SCHEMA_OBJECT: OnceLock<SchemaObject> = OnceLock::new();
+    static FALSE_SCHEMA_OBJECT: OnceLock<SchemaObject> = OnceLock::new();
 
     let schema_object = match schema {
         Schema::Bool(bool) => {

lib/vector-core/src/config/log_schema.rs

@@ -1,9 +1,11 @@
+use std::sync::OnceLock;
+
 use lookup::lookup_v2::{parse_target_path, OptionalValuePath};
 use lookup::{owned_value_path, OwnedTargetPath, OwnedValuePath};
-use once_cell::sync::{Lazy, OnceCell};
+use once_cell::sync::Lazy;
 use vector_config::configurable_component;
 
-static LOG_SCHEMA: OnceCell<LogSchema> = OnceCell::new();
+static LOG_SCHEMA: OnceLock<LogSchema> = OnceLock::new();
 static LOG_SCHEMA_DEFAULT: Lazy<LogSchema> = Lazy::new(LogSchema::default);
 
 /// Loads Log Schema from configurations and sets global schema. Once this is

lib/vector-core/src/metrics/mod.rs

@@ -4,13 +4,12 @@ mod recency;
 mod recorder;
 mod storage;
 
-use std::time::Duration;
+use std::{sync::OnceLock, time::Duration};
 
 use chrono::Utc;
 use metrics::Key;
 use metrics_tracing_context::TracingContextLayer;
 use metrics_util::layers::Layer;
-use once_cell::sync::OnceCell;
 use snafu::Snafu;
 
 pub use self::ddsketch::{AgentDDSketch, BinMap, Config};
@@ -29,7 +28,7 @@ pub enum Error {
     TimeoutMustBePositive { timeout: f64 },
 }
 
-static CONTROLLER: OnceCell<Controller> = OnceCell::new();
+static CONTROLLER: OnceLock<Controller> = OnceLock::new();
 
 // Cardinality counter parameters, expose the internal metrics registry
 // cardinality. Useful for the end users to help understand the characteristics

lib/vector-core/src/metrics/recorder.rs

@@ -1,10 +1,9 @@
 use std::sync::{atomic::Ordering, Arc, RwLock};
-use std::time::Duration;
+use std::{cell::OnceCell, time::Duration};
 
 use chrono::Utc;
 use metrics::{Counter, Gauge, Histogram, Key, KeyName, Recorder, SharedString, Unit};
 use metrics_util::{registry::Registry as MetricsRegistry, MetricKindMask};
-use once_cell::unsync::OnceCell;
 use quanta::Clock;
 
 use super::recency::{GenerationalStorage, Recency};

rust-toolchain.toml

@@ -1,3 +1,3 @@
 [toolchain]
-channel = "1.69.0"
+channel = "1.70.0"
 profile = "default"

src/aws/mod.rs

@@ -5,7 +5,7 @@ pub mod region;
 use std::future::Future;
 use std::pin::Pin;
 use std::sync::atomic::{AtomicUsize, Ordering};
-use std::sync::Arc;
+use std::sync::{Arc, OnceLock};
 use std::task::{Context, Poll};
 use std::time::SystemTime;
 
@@ -26,7 +26,6 @@ use aws_types::credentials::{ProvideCredentials, SharedCredentialsProvider};
 use aws_types::region::Region;
 use aws_types::SdkConfig;
 use bytes::Bytes;
-use once_cell::sync::OnceCell;
 use regex::RegexSet;
 pub use region::RegionOrEndpoint;
 use tower::{Layer, Service, ServiceBuilder};
@@ -36,7 +35,7 @@ use crate::http::{build_proxy_connector, build_tls_connector};
 use crate::internal_events::AwsBytesSent;
 use crate::tls::{MaybeTlsSettings, TlsConfig};
 
-static RETRIABLE_CODES: OnceCell<RegexSet> = OnceCell::new();
+static RETRIABLE_CODES: OnceLock<RegexSet> = OnceLock::new();
 
 pub fn is_retriable_error<T>(error: &SdkError<T>) -> bool {
     match error {

src/sinks/util/service.rs

@@ -357,7 +357,6 @@ impl TowerRequestSettings {
         S::Future: Send + 'static,
     {
         let policy = self.retry_policy(retry_logic.clone());
-        let settings = self.clone();
 
         // Build services
         let open = OpenGauge::new();
@@ -368,16 +367,14 @@ impl TowerRequestSettings {
                 // Build individual service
                 ServiceBuilder::new()
                     .layer(AdaptiveConcurrencyLimitLayer::new(
-                        settings.concurrency,
-                        settings.adaptive_concurrency,
+                        self.concurrency,
+                        self.adaptive_concurrency,
                         retry_logic.clone(),
                     ))
                     .service(
                         health_config.build(
                             health_logic.clone(),
-                            ServiceBuilder::new()
-                                .timeout(settings.timeout)
-                                .service(inner),
+                            ServiceBuilder::new().timeout(self.timeout).service(inner),
                             open.clone(),
                             endpoint,
                         ), // NOTE: there is a version conflict for crate `tracing` between `tracing_tower` crate

src/sources/kafka.rs

@@ -1,7 +1,7 @@
 use std::{
     collections::{BTreeMap, HashMap},
     io::Cursor,
-    sync::Arc,
+    sync::{Arc, OnceLock},
     time::Duration,
 };
 
@@ -14,7 +14,6 @@ use codecs::{
 };
 use futures::{Stream, StreamExt};
 use lookup::{lookup_v2::OptionalValuePath, owned_value_path, path, OwnedValuePath};
-use once_cell::sync::OnceCell;
 use rdkafka::{
     consumer::{CommitMode, Consumer, ConsumerContext, Rebalance, StreamConsumer},
     message::{BorrowedMessage, Headers as _, Message},
@@ -724,7 +723,7 @@ fn create_consumer(config: &KafkaSourceConfig) -> crate::Result<StreamConsumer<C
 #[derive(Default)]
 struct CustomContext {
     stats: kafka::KafkaStatisticsContext,
-    finalizer: OnceCell<Arc<OrderedFinalizer<FinalizerEntry>>>,
+    finalizer: OnceLock<Arc<OrderedFinalizer<FinalizerEntry>>>,
 }
 
 impl CustomContext {

src/trace.rs

@@ -5,14 +5,13 @@ use std::{
     str::FromStr,
     sync::{
         atomic::{AtomicBool, Ordering},
-        Mutex, MutexGuard,
+        Mutex, MutexGuard, OnceLock,
     },
 };
 
 use futures_util::{future::ready, Stream, StreamExt};
 use lookup::event_path;
 use metrics_tracing_context::MetricsLayer;
-use once_cell::sync::OnceCell;
 use tokio::sync::{
     broadcast::{self, Receiver, Sender},
     oneshot,
@@ -51,7 +50,7 @@ static SUBSCRIBERS: Mutex<Option<Vec<oneshot::Sender<Vec<LogEvent>>>>> =
 
 /// SENDER holds the sender/receiver handle that will receive a copy of all the internal log events *after* the topology
 /// has been initialized.
-static SENDER: OnceCell<Sender<LogEvent>> = OnceCell::new();
+static SENDER: OnceLock<Sender<LogEvent>> = OnceLock::new();
 
 fn metrics_layer_enabled() -> bool {
     !matches!(std::env::var("DISABLE_INTERNAL_METRICS_TRACING_INTEGRATION"), Ok(x) if x == "true")

vdev/src/app.rs

@@ -1,13 +1,14 @@
 use std::ffi::{OsStr, OsString};
 pub use std::process::Command;
 use std::{
-    borrow::Cow, env, io::Read, path::PathBuf, process::ExitStatus, process::Stdio, time::Duration,
+    borrow::Cow, env, io::Read, path::PathBuf, process::ExitStatus, process::Stdio, sync::OnceLock,
+    time::Duration,
 };
 
 use anyhow::{bail, Context as _, Result};
 use indicatif::{ProgressBar, ProgressStyle};
 use log::LevelFilter;
-use once_cell::sync::{Lazy, OnceCell};
+use once_cell::sync::Lazy;
 
 use crate::{config::Config, git, platform, util};
 
@@ -25,9 +26,9 @@ const DEFAULT_SHELL: &str = "/bin/sh";
 pub static SHELL: Lazy<OsString> =
     Lazy::new(|| (env::var_os("SHELL").unwrap_or_else(|| DEFAULT_SHELL.into())));
 
-static VERBOSITY: OnceCell<LevelFilter> = OnceCell::new();
-static CONFIG: OnceCell<Config> = OnceCell::new();
-static PATH: OnceCell<String> = OnceCell::new();
+static VERBOSITY: OnceLock<LevelFilter> = OnceLock::new();
+static CONFIG: OnceLock<Config> = OnceLock::new();
+static PATH: OnceLock<String> = OnceLock::new();
 
 pub fn verbosity() -> &'static LevelFilter {
     VERBOSITY.get().expect("verbosity is not initialized")