Rust: edition 2024 with resolver 3

Cargo.toml

@@ -1,7 +1,7 @@
 [workspace]
 
-# Resolver 2 is default in the 2021 Edition, but the workspace doesn't know that
-resolver = "2"
+# Resolver 3 is default in the 2024 Edition, but the workspace doesn't know that
+resolver = "3"
 
 members = [
     "opsqueue/",

libs/opsqueue_python/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "opsqueue_python"
 version.workspace = true
-edition = "2021"
+edition = "2024"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 [lib]

libs/opsqueue_python/src/async_util.rs

@@ -4,7 +4,7 @@ use pyo3::{Bound, IntoPyObject, PyAny, PyResult, Python};
 use pyo3_async_runtimes::TaskLocals;
 use std::{
     future::Future,
-    pin::{pin, Pin},
+    pin::{Pin, pin},
     task::{Context, Poll},
 };
 

libs/opsqueue_python/src/common.rs

@@ -4,9 +4,9 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use chrono::{DateTime, Utc};
+use opsqueue::common::StrategicMetadataMap;
 use opsqueue::common::errors::TryFromIntError;
 use opsqueue::common::submission::Metadata;
-use opsqueue::common::StrategicMetadataMap;
 use opsqueue::object_store::{ChunkRetrievalError, ChunkType, ObjectStoreClient};
 use opsqueue::tracing::CarrierMap;
 use pyo3::prelude::*;
@@ -227,7 +227,12 @@ impl Chunk {
             Some(bytes) => (bytes, None),
             None => {
                 let prefix = s.prefix.unwrap();
-                tracing::debug!("Fetching chunk content from object store: submission_id={}, prefix={}, chunk_index={}", c.submission_id, prefix, c.chunk_index);
+                tracing::debug!(
+                    "Fetching chunk content from object store: submission_id={}, prefix={}, chunk_index={}",
+                    c.submission_id,
+                    prefix,
+                    c.chunk_index
+                );
                 let res = object_store_client
                     .retrieve_chunk(&prefix, c.chunk_index, ChunkType::Input)
                     .await?;
@@ -431,16 +436,30 @@ impl SubmissionCompleted {
 #[pymethods]
 impl SubmissionFailed {
     fn __repr__(&self) -> String {
-        format!("SubmissionFailed(id={0}, chunks_total={1}, failed_at={2}, failed_chunk_id={3}, metadata={4:?}, strategic_metadata={5:?})",
-        self.id.__repr__(), self.chunks_total, self.failed_at, self.failed_chunk_id, self.metadata, self.strategic_metadata)
+        format!(
+            "SubmissionFailed(id={0}, chunks_total={1}, failed_at={2}, failed_chunk_id={3}, metadata={4:?}, strategic_metadata={5:?})",
+            self.id.__repr__(),
+            self.chunks_total,
+            self.failed_at,
+            self.failed_chunk_id,
+            self.metadata,
+            self.strategic_metadata
+        )
     }
 }
 
 #[pymethods]
 impl SubmissionCancelled {
     fn __repr__(&self) -> String {
-        format!("SubmissionCancelled(id={0}, chunks_total={1}, chunks_done={2}, metadata={3:?}, strategic_metadata={4:?}, cancelled_at={5})",
-        self.id.__repr__(), self.chunks_total, self.chunks_done, self.metadata, self.strategic_metadata, self.cancelled_at)
+        format!(
+            "SubmissionCancelled(id={0}, chunks_total={1}, chunks_done={2}, metadata={3:?}, strategic_metadata={4:?}, cancelled_at={5})",
+            self.id.__repr__(),
+            self.chunks_total,
+            self.chunks_done,
+            self.metadata,
+            self.strategic_metadata,
+            self.cancelled_at
+        )
     }
 }
 

libs/opsqueue_python/src/consumer.rs

@@ -2,18 +2,18 @@ use std::future::IntoFuture;
 use std::sync::Arc;
 use std::time::Duration;
 
-use futures::{stream, StreamExt, TryStreamExt};
+use futures::{StreamExt, TryStreamExt, stream};
 use opsqueue::{
+    E,
     common::errors::{
-        IncorrectUsage, LimitIsZero,
         E::{self, L, R},
+        IncorrectUsage, LimitIsZero,
     },
     consumer::client::InternalConsumerClientError,
     object_store::{
         ChunkRetrievalError, ChunkStorageError, ChunkType, NewObjectStoreClientError,
         ObjectStoreClient,
     },
-    E,
 };
 use pyo3::{
     create_exception,
@@ -350,18 +350,23 @@ impl ConsumerClient {
             let submission_prefix = chunk.submission_prefix.clone();
             let chunk_index = chunk.chunk_index;
             tracing::debug!(
-            "Running fun for chunk: submission_id={:?}, chunk_index={:?}, submission_prefix={:?}",
-            submission_id,
-            chunk_index,
-            &submission_prefix
-        );
+                "Running fun for chunk: submission_id={:?}, chunk_index={:?}, submission_prefix={:?}",
+                submission_id,
+                chunk_index,
+                &submission_prefix
+            );
             let res = Python::attach(|py| {
                 let res = unbound_fun.bind(py).call1((chunk,))?;
                 res.extract()
             });
             match res {
                 Ok(res) => {
-                    tracing::debug!("Completing chunk: submission_id={:?}, chunk_index={:?}, submission_prefix={:?}", submission_id, chunk_index, &submission_prefix);
+                    tracing::debug!(
+                        "Completing chunk: submission_id={:?}, chunk_index={:?}, submission_prefix={:?}",
+                        submission_id,
+                        chunk_index,
+                        &submission_prefix
+                    );
                     self.complete_chunk_gilless(
                         submission_id,
                         submission_prefix.clone(),
@@ -374,15 +379,20 @@ impl ConsumerClient {
                         CError(R(R(e))) => CError(R(R(R(L(e))))),
                     })?;
                     tracing::debug!(
-                    "Completed chunk: submission_id={:?}, chunk_index={:?}, submission_prefix={:?}",
-                    submission_id,
-                    chunk_index,
-                    &submission_prefix
-                );
+                        "Completed chunk: submission_id={:?}, chunk_index={:?}, submission_prefix={:?}",
+                        submission_id,
+                        chunk_index,
+                        &submission_prefix
+                    );
                 }
                 Err(failure) => {
                     let failure_str = crate::common::format_pyerr(&failure);
-                    tracing::warn!("Failing chunk: submission_id={:?}, chunk_index={:?}, submission_prefix={:?}, reason: {failure_str}", submission_id, chunk_index, &submission_prefix);
+                    tracing::warn!(
+                        "Failing chunk: submission_id={:?}, chunk_index={:?}, submission_prefix={:?}, reason: {failure_str}",
+                        submission_id,
+                        chunk_index,
+                        &submission_prefix
+                    );
                     self.fail_chunk_gilless(
                         submission_id,
                         submission_prefix.clone(),
@@ -394,11 +404,11 @@ impl ConsumerClient {
                         CError(R(e)) => CError(R(R(R(L(e))))),
                     })?;
                     tracing::warn!(
-                    "Failed chunk: submission_id={:?}, chunk_index={:?}, submission_prefix={:?}",
-                    submission_id,
-                    chunk_index,
-                    &submission_prefix
-                );
+                        "Failed chunk: submission_id={:?}, chunk_index={:?}, submission_prefix={:?}",
+                        submission_id,
+                        chunk_index,
+                        &submission_prefix
+                    );
 
                     // On exceptions that are not PyExceptions (but PyBaseExceptions), like KeyboardInterrupt etc, return.
                     if !Python::attach(|py| failure.is_instance_of::<PyException>(py)) {

libs/opsqueue_python/src/errors.rs

@@ -4,11 +4,11 @@ use std::error::Error;
 
 use opsqueue::common::chunk::ChunkId;
 use opsqueue::common::errors::{
-    ChunkNotFound, IncorrectUsage, SubmissionNotCancellable, SubmissionNotFound,
-    UnexpectedOpsqueueConsumerServerResponse, E,
+    ChunkNotFound, E, IncorrectUsage, SubmissionNotCancellable, SubmissionNotFound,
+    UnexpectedOpsqueueConsumerServerResponse,
 };
 use pyo3::exceptions::PyBaseException;
-use pyo3::{import_exception, Bound, PyErr, Python};
+use pyo3::{Bound, PyErr, Python, import_exception};
 
 use crate::common;
 use crate::common::{ChunkIndex, SubmissionId};
@@ -74,7 +74,7 @@ pub struct FatalPythonException(#[from] pub PyErr);
 
 impl From<CError<FatalPythonException>> for PyErr {
     fn from(value: CError<FatalPythonException>) -> Self {
-        value.0 .0
+        value.0.0
     }
 }
 
@@ -141,7 +141,7 @@ impl From<CError<SubmissionNotCancellable>> for PyErr {
 
 impl From<CError<SubmissionNotFound>> for PyErr {
     fn from(value: CError<SubmissionNotFound>) -> Self {
-        let submission_id = value.0 .0;
+        let submission_id = value.0.0;
         SubmissionNotFoundError::new_err(u64::from(submission_id))
     }
 }
@@ -153,15 +153,15 @@ pub struct SubmissionFailed(
 
 impl From<CError<SubmissionFailed>> for PyErr {
     fn from(value: CError<SubmissionFailed>) -> Self {
-        let submission: crate::common::SubmissionFailed = value.0 .0;
-        let chunk: crate::common::ChunkFailed = value.0 .1;
+        let submission: crate::common::SubmissionFailed = value.0.0;
+        let chunk: crate::common::ChunkFailed = value.0.1;
         SubmissionFailedError::new_err((submission, chunk))
     }
 }
 
 impl From<CError<crate::producer::SubmissionNotCompletedYetError>> for PyErr {
     fn from(value: CError<crate::producer::SubmissionNotCompletedYetError>) -> Self {
-        let submission_id = value.0 .0;
+        let submission_id = value.0.0;
         SubmissionNotCompletedYetError::new_err((value.0.to_string(), submission_id))
     }
 }
@@ -171,7 +171,7 @@ impl From<CError<ChunkNotFound>> for PyErr {
         let ChunkId {
             submission_id,
             chunk_index,
-        } = value.0 .0;
+        } = value.0.0;
         ChunkNotFoundError::new_err((
             value.0.to_string(),
             (

libs/opsqueue_python/src/producer.rs

@@ -7,25 +7,23 @@ use pyo3::{
     types::PyIterator,
 };
 
-use futures::{stream::BoxStream, StreamExt, TryStreamExt};
+use futures::{StreamExt, TryStreamExt, stream::BoxStream};
 use opsqueue::{
+    E,
     common::errors::E::{self, L, R},
     common::errors::{SubmissionNotCancellable, SubmissionNotFound},
-    object_store::{ChunksStorageError, NewObjectStoreClientError},
-    producer::client::{Client as ActualClient, InternalProducerClientError},
-};
-use opsqueue::{
-    common::{chunk, submission, StrategicMetadataMap},
+    common::{StrategicMetadataMap, chunk, submission},
     object_store::{ChunkRetrievalError, ChunkType},
+    object_store::{ChunksStorageError, NewObjectStoreClientError},
     producer::ChunkContents,
+    producer::client::{Client as ActualClient, InternalProducerClientError},
     tracing::CarrierMap,
-    E,
 };
 use ux::u63;
 
 use crate::{
     async_util,
-    common::{run_unless_interrupted, start_runtime, SubmissionId, SubmissionStatus},
+    common::{SubmissionId, SubmissionStatus, run_unless_interrupted, start_runtime},
     errors::{self, CError, CPyResult, FatalPythonException},
 };
 

opsqueue/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "opsqueue"
 version.workspace = true
-edition = "2021"
+edition = "2024"
 description = "lightweight batch processing queue for heavy loads"
 repository = "https://github.com/channable/opsqueue"
 license = "MIT"

opsqueue/app/main.rs

@@ -7,7 +7,7 @@ use opentelemetry_sdk::trace::{RandomIdGenerator, Sampler, SdkTracerProvider};
 use opsqueue::{common::submission::db::periodically_cleanup_old, config::Config, prometheus};
 use std::{
     error::Error,
-    sync::{atomic::AtomicBool, Arc},
+    sync::{Arc, atomic::AtomicBool},
     time::Duration,
 };
 use tokio_util::sync::CancellationToken;

opsqueue/src/common/chunk.rs

@@ -218,11 +218,11 @@ impl Chunk {
 #[cfg(feature = "server-logic")]
 pub mod db {
     use super::*;
-    use crate::common::errors::{ChunkNotFound, DatabaseError, SubmissionNotFound, E};
+    use crate::common::errors::{ChunkNotFound, DatabaseError, E, SubmissionNotFound};
     use crate::db::{Connection, True, WriterConnection};
     use axum_prometheus::metrics::{counter, gauge};
-    use sqlx::{query, query_as};
     use sqlx::{QueryBuilder, Sqlite};
+    use sqlx::{query, query_as};
 
     impl<'q> sqlx::Encode<'q, Sqlite> for super::ChunkIndex {
         fn encode_by_ref(
@@ -597,9 +597,9 @@ pub mod db {
 #[cfg(test)]
 #[cfg(feature = "server-logic")]
 pub mod test {
+    use crate::common::StrategicMetadataMap;
     use crate::common::submission::db::insert_submission_raw;
     use crate::common::submission::{Submission, SubmissionStatus};
-    use crate::common::StrategicMetadataMap;
     use crate::db::{Connection as _, WriterPool};
 
     use super::db::*;

opsqueue/src/common/errors.rs

@@ -50,7 +50,9 @@ pub enum SubmissionNotCancellable {
 }
 
 #[derive(Error, Debug)]
-#[error("Unexpected opsqueue consumer server response. This indicates an error inside Opsqueue itself: {0:?}")]
+#[error(
+    "Unexpected opsqueue consumer server response. This indicates an error inside Opsqueue itself: {0:?}"
+)]
 pub struct UnexpectedOpsqueueConsumerServerResponse(pub SyncServerToClientResponse);
 
 /// We roll our own version of `either::E` so that we're not limited by the orphan rule.

opsqueue/src/common/submission.rs

@@ -2,9 +2,9 @@
 use std::fmt::Display;
 use std::time::Duration;
 
-use crate::common::StrategicMetadataMap;
 #[cfg(feature = "server-logic")]
 use crate::E;
+use crate::common::StrategicMetadataMap;
 use chrono::{DateTime, Utc};
 use ux::u63;
 
@@ -270,13 +270,13 @@ impl Submission {
 pub mod db {
     use crate::{
         common::{
-            errors::{DatabaseError, SubmissionNotCancellable, SubmissionNotFound, E},
             StrategicMetadataMap,
+            errors::{DatabaseError, E, SubmissionNotCancellable, SubmissionNotFound},
         },
         db::{Connection, True, WriterConnection, WriterPool},
     };
     use chunk::ChunkSize;
-    use sqlx::{query, Sqlite};
+    use sqlx::{Sqlite, query};
 
     use axum_prometheus::metrics::{counter, histogram};
 
@@ -508,7 +508,7 @@ pub mod db {
         id: SubmissionId,
         mut conn: impl Connection,
     ) -> Result<StrategicMetadataMap, DatabaseError> {
-        use futures::{future, TryStreamExt};
+        use futures::{TryStreamExt, future};
         let metadata = query!(
             r#"
         SELECT metadata_key, metadata_value FROM submissions_metadata