From c57cea41a849607e5962a641d29f038e56ca0e8b Mon Sep 17 00:00:00 2001 From: Reinier Maas Date: Thu, 16 Jul 2026 11:12:22 +0200 Subject: [PATCH] Rust: edition 2024 with resolver 3 --- Cargo.toml | 4 +- libs/opsqueue_python/Cargo.toml | 2 +- libs/opsqueue_python/src/async_util.rs | 2 +- libs/opsqueue_python/src/common.rs | 31 +++++++++--- libs/opsqueue_python/src/consumer.rs | 50 +++++++++++-------- libs/opsqueue_python/src/errors.rs | 18 +++---- libs/opsqueue_python/src/producer.rs | 14 +++--- opsqueue/Cargo.toml | 2 +- opsqueue/app/main.rs | 2 +- opsqueue/src/common/chunk.rs | 6 +-- opsqueue/src/common/errors.rs | 4 +- opsqueue/src/common/submission.rs | 8 +-- opsqueue/src/consumer/client.rs | 24 ++++----- opsqueue/src/consumer/dispatcher/mod.rs | 2 +- opsqueue/src/consumer/dispatcher/reserver.rs | 2 +- opsqueue/src/consumer/server/conn.rs | 8 +-- opsqueue/src/consumer/server/mod.rs | 2 +- opsqueue/src/consumer/server/state.rs | 2 +- opsqueue/src/consumer/strategy.rs | 18 +++++-- opsqueue/src/db/conn.rs | 4 +- opsqueue/src/db/mod.rs | 2 +- opsqueue/src/object_store/mod.rs | 10 ++-- opsqueue/src/producer/client.rs | 4 +- opsqueue/src/prometheus.rs | 52 ++++++++++++++++---- opsqueue/src/server.rs | 15 +++--- opsqueue/src/tracing.rs | 2 +- 26 files changed, 183 insertions(+), 107 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 7bf754dd..eb269fea 100644 --- a/Cargo.toml +++ b/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/", diff --git a/libs/opsqueue_python/Cargo.toml b/libs/opsqueue_python/Cargo.toml index 7cefb854..3b063cd9 100644 --- a/libs/opsqueue_python/Cargo.toml +++ b/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] diff --git a/libs/opsqueue_python/src/async_util.rs b/libs/opsqueue_python/src/async_util.rs index 1ee4c130..a3e4a59c 100644 --- a/libs/opsqueue_python/src/async_util.rs +++ b/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}, }; diff --git a/libs/opsqueue_python/src/common.rs b/libs/opsqueue_python/src/common.rs index 3b3c43db..f24305b4 100644 --- a/libs/opsqueue_python/src/common.rs +++ b/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 + ) } } diff --git a/libs/opsqueue_python/src/consumer.rs b/libs/opsqueue_python/src/consumer.rs index ad6108e5..fe85e0c3 100644 --- a/libs/opsqueue_python/src/consumer.rs +++ b/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::(py)) { diff --git a/libs/opsqueue_python/src/errors.rs b/libs/opsqueue_python/src/errors.rs index f0c68508..c3ddc2da 100644 --- a/libs/opsqueue_python/src/errors.rs +++ b/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> for PyErr { fn from(value: CError) -> Self { - value.0 .0 + value.0.0 } } @@ -141,7 +141,7 @@ impl From> for PyErr { impl From> for PyErr { fn from(value: CError) -> 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> for PyErr { fn from(value: CError) -> 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> for PyErr { fn from(value: CError) -> 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> for PyErr { let ChunkId { submission_id, chunk_index, - } = value.0 .0; + } = value.0.0; ChunkNotFoundError::new_err(( value.0.to_string(), ( diff --git a/libs/opsqueue_python/src/producer.rs b/libs/opsqueue_python/src/producer.rs index 3451cdf7..c60ec56f 100644 --- a/libs/opsqueue_python/src/producer.rs +++ b/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}, }; diff --git a/opsqueue/Cargo.toml b/opsqueue/Cargo.toml index 486ea652..a40cd064 100644 --- a/opsqueue/Cargo.toml +++ b/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" diff --git a/opsqueue/app/main.rs b/opsqueue/app/main.rs index 7461f121..9604c940 100644 --- a/opsqueue/app/main.rs +++ b/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; diff --git a/opsqueue/src/common/chunk.rs b/opsqueue/src/common/chunk.rs index b5b84414..534b6569 100644 --- a/opsqueue/src/common/chunk.rs +++ b/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::*; diff --git a/opsqueue/src/common/errors.rs b/opsqueue/src/common/errors.rs index f30436eb..86d0086f 100644 --- a/opsqueue/src/common/errors.rs +++ b/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. diff --git a/opsqueue/src/common/submission.rs b/opsqueue/src/common/submission.rs index 30c41f7e..c51e9958 100644 --- a/opsqueue/src/common/submission.rs +++ b/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 { - use futures::{future, TryStreamExt}; + use futures::{TryStreamExt, future}; let metadata = query!( r#" SELECT metadata_key, metadata_value FROM submissions_metadata diff --git a/opsqueue/src/consumer/client.rs b/opsqueue/src/consumer/client.rs index b0c74fb1..b92d7d48 100644 --- a/opsqueue/src/consumer/client.rs +++ b/opsqueue/src/consumer/client.rs @@ -3,24 +3,24 @@ use std::{ error::Error, str::FromStr, sync::{ - atomic::{AtomicBool, AtomicUsize, Ordering}, Arc, + atomic::{AtomicBool, AtomicUsize, Ordering}, }, time::Duration, }; use arc_swap::ArcSwapOption; -use futures::{stream::SplitSink, SinkExt, Stream, StreamExt}; +use futures::{SinkExt, Stream, StreamExt, stream::SplitSink}; use http::Uri; use tokio::{net::TcpStream, sync::oneshot::error::RecvError}; use tokio::{ select, - sync::{oneshot, Mutex}, + sync::{Mutex, oneshot}, task::yield_now, }; use tokio_tungstenite::{ - tungstenite::{self, Message}, MaybeTlsStream, WebSocketStream, + tungstenite::{self, Message}, }; use tokio_util::sync::{CancellationToken, DropGuard}; // use tokio_websockets::{MaybeTlsStream, Message, WebSocketStream}; @@ -28,7 +28,7 @@ use tokio_util::sync::{CancellationToken, DropGuard}; use crate::{ common::{ chunk::{self, Chunk, ChunkId}, - errors::{IncorrectUsage, LimitIsZero, E}, + errors::{E, IncorrectUsage, LimitIsZero}, submission::Submission, }, consumer::common::{AsyncServerToClientMessage, Envelope}, @@ -177,14 +177,14 @@ struct InFlightRequests( impl InFlightRequests { fn next_nonce(&self) -> usize { - self.0 .0.fetch_add(1, Ordering::SeqCst) + self.0.0.fetch_add(1, Ordering::SeqCst) } async fn next_nonce_with_oneshot( &self, ) -> (usize, oneshot::Receiver) { let (oneshot_sender, oneshot_receiver) = oneshot::channel(); - let mut guard = self.0 .1.lock().await; + let mut guard = self.0.1.lock().await; // This is called within the lock, so we know the nonce and the oneshot_sender are inserted atomically. let nonce = self.next_nonce(); guard.insert(nonce, oneshot_sender); @@ -195,7 +195,7 @@ impl InFlightRequests { &self, envelop: Envelope, ) -> Result<(), SyncServerToClientResponse> { - let mut in_flight_requests = self.0 .1.lock().await; + let mut in_flight_requests = self.0.1.lock().await; let oneshot_sender = in_flight_requests .remove(&envelop.nonce) .expect("Received response with nonce that matches none of the open requests"); @@ -203,9 +203,9 @@ impl InFlightRequests { } async fn clear(&self) { - let mut in_flight_requests = self.0 .1.lock().await; + let mut in_flight_requests = self.0.1.lock().await; // This is called within the lock, so we know the nonce and the `onceshot_sender`s are cleared atomically. - self.0 .0.store(0, Ordering::SeqCst); + self.0.0.store(0, Ordering::SeqCst); in_flight_requests.clear(); } } @@ -469,7 +469,9 @@ impl Client { pub enum InternalConsumerClientError { #[error("Low-level error in the websocket connection: {0}")] LowLevelWebsocketError(#[from] tokio_tungstenite::tungstenite::Error), - #[error("The oneshot channel to receive a sync response to an earlier request was dropped before a response was received: {0}")] + #[error( + "The oneshot channel to receive a sync response to an earlier request was dropped before a response was received: {0}" + )] OneshotSenderDropped(#[from] RecvError), #[error("Expected the sync response of kind {expected} but received {actual:?}")] UnexpectedSyncResponse { diff --git a/opsqueue/src/consumer/dispatcher/mod.rs b/opsqueue/src/consumer/dispatcher/mod.rs index 26514b68..e3e139a7 100644 --- a/opsqueue/src/consumer/dispatcher/mod.rs +++ b/opsqueue/src/consumer/dispatcher/mod.rs @@ -6,7 +6,7 @@ use crate::{ chunk::{Chunk, ChunkId}, submission::Submission, }, - db::{magic::Bool, Connection, Pool, ReaderPool}, + db::{Connection, Pool, ReaderPool, magic::Bool}, }; use futures::stream::{StreamExt as _, TryStreamExt as _}; use metastate::MetaState; diff --git a/opsqueue/src/consumer/dispatcher/reserver.rs b/opsqueue/src/consumer/dispatcher/reserver.rs index 24d7ab1d..e1d9dbd5 100644 --- a/opsqueue/src/consumer/dispatcher/reserver.rs +++ b/opsqueue/src/consumer/dispatcher/reserver.rs @@ -8,8 +8,8 @@ use std::{ use axum_prometheus::metrics::{counter, gauge}; use moka::{notification::RemovalCause, sync::Cache}; use rustc_hash::FxBuildHasher; -use tokio::sync::mpsc::UnboundedSender; use tokio::sync::Mutex; +use tokio::sync::mpsc::UnboundedSender; use tokio_util::sync::CancellationToken; use tokio_util::time::DelayQueue; diff --git a/opsqueue/src/consumer/server/conn.rs b/opsqueue/src/consumer/server/conn.rs index 5e429b22..4530072f 100644 --- a/opsqueue/src/consumer/server/conn.rs +++ b/opsqueue/src/consumer/server/conn.rs @@ -1,7 +1,7 @@ use std::{ sync::{ - atomic::{AtomicUsize, Ordering}, Arc, + atomic::{AtomicUsize, Ordering}, }, time::Duration, }; @@ -11,8 +11,8 @@ use futures::SinkExt; use tokio::{ select, sync::{ - mpsc::{UnboundedReceiver, UnboundedSender}, Notify, + mpsc::{UnboundedReceiver, UnboundedSender}, }, }; use tokio_util::sync::CancellationToken; @@ -31,7 +31,7 @@ use crate::{ }, }; -use super::{state::ConsumerState, ServerState}; +use super::{ServerState, state::ConsumerState}; #[derive(Debug, Clone)] pub struct RetryReservation { @@ -192,7 +192,7 @@ impl ConsumerConn { _ => { return Err(ConsumerConnError::UnexpectedWSMessageType( anyhow::format_err!("Unexpected message format {:?}", msg), - )) + )); } } diff --git a/opsqueue/src/consumer/server/mod.rs b/opsqueue/src/consumer/server/mod.rs index 1837109c..044a2855 100644 --- a/opsqueue/src/consumer/server/mod.rs +++ b/opsqueue/src/consumer/server/mod.rs @@ -5,9 +5,9 @@ use std::{ }; use axum::{ + Router, extract::{State, WebSocketUpgrade}, routing::get, - Router, }; use axum_prometheus::metrics::{gauge, histogram}; use tokio::{select, sync::Notify}; diff --git a/opsqueue/src/consumer/server/state.rs b/opsqueue/src/consumer/server/state.rs index d9388c29..86158a05 100644 --- a/opsqueue/src/consumer/server/state.rs +++ b/opsqueue/src/consumer/server/state.rs @@ -16,7 +16,7 @@ use crate::consumer::strategy; use super::CompleterMessage; use super::ServerState; -use crate::common::errors::{IncorrectUsage, LimitIsZero, E}; +use crate::common::errors::{E, IncorrectUsage, LimitIsZero}; #[derive(Debug, Clone)] pub struct ConsumerState { diff --git a/opsqueue/src/consumer/strategy.rs b/opsqueue/src/consumer/strategy.rs index 68f0493b..0a9f197c 100644 --- a/opsqueue/src/consumer/strategy.rs +++ b/opsqueue/src/consumer/strategy.rs @@ -75,7 +75,9 @@ impl Strategy { let taken_values: Vec<_> = field.too_high_counts(1).collect(); let taken_values_string = serde_json::to_string(&taken_values).expect("Always valid JSON"); - tracing::trace!("Taken values that are left out of PreferDistinct: {taken_values_string:?}"); + tracing::trace!( + "Taken values that are left out of PreferDistinct: {taken_values_string:?}" + ); qb.push_bind(taken_values_string); } } @@ -96,12 +98,12 @@ pub type ChunkStream<'a> = BoxStream<'a, Result>; #[cfg(test)] #[cfg(feature = "server-logic")] pub mod test { - use crate::common::chunk::ChunkSize; use crate::common::StrategicMetadataMap; + use crate::common::chunk::ChunkSize; use super::*; use itertools::Itertools; - use sqlformat::{format, FormatOptions, QueryParams}; + use sqlformat::{FormatOptions, QueryParams, format}; use sqlx::Row; use sqlx::{QueryBuilder, Sqlite, SqliteConnection}; @@ -127,8 +129,14 @@ pub mod test { fn assert_streaming_query(qb: &sqlx::QueryBuilder<'_, Sqlite>, explained: &str) { let query = qb.sql(); - assert!(!explained.contains("MATERIALIZED"), "Query should contain no materialization, but it did\n\nQuery: {query}\n\nPlan: \n\n {explained}"); - assert!(!explained.contains("B-TREE"), "Query should contain no temporary B-tree construction, but it did.\n\nQuery: {query}\n\nPlan: \n\n{explained}"); + assert!( + !explained.contains("MATERIALIZED"), + "Query should contain no materialization, but it did\n\nQuery: {query}\n\nPlan: \n\n {explained}" + ); + assert!( + !explained.contains("B-TREE"), + "Query should contain no temporary B-tree construction, but it did.\n\nQuery: {query}\n\nPlan: \n\n{explained}" + ); } #[sqlx::test] diff --git a/opsqueue/src/db/conn.rs b/opsqueue/src/db/conn.rs index 30b24ff8..e24f586a 100644 --- a/opsqueue/src/db/conn.rs +++ b/opsqueue/src/db/conn.rs @@ -2,9 +2,9 @@ use std::marker::PhantomData; -use sqlx::{pool::PoolConnection, Sqlite, SqliteConnection}; +use sqlx::{Sqlite, SqliteConnection, pool::PoolConnection}; -use super::{magic::*, Connection}; +use super::{Connection, magic::*}; /// A connection to the database. /// diff --git a/opsqueue/src/db/mod.rs b/opsqueue/src/db/mod.rs index 034ceebd..6b4e10cb 100644 --- a/opsqueue/src/db/mod.rs +++ b/opsqueue/src/db/mod.rs @@ -58,10 +58,10 @@ use std::{marker::PhantomData, num::NonZero, time::Duration}; use futures::future::BoxFuture; use magic::Bool; use sqlx::{ + Connection as _, Sqlite, SqliteConnection, SqlitePool, migrate::MigrateDatabase, sqlite::SqlitePoolOptions, sqlite::{SqliteConnectOptions, SqliteJournalMode, SqliteSynchronous}, - Connection as _, Sqlite, SqliteConnection, SqlitePool, }; use conn::{Conn, NoTransaction, Reader, Tx, Writer}; diff --git a/opsqueue/src/object_store/mod.rs b/opsqueue/src/object_store/mod.rs index bf87edf2..585dbfa1 100644 --- a/opsqueue/src/object_store/mod.rs +++ b/opsqueue/src/object_store/mod.rs @@ -2,9 +2,9 @@ use std::sync::Arc; use crate::common::chunk; use futures::stream::{self, TryStreamExt}; -use object_store::path::Path; use object_store::DynObjectStore; use object_store::ObjectStoreExt; +use object_store::path::Path; use reqwest::Url; use ux::u63; @@ -49,7 +49,9 @@ impl std::fmt::Display for ChunkType { #[derive(thiserror::Error, Debug)] pub enum ChunkRetrievalError { - #[error("Failed to retrieve chunk ({submission_prefix}, {chunk_index}, {chunk_type}) from object store: {source}")] + #[error( + "Failed to retrieve chunk ({submission_prefix}, {chunk_index}, {chunk_type}) from object store: {source}" + )] ObjectStoreError { source: object_store::Error, submission_prefix: Box, @@ -60,7 +62,9 @@ pub enum ChunkRetrievalError { #[derive(thiserror::Error, Debug)] pub enum ChunkStorageError { - #[error("Failed to store chunk ({submission_prefix}, {chunk_index}, {chunk_type}) to object store: {source}")] + #[error( + "Failed to store chunk ({submission_prefix}, {chunk_index}, {chunk_type}) to object store: {source}" + )] ObjectStoreError { source: object_store::Error, submission_prefix: Box, diff --git a/opsqueue/src/producer/client.rs b/opsqueue/src/producer/client.rs index b09e20b0..55ce0209 100644 --- a/opsqueue/src/producer/client.rs +++ b/opsqueue/src/producer/client.rs @@ -6,13 +6,13 @@ use backon::Retryable; use http::StatusCode; use crate::{ + E, common::{ errors::E::{L, R}, errors::{SubmissionNotCancellable, SubmissionNotFound}, submission::{SubmissionId, SubmissionStatus}, }, tracing::CarrierMap, - E, }; use super::common::InsertSubmission; @@ -293,9 +293,9 @@ impl InternalProducerClientError { mod tests { use crate::{ common::{ + StrategicMetadataMap, chunk::ChunkSize, submission::{self, SubmissionStatus}, - StrategicMetadataMap, }, db::{DBPools, WriterPool}, producer::common::ChunkContents, diff --git a/opsqueue/src/prometheus.rs b/opsqueue/src/prometheus.rs index d17ebd8d..d0c3dda5 100644 --- a/opsqueue/src/prometheus.rs +++ b/opsqueue/src/prometheus.rs @@ -5,10 +5,10 @@ //! Note that we explicitly have a separate endpoint to check the queue health, //! which is more fine-grained than Prometheus' way to check whether a service is 'up'. use axum_prometheus::{ - metrics::{describe_counter, describe_gauge, describe_histogram, gauge, Unit}, + AXUM_HTTP_REQUESTS_DURATION_SECONDS, GenericMetricLayer, PrometheusMetricLayer, + metrics::{Unit, describe_counter, describe_gauge, describe_histogram, gauge}, metrics_exporter_prometheus::{Matcher, PrometheusBuilder, PrometheusHandle}, utils::SECONDS_DURATION_BUCKETS, - GenericMetricLayer, PrometheusMetricLayer, AXUM_HTTP_REQUESTS_DURATION_SECONDS, }; use tokio_util::sync::CancellationToken; @@ -45,7 +45,11 @@ pub const CONSUMER_FAIL_CHUNK_DURATION: &str = "consumer_fail_chunk_duration_sec pub const OPERATIONS_BACKLOG_GAUGE: &str = "operations_in_backlog_count"; pub fn describe_metrics() { - describe_counter!(SUBMISSIONS_TOTAL_COUNTER, Unit::Count, "Total count of submissions (in backlog + completed + failed), i.e. total that ever entered the system"); + describe_counter!( + SUBMISSIONS_TOTAL_COUNTER, + Unit::Count, + "Total count of submissions (in backlog + completed + failed), i.e. total that ever entered the system" + ); describe_counter!( SUBMISSIONS_COMPLETED_COUNTER, Unit::Count, @@ -61,7 +65,11 @@ pub fn describe_metrics() { Unit::Count, "Number of submissions cancelled (client-requested cancellation, not failure) permanently" ); - describe_histogram!(SUBMISSIONS_DURATION_COMPLETE_HISTOGRAM, Unit::Seconds, "Time between a submission entering the system and its final chunk being completed. Does not count failed submissions."); + describe_histogram!( + SUBMISSIONS_DURATION_COMPLETE_HISTOGRAM, + Unit::Seconds, + "Time between a submission entering the system and its final chunk being completed. Does not count failed submissions." + ); describe_histogram!( SUBMISSIONS_DURATION_FAIL_HISTOGRAM, Unit::Seconds, @@ -93,22 +101,46 @@ pub fn describe_metrics() { Unit::Count, "Number of chunks skipped (because another chunk in the submission failed)" ); - describe_counter!(CHUNKS_TOTAL_COUNTER, Unit::Count, "Total count of chunks (in backlog + completed + failed), i.e. total that ever entered the system"); + describe_counter!( + CHUNKS_TOTAL_COUNTER, + Unit::Count, + "Total count of chunks (in backlog + completed + failed), i.e. total that ever entered the system" + ); // We could calculate the backlog size from TOTAL - COMPLETED - FAILED - SKIPPED // but since it will commonly be used for checking whether we should autoscale, // it's much nicer to measure/expose it directly - describe_gauge!(CHUNKS_BACKLOG_GAUGE, Unit::Count, "Number of chunks in the backlog. Note that this is a _gauge_ reflecting the accurate state of the DB"); - describe_histogram!(CHUNKS_DURATION_COMPLETED_HISTOGRAM, Unit::Seconds, "How long it took from the moment a chunk was reserved until 'complete_chunk' was called, as measured from Opsqueue (so including network overhead and reading/writing to the object_store to get the chunk data)"); - describe_histogram!(CHUNKS_DURATION_FAILED_HISTOGRAM, Unit::Seconds, "How long it took from the moment a chunk was reserved until 'fail_chunk' was called, as measured from Opsqueue (so including network overhead and reading/writing to the object_store to get the chunk data). Includes chunks that are retried."); + describe_gauge!( + CHUNKS_BACKLOG_GAUGE, + Unit::Count, + "Number of chunks in the backlog. Note that this is a _gauge_ reflecting the accurate state of the DB" + ); + describe_histogram!( + CHUNKS_DURATION_COMPLETED_HISTOGRAM, + Unit::Seconds, + "How long it took from the moment a chunk was reserved until 'complete_chunk' was called, as measured from Opsqueue (so including network overhead and reading/writing to the object_store to get the chunk data)" + ); + describe_histogram!( + CHUNKS_DURATION_FAILED_HISTOGRAM, + Unit::Seconds, + "How long it took from the moment a chunk was reserved until 'fail_chunk' was called, as measured from Opsqueue (so including network overhead and reading/writing to the object_store to get the chunk data). Includes chunks that are retried." + ); - describe_gauge!(RESERVER_CHUNKS_RESERVED_GAUGE, Unit::Count, "Number of chunks currently reserved by the reserver, i.e. being worked on by the consumers"); + describe_gauge!( + RESERVER_CHUNKS_RESERVED_GAUGE, + Unit::Count, + "Number of chunks currently reserved by the reserver, i.e. being worked on by the consumers" + ); describe_gauge!( CONSUMERS_CONNECTED_GAUGE, Unit::Count, "Number of healthy websocket connections between the system and consumers" ); - describe_histogram!(CONSUMER_FETCH_AND_RESERVE_CHUNKS_HISTOGRAM, Unit::Seconds, "Time spent by Opsqueue (SQLite + reserver) to reserve `limit` chunks for a consumer using strategy `strategy`"); + describe_histogram!( + CONSUMER_FETCH_AND_RESERVE_CHUNKS_HISTOGRAM, + Unit::Seconds, + "Time spent by Opsqueue (SQLite + reserver) to reserve `limit` chunks for a consumer using strategy `strategy`" + ); describe_histogram!( CONSUMER_COMPLETE_CHUNK_DURATION, Unit::Seconds, diff --git a/opsqueue/src/server.rs b/opsqueue/src/server.rs index bc6292c8..2fbb4000 100644 --- a/opsqueue/src/server.rs +++ b/opsqueue/src/server.rs @@ -1,13 +1,13 @@ //! Defines the HTTP endpoints that are used by both the `producer` and `consumer` APIs use std::{ any::Any, - sync::{atomic::AtomicBool, Arc}, + sync::{Arc, atomic::AtomicBool}, time::Duration, }; -use axum::{routing::get, Router}; +use axum::{Router, routing::get}; use backon::{BackoffBuilder, FibonacciBuilder}; -use http::{header, Response, StatusCode}; +use http::{Response, StatusCode, header}; use crate::db::DBPools; use tokio::select; @@ -47,15 +47,14 @@ pub async fn serve_producer_and_consumer( match listener.local_addr() { Ok(addr) => { tracing::info!("Server listening on {addr}"); - if let Some(pipe) = config.report_bound_port_pipe.take() { - if let Err(err) = pipe.write_port(addr.port()) { + if let Some(pipe) = config.report_bound_port_pipe.take() + && let Err(err) = pipe.write_port(addr.port()) { tracing::warn!( "Failed to write bound port {} to pipe: {}", addr.port(), err ); } - } } Err(err) => tracing::warn!( "Could not get locally bound address of the server, tried binding on {server_addr}: {err}" @@ -137,7 +136,9 @@ pub fn build_router( } async fn intentionally_panic_for_tests() { - panic!("Boom! A big explosion! This allows us to test the panic handler + trace/sentry integration") + panic!( + "Boom! A big explosion! This allows us to test the panic handler + trace/sentry integration" + ) } pub async fn version_endpoint() -> String { diff --git a/opsqueue/src/tracing.rs b/opsqueue/src/tracing.rs index 705e067c..8d4ce310 100644 --- a/opsqueue/src/tracing.rs +++ b/opsqueue/src/tracing.rs @@ -1,6 +1,6 @@ //! Helpers to read/write OpenTelemetry Tracing contexts from inside submissions stored in the queue use opentelemetry::propagation::TextMapPropagator; -use opentelemetry::{propagation::TextMapCompositePropagator, Context}; +use opentelemetry::{Context, propagation::TextMapCompositePropagator}; use opentelemetry_http::{HeaderExtractor, HeaderInjector}; use opentelemetry_sdk::propagation::{BaggagePropagator, TraceContextPropagator}; use rustc_hash::FxHashMap;