chore(deps): Upgrade rust to 1.69.0 (#17194)

* Bump rust to 1.69.0

* Fix name of `derived_hash_with_manual_eq` lint

* Rewrite all `let _ =` to just `_ =`

* Fix needless borrows

* Update is-terminal to 0.4.7

* Patch ntapi to pull in fix for Windows unaligned access

* Add ntapi spelling lookaside

* Ignore failing file watcher assertion on Windows

.github/actions/spelling/allow.txt

@@ -368,6 +368,7 @@ nixpkgs
 nokia
 nslookup
 nsupdate
+ntapi
 ntfs
 opendal
 opensearch

Cargo.toml

@@ -372,6 +372,9 @@ chrono = { git = "https://github.com/vectordotdev/chrono.git", tag = "v0.4.24-no
 # The upgrade for `tokio-util` >= 0.6.9 is blocked on https://github.com/vectordotdev/vector/issues/11257.
 tokio-util = { git = "https://github.com/vectordotdev/tokio", branch = "tokio-util-0.7.4-framed-read-continue-on-error" }
 nix = { git = "https://github.com/vectordotdev/nix.git", branch = "memfd/gnu/musl" }
+# The `heim` crates depend on `ntapi` 0.3.7 on Windows, but that version has an
+# unaligned access bug fixed in the following revision.
+ntapi = { git = "https://github.com/MSxDOS/ntapi.git", rev = "24fc1e47677fc9f6e38e5f154e6011dc9b270da6" }
 
 [features]
 # Default features for *-unknown-linux-gnu and *-apple-darwin

Tiltfile

@@ -7,7 +7,7 @@ load('ext://helm_resource', 'helm_resource', 'helm_repo')
 docker_build(
     ref='timberio/vector',
     context='.',
-    build_args={'RUST_VERSION': '1.66.1'},
+    build_args={'RUST_VERSION': '1.69.0'},
     dockerfile='tilt/Dockerfile'
     )
 

lib/codecs/src/decoding/framing/octet_counting.rs

@@ -406,7 +406,7 @@ mod tests {
         let mut buffer = BytesMut::with_capacity(32);
 
         buffer.put(&b"32thisshouldbelongerthanthmaxframeasizewhichmeansthesyslogparserwillnotbeabletodecodeit"[..]);
-        let _ = decoder.decode(&mut buffer);
+        _ = decoder.decode(&mut buffer);
 
         assert_eq!(decoder.octet_decoding, Some(State::DiscardingToEol));
         buffer.put(&b"wemustcontinuetodiscard\n32 something valid"[..]);

lib/file-source/benches/buffer.rs

@@ -57,7 +57,7 @@ fn read_until_bench(c: &mut Criterion) {
         let delimiter: [u8; 1] = [param.delim];
         group.bench_with_input(BenchmarkId::new("read_until", param), &param, |b, _| {
             b.iter(|| {
-                let _ = read_until_with_max_size(
+                _ = read_until_with_max_size(
                     &mut reader,
                     &mut position,
                     &delimiter,

lib/file-source/src/file_watcher/tests/experiment.rs

@@ -48,7 +48,8 @@ fn experiment(actions: Vec<FileWatcherAction>) {
     for action in actions.iter() {
         match *action {
             FileWatcherAction::DeleteFile => {
-                let _ = fs::remove_file(&path);
+                _ = fs::remove_file(&path);
+                #[cfg(not(windows))] // Windows will only remove after the file is closed.
                 assert!(!path.exists());
                 fwfiles[0].reset();
                 break;

lib/file-source/src/file_watcher/tests/experiment_no_truncations.rs

@@ -32,7 +32,8 @@ fn experiment_no_truncations(actions: Vec<FileWatcherAction>) {
     for action in actions.iter() {
         match *action {
             FileWatcherAction::DeleteFile => {
-                let _ = fs::remove_file(&path);
+                _ = fs::remove_file(&path);
+                #[cfg(not(windows))] // Windows will only remove after the file is closed.
                 assert!(!path.exists());
                 fwfiles[0].reset();
                 break;

lib/k8s-e2e-tests/src/lib.rs

@@ -17,7 +17,7 @@ pub mod metrics;
 pub const BUSYBOX_IMAGE: &str = "busybox:1.28";
 
 pub fn init() {
-    let _ = env_logger::builder().is_test(true).try_init();
+    _ = env_logger::builder().is_test(true).try_init();
 }
 
 pub fn get_namespace() -> String {

lib/k8s-test-framework/src/temp_file.rs

@@ -23,7 +23,7 @@ impl TempFile {
 impl Drop for TempFile {
     fn drop(&mut self) {
         if let Some(dir) = self.path.parent() {
-            let _ = std::fs::remove_dir_all(dir);
+            _ = std::fs::remove_dir_all(dir);
         }
     }
 }

lib/tracing-limit/benches/limit.rs

@@ -97,7 +97,7 @@ struct Visitor<'a>(MutexGuard<'a, String>);
 impl<'a> field::Visit for Visitor<'a> {
     fn record_debug(&mut self, _field: &field::Field, value: &dyn fmt::Debug) {
         use std::fmt::Write;
-        let _ = write!(&mut *self.0, "{:?}", value);
+        _ = write!(&mut *self.0, "{:?}", value);
     }
 }
 
@@ -110,7 +110,7 @@ where
     }
 
     fn enabled(&self, metadata: &Metadata<'_>, _ctx: Context<'_, S>) -> bool {
-        let _ = metadata;
+        _ = metadata;
         true
     }
 
@@ -130,23 +130,23 @@ where
     }
 
     fn on_follows_from(&self, id: &span::Id, follows: &span::Id, _ctx: Context<'_, S>) {
-        let _ = (id, follows);
+        _ = (id, follows);
     }
 
     fn on_enter(&self, id: &span::Id, _ctx: Context<'_, S>) {
-        let _ = id;
+        _ = id;
     }
 
     fn on_exit(&self, id: &span::Id, _ctx: Context<'_, S>) {
-        let _ = id;
+        _ = id;
     }
 
     fn on_close(&self, id: span::Id, _ctx: Context<'_, S>) {
-        let _ = id;
+        _ = id;
     }
 
     fn on_id_change(&self, old: &span::Id, new: &span::Id, _ctx: Context<'_, S>) {
-        let _ = (old, new);
+        _ = (old, new);
     }
 }
 

lib/vector-api-client/src/subscription.rs

@@ -108,7 +108,7 @@ impl SubscriptionClient {
                     _ = &mut shutdown_rx => {
                         let subscriptions = subscriptions_clone.lock().unwrap();
                         for id in subscriptions.keys() {
-                            let _ = tx_clone.send(Payload::stop(*id));
+                            _ = tx_clone.send(Payload::stop(*id));
                         }
                         break
                     },
@@ -120,7 +120,7 @@ impl SubscriptionClient {
                                 let subscriptions = subscriptions_clone.lock().unwrap();
                                 let s: Option<&Sender<Payload>> = subscriptions.get::<Uuid>(&p.id);
                                 if let Some(s) = s {
-                                    let _ = s.send(p);
+                                    _ = s.send(p);
                                 }
                             }
                             None => {
@@ -159,8 +159,8 @@ impl SubscriptionClient {
         self.subscriptions.lock().unwrap().insert(id, tx);
 
         // Initialize the connection with the relevant control messages.
-        let _ = self.tx.send(Payload::init(id));
-        let _ = self.tx.send(Payload::start::<T>(id, request_body));
+        _ = self.tx.send(Payload::init(id));
+        _ = self.tx.send(Payload::start::<T>(id, request_body));
 
         Box::pin(
             BroadcastStream::new(rx)
@@ -185,7 +185,7 @@ pub async fn connect_subscription_client(
     // Forwarded received messages back upstream to the GraphQL server
     tokio::spawn(async move {
         while let Some(p) = send_rx.recv().await {
-            let _ = ws_tx
+            _ = ws_tx
                 .send(Message::Text(serde_json::to_string(&p).unwrap()))
                 .await;
         }
@@ -195,7 +195,7 @@ pub async fn connect_subscription_client(
     tokio::spawn(async move {
         while let Some(Ok(Message::Text(m))) = ws_rx.next().await {
             if let Ok(p) = serde_json::from_str::<Payload>(&m) {
-                let _ = recv_tx.send(p);
+                _ = recv_tx.send(p);
             }
         }
     });

lib/vector-buffers/benches/common.rs

@@ -102,7 +102,7 @@ impl<const N: usize> FixedEncodable for Message<N> {
         let id = buffer.get_u64();
         for _ in 0..N {
             // this covers self._padding
-            let _ = buffer.get_u64();
+            _ = buffer.get_u64();
         }
         Ok(Message::new(id))
     }
@@ -177,6 +177,6 @@ pub async fn war_measurement<const N: usize>(
 ) {
     for msg in messages.into_iter() {
         sender.send(msg).await.unwrap();
-        let _ = receiver.next().await.unwrap();
+        _ = receiver.next().await.unwrap();
     }
 }

lib/vector-buffers/src/topology/channel/tests.rs

@@ -44,7 +44,7 @@ where
 
         // Synchronize with sender and then wait for a small period of time to simulate a
         // blocking delay.
-        let _ = recv_baton.wait().await;
+        _ = recv_baton.wait().await;
         sleep(recv_delay).await;
 
         // Grab all messages and then return the results.
@@ -58,7 +58,7 @@ where
     // task correctly exits.  If we didn't drop it, the receiver task would just assume that we
     // had no more messages to send, waiting for-ev-er for the next one.
     let start = Instant::now();
-    let _ = send_baton.wait().await;
+    _ = send_baton.wait().await;
     assert!(sender.send(send_value.into()).await.is_ok());
     let send_delay = start.elapsed();
     assert!(send_delay > recv_delay);

lib/vector-buffers/src/variants/disk_v2/backed_archive.rs

@@ -71,7 +71,7 @@ where
         for<'a> T::Archived: CheckBytes<DefaultValidator<'a>>,
     {
         // Validate that the input is, well, valid.
-        let _ = check_archived_root::<T>(backing.as_ref())?;
+        _ = check_archived_root::<T>(backing.as_ref())?;
 
         // Now that we know the buffer fits T, we're good to go!
         Ok(Self {
@@ -110,7 +110,7 @@ where
     {
         // Serialize our value so we can shove it into the backing.
         let mut serializer = DefaultSerializer::default();
-        let _ = serializer
+        _ = serializer
             .serialize_value(&value)
             .map_err(|e| SerializeError::FailedToSerialize(e.to_string()))?;
 

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

@@ -349,7 +349,7 @@ impl ReaderModel {
                 // We have enough unconsumed event acknowledgements to fully acknowledge this
                 // record. Remove it, consume the event acknowledgements, add a record
                 // acknowledgement, and update the buffer size.
-                let _ = self.pending_record_acks.pop_front().unwrap();
+                _ = self.pending_record_acks.pop_front().unwrap();
                 self.unconsumed_event_acks -= 1;
                 self.unconsumed_record_acks += 1;
 
@@ -369,7 +369,7 @@ impl ReaderModel {
             if self.unconsumed_record_acks >= required_record_acks {
                 // We have enough unconsumed record acknowledgements to fully acknowledge this data
                 // file. Remove it, consume the record acknowledgements, and delete the data file.
-                let _ = self.pending_data_file_acks.pop_front().unwrap();
+                _ = self.pending_data_file_acks.pop_front().unwrap();
                 self.unconsumed_record_acks -= required_record_acks;
 
                 assert!(

lib/vector-common/src/event_test_util.rs

@@ -21,7 +21,7 @@ pub fn contains_name_once(pattern: &str) -> Result<(), String> {
                     names.push_str(", ");
                 }
                 n_events += 1;
-                let _ = write!(names, "`{event}`");
+                _ = write!(names, "`{event}`");
             }
         }
         if n_events == 0 {

lib/vector-common/src/finalization.rs

@@ -266,7 +266,7 @@ impl OwnedBatchNotifier {
             let status = self.status.load();
             // Ignore the error case, as it will happen during normal
             // source shutdown and we can't detect that here.
-            let _ = notifier.send(status);
+            _ = notifier.send(status);
         }
     }
 }

lib/vector-core/src/event/metric/tags.rs

@@ -269,7 +269,7 @@ impl<'a> IntoIterator for &'a TagValueSet {
 
 // The impl for `Hash` here follows the guarantees for the derived `PartialEq`, The resulting hash
 // will always be the same if the contents compare equal, so we can ignore the clippy lint.
-#[allow(clippy::derive_hash_xor_eq)]
+#[allow(clippy::derived_hash_with_manual_eq)]
 impl Hash for TagValueSet {
     fn hash<H: Hasher>(&self, hasher: &mut H) {
         match self {

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

@@ -110,7 +110,7 @@ impl<T> Generational<T> {
         F: Fn(&T) -> V,
     {
         let result = f(&self.inner);
-        let _ = self.gen.fetch_add(1, Ordering::AcqRel);
+        _ = self.gen.fetch_add(1, Ordering::AcqRel);
         result
     }
 }

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

@@ -6,7 +6,7 @@ use crate::event::Event;
 #[ignore]
 #[test]
 fn test_labels_injection() {
-    let _ = super::init();
+    _ = super::init();
 
     let span = span!(
         Level::ERROR,
@@ -46,7 +46,7 @@ fn test_labels_injection() {
 
 #[test]
 fn test_cardinality_metric() {
-    let _ = super::init();
+    _ = super::init();
 
     let capture_value = || {
         let metric = super::Controller::get()

rust-toolchain.toml

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