chore(deps): Bump Rust to 1.75.0

Tiltfile

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

lib/dnsmsg-parser/src/dns_message_parser.rs

@@ -131,7 +131,7 @@ impl DnsMessageParser {
             .collect::<Vec<ZoneInfo>>();
 
         zones
-            .get(0)
+            .first()
             .cloned()
             .ok_or_else(|| DnsMessageParserError::SimpleError {
                 cause: format!(

lib/prometheus-parser/src/lib.rs

@@ -282,7 +282,7 @@ impl MetricGroup {
 }
 
 fn matching_group<T: Default>(values: &mut MetricMap<T>, group: GroupKey) -> &mut T {
-    values.entry(group).or_insert_with(T::default)
+    values.entry(group).or_default()
 }
 
 /// Parse the given text input, and group the result into higher-level

lib/vector-buffers/src/config.rs

@@ -425,19 +425,19 @@ mod test {
 
     #[test]
     fn parse_partial_invalid_keys() {
-        let source = r#"max_size: 100
+        let source = r"max_size: 100
 max_events: 42
-"#;
+";
         let error = serde_yaml::from_str::<BufferConfig>(source).unwrap_err();
         assert_eq!(error.to_string(), BUFFER_CONFIG_NO_MATCH_ERR);
     }
 
     #[test]
     fn parse_without_type_tag() {
         check_single_stage(
-            r#"
+            r"
           max_events: 100
-          "#,
+          ",
             BufferType::Memory {
                 max_events: NonZeroUsize::new(100).unwrap(),
                 when_full: WhenFull::Block,
@@ -448,11 +448,11 @@ max_events: 42
     #[test]
     fn parse_multiple_stages() {
         check_multiple_stages(
-            r#"
+            r"
           - max_events: 42
           - max_events: 100
             when_full: drop_newest
-          "#,
+          ",
             &[
                 BufferType::Memory {
                     max_events: NonZeroUsize::new(42).unwrap(),
@@ -469,53 +469,53 @@ max_events: 42
     #[test]
     fn ensure_field_defaults_for_all_types() {
         check_single_stage(
-            r#"
+            r"
           type: memory
-          "#,
+          ",
             BufferType::Memory {
                 max_events: NonZeroUsize::new(500).unwrap(),
                 when_full: WhenFull::Block,
             },
         );
 
         check_single_stage(
-            r#"
+            r"
           type: memory
           max_events: 100
-          "#,
+          ",
             BufferType::Memory {
                 max_events: NonZeroUsize::new(100).unwrap(),
                 when_full: WhenFull::Block,
             },
         );
 
         check_single_stage(
-            r#"
+            r"
           type: memory
           when_full: drop_newest
-          "#,
+          ",
             BufferType::Memory {
                 max_events: NonZeroUsize::new(500).unwrap(),
                 when_full: WhenFull::DropNewest,
             },
         );
 
         check_single_stage(
-            r#"
+            r"
           type: memory
           when_full: overflow
-          "#,
+          ",
             BufferType::Memory {
                 max_events: NonZeroUsize::new(500).unwrap(),
                 when_full: WhenFull::Overflow,
             },
         );
 
         check_single_stage(
-            r#"
+            r"
           type: disk
           max_size: 1024
-          "#,
+          ",
             BufferType::DiskV2 {
                 max_size: NonZeroU64::new(1024).unwrap(),
                 when_full: WhenFull::Block,

lib/vector-buffers/src/encoding.rs

@@ -177,7 +177,7 @@ impl<T: FixedEncodable> Encodable for T {
 
     fn get_metadata() -> Self::Metadata {}
 
-    fn can_decode(_: Self::Metadata) -> bool {
+    fn can_decode((): Self::Metadata) -> bool {
         true
     }
 
@@ -189,7 +189,7 @@ impl<T: FixedEncodable> Encodable for T {
         FixedEncodable::encoded_size(self)
     }
 
-    fn decode<B: Buf + Clone>(_: Self::Metadata, buffer: B) -> Result<Self, Self::DecodeError> {
+    fn decode<B: Buf + Clone>((): Self::Metadata, buffer: B) -> Result<Self, Self::DecodeError> {
         <Self as FixedEncodable>::decode(buffer)
     }
 }

lib/vector-buffers/src/test/variant.rs

@@ -103,12 +103,8 @@ impl Arbitrary for Variant {
         let use_memory_buffer = bool::arbitrary(g);
 
         // Using a u16 ensures we avoid any allocation errors for our holding buffers, etc.
-        let max_events = NonZeroU16::arbitrary(g)
-            .try_into()
-            .expect("we don't support 16-bit platforms");
-        let max_size = NonZeroU16::arbitrary(g)
-            .try_into()
-            .expect("we don't support 16-bit platforms");
+        let max_events = NonZeroU16::arbitrary(g).into();
+        let max_size = NonZeroU16::arbitrary(g).into();
 
         let when_full = WhenFull::arbitrary(g);
 

lib/vector-buffers/src/topology/acks.rs

@@ -730,7 +730,7 @@ mod tests {
     }
 
     #[test]
-    #[should_panic]
+    #[should_panic(expected = "overflowing unclaimed acknowledgements is a serious bug")]
     fn panic_when_unclaimed_acks_overflows() {
         let actions = vec![Action::Acknowledge(u64::MAX), Action::Acknowledge(1)];
 

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

@@ -24,7 +24,7 @@ pub enum ReceiverAdapter<T: Bufferable> {
     InMemory(LimitedReceiver<T>),
 
     /// The disk v2 buffer.
-    DiskV2(disk_v2::Reader<T, ProductionFilesystem>),
+    DiskV2(disk_v2::BufferReader<T, ProductionFilesystem>),
 }
 
 impl<T: Bufferable> From<LimitedReceiver<T>> for ReceiverAdapter<T> {
@@ -33,8 +33,8 @@ impl<T: Bufferable> From<LimitedReceiver<T>> for ReceiverAdapter<T> {
     }
 }
 
-impl<T: Bufferable> From<disk_v2::Reader<T, ProductionFilesystem>> for ReceiverAdapter<T> {
-    fn from(v: disk_v2::Reader<T, ProductionFilesystem>) -> Self {
+impl<T: Bufferable> From<disk_v2::BufferReader<T, ProductionFilesystem>> for ReceiverAdapter<T> {
+    fn from(v: disk_v2::BufferReader<T, ProductionFilesystem>) -> Self {
         Self::DiskV2(v)
     }
 }

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

@@ -21,7 +21,7 @@ pub enum SenderAdapter<T: Bufferable> {
     InMemory(LimitedSender<T>),
 
     /// The disk v2 buffer.
-    DiskV2(Arc<Mutex<disk_v2::Writer<T, ProductionFilesystem>>>),
+    DiskV2(Arc<Mutex<disk_v2::BufferWriter<T, ProductionFilesystem>>>),
 }
 
 impl<T: Bufferable> From<LimitedSender<T>> for SenderAdapter<T> {
@@ -30,8 +30,8 @@ impl<T: Bufferable> From<LimitedSender<T>> for SenderAdapter<T> {
     }
 }
 
-impl<T: Bufferable> From<disk_v2::Writer<T, ProductionFilesystem>> for SenderAdapter<T> {
-    fn from(v: disk_v2::Writer<T, ProductionFilesystem>) -> Self {
+impl<T: Bufferable> From<disk_v2::BufferWriter<T, ProductionFilesystem>> for SenderAdapter<T> {
+    fn from(v: disk_v2::BufferWriter<T, ProductionFilesystem>) -> Self {
         Self::DiskV2(Arc::new(Mutex::new(v)))
     }
 }

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

@@ -428,7 +428,7 @@ mod tests {
     };
 
     #[test]
-    #[should_panic]
+    #[should_panic(expected = "`amount` must be less than `MAX_ALIGNABLE_AMOUNT`")]
     fn test_align16_too_large() {
         // We forcefully panic if the input to `align16` is too large to align without overflow, primarily because
         // that's a huge amount even on 32-bit systems and in non-test code, we only use `align16` in a const context,

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

@@ -92,16 +92,16 @@ pub enum LedgerLoadCreateError {
 pub struct LedgerState {
     /// Next record ID to use when writing a record.
     #[with(Atomic)]
-    writer_next_record_id: AtomicU64,
+    writer_next_record: AtomicU64,
     /// The current data file ID being written to.
     #[with(Atomic)]
-    writer_current_data_file_id: AtomicU16,
+    writer_current_data_file: AtomicU16,
     /// The current data file ID being read from.
     #[with(Atomic)]
-    reader_current_data_file_id: AtomicU16,
+    reader_current_data_file: AtomicU16,
     /// The last record ID read by the reader.
     #[with(Atomic)]
-    reader_last_record_id: AtomicU64,
+    reader_last_record: AtomicU64,
 }
 
 impl Default for LedgerState {
@@ -110,41 +110,39 @@ impl Default for LedgerState {
             // First record written is always 1, so that our default of 0 for
             // `reader_last_record_id` ensures we start up in a state of "alright, waiting to read
             // record #1 next".
-            writer_next_record_id: AtomicU64::new(1),
-            writer_current_data_file_id: AtomicU16::new(0),
-            reader_current_data_file_id: AtomicU16::new(0),
-            reader_last_record_id: AtomicU64::new(0),
+            writer_next_record: AtomicU64::new(1),
+            writer_current_data_file: AtomicU16::new(0),
+            reader_current_data_file: AtomicU16::new(0),
+            reader_last_record: AtomicU64::new(0),
         }
     }
 }
 
 impl ArchivedLedgerState {
     fn get_current_writer_file_id(&self) -> u16 {
-        self.writer_current_data_file_id.load(Ordering::Acquire)
+        self.writer_current_data_file.load(Ordering::Acquire)
     }
 
     fn get_next_writer_file_id(&self) -> u16 {
         (self.get_current_writer_file_id() + 1) % MAX_FILE_ID
     }
 
     pub(super) fn increment_writer_file_id(&self) {
-        self.writer_current_data_file_id
+        self.writer_current_data_file
             .store(self.get_next_writer_file_id(), Ordering::Release);
     }
 
     pub(super) fn get_next_writer_record_id(&self) -> u64 {
-        self.writer_next_record_id.load(Ordering::Acquire)
+        self.writer_next_record.load(Ordering::Acquire)
     }
 
     pub(super) fn increment_next_writer_record_id(&self, amount: u64) -> u64 {
-        let previous = self
-            .writer_next_record_id
-            .fetch_add(amount, Ordering::AcqRel);
+        let previous = self.writer_next_record.fetch_add(amount, Ordering::AcqRel);
         previous.wrapping_add(amount)
     }
 
     fn get_current_reader_file_id(&self) -> u16 {
-        self.reader_current_data_file_id.load(Ordering::Acquire)
+        self.reader_current_data_file.load(Ordering::Acquire)
     }
 
     fn get_next_reader_file_id(&self) -> u16 {
@@ -157,18 +155,17 @@ impl ArchivedLedgerState {
 
     fn increment_reader_file_id(&self) -> u16 {
         let value = self.get_next_reader_file_id();
-        self.reader_current_data_file_id
+        self.reader_current_data_file
             .store(value, Ordering::Release);
         value
     }
 
     pub(super) fn get_last_reader_record_id(&self) -> u64 {
-        self.reader_last_record_id.load(Ordering::Acquire)
+        self.reader_last_record.load(Ordering::Acquire)
     }
 
     pub(super) fn increment_last_reader_record_id(&self, amount: u64) {
-        self.reader_last_record_id
-            .fetch_add(amount, Ordering::AcqRel);
+        self.reader_last_record.fetch_add(amount, Ordering::AcqRel);
     }
 
     #[cfg(test)]
@@ -184,7 +181,7 @@ impl ArchivedLedgerState {
         //
         // Despite it being test-only, we're really amping up the "this is only for testing!" factor
         // by making it an actual `unsafe` function, and putting "unsafe" in the name. :)
-        self.writer_next_record_id.store(id, Ordering::Release);
+        self.writer_next_record.store(id, Ordering::Release);
     }
 
     #[cfg(test)]
@@ -200,7 +197,7 @@ impl ArchivedLedgerState {
         //
         // Despite it being test-only, we're really amping up the "this is only for testing!" factor
         // by making it an actual `unsafe` function, and putting "unsafe" in the name. :)
-        self.reader_last_record_id.store(id, Ordering::Release);
+        self.reader_last_record.store(id, Ordering::Release);
     }
 }
 
@@ -213,7 +210,7 @@ where
     config: DiskBufferConfig<FS>,
     // Advisory lock for this buffer directory.
     #[allow(dead_code)]
-    ledger_lock: LockFile,
+    lock: LockFile,
     // Ledger state.
     state: BackedArchive<FS::MutableMemoryMap, LedgerState>,
     // The total size, in bytes, of all unread records in the buffer.
@@ -576,8 +573,8 @@ where
         // file I/O, but the code is so specific, including the drop guard for the lock file, that I
         // don't know if it's worth it.
         let ledger_lock_path = config.data_dir.join("buffer.lock");
-        let mut ledger_lock = LockFile::open(&ledger_lock_path).context(IoSnafu)?;
-        if !ledger_lock.try_lock().context(IoSnafu)? {
+        let mut lock = LockFile::open(&ledger_lock_path).context(IoSnafu)?;
+        if !lock.try_lock().context(IoSnafu)? {
             return Err(LedgerLoadCreateError::LedgerLockAlreadyHeld);
         }
 
@@ -640,7 +637,7 @@ where
         // what not.
         let mut ledger = Ledger {
             config,
-            ledger_lock,
+            lock,
             state: ledger_state,
             total_buffer_size: AtomicU64::new(0),
             reader_notify: Notify::new(),