chore: bump Rust to 1.95

Dockerfile

@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 
-ARG RUST_VERSION=1.94
+ARG RUST_VERSION=1.95
 
 FROM rust:${RUST_VERSION}-slim-bookworm AS builder
 

bdd/java/Dockerfile

@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 
-ARG RUST_VERSION=1.94
+ARG RUST_VERSION=1.95
 FROM rust:${RUST_VERSION}
 
 FROM gradle:9.3.1-jdk17

bdd/python/Dockerfile

@@ -16,7 +16,7 @@
 # under the License.
 
 # syntax=docker/dockerfile:1
-ARG RUST_VERSION=1.94
+ARG RUST_VERSION=1.95
 FROM rust:${RUST_VERSION}-slim-trixie
 
 RUN apt-get update && \

bdd/rust/Dockerfile

@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 
-ARG RUST_VERSION=1.94
+ARG RUST_VERSION=1.95
 FROM rust:${RUST_VERSION}
 
 WORKDIR /app

core/ai/mcp/Dockerfile

@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 
-ARG RUST_VERSION=1.94
+ARG RUST_VERSION=1.95
 ARG ALPINE_VERSION=3.22
 
 FROM --platform=$BUILDPLATFORM lukemathwalker/cargo-chef:latest-rust-${RUST_VERSION}-alpine AS chef

core/bench/dashboard/server/Dockerfile

@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 
-ARG RUST_VERSION=1.94
+ARG RUST_VERSION=1.95
 
 # Build stage
 FROM rust:${RUST_VERSION}-slim-trixie AS builder

core/bench/src/args/common.rs

@@ -422,7 +422,7 @@ impl IggyBenchArgs {
                 self.consumers()
             ),
             BenchmarkKindCommand::EndToEndProducingConsumer(_) => {
-                format!("{} producing consumers", self.producers(),)
+                format!("{} producing consumers", self.producers())
             }
             BenchmarkKindCommand::EndToEndProducingConsumerGroup(_) => {
                 format!(

core/bench/src/benchmarks/benchmark.rs

@@ -195,6 +195,6 @@ pub trait Benchmarkable: Send {
             .map(|rl| format!(" global rate limit: {rl}/s"))
             .unwrap_or_default();
 
-        format!("{message_size}{messages_per_batch}{data}{rate_limit}",)
+        format!("{message_size}{messages_per_batch}{data}{rate_limit}")
     }
 }

core/common/src/types/message/messages_batch_mut.rs

@@ -448,8 +448,7 @@ impl IggyMessagesBatchMut {
         &self,
         absolute_start_offset: u64,
     ) -> Result<(), IggyError> {
-        let mut current_offset = absolute_start_offset;
-        for message in self.iter() {
+        for (current_offset, message) in (absolute_start_offset..).zip(self.iter()) {
             let calculated_checksum = message.calculate_checksum();
             let actual_checksum = message.header().checksum();
             let offset = message.header().offset();
@@ -463,7 +462,6 @@ impl IggyMessagesBatchMut {
                     offset,
                 ));
             }
-            current_offset += 1;
         }
         Ok(())
     }

core/connectors/runtime/Dockerfile

@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 
-ARG RUST_VERSION=1.94
+ARG RUST_VERSION=1.95
 ARG ALPINE_VERSION=3.22
 
 FROM --platform=$BUILDPLATFORM lukemathwalker/cargo-chef:latest-rust-${RUST_VERSION}-alpine AS chef

core/consensus/src/impls.rs

@@ -331,17 +331,15 @@ impl LocalPipeline {
 
         // Verify prepare queue hash chain
         if let Some(head) = self.prepare_queue.front() {
-            let mut expected_op = head.header.op;
             let mut expected_parent = head.header.parent;
 
-            for entry in &self.prepare_queue {
+            for (expected_op, entry) in (head.header.op..).zip(self.prepare_queue.iter()) {
                 let header = &entry.header;
 
                 assert_eq!(header.op, expected_op, "ops must be sequential");
                 assert_eq!(header.parent, expected_parent, "must be hash-chained");
 
                 expected_parent = header.checksum;
-                expected_op += 1;
             }
         }
     }

core/integration/tests/server/scenarios/consumer_group_with_single_client_polling_messages_scenario.rs

@@ -187,7 +187,6 @@ async fn execute_using_none_key(client: &IggyClient) {
     let consumer = Consumer::group(Identifier::named(CONSUMER_GROUP_NAME).unwrap());
     let mut partition_id = 1;
     let mut offset = 0;
-    let mut entity_id = 1;
     for i in 1..=PARTITIONS_COUNT * MESSAGES_COUNT {
         let polled_messages = client
             .poll_messages(
@@ -210,12 +209,8 @@ async fn execute_using_none_key(client: &IggyClient) {
         let message = &polled_messages.messages[0];
         assert_eq!(message.header.offset, offset);
         let payload = from_utf8(&message.payload).unwrap();
-        assert_eq!(
-            payload,
-            &create_extended_message_payload(partition_id, entity_id)
-        );
+        assert_eq!(payload, &create_extended_message_payload(partition_id, i));
         partition_id += 1;
-        entity_id += 1;
         if partition_id > PARTITIONS_COUNT {
             partition_id = 1;
             offset += 1;

core/partitions/src/types.rs

@@ -226,7 +226,7 @@ impl PartitionsConfig {
         topic_id: usize,
         partition_id: usize,
     ) -> String {
-        format!("/tmp/iggy_stub/streams/{stream_id}/topics/{topic_id}/partitions/{partition_id}",)
+        format!("/tmp/iggy_stub/streams/{stream_id}/topics/{topic_id}/partitions/{partition_id}")
     }
 
     /// Constructs the file path for segment messages.

core/server-ng/Dockerfile

@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 
-ARG RUST_VERSION=1.94
+ARG RUST_VERSION=1.95
 ARG ALPINE_VERSION=3.22
 
 # ── from-source path ─────────────────────────────────────────────────────────

core/server/Dockerfile

@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 
-ARG RUST_VERSION=1.94
+ARG RUST_VERSION=1.95
 ARG ALPINE_VERSION=3.22
 
 # ── from-source path ─────────────────────────────────────────────────────────

core/server/src/http/mapper.rs

@@ -46,7 +46,7 @@ pub fn map_users(users: &[&User]) -> Vec<UserInfo> {
         };
         users_data.push(user);
     }
-    users_data.sort_by(|a, b| a.id.cmp(&b.id));
+    users_data.sort_by_key(|u| u.id);
     users_data
 }
 
@@ -97,7 +97,7 @@ pub fn map_clients(clients: &[Client]) -> Vec<iggy_common::ClientInfo> {
         all_clients.push(client);
     }
 
-    all_clients.sort_by(|a, b| a.client_id.cmp(&b.client_id));
+    all_clients.sort_by_key(|c| c.client_id);
     all_clients
 }
 

core/server/src/shard/system/segments.rs

@@ -439,7 +439,7 @@ impl IggyShard {
             (segments, storages, partition.stats.clone())
         };
 
-        for (mut storage, segment) in storages.into_iter().zip(segments.into_iter()) {
+        for (mut storage, segment) in storages.into_iter().zip(segments) {
             let (msg_writer, index_writer) = storage.shutdown();
             let start_offset = segment.start_offset;
 

core/server/src/streaming/partitions/ops_tests.rs

@@ -137,7 +137,7 @@ mod tests {
 
         let mut journal_batch = create_batch(journal_count);
         journal_batch
-            .prepare_for_persistence(0, journal_base, in_flight_size as u32, None)
+            .prepare_for_persistence(0, journal_base, in_flight_size, None)
             .await;
         partition.log.journal_mut().append(journal_batch).unwrap();
 

core/server/src/streaming/partitions/storage.rs

@@ -216,7 +216,7 @@ pub fn load_consumer_offsets(path: &str) -> Result<Vec<ConsumerOffset>, IggyErro
         });
     }
 
-    consumer_offsets.sort_by(|a, b| a.consumer_id.cmp(&b.consumer_id));
+    consumer_offsets.sort_by_key(|o| o.consumer_id);
     Ok(consumer_offsets)
 }
 

examples/rust/src/multi-tenant/consumer/main.rs

@@ -122,11 +122,9 @@ async fn main() -> anyhow::Result<(), Box<dyn Error>> {
 
     print_info("Creating clients for each tenant");
     let mut tenants = Vec::new();
-    let mut tenant_id = 1;
-    for (stream, user) in streams_with_users.into_iter() {
+    for (tenant_id, (stream, user)) in (1u32..).zip(streams_with_users.into_iter()) {
         let client = create_client(&address, &user, PASSWORD).await?;
         tenants.push(Tenant::new(tenant_id, stream, user, client));
-        tenant_id += 1;
     }
 
     if ensure_access {

examples/rust/src/multi-tenant/producer/main.rs

@@ -125,11 +125,9 @@ async fn main() -> anyhow::Result<(), Box<dyn Error>> {
 
     print_info("Creating clients for each tenant");
     let mut tenants = Vec::new();
-    let mut tenant_id = 1;
-    for (stream, user) in streams_with_users.into_iter() {
+    for (tenant_id, (stream, user)) in (1u32..).zip(streams_with_users.into_iter()) {
         let client = create_client(&address, &user, PASSWORD).await?;
         tenants.push(Tenant::new(tenant_id, stream, user, client));
-        tenant_id += 1;
     }
 
     if ensure_access {

rust-toolchain.toml

@@ -16,6 +16,6 @@
 # under the License.
 
 [toolchain]
-channel = "1.94.0"
+channel = "1.95.0"
 components = ["rustfmt", "clippy"]
 profile = "default"