Update Rust toolchain to 1.84.1 (#2611)

crates/admin/src/cluster_controller/logs_controller.rs

@@ -1306,9 +1306,7 @@ pub struct LogsBasedPartitionProcessorPlacementHints<'a> {
     logs_controller: &'a LogsController,
 }
 
-impl<'a> scheduler::PartitionProcessorPlacementHints
-    for LogsBasedPartitionProcessorPlacementHints<'a>
-{
+impl scheduler::PartitionProcessorPlacementHints for LogsBasedPartitionProcessorPlacementHints<'_> {
     fn preferred_nodes(&self, partition_id: &PartitionId) -> impl Iterator<Item = &PlainNodeId> {
         let log_id = LogId::from(*partition_id);
 

crates/admin/src/cluster_controller/scheduler.rs

@@ -551,13 +551,13 @@ impl<'a> TargetPartitionPlacementState<'a> {
     }
 }
 
-impl<'a> TargetPartitionPlacementState<'a> {
+impl TargetPartitionPlacementState<'_> {
     pub fn contains(&self, value: PlainNodeId) -> bool {
         self.node_set.contains(value)
     }
 }
 
-impl<'a> Drop for TargetPartitionPlacementState<'a> {
+impl Drop for TargetPartitionPlacementState<'_> {
     fn drop(&mut self) {
         if let Some(node_id) = self.leader.take() {
             self.node_set.set_leader(node_id);
@@ -917,7 +917,7 @@ mod tests {
         partition_table: &'a PartitionTable,
     }
 
-    impl<'a> Matcher for PartitionTableMatcher<'a> {
+    impl Matcher for PartitionTableMatcher<'_> {
         type ActualT = ObservedClusterState;
 
         fn matches(&self, actual: &Self::ActualT) -> MatcherResult {

crates/bifrost/src/providers/replicated_loglet/sequencer/appender.rs

@@ -407,7 +407,7 @@ struct LogServerStoreTask<'a, T> {
     timeout_at: MillisSinceEpoch,
 }
 
-impl<'a, T: TransportConnect> LogServerStoreTask<'a, T> {
+impl<T: TransportConnect> LogServerStoreTask<'_, T> {
     #[instrument(
         skip_all,
         fields(

crates/core/src/network/connection.rs

@@ -43,7 +43,7 @@ pub struct SendPermit<'a, M> {
     _phantom: std::marker::PhantomData<M>,
 }
 
-impl<'a, M> SendPermit<'a, M>
+impl<M> SendPermit<'_, M>
 where
     M: WireEncode + Targeted,
 {
@@ -76,7 +76,7 @@ where
     }
 }
 
-impl<'a, M> SendPermit<'a, M> {
+impl<M> SendPermit<'_, M> {
     /// Sends a raw pre-serialized message over this permit.
     ///
     /// Note that sending messages over this permit won't use the peer information nor the connection

crates/core/src/partitions.rs

@@ -184,10 +184,10 @@ impl PartitionRoutingRefresher {
     }
 
     fn spawn_sync_routing_information_task(&mut self) {
-        if !self
+        if self
             .inflight_refresh_task
             .as_ref()
-            .is_some_and(|t| !t.is_finished())
+            .is_none_or(|t| t.is_finished())
         {
             let task = TaskCenter::spawn_unmanaged(
                 TaskKind::Disposable,

crates/ingress-http/src/layers/tracing_context_extractor.rs

@@ -54,7 +54,7 @@ where
 // https://github.com/open-telemetry/opentelemetry-rust/blob/ef4701055cc39d3448d5e5392812ded00cdd4476/opentelemetry-http/src/lib.rs#L14 License APL 2.0
 pub struct HeaderExtractor<'a>(pub &'a HeaderMap);
 
-impl<'a> Extractor for HeaderExtractor<'a> {
+impl Extractor for HeaderExtractor<'_> {
     /// Get a value for a key from the HeaderMap.  If the value is not valid ASCII, returns None.
     fn get(&self, key: &str) -> Option<&str> {
         self.0.get(key).and_then(|value| value.to_str().ok())

crates/log-server/src/rocksdb_logstore/record_format.rs

@@ -77,7 +77,7 @@ impl<'a> DataRecordDecoder<'a> {
 #[derive(derive_more::From)]
 pub(super) struct DataRecordEncoder<'a>(&'a Record);
 
-impl<'a> DataRecordEncoder<'a> {
+impl DataRecordEncoder<'_> {
     /// On-disk record layout.
     /// For the record header, byte-order is little-endian.
     ///

crates/partition-store/src/deduplication_table/mod.rs

@@ -88,7 +88,7 @@ impl ReadOnlyDeduplicationTable for PartitionStore {
     }
 }
 
-impl<'a> ReadOnlyDeduplicationTable for PartitionStoreTransaction<'a> {
+impl ReadOnlyDeduplicationTable for PartitionStoreTransaction<'_> {
     async fn get_dedup_sequence_number(
         &mut self,
         producer_id: &ProducerId,
@@ -101,7 +101,7 @@ impl<'a> ReadOnlyDeduplicationTable for PartitionStoreTransaction<'a> {
     }
 }
 
-impl<'a> DeduplicationTable for PartitionStoreTransaction<'a> {
+impl DeduplicationTable for PartitionStoreTransaction<'_> {
     async fn put_dedup_seq_number(
         &mut self,
         producer_id: ProducerId,

crates/partition-store/src/fsm_table/mod.rs

@@ -82,7 +82,7 @@ impl ReadOnlyFsmTable for PartitionStore {
     }
 }
 
-impl<'a> ReadOnlyFsmTable for PartitionStoreTransaction<'a> {
+impl ReadOnlyFsmTable for PartitionStoreTransaction<'_> {
     async fn get_inbox_seq_number(&mut self) -> Result<MessageIndex> {
         get::<SequenceNumber, _>(self, self.partition_id(), fsm_variable::INBOX_SEQ_NUMBER)
             .map(|opt| opt.map(Into::into).unwrap_or_default())
@@ -99,7 +99,7 @@ impl<'a> ReadOnlyFsmTable for PartitionStoreTransaction<'a> {
     }
 }
 
-impl<'a> FsmTable for PartitionStoreTransaction<'a> {
+impl FsmTable for PartitionStoreTransaction<'_> {
     async fn put_applied_lsn(&mut self, lsn: Lsn) {
         put(
             self,

crates/partition-store/src/idempotency_table/mod.rs

@@ -120,7 +120,7 @@ impl ReadOnlyIdempotencyTable for PartitionStore {
     }
 }
 
-impl<'a> ReadOnlyIdempotencyTable for PartitionStoreTransaction<'a> {
+impl ReadOnlyIdempotencyTable for PartitionStoreTransaction<'_> {
     async fn get_idempotency_metadata(
         &mut self,
         idempotency_id: &IdempotencyId,
@@ -137,7 +137,7 @@ impl<'a> ReadOnlyIdempotencyTable for PartitionStoreTransaction<'a> {
     }
 }
 
-impl<'a> IdempotencyTable for PartitionStoreTransaction<'a> {
+impl IdempotencyTable for PartitionStoreTransaction<'_> {
     async fn put_idempotency_metadata(
         &mut self,
         idempotency_id: &IdempotencyId,

crates/partition-store/src/inbox_table/mod.rs

@@ -119,7 +119,7 @@ impl ReadOnlyInboxTable for PartitionStore {
     }
 }
 
-impl<'a> ReadOnlyInboxTable for PartitionStoreTransaction<'a> {
+impl ReadOnlyInboxTable for PartitionStoreTransaction<'_> {
     fn peek_inbox(
         &mut self,
         service_id: &ServiceId,
@@ -144,7 +144,7 @@ impl<'a> ReadOnlyInboxTable for PartitionStoreTransaction<'a> {
     }
 }
 
-impl<'a> InboxTable for PartitionStoreTransaction<'a> {
+impl InboxTable for PartitionStoreTransaction<'_> {
     async fn put_inbox_entry(
         &mut self,
         inbox_sequence_number: MessageIndex,

crates/partition-store/src/invocation_status_table/mod.rs

@@ -324,7 +324,7 @@ impl ReadOnlyInvocationStatusTable for PartitionStore {
     }
 }
 
-impl<'a> ReadOnlyInvocationStatusTable for PartitionStoreTransaction<'a> {
+impl ReadOnlyInvocationStatusTable for PartitionStoreTransaction<'_> {
     async fn get_invocation_status(
         &mut self,
         invocation_id: &InvocationId,
@@ -350,7 +350,7 @@ impl<'a> ReadOnlyInvocationStatusTable for PartitionStoreTransaction<'a> {
     }
 }
 
-impl<'a> InvocationStatusTable for PartitionStoreTransaction<'a> {
+impl InvocationStatusTable for PartitionStoreTransaction<'_> {
     async fn put_invocation_status(
         &mut self,
         invocation_id: &InvocationId,

crates/partition-store/src/journal_table/mod.rs

@@ -163,7 +163,7 @@ impl ReadOnlyJournalTable for PartitionStore {
     }
 }
 
-impl<'a> ReadOnlyJournalTable for PartitionStoreTransaction<'a> {
+impl ReadOnlyJournalTable for PartitionStoreTransaction<'_> {
     async fn get_journal_entry(
         &mut self,
         invocation_id: &InvocationId,
@@ -191,7 +191,7 @@ impl<'a> ReadOnlyJournalTable for PartitionStoreTransaction<'a> {
     }
 }
 
-impl<'a> JournalTable for PartitionStoreTransaction<'a> {
+impl JournalTable for PartitionStoreTransaction<'_> {
     async fn put_journal_entry(
         &mut self,
         invocation_id: &InvocationId,

crates/partition-store/src/journal_table_v2/mod.rs

@@ -346,7 +346,7 @@ impl ReadOnlyJournalTable for PartitionStore {
     }
 }
 
-impl<'a> ReadOnlyJournalTable for PartitionStoreTransaction<'a> {
+impl ReadOnlyJournalTable for PartitionStoreTransaction<'_> {
     async fn get_journal_entry(
         &mut self,
         invocation_id: InvocationId,
@@ -389,7 +389,7 @@ impl<'a> ReadOnlyJournalTable for PartitionStoreTransaction<'a> {
     }
 }
 
-impl<'a> JournalTable for PartitionStoreTransaction<'a> {
+impl JournalTable for PartitionStoreTransaction<'_> {
     async fn put_journal_entry(
         &mut self,
         invocation_id: InvocationId,

crates/partition-store/src/keys.rs

@@ -439,7 +439,7 @@ impl<T: KeyCodec> KeyCodec for Option<T> {
     }
 }
 
-impl<'a> KeyCodec for &'a [u8] {
+impl KeyCodec for &[u8] {
     fn encode<B: BufMut>(&self, target: &mut B) {
         target.put(*self);
     }

crates/partition-store/src/outbox_table/mod.rs

@@ -150,13 +150,13 @@ impl OutboxTable for PartitionStore {
     }
 }
 
-impl<'a> ReadOnlyOutboxTable for PartitionStoreTransaction<'a> {
+impl ReadOnlyOutboxTable for PartitionStoreTransaction<'_> {
     async fn get_outbox_head_seq_number(&mut self) -> Result<Option<u64>> {
         get_outbox_head_seq_number(self, self.partition_id())
     }
 }
 
-impl<'a> OutboxTable for PartitionStoreTransaction<'a> {
+impl OutboxTable for PartitionStoreTransaction<'_> {
     async fn put_outbox_message(&mut self, message_index: u64, outbox_message: &OutboxMessage) {
         add_message(self, self.partition_id(), message_index, outbox_message)
     }

crates/partition-store/src/owned_iter.rs

@@ -25,7 +25,7 @@ impl<'a, DB: DBAccess> OwnedIterator<'a, DB> {
     }
 }
 
-impl<'a, DB: DBAccess> Iterator for OwnedIterator<'a, DB> {
+impl<DB: DBAccess> Iterator for OwnedIterator<'_, DB> {
     type Item = (Bytes, Bytes);
 
     #[inline]

crates/partition-store/src/partition_store.rs

@@ -557,7 +557,7 @@ pub struct PartitionStoreTransaction<'a> {
     value_buffer: &'a mut BytesMut,
 }
 
-impl<'a> PartitionStoreTransaction<'a> {
+impl PartitionStoreTransaction<'_> {
     pub(crate) fn prefix_iterator(
         &self,
         table: TableKind,
@@ -627,7 +627,7 @@ fn assert_partition_key(
             "Partition key '{partition_key}' is not part of PartitionStore's partition '{partition_key_range:?}'. This indicates a bug.");
 }
 
-impl<'a> Transaction for PartitionStoreTransaction<'a> {
+impl Transaction for PartitionStoreTransaction<'_> {
     async fn commit(self) -> Result<()> {
         // We cannot directly commit the txn because it might fail because of unrelated concurrent
         // writes to RocksDB. However, it is safe to write the WriteBatch for a given partition,
@@ -660,7 +660,7 @@ impl<'a> Transaction for PartitionStoreTransaction<'a> {
     }
 }
 
-impl<'a> StorageAccess for PartitionStoreTransaction<'a> {
+impl StorageAccess for PartitionStoreTransaction<'_> {
     type DBAccess<'b>
         = DB
     where

crates/partition-store/src/promise_table/mod.rs

@@ -126,7 +126,7 @@ impl ReadOnlyPromiseTable for PartitionStore {
     }
 }
 
-impl<'a> ReadOnlyPromiseTable for PartitionStoreTransaction<'a> {
+impl ReadOnlyPromiseTable for PartitionStoreTransaction<'_> {
     async fn get_promise(
         &mut self,
         service_id: &ServiceId,
@@ -144,7 +144,7 @@ impl<'a> ReadOnlyPromiseTable for PartitionStoreTransaction<'a> {
     }
 }
 
-impl<'a> PromiseTable for PartitionStoreTransaction<'a> {
+impl PromiseTable for PartitionStoreTransaction<'_> {
     async fn put_promise(&mut self, service_id: &ServiceId, key: &ByteString, promise: &Promise) {
         self.assert_partition_key(service_id);
         put_promise(self, service_id, key, promise)

crates/partition-store/src/service_status_table/mod.rs

@@ -118,7 +118,7 @@ impl ReadOnlyVirtualObjectStatusTable for PartitionStore {
     }
 }
 
-impl<'a> ReadOnlyVirtualObjectStatusTable for PartitionStoreTransaction<'a> {
+impl ReadOnlyVirtualObjectStatusTable for PartitionStoreTransaction<'_> {
     async fn get_virtual_object_status(
         &mut self,
         service_id: &ServiceId,
@@ -135,7 +135,7 @@ impl<'a> ReadOnlyVirtualObjectStatusTable for PartitionStoreTransaction<'a> {
     }
 }
 
-impl<'a> VirtualObjectStatusTable for PartitionStoreTransaction<'a> {
+impl VirtualObjectStatusTable for PartitionStoreTransaction<'_> {
     async fn put_virtual_object_status(
         &mut self,
         service_id: &ServiceId,

crates/partition-store/src/state_table/mod.rs

@@ -159,7 +159,7 @@ impl ReadOnlyStateTable for PartitionStore {
     }
 }
 
-impl<'a> ReadOnlyStateTable for PartitionStoreTransaction<'a> {
+impl ReadOnlyStateTable for PartitionStoreTransaction<'_> {
     fn get_user_state(
         &mut self,
         service_id: &ServiceId,
@@ -182,7 +182,7 @@ impl<'a> ReadOnlyStateTable for PartitionStoreTransaction<'a> {
     }
 }
 
-impl<'a> StateTable for PartitionStoreTransaction<'a> {
+impl StateTable for PartitionStoreTransaction<'_> {
     fn put_user_state(
         &mut self,
         service_id: &ServiceId,

crates/partition-store/src/timer_table/mod.rs

@@ -195,7 +195,7 @@ impl TimerTable for PartitionStore {
     }
 }
 
-impl<'a> TimerTable for PartitionStoreTransaction<'a> {
+impl TimerTable for PartitionStoreTransaction<'_> {
     async fn put_timer(&mut self, key: &TimerKey, timer: &Timer) {
         add_timer(self, self.partition_id(), key, timer)
     }

crates/serde-util/src/byte_count.rs

@@ -176,7 +176,7 @@ struct ByteCountVisitor<const CAN_BE_ZERO: bool>;
 // it's actually compile-time, compiler will inline the comparison but we need
 // to make clippy happy.
 #[allow(clippy::absurd_extreme_comparisons)]
-impl<'de, const CAN_BE_ZERO: bool> Visitor<'de> for ByteCountVisitor<CAN_BE_ZERO> {
+impl<const CAN_BE_ZERO: bool> Visitor<'_> for ByteCountVisitor<CAN_BE_ZERO> {
     type Value = ByteCount<CAN_BE_ZERO>;
     fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
         formatter.write_str("an integer or string")

crates/service-client/src/http.rs

@@ -238,7 +238,7 @@ impl From<hyper_util::client::legacy::Error> for HttpError {
 
 struct FormatHyperError<'a>(&'a hyper_util::client::legacy::Error);
 
-impl<'a> fmt::Display for FormatHyperError<'a> {
+impl fmt::Display for FormatHyperError<'_> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         if let Some(source) = self.0.source() {
             write!(f, "{}, {}", self.0, source)

crates/service-client/src/request_identity/v1.rs

@@ -121,7 +121,7 @@ impl<'key, 'aud> Signer<'key, 'aud> {
     }
 }
 
-impl<'key, 'aud> super::SignRequest for Signer<'key, 'aud> {
+impl super::SignRequest for Signer<'_, '_> {
     type Error = jsonwebtoken::errors::Error;
     fn insert_identity(self, mut headers: HeaderMap) -> Result<HeaderMap, Self::Error> {
         let jwt = jsonwebtoken::encode(

crates/storage-query-datafusion/src/physical_optimizer.rs

@@ -32,7 +32,7 @@ fn is_partition_key_column(arg: &PhysicalExprRef) -> bool {
     let Some(col) = arg.as_any().downcast_ref::<Column>() else {
         return false;
     };
-    return col.name() == "partition_key";
+    col.name() == "partition_key"
 }
 
 impl PhysicalOptimizerRule for JoinRewrite {

crates/timer/src/service/tests.rs

@@ -325,7 +325,6 @@ async fn earlier_timers_replace_older_ones() {
 }
 
 async fn yield_to_timer_service<
-    'a,
     Timer: crate::Timer + Debug + 'static,
     Clock: crate::Clock,
     TimerReader: crate::TimerReader<Timer> + Send + 'static,