From 813f56a9699bc84c126df379edec33705be6db45 Mon Sep 17 00:00:00 2001 From: Reinier Maas Date: Fri, 17 Jul 2026 15:02:30 +0200 Subject: [PATCH 01/10] Clippy: Deny multiple crate versions --- Cargo.toml | 4 ++++ clippy.toml | 25 +++++++++++++++++++++++++ libs/opsqueue_python/Cargo.toml | 3 +++ opsqueue/Cargo.toml | 3 +++ 4 files changed, 35 insertions(+) create mode 100644 clippy.toml diff --git a/Cargo.toml b/Cargo.toml index 9be2ce9..17b6eae 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -27,6 +27,10 @@ restriction = { level = "allow", priority = -1 } style = { level = "warn", priority = -1 } suspicious = { level = "warn", priority = -1 } +# Prevent duplicate versions of crates from being used in the workspace. If an easy fix is not +# possible, you can add the crate to the `allowed-duplicate-crates` list in `clippy.toml`. +multiple_crate_versions = "deny" + [profile.release] # Full fat LTO makes a big difference in performance, at the cost of a large # compile time hit. You should only use release builds when performance matters. diff --git a/clippy.toml b/clippy.toml new file mode 100644 index 0000000..82475bf --- /dev/null +++ b/clippy.toml @@ -0,0 +1,25 @@ +allowed-duplicate-crates = [ + "cpufeatures", + "foldhash", + "getrandom", + "hashbrown", + "jni-sys", + "nix", + "r-efi", + "rand", + "rand_chacha", + "rand_core", + "redox_syscall", + "thiserror", + "thiserror-impl", + "windows-sys", + "windows-targets", + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_gnullvm", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", +] diff --git a/libs/opsqueue_python/Cargo.toml b/libs/opsqueue_python/Cargo.toml index 3b063cd..67d1054 100644 --- a/libs/opsqueue_python/Cargo.toml +++ b/libs/opsqueue_python/Cargo.toml @@ -37,3 +37,6 @@ once_cell = "1.21.4" # Only currently used for `unsync::OnceCell` as part of PyO # Logging/tracing: pyo3-log = "0.13.3" tracing = { version = "0.1.41", features = ["log"] } + +[lints] +workspace = true diff --git a/opsqueue/Cargo.toml b/opsqueue/Cargo.toml index a40cd06..8ac194d 100644 --- a/opsqueue/Cargo.toml +++ b/opsqueue/Cargo.toml @@ -79,6 +79,9 @@ sqlformat = "0.5.0" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html +[lints] +workspace = true + [dev-dependencies] criterion = {version = "0.8", features = ["async_tokio"]} insta = { version = "1.47.2" } From 02f484d2456e7db8c044284f6061f96d78949fb6 Mon Sep 17 00:00:00 2001 From: Reinier Maas Date: Fri, 17 Jul 2026 15:13:29 +0200 Subject: [PATCH 02/10] Run: `just clippy` --- justfile | 12 ++--- libs/opsqueue_python/src/async_util.rs | 4 +- libs/opsqueue_python/src/common.rs | 41 ++++++++-------- libs/opsqueue_python/src/consumer.rs | 6 +-- libs/opsqueue_python/src/errors.rs | 8 +-- libs/opsqueue_python/src/producer.rs | 11 +++-- opsqueue/app/main.rs | 16 +++--- opsqueue/src/common/chunk.rs | 10 ++-- opsqueue/src/common/mod.rs | 6 +-- opsqueue/src/common/submission.rs | 49 ++++++++++--------- opsqueue/src/config.rs | 7 +-- opsqueue/src/consumer/client.rs | 17 ++++--- opsqueue/src/consumer/common.rs | 6 +-- opsqueue/src/consumer/dispatcher/metastate.rs | 10 ++-- opsqueue/src/consumer/dispatcher/mod.rs | 3 ++ opsqueue/src/consumer/dispatcher/reserver.rs | 5 +- opsqueue/src/consumer/server/conn.rs | 16 +++--- opsqueue/src/consumer/server/mod.rs | 8 +-- opsqueue/src/consumer/server/state.rs | 1 + opsqueue/src/consumer/strategy.rs | 14 +++--- opsqueue/src/db/conn.rs | 2 +- opsqueue/src/db/mod.rs | 20 ++++---- opsqueue/src/lib.rs | 3 +- opsqueue/src/object_store/mod.rs | 8 +-- opsqueue/src/producer/client.rs | 8 +-- opsqueue/src/producer/server.rs | 4 +- opsqueue/src/prometheus.rs | 9 ++-- opsqueue/src/server.rs | 2 +- opsqueue/src/tracing.rs | 10 +++- 29 files changed, 173 insertions(+), 143 deletions(-) diff --git a/justfile b/justfile index 5a5c542..2f74471 100644 --- a/justfile +++ b/justfile @@ -86,12 +86,12 @@ lint-heavy: clippy mypy # Rust static analysis [group('lint')] clippy: - cargo clippy --no-deps --fix --allow-dirty --allow-staged -- -Dwarnings - cargo clippy --no-deps -- -Dwarnings - cargo clippy --no-deps --fix --allow-dirty --allow-staged --no-default-features -- -Dwarnings - cargo clippy --no-deps --no-default-features -- -Dwarnings - cargo clippy --no-deps --fix --allow-dirty --allow-staged --all-features -- -Dwarnings - cargo clippy --no-deps --all-features -- -Dwarnings + cargo clippy --no-deps --all-targets --fix --allow-dirty --allow-staged -- -Dwarnings + cargo clippy --no-deps --all-targets -- -Dwarnings + cargo clippy --no-deps --all-targets --no-default-features --fix --allow-dirty --allow-staged -- -Dwarnings + cargo clippy --no-deps --all-targets --no-default-features -- -Dwarnings + cargo clippy --no-deps --all-targets --all-features --fix --allow-dirty --allow-staged -- -Dwarnings + cargo clippy --no-deps --all-targets --all-features -- -Dwarnings # Python static analysis type-checker [group('lint')] diff --git a/libs/opsqueue_python/src/async_util.rs b/libs/opsqueue_python/src/async_util.rs index a3e4a59..bdf7826 100644 --- a/libs/opsqueue_python/src/async_util.rs +++ b/libs/opsqueue_python/src/async_util.rs @@ -1,4 +1,4 @@ -//! Helpers to bridge async Rust and PyO3's flavour of async Python +//! Helpers to bridge async Rust and `PyO3`'s flavour of async Python //! use pyo3::{Bound, IntoPyObject, PyAny, PyResult, Python}; use pyo3_async_runtimes::TaskLocals; @@ -12,7 +12,7 @@ use std::{ /// /// Essentially `py.detach` but for async code. /// -/// Based on https://pyo3.rs/v0.25.1/async-await.html#release-the-gil-across-await +/// Based on pub struct AsyncAllowThreads(F); pub fn async_detach(fut: F) -> AsyncAllowThreads diff --git a/libs/opsqueue_python/src/common.rs b/libs/opsqueue_python/src/common.rs index f24305b..cefc6e1 100644 --- a/libs/opsqueue_python/src/common.rs +++ b/libs/opsqueue_python/src/common.rs @@ -187,7 +187,7 @@ impl From<&Strategy> for strategy::Strategy { } => { let underlying = strategy::Strategy::from(underlying.get()); strategy::Strategy::PreferDistinct { - meta_key: meta_key.to_string(), + meta_key: meta_key.clone(), underlying: Box::new(underlying), } } @@ -223,22 +223,19 @@ impl Chunk { s: submission::Submission, object_store_client: &ObjectStoreClient, ) -> Result { - let (content, prefix) = match c.input_content { - 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 - ); - let res = object_store_client - .retrieve_chunk(&prefix, c.chunk_index, ChunkType::Input) - .await?; - tracing::debug!("Fetched chunk content: {res:?}"); - (res, Some(prefix)) - } + let (content, prefix) = if let Some(bytes) = c.input_content { (bytes, None) } else { + 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 + ); + let res = object_store_client + .retrieve_chunk(&prefix, c.chunk_index, ChunkType::Input) + .await?; + tracing::debug!("Fetched chunk content: {res:?}"); + (res, Some(prefix)) }; Ok(Chunk { submission_id: c.submission_id.into(), @@ -273,6 +270,7 @@ pub struct ChunkFailed { } impl ChunkFailed { + #[must_use] pub fn from_internal(c: chunk::ChunkFailed, _s: &submission::SubmissionFailed) -> Self { ChunkFailed { submission_id: c.submission_id.into(), @@ -356,7 +354,7 @@ pub enum SubmissionStatus { impl From for SubmissionStatus { fn from(value: opsqueue::common::submission::SubmissionStatus) -> Self { - use opsqueue::common::submission::SubmissionStatus::*; + use opsqueue::common::submission::SubmissionStatus::{InProgress, Completed, Failed, Cancelled}; match value { InProgress(s) => SubmissionStatus::InProgress { submission: s.into(), @@ -515,7 +513,7 @@ pub enum SubmissionNotCancellable { impl From for SubmissionNotCancellable { fn from(value: opsqueue::common::errors::SubmissionNotCancellable) -> Self { - use opsqueue::common::errors::SubmissionNotCancellable::*; + use opsqueue::common::errors::SubmissionNotCancellable::{Completed, Failed, Cancelled}; match value { Completed(s) => SubmissionNotCancellable::Completed { submission: s.into(), @@ -575,7 +573,7 @@ pub async fn check_signals_in_background() -> FatalPythonException { /// Note that we very intentionally use the multi-threaded scheduler /// but with a single thread. /// -/// We **cannot** use the current_thread scheduler, +/// We **cannot** use the `current_thread` scheduler, /// since that would result in the Python task scheduler (e.g. `asyncio`) /// and the Rust task scheduler (`Tokio`) to run on the same thread. /// This seems to work fine, until you end up with a Python future @@ -590,6 +588,7 @@ pub async fn check_signals_in_background() -> FatalPythonException { /// 'Tokio scheduler thread' is preemptive, /// the same problem now no longer occurs: /// Both schedulers are able to make forward progress (even on a 1-CPU machine). +#[must_use] pub fn start_runtime() -> Arc { let runtime = tokio::runtime::Builder::new_multi_thread() .worker_threads(1) @@ -600,7 +599,7 @@ pub fn start_runtime() -> Arc { } /// Formats a Python exception -/// in similar fashion as the traceback.format_exc() +/// in similar fashion as the `traceback.format_exc()` /// would do it. /// /// Internally acquires the GIL! diff --git a/libs/opsqueue_python/src/consumer.rs b/libs/opsqueue_python/src/consumer.rs index fe85e0c..14afbc7 100644 --- a/libs/opsqueue_python/src/consumer.rs +++ b/libs/opsqueue_python/src/consumer.rs @@ -48,11 +48,11 @@ impl ConsumerClient { /// /// :param address: The HTTP address where the opsqueue instance is running. /// - /// :param object_store_url: The URL used to upload/download objects from e.g. GCS. + /// :param `object_store_url`: The URL used to upload/download objects from e.g. GCS. /// use `file:///tmp/my/local/path` to use a local file when running small examples in development. /// use `gs://bucket-name/path/inside/bucket` to connect to GCS in production. /// Supports the formats listed here: - /// :param object_store_options: A list of key-value strings as extra options for the chosen object store. + /// :param `object_store_options`: A list of key-value strings as extra options for the chosen object store. /// For example, for GCS, see #[new] #[pyo3(signature = (address, object_store_url, object_store_options=vec![]))] @@ -240,7 +240,7 @@ impl ConsumerClient { match res { Err(e) => return Err(e), Ok(chunks) if chunks.is_empty() => { - self.sleep_unless_interrupted::(POLL_INTERVAL)? + self.sleep_unless_interrupted::(POLL_INTERVAL)?; } Ok(chunks) => return Ok(chunks), } diff --git a/libs/opsqueue_python/src/errors.rs b/libs/opsqueue_python/src/errors.rs index 760c2d7..76b4172 100644 --- a/libs/opsqueue_python/src/errors.rs +++ b/libs/opsqueue_python/src/errors.rs @@ -39,7 +39,7 @@ import_exception!(opsqueue.exceptions, InternalConsumerClientError); import_exception!(opsqueue.exceptions, InternalProducerClientError); /// A newtype so we can write From/Into implementations turning various error types -/// into PyErr, including those defined in other crates. +/// into `PyErr`, including those defined in other crates. /// /// This follows the 'newtype wrapper' approach from /// @@ -53,7 +53,7 @@ impl From for CError { } /// Result type alias to help with the automatic conversion of error types -/// into PyErr. +/// into `PyErr`. /// /// This follows the 'newtype wrapper' approach from /// @@ -61,13 +61,13 @@ impl From for CError { /// The 'C' stands for 'Convertible'. pub type CPyResult = Result>; -/// Indicates a 'fatal' PyErr: Any Python exception which is _not_ a subclass of `PyException`. +/// Indicates a 'fatal' `PyErr`: Any Python exception which is _not_ a subclass of `PyException`. /// /// These are known as 'fatal' exceptions in Python. /// c.f. /// /// We don't consume/wrap these errors but propagate them, -/// allowing things like KeyboardInterrupt, SystemExit or MemoryError, +/// allowing things like `KeyboardInterrupt`, `SystemExit` or `MemoryError`, /// to trigger cleanup-and-exit. #[derive(thiserror::Error, Debug)] #[error("Fatal Python exception: {0}")] diff --git a/libs/opsqueue_python/src/producer.rs b/libs/opsqueue_python/src/producer.rs index 03dabc3..3a18306 100644 --- a/libs/opsqueue_python/src/producer.rs +++ b/libs/opsqueue_python/src/producer.rs @@ -28,7 +28,7 @@ use crate::{ create_exception!(opsqueue_internal, ProducerClientError, PyException); -const SUBMISSION_POLLING_INTERVAL: Duration = Duration::from_millis(5000); +const SUBMISSION_POLLING_INTERVAL: Duration = Duration::from_secs(5); // NOTE: ProducerClient is reasonably cheap to clone, as most of its fields are behind Arcs. #[pyclass(from_py_object, module = "opsqueue")] @@ -45,11 +45,11 @@ impl ProducerClient { /// /// :param address: The HTTP address where the opsqueue instance is running. /// - /// :param object_store_url: The URL used to upload/download objects from e.g. GCS. + /// :param `object_store_url`: The URL used to upload/download objects from e.g. GCS. /// use `file:///tmp/my/local/path` to use a local file when running small examples in development. /// use `gs://bucket-name/path/inside/bucket` to connect to GCS in production. /// Supports the formats listed here: - /// :param object_store_options: A list of key-value strings as extra options for the chosen object store. + /// :param `object_store_options`: A list of key-value strings as extra options for the chosen object store. /// For example, for GCS, see #[new] #[pyo3(signature = (address, object_store_url, object_store_options=vec![]))] @@ -76,6 +76,7 @@ impl ProducerClient { }) } + #[must_use] pub fn __repr__(&self) -> String { format!( "", @@ -146,7 +147,7 @@ impl ProducerClient { /// Retrieve the status (in progress, completed or failed) of a specific submission. /// - /// The returned SubmissionStatus object also includes the number of chunks finished so far, + /// The returned `SubmissionStatus` object also includes the number of chunks finished so far, /// when the submission was started/completed/failed, etc. /// /// This call does _not_ fetch the submission's chunk contents on its own. @@ -214,7 +215,7 @@ impl ProducerClient { /// Directly inserts a submission without sending the chunks to GCS /// (but immediately embedding them in the DB). - /// NOTE: This does not support StrategicMetadata currently + /// NOTE: This does not support `StrategicMetadata` currently #[pyo3(signature = (chunk_contents, metadata=None, chunk_size=None, otel_trace_carrier=CarrierMap::default()))] pub fn insert_submission_direct( &self, diff --git a/opsqueue/app/main.rs b/opsqueue/app/main.rs index 9604c94..8154c87 100644 --- a/opsqueue/app/main.rs +++ b/opsqueue/app/main.rs @@ -21,7 +21,7 @@ fn main() { // Sentry has to be initialized before starting the Tokio runtime let _sentry_guard = init_sentry(); - async_main() + async_main(); } #[tokio::main] @@ -81,26 +81,26 @@ pub async fn async_main() { ); tokio::select! { - _ = checkpoint_handle => { + () = checkpoint_handle => { tracing::error!("Checkpointing task exited unexpectedly"); }, _ = server_handle => { tracing::error!("Server task exited unexpectedly"); }, - _ = cleanup_handle => { + () = cleanup_handle => { tracing::error!("Cleanup task exited unexpectedly"); }, - _ = prometheus_handle => { + () = prometheus_handle => { tracing::error!("Prometheus metrics calculation task exited unexpectedly"); }, - _ = watchdog_handle => { + () = watchdog_handle => { tracing::error!("Watchdog task exited unexpectedly"); }, res = tokio::signal::ctrl_c() => { match res { - Ok(_) => tracing::warn!("Received Ctrl-C signal"), + Ok(()) => tracing::warn!("Received Ctrl-C signal"), Err(ref err) => tracing::error!(error = err as &dyn Error, "Error while waiting for Ctrl-C signal"), - }; + } }, }; @@ -124,7 +124,7 @@ pub async fn async_main() { /// Starts up the Sentry client to forward errors/panics to it. /// -/// SENTRY_DSN, SENTRY_ENVIRONMENT and SENTRY_RELEASE +/// `SENTRY_DSN`, `SENTRY_ENVIRONMENT` and `SENTRY_RELEASE` /// are expected to be set as environment variables /// (and if unset, Sentry support is turned off) fn init_sentry() -> sentry::ClientInitGuard { diff --git a/opsqueue/src/common/chunk.rs b/opsqueue/src/common/chunk.rs index 740fc61..7d245ab 100644 --- a/opsqueue/src/common/chunk.rs +++ b/opsqueue/src/common/chunk.rs @@ -41,9 +41,9 @@ impl<'a> serde::Deserialize<'a> for ChunkIndex { } } -/// Another name for ChunkIndex, indicating that we're dealing with the _total count_ of chunks. -/// i.e. when you have a value of type ChunkCount, there is a high likelyhood -/// that there are [0..chunk_count) (note the half-open range) chunks to select from. +/// Another name for `ChunkIndex`, indicating that we're dealing with the _total count_ of chunks. +/// i.e. when you have a value of type `ChunkCount`, there is a high likelyhood +/// that there are [`0..chunk_count`) (note the half-open range) chunks to select from. pub type ChunkCount = ChunkIndex; /// The count of entries in each chunk. @@ -81,6 +81,7 @@ impl ChunkIndex { { Self::try_from(index) } + #[must_use] pub fn zero() -> Self { Self(u63::new(0)) } @@ -205,6 +206,7 @@ pub struct ChunkFailed { } impl Chunk { + #[must_use] pub fn new(submission_id: SubmissionId, chunk_index: ChunkIndex, uri: Content) -> Self { Chunk { submission_id, @@ -217,7 +219,7 @@ impl Chunk { #[cfg(feature = "server-logic")] pub mod db { - use super::*; + use super::{ChunkIndex, Chunk, ChunkId, ChunkSize, SubmissionId, ChunkCompleted, DateTime, Utc, ChunkFailed, u63}; use crate::common::errors::{ChunkNotFound, DatabaseError, E, SubmissionNotFound}; use crate::db::{Connection, True, WriterConnection}; use axum_prometheus::metrics::{counter, gauge}; diff --git a/opsqueue/src/common/mod.rs b/opsqueue/src/common/mod.rs index 4a30509..4e06618 100644 --- a/opsqueue/src/common/mod.rs +++ b/opsqueue/src/common/mod.rs @@ -6,16 +6,16 @@ pub mod chunk; pub mod errors; pub mod submission; -/// As values, we support the largest number value SQLite supports by itself, +/// As values, we support the largest number value `SQLite` supports by itself, /// which should be sufficient for most 'ID' fields, which is what this feature is intended for. /// /// If you really need to use strings or UUIDs with a `PreferDistinct` strategy, -/// consider hashing them and using that hash as MetaStateVal. +/// consider hashing them and using that hash as `MetaStateVal`. pub type MetaStateVal = i64; pub type StrategicMetadataMap = FxHashMap; /// Maximum number of submissions a lookup may return. -/// Guarantees: 0 < MaxSubmissions 1 < i64::MAX; +/// Guarantees: 0 < `MaxSubmissions` 1 < `i64::MAX`; #[derive(Debug, Clone, Copy)] pub struct MaxSubmissions(NonZero); diff --git a/opsqueue/src/common/submission.rs b/opsqueue/src/common/submission.rs index 1ede03e..3bc491f 100644 --- a/opsqueue/src/common/submission.rs +++ b/opsqueue/src/common/submission.rs @@ -67,6 +67,7 @@ impl SubmissionId { .expect("Invalid timestamp extracted from snowflake ID") } /// Get the time at which the submission ID was generated as a [`chrono::DateTime`]. + #[must_use] pub fn timestamp(self) -> chrono::DateTime { self.system_time().into() } @@ -221,6 +222,7 @@ impl Default for Submission { } impl Submission { + #[must_use] pub fn new() -> Self { let otel_trace_carrier = crate::tracing::current_context_to_json(); Submission { @@ -235,6 +237,7 @@ impl Submission { } } + #[must_use] pub fn from_vec( chunks: Vec, metadata: Option, @@ -282,7 +285,7 @@ pub mod db { use chunk::ChunkSize; use sqlx::{QueryBuilder, Sqlite, query, query_scalar}; - use super::*; + use super::{chunk, E, SubmissionId, Submission, Chunk, Metadata, ChunkCount, SubmissionStatus, DateTime, Utc, SubmissionCompleted, ChunkIndex, SubmissionFailed, SubmissionCancelled, Duration}; impl<'q> sqlx::Encode<'q, Sqlite> for SubmissionId { fn encode_by_ref( @@ -448,16 +451,16 @@ pub mod db { Err(E::L(e)) => return Err(e), // If the submission ID can't be found, that's too bad, but it's not our problem anymore i guess. Err(E::R(_)) => { - tracing::warn!(%submission_id, "Presumed zero-length submission not found") + tracing::warn!(%submission_id, "Presumed zero-length submission not found"); } // If everything went OK, this *could* still indicate a bug in producer code, so let's just log it. // Our future selves might thank us. Ok(true) => { - tracing::debug!(%submission_id, "Zero-length submission marked as completed") + tracing::debug!(%submission_id, "Zero-length submission marked as completed"); } // This should never happen. If it does, better log it. Ok(false) => { - tracing::warn!(%submission_id, "Zero-length submission wasn't zero-length?!") + tracing::warn!(%submission_id, "Zero-length submission wasn't zero-length?!"); } } } @@ -578,7 +581,8 @@ pub mod db { } } - /// The query in 'lookup_ids_by_strategic_metadata', extracted for testing. + /// The query in '`lookup_ids_by_strategic_metadata`', extracted for testing. + #[must_use] pub fn lookup_ids_by_strategic_metadata_query( strategic_metadata: &StrategicMetadataMap, limit: i64, @@ -794,7 +798,7 @@ pub mod db { match submission_status(id, &mut tx).await { Ok(None) => Err(E::R(E::L(not_found_err))), Ok(Some(SubmissionStatus::InProgress(submission))) => { - panic!("Failed to cancel in progress submission {:?}", submission) + panic!("Failed to cancel in progress submission {submission:?}") } Ok(Some(SubmissionStatus::Completed(submission))) => { Err(E::R(E::R(SubmissionNotCancellable::Completed(submission)))) @@ -845,15 +849,12 @@ pub mod db { ) .fetch_optional(conn.get_inner()) .await?; - match submission_opt { - None => Err(E::R(SubmissionNotFound(id))), - Some(_) => { - counter!(crate::prometheus::SUBMISSIONS_CANCELLED_COUNTER).increment(1); - histogram!(crate::prometheus::SUBMISSIONS_DURATION_CANCEL_HISTOGRAM).record( - crate::prometheus::time_delta_as_f64(Utc::now() - id.timestamp()), - ); - Ok(()) - } + if submission_opt.is_none() { Err(E::R(SubmissionNotFound(id))) } else { + counter!(crate::prometheus::SUBMISSIONS_CANCELLED_COUNTER).increment(1); + histogram!(crate::prometheus::SUBMISSIONS_DURATION_CANCEL_HISTOGRAM).record( + crate::prometheus::time_delta_as_f64(Utc::now() - id.timestamp()), + ); + Ok(()) } } @@ -1066,7 +1067,7 @@ pub mod db { } pub async fn periodically_cleanup_old(db: &WriterPool, max_age: Duration) { - const PERIODIC_CLEANUP_INTERVAL: Duration = Duration::from_secs(60); + const PERIODIC_CLEANUP_INTERVAL: Duration = Duration::from_mins(1); loop { let cutoff = Utc::now() - max_age; let res: sqlx::Result<()> = async move { @@ -1164,7 +1165,7 @@ pub mod test { #[sqlx::test(migrator = "crate::MIGRATOR")] pub async fn test_query_plan_submission_status_in_progress(db: sqlx::SqlitePool) { let mut conn = db.acquire().await.unwrap(); - let query = r#" + let query = r" SELECT id , prefix @@ -1178,7 +1179,7 @@ pub mod test { ) AS strategic_metadata , otel_trace_carrier FROM submissions WHERE id = 1 - "#; + "; let explained = explain_query_plan(query, &mut conn).await; assert_non_regressing_query_plan(query, &explained); @@ -1192,7 +1193,7 @@ pub mod test { #[sqlx::test(migrator = "crate::MIGRATOR")] pub async fn test_query_plan_submission_status_completed(db: sqlx::SqlitePool) { let mut conn = db.acquire().await.unwrap(); - let query = r#" + let query = r" SELECT id , prefix @@ -1206,7 +1207,7 @@ pub mod test { , completed_at , otel_trace_carrier FROM submissions_completed WHERE id = 1 - "#; + "; let explained = explain_query_plan(query, &mut conn).await; assert_non_regressing_query_plan(query, &explained); @@ -1220,7 +1221,7 @@ pub mod test { #[sqlx::test(migrator = "crate::MIGRATOR")] pub async fn test_query_plan_submission_status_failed(db: sqlx::SqlitePool) { let mut conn = db.acquire().await.unwrap(); - let query = r#" + let query = r" SELECT id , prefix @@ -1236,7 +1237,7 @@ pub mod test { , failed_chunk_id , otel_trace_carrier FROM submissions_failed WHERE id = 1 - "#; + "; let explained = explain_query_plan(query, &mut conn).await; assert_non_regressing_query_plan(query, &explained); @@ -1250,7 +1251,7 @@ pub mod test { #[sqlx::test(migrator = "crate::MIGRATOR")] pub async fn test_query_plan_submission_status_cancelled(db: sqlx::SqlitePool) { let mut conn = db.acquire().await.unwrap(); - let query = r#" + let query = r" SELECT id , prefix @@ -1263,7 +1264,7 @@ pub mod test { ) AS strategic_metadata , cancelled_at FROM submissions_cancelled WHERE id = 1 - "#; + "; let explained = explain_query_plan(query, &mut conn).await; assert_non_regressing_query_plan(query, &explained); diff --git a/opsqueue/src/config.rs b/opsqueue/src/config.rs index 92c2d69..68431f3 100644 --- a/opsqueue/src/config.rs +++ b/opsqueue/src/config.rs @@ -40,7 +40,7 @@ pub struct Config { #[arg(long, value_parser = parse_report_bound_port_fd, default_value = "-1")] pub report_bound_port_pipe: ReportBoundPortPipe, - /// Name of the SQLite database file used by this opsqueue. + /// Name of the `SQLite` database file used by this opsqueue. /// /// Configure this to different values when you run multiple opsqueues /// for different purposes. @@ -63,7 +63,7 @@ pub struct Config { #[arg(long, default_value = "10 minutes")] pub reservation_expiration: humantime::Duration, - /// Maximum number of SQLite connections to keep in memory for reading. + /// Maximum number of `SQLite` connections to keep in memory for reading. /// Connections will only be opened when needed. /// /// For the best performance, keep this number as high @@ -80,7 +80,7 @@ pub struct Config { /// Note that special heartbeat messages are only sent if /// there was no other message sent/received within the given interval; /// any message sent/received on the connection will reset the interval - /// as well as the 'missed_heartbeats' counter. + /// as well as the '`missed_heartbeats`' counter. /// /// It is recommended to keep this value in the seconds range. #[arg(long, default_value = "10 seconds")] @@ -141,6 +141,7 @@ impl Default for Config { pub struct ReportBoundPortPipe(Arc>>); impl ReportBoundPortPipe { + #[must_use] pub fn take(&self) -> Option { self.0.lock().expect("No poison").take() } diff --git a/opsqueue/src/consumer/client.rs b/opsqueue/src/consumer/client.rs index 3b2c4ae..75e50e1 100644 --- a/opsqueue/src/consumer/client.rs +++ b/opsqueue/src/consumer/client.rs @@ -54,6 +54,7 @@ type WebsocketTcpStream = WebSocketStream>; pub struct OuterClient(ArcSwapOption, Box); impl OuterClient { + #[must_use] pub fn new(url: &str) -> Self { Self(None.into(), url.into()) } @@ -256,7 +257,7 @@ impl Client { "Careful! Consumer and Server use different Opsqueue library versions! Client is version {} whereas Server is version {}.", crate::version_info(), config.version_info, - ) + ); } let healthy = Arc::new(AtomicBool::new(true)); @@ -278,6 +279,7 @@ impl Client { Ok(me) } + #[must_use] pub fn is_healthy(&self) -> bool { self.healthy.load(std::sync::atomic::Ordering::Relaxed) } @@ -295,7 +297,7 @@ impl Client { loop { yield_now().await; select! { - _ = cancellation_token.cancelled() => break, + () = cancellation_token.cancelled() => break, _ = heartbeat_interval.tick() => { if heartbeats_missed > config.max_missable_heartbeats { tracing::warn!("We missed too many heartbeats! Closing connection and marking client as unhealthy."); @@ -305,11 +307,10 @@ impl Client { let _ = ws_sink.lock().await.close().await; // And now exit the background task, which means all remaining in-flight requests immediately fail as well break - } else { - // NOTE: We don't need to send a heartbeat as client; only the server needs to. - // We only need to track missed heartbeats. - heartbeats_missed += 1; } + // NOTE: We don't need to send a heartbeat as client; only the server needs to. + // We only need to track missed heartbeats. + heartbeats_missed += 1; }, msg = ws_stream.next() => { heartbeat_interval.reset(); @@ -340,7 +341,7 @@ impl Client { match contents { SyncServerToClientResponse::ChunksReserved(reservation) => { let Ok(chunks) = reservation else { continue; }; - for chunk in chunks.iter() { + for chunk in &chunks { let chunk_id = ChunkId::from((chunk.0.submission_id, chunk.0.chunk_index)); // Send message to the server to indicate that we are no longer // reserving this chunk, so that it can be re-reserved by other @@ -530,7 +531,7 @@ mod tests { db_pools, uri.into(), cancellation_token, - Duration::from_secs(60), + Duration::from_mins(1), )); yield_now().await; diff --git a/opsqueue/src/consumer/common.rs b/opsqueue/src/consumer/common.rs index e4614b4..35cd537 100644 --- a/opsqueue/src/consumer/common.rs +++ b/opsqueue/src/consumer/common.rs @@ -38,7 +38,7 @@ pub enum ServerToClientMessage { Init(ConsumerConfig), } -/// Responses to earlier ClientToServerMessages +/// Responses to earlier `ClientToServerMessages` #[derive(Debug, Clone, Serialize, Deserialize)] pub enum SyncServerToClientResponse { #[allow(clippy::type_complexity)] @@ -61,12 +61,12 @@ pub struct Envelope { /// the opsqueue server. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ConsumerConfig { - /// Specifies how long the heartbeat window is. See also [Config::max_missable_heartbeats]. + /// Specifies how long the heartbeat window is. See also [`Config::max_missable_heartbeats`]. /// /// [Config::max_missable_heartbeats]: crate::config::Config::max_missable_heartbeats pub max_missable_heartbeats: usize, /// Specifies how long the heartbeat window is. This value is determined by the - /// [Config::heartbeat_interval] option set on the server. + /// [`Config::heartbeat_interval`] option set on the server. /// /// [Config::heartbeat_interval]: crate::config::Config::heartbeat_interval pub heartbeat_interval: Duration, diff --git a/opsqueue/src/consumer/dispatcher/metastate.rs b/opsqueue/src/consumer/dispatcher/metastate.rs index 08665f6..fad8e4d 100644 --- a/opsqueue/src/consumer/dispatcher/metastate.rs +++ b/opsqueue/src/consumer/dispatcher/metastate.rs @@ -7,6 +7,7 @@ use tracing; pub struct MetaState(DashMap); impl MetaState { + #[must_use] pub fn new() -> Self { Default::default() } @@ -46,10 +47,12 @@ impl MetaState { } } + #[must_use] pub fn is_empty(&self) -> bool { self.0.is_empty() } + #[must_use] pub fn get(&self, key: &str) -> Option> { self.0.get(key) } @@ -57,11 +60,11 @@ impl MetaState { pub type Bytes = Vec; -/// As values, we support the largest number value SQLite supports by itself, +/// As values, we support the largest number value `SQLite` supports by itself, /// which should be sufficient for most 'ID' fields, which is what this feature is intended for. /// /// If you really need to use strings or UUIDs with a `PreferDistinct` strategy, -/// consider hashing them and using that hash as MetaStateVal. +/// consider hashing them and using that hash as `MetaStateVal`. pub type MetaStateVal = i64; #[derive(Debug, Default)] @@ -71,6 +74,7 @@ pub struct MetaStateField { } impl MetaStateField { + #[must_use] pub fn new() -> Self { Default::default() } @@ -113,7 +117,7 @@ impl MetaStateField { self.counts_to_vals.insert(set_entry); } } - }; + } } pub fn is_empty(&self) -> bool { diff --git a/opsqueue/src/consumer/dispatcher/mod.rs b/opsqueue/src/consumer/dispatcher/mod.rs index e3e139a..76e20d8 100644 --- a/opsqueue/src/consumer/dispatcher/mod.rs +++ b/opsqueue/src/consumer/dispatcher/mod.rs @@ -28,6 +28,7 @@ pub struct Dispatcher { } impl Dispatcher { + #[must_use] pub fn new(reservation_expiration: Duration) -> Self { let reserver = Reserver::new(reservation_expiration); let metastate = Arc::new(MetaState::default()); @@ -38,10 +39,12 @@ impl Dispatcher { } } + #[must_use] pub fn metastate(&self) -> &MetaState { &self.metastate } + #[must_use] pub fn reserver(&self) -> &Reserver { &self.reserver } diff --git a/opsqueue/src/consumer/dispatcher/reserver.rs b/opsqueue/src/consumer/dispatcher/reserver.rs index e1d9dbd..5fab097 100644 --- a/opsqueue/src/consumer/dispatcher/reserver.rs +++ b/opsqueue/src/consumer/dispatcher/reserver.rs @@ -37,6 +37,7 @@ where K: Hash + Eq + Send + Sync + Debug + Copy + 'static, V: Send + Sync + Clone + 'static, { + #[must_use] pub fn new(reservation_expiration: Duration) -> Self { let reservations = Cache::builder() .time_to_live(reservation_expiration) @@ -148,7 +149,7 @@ where tracing::trace!("Removing outdated reservation: {entry:?}"); self.reservations.remove(entry.get_ref()); } - _ = async {} => break, + () = async {} => break, } } } @@ -164,7 +165,7 @@ where bg_reserver_handle.run_pending_tasks().await; tokio::select! { () = cancellation_token.cancelled() => break, - _ = tokio::time::sleep(Duration::from_millis(10)) => {} + () = tokio::time::sleep(Duration::from_millis(10)) => {} } } }); diff --git a/opsqueue/src/consumer/server/conn.rs b/opsqueue/src/consumer/server/conn.rs index 4530072..bfd4506 100644 --- a/opsqueue/src/consumer/server/conn.rs +++ b/opsqueue/src/consumer/server/conn.rs @@ -123,7 +123,7 @@ impl ConsumerConn { // When we are notified that we can retry an earlier failing reservation: Some(RetryReservation{nonce, max, strategy}) = self.rx2.recv() => { tracing::debug!("Retrying reservation for nonce {nonce} / {max} / {strategy:?}"); - self.handle_incoming_client_message(Envelope {nonce, contents: ClientToServerMessage::WantToReserveChunks { max, strategy }}).await? + self.handle_incoming_client_message(Envelope {nonce, contents: ClientToServerMessage::WantToReserveChunks { max, strategy }}).await?; } // When a message from elsewhere in the app is received, pass it forward // (Currently there is only one kind of these, namely a chunk reservation having expired) @@ -191,7 +191,7 @@ impl ConsumerConn { } _ => { return Err(ConsumerConnError::UnexpectedWSMessageType( - anyhow::format_err!("Unexpected message format {:?}", msg), + anyhow::format_err!("Unexpected message format {msg:?}"), )); } } @@ -203,8 +203,8 @@ impl ConsumerConn { &mut self, msg: Envelope, ) -> Result<(), ConsumerConnError> { - use ClientToServerMessage::*; - use SyncServerToClientResponse::*; + use ClientToServerMessage::{WantToReserveChunks, CompleteChunk, FailChunk}; + use SyncServerToClientResponse::ChunksReserved; let maybe_response = match msg.contents { WantToReserveChunks { max, strategy } => { let chunks_or_err = self @@ -260,7 +260,7 @@ impl ConsumerConn { }; self.ws_stream .send(ServerToClientMessage::Sync(enveloped_response).into()) - .await? + .await?; } Ok(()) @@ -282,6 +282,7 @@ pub enum ConsumerConnError { } impl ConsumerConnError { + #[must_use] pub fn is_internal_error(&self) -> bool { matches!( self, @@ -321,6 +322,7 @@ pub struct ConsumerConnectionId(String); impl ConsumerConnectionId { /// Generate the consumer id from a UUID. + #[must_use] pub fn uuid() -> ConsumerConnectionId { ConsumerConnectionId(uuid::Uuid::now_v7().to_string()) } @@ -362,9 +364,7 @@ impl ConsumerConnectionId { COLORS .get(i) - .map(|c| String::from(*c)) - .map(ConsumerConnectionId) - .unwrap_or_else(Self::uuid) + .map(|c| String::from(*c)).map_or_else(Self::uuid, ConsumerConnectionId) } } diff --git a/opsqueue/src/consumer/server/mod.rs b/opsqueue/src/consumer/server/mod.rs index 044a285..b804aa3 100644 --- a/opsqueue/src/consumer/server/mod.rs +++ b/opsqueue/src/consumer/server/mod.rs @@ -47,7 +47,7 @@ pub async fn serve_for_tests( tracing::info!("Consumer WebSocket server listening at {server_addr}..."); select! { - _ = cancellation_token.cancelled() => {}, + () = cancellation_token.cancelled() => {}, res = axum::serve(listener, app) => res.expect("Failed to start consumer server"), } } @@ -85,6 +85,7 @@ impl ServerState { } } + #[must_use] pub fn run_background(mut self) -> Self { self.dispatcher .run_pending_tasks_periodically(self.cancellation_token.clone()); @@ -175,6 +176,7 @@ pub struct Completer { } impl Completer { + #[must_use] pub fn new( pool: &db::WriterPool, dispatcher: &Dispatcher, @@ -233,7 +235,7 @@ impl Completer { .await { histogram!(crate::prometheus::CHUNKS_DURATION_COMPLETED_HISTOGRAM) - .record(started_at.elapsed()) + .record(started_at.elapsed()); } histogram!(crate::prometheus::CONSUMER_COMPLETE_CHUNK_DURATION) .record(start.elapsed()); @@ -272,7 +274,7 @@ impl Completer { .await; if let Some(started_at) = maybe_started_at { histogram!(crate::prometheus::CHUNKS_DURATION_FAILED_HISTOGRAM) - .record(started_at.elapsed()) + .record(started_at.elapsed()); } histogram!(crate::prometheus::CONSUMER_FAIL_CHUNK_DURATION) diff --git a/opsqueue/src/consumer/server/state.rs b/opsqueue/src/consumer/server/state.rs index 86158a0..17ad0d7 100644 --- a/opsqueue/src/consumer/server/state.rs +++ b/opsqueue/src/consumer/server/state.rs @@ -38,6 +38,7 @@ impl Drop for ConsumerState { } impl ConsumerState { + #[must_use] pub fn new(server_state: &Arc) -> Self { Self { reservations: Arc::new(Mutex::new(HashSet::new())), diff --git a/opsqueue/src/consumer/strategy.rs b/opsqueue/src/consumer/strategy.rs index 573475d..18e3d98 100644 --- a/opsqueue/src/consumer/strategy.rs +++ b/opsqueue/src/consumer/strategy.rs @@ -38,7 +38,7 @@ impl Strategy { qb: &'a mut QueryBuilder<'a, Sqlite>, metastate: &MetaState, ) -> &'a mut QueryBuilder<'a, Sqlite> { - use Strategy::*; + use Strategy::{Oldest, Newest, Random, PreferDistinct}; match self { Oldest => qb.push("SELECT * FROM chunks ORDER BY submission_id ASC"), Newest => qb.push("SELECT * FROM chunks ORDER BY submission_id DESC"), @@ -57,15 +57,15 @@ impl Strategy { let qb = qb.push(format_args!("WITH inner_{meta_key} AS NOT MATERIALIZED (")); let qb = underlying.build_query_snippet(qb, metastate); qb.push(format_args!( - r#"), + r"), taken_{meta_key} AS ( SELECT * FROM submissions_metadata WHERE - submissions_metadata.metadata_key = "#, + submissions_metadata.metadata_key = ", )); qb.push_bind(meta_key); qb.push( - r#" AND submissions_metadata.metadata_value IN (SELECT value FROM json_each("#, + r" AND submissions_metadata.metadata_value IN (SELECT value FROM json_each(", ); match metastate.get(meta_key) { None => { @@ -113,10 +113,10 @@ pub mod test { ) -> String { let formatted_query = format(qb.sql(), &QueryParams::None, &FormatOptions::default()); - sqlx::raw_sql(&format!("EXPLAIN QUERY PLAN {}", formatted_query)) + sqlx::raw_sql(&format!("EXPLAIN QUERY PLAN {formatted_query}")) .fetch_all(&mut *conn) .await - .unwrap_or_else(|_| panic!("Invalid query: \n{}\n", formatted_query)) + .unwrap_or_else(|_| panic!("Invalid query: \n{formatted_query}\n")) .into_iter() .map(|row| { let id = row.get::("id"); @@ -767,6 +767,6 @@ pub mod test { .await .unwrap(); - assert!(vals1 != vals2) + assert!(vals1 != vals2); } } diff --git a/opsqueue/src/db/conn.rs b/opsqueue/src/db/conn.rs index e24f586..29b5c96 100644 --- a/opsqueue/src/db/conn.rs +++ b/opsqueue/src/db/conn.rs @@ -4,7 +4,7 @@ use std::marker::PhantomData; use sqlx::{Sqlite, SqliteConnection, pool::PoolConnection}; -use super::{Connection, magic::*}; +use super::{Connection, magic::{Bool, True, False}}; /// A connection to the database. /// diff --git a/opsqueue/src/db/mod.rs b/opsqueue/src/db/mod.rs index 6b4e10c..7a1aac1 100644 --- a/opsqueue/src/db/mod.rs +++ b/opsqueue/src/db/mod.rs @@ -176,7 +176,7 @@ where /// /// The write pool only contains a single connection, /// ensuring that any write-contention happens *in async Rust* rather than -/// on a blocking background thread that deals with the blocking SQLite API. +/// on a blocking background thread that deals with the blocking `SQLite` API. /// /// Conversely, we have _many_ read connections, /// ensuring that usually there need not be any waiting for any @@ -222,10 +222,12 @@ impl DBPools { } } /// Access the pool containing the reader connections. + #[must_use] pub fn reader_pool(&self) -> &ReaderPool { &self.read_pool } /// Access the pool containing the writer connection. + #[must_use] pub fn writer_pool(&self) -> &WriterPool { &self.write_pool } @@ -270,6 +272,7 @@ where W: magic::Bool, { /// Wrap the [`SqlitePool`] to add a type-level tag identifying it as either a reader or a writer pool. + #[must_use] pub fn new(pool: SqlitePool) -> Pool { Pool { inner: pool, @@ -301,7 +304,7 @@ impl WriterPool { /// Performs an explicit, non-passive WAL checkpoint /// We use the 'RESTART' strategy, which will do the most work but will briefly block the writer *and* all readers /// -/// c.f. https://www.sqlite.org/pragma.html#pragma_wal_checkpoint +/// c.f. pub async fn perform_explicit_wal_checkpoint(mut conn: impl WriterConnection) -> sqlx::Result<()> { let res: (i32, i32, i32) = sqlx::query_as("PRAGMA wal_checkpoint(RESTART);") .fetch_one(conn.get_inner()) @@ -369,7 +372,7 @@ pub mod magic { impl Bool for False {} } -/// Connects to the SQLite database, creating it if it doesn't exist yet, and migrating it +/// Connects to the `SQLite` database, creating it if it doesn't exist yet, and migrating it /// if it isn't up to date. /// /// This function should be called on app startup; it will panic when the database cannot be @@ -405,17 +408,16 @@ fn db_options(database_filename: &str) -> SqliteConnectOptions { } async fn ensure_db_exists(database_filename: &str) { - if !Sqlite::database_exists(database_filename) + if Sqlite::database_exists(database_filename) .await - .unwrap_or(false) - { + .unwrap_or(false) { + tracing::info!("Starting up using existing sqlite DB {}", database_filename); + } else { tracing::info!("Creating backing sqlite DB {}", database_filename); Sqlite::create_database(database_filename) .await .expect("Could not create backing sqlite DB"); tracing::info!("Finished creating backing sqlite DB {}", database_filename); - } else { - tracing::info!("Starting up using existing sqlite DB {}", database_filename); } } @@ -447,7 +449,7 @@ async fn db_connect_read_pool( /// Connect the writer pool. /// /// It intentionally only contains a single connection -/// as SQLite only allows a single write at a time +/// as `SQLite` only allows a single write at a time async fn db_connect_write_pool(database_filename: &str) -> WriterPool { SqlitePoolOptions::new() .min_connections(1) diff --git a/opsqueue/src/lib.rs b/opsqueue/src/lib.rs index 6668cef..7adfd4c 100644 --- a/opsqueue/src/lib.rs +++ b/opsqueue/src/lib.rs @@ -44,12 +44,13 @@ pub mod config; pub const VERSION_CARGO_SEMVER: &str = env!("CARGO_PKG_VERSION"); #[allow(clippy::const_is_empty)] +#[must_use] pub fn version_info() -> String { format!("v{VERSION_CARGO_SEMVER}") } /// Shared constant with the migrations that all the tests can reference, to avoid the generated /// code bloat as described in -/// https://kobzol.github.io/rust/2026/06/21/optimizing-sqlx-test-rebuild-time.html +/// #[cfg(all(test, feature = "server-logic"))] const MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!(); diff --git a/opsqueue/src/object_store/mod.rs b/opsqueue/src/object_store/mod.rs index 585dbfa..449f267 100644 --- a/opsqueue/src/object_store/mod.rs +++ b/opsqueue/src/object_store/mod.rs @@ -124,7 +124,7 @@ impl ObjectStoreClient { chunk_type: ChunkType, chunk_contents: impl TryStreamExt, Error = anyhow::Error>, ) -> Result { - use ChunksStorageError::*; + use ChunksStorageError::ChunkContentsEvalError; let chunk_count = chunk_contents .try_fold(u63::new(0), |chunk_index, chunk_content| async move { self.store_chunk( @@ -161,7 +161,7 @@ impl ObjectStoreClient { chunk_type: ChunkType, content: Vec, ) -> Result<(), ChunkStorageError> { - use ChunkStorageError::*; + use ChunkStorageError::ObjectStoreError; let path = self.chunk_path(submission_prefix, chunk_index, chunk_type); self.0 .object_store @@ -182,7 +182,7 @@ impl ObjectStoreClient { chunk_index: chunk::ChunkIndex, chunk_type: ChunkType, ) -> Result, ChunkRetrievalError> { - use ChunkRetrievalError::*; + use ChunkRetrievalError::ObjectStoreError; let res = async move { let bytes = self .0 @@ -223,6 +223,7 @@ impl ObjectStoreClient { }) } + #[must_use] pub fn base_path(&self) -> &Path { &self.0.base_path } @@ -239,6 +240,7 @@ impl ObjectStoreClient { )) } + #[must_use] pub fn url(&self) -> &str { &self.0.url } diff --git a/opsqueue/src/producer/client.rs b/opsqueue/src/producer/client.rs index b7c9983..ec67137 100644 --- a/opsqueue/src/producer/client.rs +++ b/opsqueue/src/producer/client.rs @@ -45,6 +45,7 @@ impl Client { /// panics. /// /// Examples: `0.0.0.0:1312`, `my.opsqueue.instance.example.com`, `https://services.example.com/opsqueue` + #[must_use] pub fn new(host: &str) -> Self { let http_client = reqwest::Client::new(); let base_url = match host.split_once("://") { @@ -57,6 +58,7 @@ impl Client { http_client, } } + #[must_use] pub fn base_url(&self) -> &str { &self.base_url } @@ -116,7 +118,7 @@ impl Client { .await } - /// Send a HTTP request to the OpsQueue server to cancel a submission. + /// Send a HTTP request to the `OpsQueue` server to cancel a submission. /// /// Will return an error if the submission is already complete, failed, or /// cancelled, or if the submission could not be found. @@ -305,6 +307,7 @@ pub enum InternalProducerClientError { } impl InternalProducerClientError { + #[must_use] pub fn is_ephemeral(&self) -> bool { match self { // In the case of an unexpected HTTP status error, developer @@ -326,8 +329,7 @@ impl InternalProducerClientError { // Any other status is considered a permanent failure however inner .status() - .map(|status| status.is_server_error()) - .unwrap_or(false) + .is_some_and(|status| status.is_server_error()) } else { // Anything else is a permanent failure. false diff --git a/opsqueue/src/producer/server.rs b/opsqueue/src/producer/server.rs index 2788d07..fb19203 100644 --- a/opsqueue/src/producer/server.rs +++ b/opsqueue/src/producer/server.rs @@ -106,7 +106,7 @@ where /// 200 if the submission was successfully cancelled. /// 404 if the submission could not be found. /// 409 if the submission could not be cancelled. -/// 500 if a DatabaseError occurred. +/// 500 if a `DatabaseError` occurred. async fn cancel_submission( State(state): State, Path(submission_id): Path, @@ -117,7 +117,7 @@ async fn cancel_submission( .await .map_err(|e| ServerError(e.into()).into_response())?; match submission::db::cancel_submission(submission_id, &mut conn).await { - Ok(_) => Ok(()), + Ok(()) => Ok(()), Err(L(db_err)) => Err(ServerError(db_err.into()).into_response()), Err(R(L(not_found_err))) => { Err((StatusCode::NOT_FOUND, Json(not_found_err)).into_response()) diff --git a/opsqueue/src/prometheus.rs b/opsqueue/src/prometheus.rs index d0c3dda..4ad4865 100644 --- a/opsqueue/src/prometheus.rs +++ b/opsqueue/src/prometheus.rs @@ -186,12 +186,13 @@ pub fn setup_prometheus() -> ( (metric_layer, metric_handle) } -/// Returns the number of seconds contained by this TimeDelta as f64, with nanosecond precision. +/// Returns the number of seconds contained by this `TimeDelta` as f64, with nanosecond precision. /// /// Adapted from +#[must_use] pub fn time_delta_as_f64(td: chrono::TimeDelta) -> f64 { const NANOS_PER_SEC: f64 = 1_000_000_000.0; - (td.num_seconds() as f64) + (td.subsec_nanos() as f64) / NANOS_PER_SEC + (td.num_seconds() as f64) + f64::from(td.subsec_nanos()) / NANOS_PER_SEC } /// Calculates the backlog-size metrics used for autoscaling. @@ -217,8 +218,8 @@ pub async fn periodically_calculate_scaling_metrics( const METRICS_INTERVAL: std::time::Duration = std::time::Duration::from_secs(5); loop { tokio::select! { - _ = cancellation_token.cancelled() => break, - _ = tokio::time::sleep(METRICS_INTERVAL) => { + () = cancellation_token.cancelled() => break, + () = tokio::time::sleep(METRICS_INTERVAL) => { if let Err(e) = calculate_scaling_metrics(db_pool).await { tracing::error!("Error calculating scaling metrics: {}", e); } diff --git a/opsqueue/src/server.rs b/opsqueue/src/server.rs index 9f1224a..f11f2ac 100644 --- a/opsqueue/src/server.rs +++ b/opsqueue/src/server.rs @@ -203,7 +203,7 @@ pub async fn app_watchdog( select! { () = cancellation_token.cancelled() => break, - _ = tokio::time::sleep(Duration::from_secs(10)) => {}, + () = tokio::time::sleep(Duration::from_secs(10)) => {}, } } // Set to unhealthy when shutting down diff --git a/opsqueue/src/tracing.rs b/opsqueue/src/tracing.rs index 8d4ce31..9f8a78b 100644 --- a/opsqueue/src/tracing.rs +++ b/opsqueue/src/tracing.rs @@ -5,11 +5,13 @@ use opentelemetry_http::{HeaderExtractor, HeaderInjector}; use opentelemetry_sdk::propagation::{BaggagePropagator, TraceContextPropagator}; use rustc_hash::FxHashMap; +#[must_use] pub fn context_from_headers(headers: &http::HeaderMap) -> Context { let propagator = propagator(); propagator.extract(&HeaderExtractor(headers)) } +#[must_use] pub fn context_to_headers(context: &Context) -> http::HeaderMap { let propagator = propagator(); let mut headers = Default::default(); @@ -17,6 +19,7 @@ pub fn context_to_headers(context: &Context) -> http::HeaderMap { headers } +#[must_use] pub fn current_context_to_json() -> String { use tracing::Span; use tracing_opentelemetry::OpenTelemetrySpanExt; @@ -24,6 +27,7 @@ pub fn current_context_to_json() -> String { context_to_json(&Span::current().context()) } +#[must_use] pub fn context_to_json(context: &Context) -> String { let propagator = propagator(); let mut map = CarrierMap::default(); @@ -31,19 +35,21 @@ pub fn context_to_json(context: &Context) -> String { serde_json::to_string(&map).unwrap_or("{}".to_string()) } +#[must_use] pub fn json_to_context(json: &str) -> Context { let propagator = propagator(); serde_json::from_str(json) - .map(|hashmap: CarrierMap| propagator.extract(&hashmap)) - .unwrap_or(Context::new()) + .map_or(Context::new(), |hashmap: CarrierMap| propagator.extract(&hashmap)) } +#[must_use] pub fn json_to_carrier(json: &str) -> CarrierMap { serde_json::from_str(json).unwrap_or_default() } pub type CarrierMap = FxHashMap; +#[must_use] pub fn propagator() -> TextMapCompositePropagator { TextMapCompositePropagator::new(vec![ Box::new(BaggagePropagator::new()), From adad4cf1b2196c5ee2ce61126d559d0f29e492d6 Mon Sep 17 00:00:00 2001 From: Reinier Maas Date: Fri, 17 Jul 2026 15:23:27 +0200 Subject: [PATCH 03/10] AI: Resolve Clippy warnings --- libs/opsqueue_python/src/producer.rs | 1 - opsqueue/app/main.rs | 5 + opsqueue/src/common/chunk.rs | 83 ++++++++++++- opsqueue/src/common/mod.rs | 6 + opsqueue/src/common/submission.rs | 109 ++++++++++++++++-- opsqueue/src/config.rs | 10 ++ opsqueue/src/consumer/client.rs | 55 ++++++++- opsqueue/src/consumer/dispatcher/metastate.rs | 26 ++--- opsqueue/src/consumer/dispatcher/mod.rs | 22 ++-- opsqueue/src/consumer/dispatcher/reserver.rs | 5 +- opsqueue/src/consumer/server/conn.rs | 9 +- opsqueue/src/consumer/server/mod.rs | 10 ++ opsqueue/src/consumer/server/state.rs | 9 ++ opsqueue/src/consumer/strategy.rs | 4 +- opsqueue/src/db/mod.rs | 29 +++++ opsqueue/src/object_store/mod.rs | 21 +++- opsqueue/src/producer/client.rs | 60 ++++++++-- opsqueue/src/producer/server.rs | 6 + opsqueue/src/prometheus.rs | 12 ++ opsqueue/src/server.rs | 13 ++- opsqueue/src/tracing.rs | 2 +- 21 files changed, 438 insertions(+), 59 deletions(-) diff --git a/libs/opsqueue_python/src/producer.rs b/libs/opsqueue_python/src/producer.rs index 3a18306..d0a3a25 100644 --- a/libs/opsqueue_python/src/producer.rs +++ b/libs/opsqueue_python/src/producer.rs @@ -529,7 +529,6 @@ impl PyChunksIter { let stream = client .object_store_client .retrieve_chunks(prefix, chunks_total, ChunkType::Output) - .await .map_err(CError) .boxed(); Self { diff --git a/opsqueue/app/main.rs b/opsqueue/app/main.rs index 8154c87..0588974 100644 --- a/opsqueue/app/main.rs +++ b/opsqueue/app/main.rs @@ -25,6 +25,11 @@ fn main() { } #[tokio::main] +/// Initialize and run the Opsqueue application runtime. +/// +/// # Panics +/// +/// Panics if database initialization times out. pub async fn async_main() { let config = Box::leak(Box::new(Config::parse())); diff --git a/opsqueue/src/common/chunk.rs b/opsqueue/src/common/chunk.rs index 7d245ab..80acdf0 100644 --- a/opsqueue/src/common/chunk.rs +++ b/opsqueue/src/common/chunk.rs @@ -1,4 +1,4 @@ -//! Dealing with `Chunk``s, the most fine-grained datatype the queue itself works with. +//! Dealing with `Chunk`s, the most fine-grained datatype the queue itself works with. //! //! While the smallest datatype is the 'Operation', the Opsqueue queue binary itself //! only deals with _chunks_ of them. The operations inside of them are hidden and only read/written @@ -75,6 +75,11 @@ impl From> for ChunkSize { } impl ChunkIndex { + /// Create a `ChunkIndex` from an integer-like value. + /// + /// # Errors + /// + /// Returns an error if `index` does not fit within the allowed range. pub fn new(index: T) -> Result where Self: TryFrom, @@ -121,7 +126,7 @@ impl From for i64 { fn from(value: ChunkIndex) -> Self { let inner: u64 = value.0.into(); // Guaranteed to fit positive signed range - inner as i64 + inner.cast_signed() } } @@ -264,6 +269,11 @@ pub mod db { } } + /// Insert one chunk row. + /// + /// # Errors + /// + /// Returns an error if the database operation fails. #[tracing::instrument(skip(conn))] pub async fn insert_chunk(chunk: Chunk, mut conn: impl WriterConnection) -> sqlx::Result<()> { query!( @@ -278,6 +288,11 @@ pub mod db { Ok(()) } + /// Mark a chunk as completed and update related submission state. + /// + /// # Errors + /// + /// Returns an error if the chunk/submission cannot be updated in the database. #[tracing::instrument(skip(conn))] pub async fn complete_chunk( chunk_id: ChunkId, @@ -308,6 +323,10 @@ pub mod db { } /// This function MUST be called inside a transaction. + /// + /// # Errors + /// + /// Returns an error if the chunk transition cannot be persisted. #[tracing::instrument(skip(tx))] pub async fn complete_chunk_raw( chunk_id: ChunkId, @@ -354,6 +373,11 @@ pub mod db { .map(|opt| opt.map(ChunkSize)) } + /// Increment retries for a chunk, or move it to failed state. + /// + /// # Errors + /// + /// Returns an error if the database transaction fails. #[tracing::instrument(skip(conn))] pub async fn retry_or_fail_chunk( chunk_id: ChunkId, @@ -402,6 +426,11 @@ pub mod db { Ok(failed_permanently) } + /// Move an in-progress chunk to failed chunks storage. + /// + /// # Errors + /// + /// Returns an error if any database statement in the savepoint fails. #[tracing::instrument(skip(conn))] pub async fn move_chunk_to_failed_chunks( chunk_id: ChunkId, @@ -437,6 +466,11 @@ pub mod db { Ok(()) } + /// Fetch a chunk by id. + /// + /// # Errors + /// + /// Returns an error if the query fails or the row does not exist. #[tracing::instrument(skip(conn))] pub async fn get_chunk( full_chunk_id: ChunkId, @@ -459,6 +493,11 @@ pub mod db { .await } + /// Fetch a completed chunk by id. + /// + /// # Errors + /// + /// Returns an error if the query fails or the row does not exist. #[tracing::instrument(skip(conn))] pub async fn get_chunk_completed( full_chunk_id: ChunkId, @@ -481,6 +520,11 @@ pub mod db { .await } + /// Fetch a failed chunk by id. + /// + /// # Errors + /// + /// Returns an error if the query fails or the row does not exist. #[tracing::instrument(skip(conn))] pub async fn get_chunk_failed( full_chunk_id: ChunkId, @@ -504,6 +548,11 @@ pub mod db { .await } + /// Insert many chunks in batches. + /// + /// # Errors + /// + /// Returns an error if any batch insert fails. #[tracing::instrument(skip(chunks, conn))] pub async fn insert_many_chunks( chunks: &[Chunk], @@ -531,6 +580,11 @@ pub mod db { Ok(()) } + /// Mark all remaining chunks for a submission as skipped/failed. + /// + /// # Errors + /// + /// Returns an error if the savepoint transaction fails. #[tracing::instrument(skip(conn))] pub async fn skip_remaining_chunks( submission_id: SubmissionId, @@ -558,30 +612,45 @@ pub mod db { Ok(()) } + /// Count chunks currently in progress. + /// + /// # Errors + /// + /// Returns an error if the count query fails. #[tracing::instrument(skip(db))] pub async fn count_chunks(mut db: impl Connection) -> sqlx::Result { let count = sqlx::query!("SELECT COUNT(1) as count FROM chunks;") .fetch_one(db.get_inner()) .await?; - let count = u63::new(count.count as u64); + let count = u63::new(count.count.cast_unsigned()); Ok(count) } + /// Count completed chunks. + /// + /// # Errors + /// + /// Returns an error if the count query fails. #[tracing::instrument(skip(db))] pub async fn count_chunks_completed(mut db: impl Connection) -> sqlx::Result { let count = sqlx::query!("SELECT COUNT(1) as count FROM chunks_completed;") .fetch_one(db.get_inner()) .await?; - let count = u63::new(count.count as u64); + let count = u63::new(count.count.cast_unsigned()); Ok(count) } + /// Count failed chunks. + /// + /// # Errors + /// + /// Returns an error if the count query fails. #[tracing::instrument(skip(db))] pub async fn count_chunks_failed(mut db: impl Connection) -> sqlx::Result { let count = sqlx::query!("SELECT COUNT(1) as count FROM chunks_failed;") .fetch_one(db.get_inner()) .await?; - let count = u63::new(count.count as u64); + let count = u63::new(count.count.cast_unsigned()); Ok(count) } @@ -590,6 +659,10 @@ pub mod db { /// An estimation that returns a slightly too high number, /// since we just count the number of chunks times their chunk size, /// meaning that we over-estimate the size of any 'final' chunk that is not full. + /// + /// # Errors + /// + /// Returns an error if the aggregation query fails. pub async fn count_ops_in_backlog_estimate(mut db: impl Connection) -> sqlx::Result { let count = sqlx::query!("SELECT COALESCE(SUM(chunk_size * 1.0), 0.0) as count FROM chunks INNER JOIN submissions ON chunks.submission_id = submissions.id").fetch_one(db.get_inner()).await?.count; Ok(count) diff --git a/opsqueue/src/common/mod.rs b/opsqueue/src/common/mod.rs index 4e06618..8c8636c 100644 --- a/opsqueue/src/common/mod.rs +++ b/opsqueue/src/common/mod.rs @@ -20,6 +20,12 @@ pub type StrategicMetadataMap = FxHashMap; pub struct MaxSubmissions(NonZero); impl MaxSubmissions { + /// Create a validated `MaxSubmissions`. + /// + /// # Errors + /// + /// Returns an error if `value` is too large to safely convert to `i64` + /// during SQL binding. pub fn new(value: NonZero) -> Result { if u64::from(value) < i64::MAX as u64 { Ok(Self(value)) diff --git a/opsqueue/src/common/submission.rs b/opsqueue/src/common/submission.rs index 3bc491f..9c063e0 100644 --- a/opsqueue/src/common/submission.rs +++ b/opsqueue/src/common/submission.rs @@ -55,6 +55,11 @@ impl SubmissionId { SubmissionId(u63::new(inner)) } /// Access the [`std::time::SystemTime`] at which the ID was generated. + /// + /// # Panics + /// + /// Panics if the timestamp encoded in the snowflake id cannot be represented + /// relative to the configured epoch. pub fn system_time(self) -> std::time::SystemTime { use snowflaked::Snowflake; let inner: u64 = self.0.into(); @@ -101,7 +106,7 @@ impl From for i64 { fn from(value: SubmissionId) -> Self { let inner: u64 = value.0.into(); // Guaranteed to fit positive signed range - inner as i64 + inner.cast_signed() } } @@ -129,7 +134,7 @@ impl TryFrom for SubmissionId { return Err(crate::common::errors::TryFromIntError(())); } - Ok(Self(u63::new(value as u64))) + Ok(Self(u63::new(value.cast_unsigned()))) } } @@ -326,6 +331,11 @@ pub mod db { } #[tracing::instrument(skip(conn))] + /// Insert the submission record into the database. + /// + /// # Errors + /// + /// Returns an error if insertion fails. pub async fn insert_submission_raw( submission: &Submission, mut conn: impl WriterConnection, @@ -349,6 +359,11 @@ pub mod db { Ok(()) } + /// Insert strategic metadata rows for a submission. + /// + /// # Errors + /// + /// Returns an error if insertion of any metadata row fails. pub async fn insert_submission_metadata_raw( submission: &Submission, strategic_metadata: &StrategicMetadataMap, @@ -412,6 +427,14 @@ pub mod db { /// Creates a new submission with the given chunks and inserts it into the database. /// /// If the number of chunks is 0, the submission is marked as completed immediately afterwards. + /// + /// # Panics + /// + /// Panics if the chunk vector length cannot be represented as `ChunkCount`. + /// + /// # Errors + /// + /// Returns an error if insertion or follow-up completion updates fail. #[tracing::instrument(skip(metadata, chunks_contents, conn))] pub async fn insert_submission_from_chunks( prefix: Option, @@ -467,6 +490,11 @@ pub mod db { Ok(submission_id) } + /// Fetch an in-progress submission by id. + /// + /// # Errors + /// + /// Returns an error if the query fails or if the submission is not found. #[tracing::instrument(skip(conn))] pub async fn get_submission( id: SubmissionId, @@ -509,6 +537,10 @@ pub mod db { /// Retrieves the earlier stored strategic metadata. /// /// Primarily for testing and introspection. + /// + /// # Errors + /// + /// Returns an error if the query fails. pub async fn get_submission_strategic_metadata( id: SubmissionId, mut conn: impl Connection, @@ -528,6 +560,11 @@ pub mod db { Ok(metadata) } + /// Look up a submission id by prefix across submission states. + /// + /// # Errors + /// + /// Returns an error if the query fails. #[tracing::instrument(skip(conn))] pub async fn lookup_id_by_prefix( prefix: &str, @@ -550,13 +587,18 @@ pub mod db { Ok(row.map(|row| row.id)) } + /// Look up submission ids matching strategic metadata constraints. + /// + /// # Errors + /// + /// Returns an error if the query fails or too many submissions match. pub async fn lookup_ids_by_strategic_metadata( strategic_metadata: StrategicMetadataMap, max_submissions: MaxSubmissions, mut conn: impl Connection, ) -> Result, E> { // MaxSubmissions provides us with the guarantee this won't overflow. - let limit = (u64::from(max_submissions) + 1) as i64; + let limit = (u64::from(max_submissions) + 1).cast_signed(); // The main query to match on strategic_metadata will fail at run-time // if strategic_metadata is empty, so we handle the empty case here. let ids = if strategic_metadata.is_empty() { @@ -604,7 +646,13 @@ pub mod db { query_builder } + /// Resolve the status of a submission id across all submission tables. + /// + /// # Errors + /// + /// Returns an error if one of the underlying queries fails. #[tracing::instrument(skip(conn))] + #[allow(clippy::too_many_lines)] pub async fn submission_status( id: SubmissionId, mut conn: impl Connection, @@ -763,6 +811,10 @@ pub mod db { /// /// Returns `true` if all chunks were completed and the submission was marked as completed. /// Otherwise, it returns `false`. + /// + /// # Errors + /// + /// Returns an error if the submission cannot be loaded or updated. pub async fn maybe_complete_submission( id: SubmissionId, mut conn: impl WriterConnection, @@ -782,6 +834,17 @@ pub mod db { .await } + /// Cancel a submission if it is still cancellable. + /// + /// # Panics + /// + /// Panics if internal state becomes inconsistent and an in-progress submission + /// cannot be cancelled even though it should still exist. + /// + /// # Errors + /// + /// Returns an error if database operations fail, the submission is missing, + /// or it is already in a terminal non-cancellable state. #[tracing::instrument(skip(conn))] pub async fn cancel_submission( id: SubmissionId, @@ -819,6 +882,10 @@ pub mod db { } /// Do not call directly! Must be called inside a transaction. + /// + /// # Errors + /// + /// Returns an error if cancellation or chunk skipping fails. pub async fn cancel_submission_notx( id: SubmissionId, mut conn: impl WriterConnection, @@ -926,6 +993,11 @@ pub mod db { Ok(()) } + /// Fail a submission and move related chunk state in one transaction. + /// + /// # Errors + /// + /// Returns an error if any of the transactional updates fail. #[tracing::instrument(skip(conn))] pub async fn fail_submission( id: SubmissionId, @@ -942,6 +1014,10 @@ pub mod db { } /// Do not call directly! Must be called inside a transaction. + /// + /// # Errors + /// + /// Returns an error if submission/chunk failure transitions cannot be persisted. pub async fn fail_submission_notx( id: SubmissionId, failed_chunk_index: ChunkIndex, @@ -959,34 +1035,53 @@ pub mod db { Ok(()) } + /// Count in-progress submissions. + /// + /// # Errors + /// + /// Returns an error if the count query fails. #[tracing::instrument(skip(db))] pub async fn count_submissions(mut db: impl Connection) -> sqlx::Result { let count = sqlx::query!("SELECT COUNT(1) as count FROM submissions;") .fetch_one(db.get_inner()) .await?; - Ok(count.count as usize) + Ok(usize::try_from(count.count.cast_unsigned()).unwrap_or(usize::MAX)) } + /// Count completed submissions. + /// + /// # Errors + /// + /// Returns an error if the count query fails. #[tracing::instrument(skip(db))] pub async fn count_submissions_completed(mut db: impl Connection) -> sqlx::Result { let count = sqlx::query!("SELECT COUNT(1) as count FROM submissions_completed;") .fetch_one(db.get_inner()) .await?; - Ok(count.count as usize) + Ok(usize::try_from(count.count.cast_unsigned()).unwrap_or(usize::MAX)) } + /// Count failed submissions. + /// + /// # Errors + /// + /// Returns an error if the count query fails. #[tracing::instrument(skip(db))] pub async fn count_submissions_failed(mut db: impl Connection) -> sqlx::Result { let count = sqlx::query!("SELECT COUNT(1) as count FROM submissions_failed;") .fetch_one(db.get_inner()) .await?; - Ok(count.count as usize) + Ok(usize::try_from(count.count.cast_unsigned()).unwrap_or(usize::MAX)) } /// Transactionally removes all completed/failed submissions, /// including all their chunks and associated strategic metadata. /// /// Submissions/chunks that are neither failed nor completed are not touched. + /// + /// # Errors + /// + /// Returns an error if any cleanup statement in the transaction fails. #[tracing::instrument(skip(conn))] pub async fn cleanup_old( mut conn: impl Connection, @@ -1312,7 +1407,7 @@ pub mod test { let fetched_submission = get_submission(submission.id, &mut conn).await.unwrap(); // When fetched from DB with no metadata rows, json_group_object returns '{}'. let submission = Submission { - strategic_metadata: Default::default(), + strategic_metadata: StrategicMetadataMap::default(), ..submission }; assert_eq!(fetched_submission, submission); diff --git a/opsqueue/src/config.rs b/opsqueue/src/config.rs index 68431f3..26249cf 100644 --- a/opsqueue/src/config.rs +++ b/opsqueue/src/config.rs @@ -142,6 +142,11 @@ pub struct ReportBoundPortPipe(Arc>>); impl ReportBoundPortPipe { #[must_use] + /// Take ownership of the optional bound-port pipe. + /// + /// # Panics + /// + /// Panics if the internal mutex is poisoned. pub fn take(&self) -> Option { self.0.lock().expect("No poison").take() } @@ -151,6 +156,11 @@ impl ReportBoundPortPipe { pub struct BoundPortPipe(File); impl BoundPortPipe { + /// Write the bound port into the configured file descriptor. + /// + /// # Errors + /// + /// Returns an error if writing or flushing the file descriptor fails. pub fn write_port(mut self, port: u16) -> io::Result<()> { self.0.write_all(&u16::to_be_bytes(port))?; self.0.flush() diff --git a/opsqueue/src/consumer/client.rs b/opsqueue/src/consumer/client.rs index 75e50e1..594f736 100644 --- a/opsqueue/src/consumer/client.rs +++ b/opsqueue/src/consumer/client.rs @@ -54,6 +54,7 @@ type WebsocketTcpStream = WebSocketStream>; pub struct OuterClient(ArcSwapOption, Box); impl OuterClient { + /// Construct a lazy, reconnecting consumer client. #[must_use] pub fn new(url: &str) -> Self { Self(None.into(), url.into()) @@ -63,6 +64,16 @@ impl OuterClient { &self.1 } + /// Reserve up to `max` chunks from the server. + /// + /// # Panics + /// + /// Panics if internal lazy initialization invariants are violated. + /// + /// # Errors + /// + /// Returns an error if input usage is invalid, or if websocket/serialization + /// communication with the server fails. pub async fn reserve_chunks( &self, max: usize, @@ -83,6 +94,15 @@ impl OuterClient { res } + /// Mark a chunk as completed. + /// + /// # Panics + /// + /// Panics if internal lazy initialization invariants are violated. + /// + /// # Errors + /// + /// Returns an error if websocket communication with the server fails. pub async fn complete_chunk( &self, id: ChunkId, @@ -102,6 +122,15 @@ impl OuterClient { res } + /// Mark a chunk as failed. + /// + /// # Panics + /// + /// Panics if internal lazy initialization invariants are violated. + /// + /// # Errors + /// + /// Returns an error if websocket communication with the server fails. pub async fn fail_chunk( &self, id: ChunkId, @@ -220,14 +249,20 @@ pub struct Client { } impl Client { + /// Construct and initialize a websocket consumer client. + /// + /// # Errors + /// + /// Returns an error if URL parsing fails, websocket connection setup fails, + /// or the initial server message is invalid. pub async fn new(url: &str) -> anyhow::Result { // Ensure that the given URL is always a websocket URL; tungstenite requires this - let endpoint_url = if url.starts_with("ws://") || url.starts_with("wss://") { + let websocket_url = if url.starts_with("ws://") || url.starts_with("wss://") { format!("{url}/consumer") } else { format!("ws://{url}/consumer") }; - let endpoint_uri = Uri::from_str(&endpoint_url)?; + let endpoint_uri = Uri::from_str(&websocket_url)?; tracing::debug!("Connecting to: {}", endpoint_uri); let in_flight_requests = @@ -426,6 +461,12 @@ impl Client { Ok(()) } + /// Request chunks to reserve. + /// + /// # Errors + /// + /// Returns an error if the server rejects the request or if websocket/serialization + /// communication fails. pub async fn reserve_chunks( &self, max: usize, @@ -439,6 +480,11 @@ impl Client { Ok(chunks) } + /// Report chunk completion to the server. + /// + /// # Errors + /// + /// Returns an error if websocket communication fails. pub async fn complete_chunk( &self, id: ChunkId, @@ -452,6 +498,11 @@ impl Client { .await } + /// Report chunk failure to the server. + /// + /// # Errors + /// + /// Returns an error if websocket communication fails. pub async fn fail_chunk( &self, id: ChunkId, diff --git a/opsqueue/src/consumer/dispatcher/metastate.rs b/opsqueue/src/consumer/dispatcher/metastate.rs index fad8e4d..f96534c 100644 --- a/opsqueue/src/consumer/dispatcher/metastate.rs +++ b/opsqueue/src/consumer/dispatcher/metastate.rs @@ -9,17 +9,17 @@ pub struct MetaState(DashMap); impl MetaState { #[must_use] pub fn new() -> Self { - Default::default() + Self::default() } - pub fn increment(&self, key: &str, val: &MetaStateVal) { + pub fn increment(&self, key: &str, val: MetaStateVal) { match self.0.get(key) { Some(meta_state_field) => meta_state_field.increment(val), None => self.0.entry(key.to_string()).or_default().increment(val), } } - pub fn decrement(&self, key: &str, val: &MetaStateVal) { + pub fn decrement(&self, key: &str, val: MetaStateVal) { let ripe_for_removal = { if let Some(meta_state_field) = self.0.get(key) { meta_state_field.decrement(val); @@ -76,11 +76,11 @@ pub struct MetaStateField { impl MetaStateField { #[must_use] pub fn new() -> Self { - Default::default() + Self::default() } - fn increment(&self, val: &MetaStateVal) { - match self.vals_to_counts.entry(*val) { + fn increment(&self, val: MetaStateVal) { + match self.vals_to_counts.entry(val) { Entry::Vacant(entry) => { self.counts_to_vals.insert((1, *entry.key())); entry.insert(1); @@ -97,8 +97,8 @@ impl MetaStateField { } } - fn decrement(&self, val: &MetaStateVal) { - match self.vals_to_counts.entry(*val) { + fn decrement(&self, val: MetaStateVal) { + match self.vals_to_counts.entry(val) { Entry::Vacant(_entry) => { unreachable!() } @@ -149,12 +149,12 @@ mod tests { let key = "company_id"; let mut vals: Vec<_> = (0..n_operations) .map(|x| x % n_groups) - .map(|val| val as i64) + .map(|val| i64::try_from(val).expect("test value fits into i64")) .collect(); // Increment in one order vals.shuffle(&mut rand::rng()); - for val in &vals { + for &val in &vals { sut.increment(key, val); } @@ -169,7 +169,7 @@ mod tests { // Decrement in a different order vals.shuffle(&mut rand::rng()); - for val in &vals { + for &val in &vals { sut.decrement(key, val); } @@ -190,9 +190,9 @@ mod tests { for val in vals { let sut = sut.clone(); task_set.spawn(async move { - sut.increment(key, &val); + sut.increment(key, val); tokio::task::yield_now().await; - sut.decrement(key, &val); + sut.decrement(key, val); }); } task_set.join_all().await; diff --git a/opsqueue/src/consumer/dispatcher/mod.rs b/opsqueue/src/consumer/dispatcher/mod.rs index 76e20d8..d275314 100644 --- a/opsqueue/src/consumer/dispatcher/mod.rs +++ b/opsqueue/src/consumer/dispatcher/mod.rs @@ -51,16 +51,21 @@ impl Dispatcher { fn insert_metadata(&self, metadata: &StrategicMetadataMap) { for (name, value) in metadata { - self.metastate.increment(name, value); + self.metastate.increment(name, *value); } } fn remove_metadata(&self, metadata: &StrategicMetadataMap) { for (name, value) in metadata { - self.metastate.decrement(name, value); + self.metastate.decrement(name, *value); } } + /// Fetch chunks according to strategy and reserve them in memory. + /// + /// # Errors + /// + /// Returns an error if querying or joining submission info from the database fails. pub async fn fetch_and_reserve_chunks( &self, pool: &ReaderPool, @@ -75,24 +80,25 @@ impl Dispatcher { .build_query_as() .fetch(conn.get_inner()); stream - .try_filter_map(|chunk| self.reserve_chunk(chunk, stale_chunks_notifier)) + .try_filter_map(|chunk| { + futures::future::ready(Ok(self.reserve_chunk(chunk, stale_chunks_notifier))) + }) .and_then(|chunk| self.join_chunk_with_submission_info(chunk, pool)) .take(limit) .try_collect() .await } - async fn reserve_chunk( + fn reserve_chunk( &self, chunk: Chunk, stale_chunks_notifier: &UnboundedSender, - ) -> Result, E> { + ) -> Option { let chunk_id = ChunkId::from((chunk.submission_id, chunk.chunk_index)); - let val = self + self .reserver .try_reserve(chunk_id, chunk_id, stale_chunks_notifier) - .map(|_| chunk); - Ok(val) + .map(|_| chunk) } async fn join_chunk_with_submission_info( diff --git a/opsqueue/src/consumer/dispatcher/reserver.rs b/opsqueue/src/consumer/dispatcher/reserver.rs index 5fab097..0699857 100644 --- a/opsqueue/src/consumer/dispatcher/reserver.rs +++ b/opsqueue/src/consumer/dispatcher/reserver.rs @@ -52,7 +52,7 @@ where // We're not worried about HashDoS as the consumers are trusted code, // so let's use a faster hash than SipHash .build_with_hasher(rustc_hash::FxBuildHasher); - let pending_removals = Default::default(); + let pending_removals = Arc::default(); Reserver { reservations, pending_removals, @@ -133,8 +133,9 @@ where // By running this immediately after 'run_pending_tasks', // we can be reasonably sure that the count is accurate (doesn't include expired entries), // c.f. documentation of moka::sync::Cache::entry_count. + #[allow(clippy::cast_precision_loss)] gauge!(crate::prometheus::RESERVER_CHUNKS_RESERVED_GAUGE) - .set(self.reservations.entry_count() as u32); + .set(self.reservations.entry_count() as f64); } /// Purges all reservations that were finished with `delayed: true` earlier, diff --git a/opsqueue/src/consumer/server/conn.rs b/opsqueue/src/consumer/server/conn.rs index bfd4506..82ee6df 100644 --- a/opsqueue/src/consumer/server/conn.rs +++ b/opsqueue/src/consumer/server/conn.rs @@ -96,6 +96,10 @@ impl ConsumerConn { /// Runs the consumer websocket connection loop /// Blocks until the loop is stopped (because the connection is closed intentionally or by network failure). + /// + /// # Errors + /// + /// Returns an error when a websocket protocol/transport failure occurs. pub async fn run(mut self) -> Result<(), ConsumerConnError> { self.initialize().await?; loop { @@ -110,10 +114,9 @@ impl ConsumerConn { // When a normal message is received, handle it msg = self.ws_stream.recv() => { match msg { - // Socket closed (stream closed before receiving WS 'close' message, ungraceful shutdown) - None => return Ok(()), // Socket received a close message (graceful WS shutdown) - Some(Ok(Message::Close(_))) => return Ok(()), + // Socket closed (stream closed before receiving WS 'close' message, ungraceful shutdown) + None | Some(Ok(Message::Close(_))) => return Ok(()), // Socket had a problem (protocol violation, sending too much data, closed, etc.) Some(Err(err)) => return Err(ConsumerConnError::LowLevelWebsocketError(err)), // Socket received a normal message diff --git a/opsqueue/src/consumer/server/mod.rs b/opsqueue/src/consumer/server/mod.rs index b804aa3..cc0166c 100644 --- a/opsqueue/src/consumer/server/mod.rs +++ b/opsqueue/src/consumer/server/mod.rs @@ -24,6 +24,11 @@ use super::dispatcher::Dispatcher; pub mod conn; pub mod state; +/// Run a test-only consumer websocket server. +/// +/// # Panics +/// +/// Panics if binding the server socket fails or if serving fails unexpectedly. pub async fn serve_for_tests( pool: DBPools, server_addr: Box, @@ -86,6 +91,11 @@ impl ServerState { } #[must_use] + /// Start background maintenance tasks for the consumer server state. + /// + /// # Panics + /// + /// Panics if called more than once for the same state instance. pub fn run_background(mut self) -> Self { self.dispatcher .run_pending_tasks_periodically(self.cancellation_token.clone()); diff --git a/opsqueue/src/consumer/server/state.rs b/opsqueue/src/consumer/server/state.rs index 17ad0d7..4288a1d 100644 --- a/opsqueue/src/consumer/server/state.rs +++ b/opsqueue/src/consumer/server/state.rs @@ -48,6 +48,15 @@ impl ConsumerState { #[tracing::instrument(skip(self, stale_chunks_notifier))] #[allow(clippy::type_complexity)] + /// Fetch and reserve chunks for this connection state. + /// + /// # Panics + /// + /// Panics if the reservations mutex is poisoned. + /// + /// # Errors + /// + /// Returns an error for invalid `limit` values or underlying database failures. pub async fn fetch_and_reserve_chunks( &mut self, strategy: strategy::Strategy, diff --git a/opsqueue/src/consumer/strategy.rs b/opsqueue/src/consumer/strategy.rs index 18e3d98..c510c39 100644 --- a/opsqueue/src/consumer/strategy.rs +++ b/opsqueue/src/consumer/strategy.rs @@ -751,7 +751,7 @@ pub mod test { let mut conn = db_pools.reader_conn().await.unwrap(); let mut query_builder = QueryBuilder::default(); let vals1: Vec = Strategy::Random - .build_query(&mut query_builder, &Default::default()) + .build_query(&mut query_builder, &MetaState::default()) .build_query_as() .fetch(conn.get_inner()) .try_collect() @@ -760,7 +760,7 @@ pub mod test { let mut query_builder = QueryBuilder::default(); let vals2: Vec = Strategy::Random - .build_query(&mut query_builder, &Default::default()) + .build_query(&mut query_builder, &MetaState::default()) .build_query_as() .fetch(conn.get_inner()) .try_collect() diff --git a/opsqueue/src/db/mod.rs b/opsqueue/src/db/mod.rs index 7a1aac1..5b04f0e 100644 --- a/opsqueue/src/db/mod.rs +++ b/opsqueue/src/db/mod.rs @@ -234,6 +234,10 @@ impl DBPools { /// Access a reader connection. /// /// Such a connection can't be used to make changes to the state in the database. + /// + /// # Errors + /// + /// Returns an error if acquiring a reader connection fails. pub async fn reader_conn(&self) -> sqlx::Result> { self.read_pool.reader_conn().await } @@ -241,6 +245,10 @@ impl DBPools { /// /// This connection can be used both for functions requiring read-only access and for functions /// that make changes to the state in the database. + /// + /// # Errors + /// + /// Returns an error if acquiring the writer connection fails. pub async fn writer_conn(&self) -> sqlx::Result> { self.write_pool.writer_conn().await } @@ -280,6 +288,10 @@ where } } /// Acquire a new connection from the underlying pool and wrap it in our typed connection. + /// + /// # Errors + /// + /// Returns an error if pool acquisition fails. pub async fn acquire(&self) -> sqlx::Result> { self.inner.acquire().await.map(Conn::new) } @@ -291,10 +303,19 @@ impl WriterPool { /// See [`DBPools`] for further explanation about readers and writers. /// /// [`DBPools`]: DBPools + /// + /// # Errors + /// + /// Returns an error if acquiring the connection fails. pub async fn writer_conn(&self) -> sqlx::Result> { self.acquire().await } + /// Perform an explicit WAL checkpoint using the writer connection. + /// + /// # Errors + /// + /// Returns an error if connection acquisition or checkpoint execution fails. pub async fn perform_explicit_wal_checkpoint(&self) -> sqlx::Result<()> { let conn = self.writer_conn().await?; perform_explicit_wal_checkpoint(conn).await @@ -305,6 +326,10 @@ impl WriterPool { /// We use the 'RESTART' strategy, which will do the most work but will briefly block the writer *and* all readers /// /// c.f. +/// +/// # Errors +/// +/// Returns an error if the checkpoint query fails. pub async fn perform_explicit_wal_checkpoint(mut conn: impl WriterConnection) -> sqlx::Result<()> { let res: (i32, i32, i32) = sqlx::query_as("PRAGMA wal_checkpoint(RESTART);") .fetch_one(conn.get_inner()) @@ -319,6 +344,10 @@ impl ReaderPool { /// See [`DBPools`] for further explanation about readers and writers. /// /// [`DBPools`]: DBPools + /// + /// # Errors + /// + /// Returns an error if acquiring the connection fails. pub async fn reader_conn(&self) -> sqlx::Result> { self.acquire().await } diff --git a/opsqueue/src/object_store/mod.rs b/opsqueue/src/object_store/mod.rs index 449f267..12ee43e 100644 --- a/opsqueue/src/object_store/mod.rs +++ b/opsqueue/src/object_store/mod.rs @@ -105,6 +105,10 @@ impl ObjectStoreClient { /// /// The given `object_store_url` recognizes the formats detailed [here](https://docs.rs/object_store/0.11.1/object_store/enum.ObjectStoreScheme.html#method.parse). /// Most importantly, we support GCS (for production usage) and local file systems (for testing). + /// + /// # Errors + /// + /// Returns an error if the URL cannot be parsed or if object store initialization fails. pub fn new( object_store_url: &str, options: Vec<(String, String)>, @@ -118,6 +122,11 @@ impl ObjectStoreClient { }))) } + /// Store a stream of chunks and return the number of stored chunks. + /// + /// # Errors + /// + /// Returns an error if evaluating the stream or uploading any chunk fails. pub async fn store_chunks( &self, submission_prefix: &str, @@ -154,6 +163,11 @@ impl ObjectStoreClient { Ok(chunk_count) } + /// Store one chunk in object storage. + /// + /// # Errors + /// + /// Returns an error if uploading fails. pub async fn store_chunk( &self, submission_prefix: &str, @@ -176,6 +190,11 @@ impl ObjectStoreClient { Ok(()) } + /// Retrieve one chunk from object storage. + /// + /// # Errors + /// + /// Returns an error if the object cannot be read. pub async fn retrieve_chunk( &self, submission_prefix: &str, @@ -202,7 +221,7 @@ impl ObjectStoreClient { chunk_type, }) } - pub async fn retrieve_chunks>( + pub fn retrieve_chunks>( &self, submission_prefix: Prefix, chunk_count: u63, diff --git a/opsqueue/src/producer/client.rs b/opsqueue/src/producer/client.rs index ec67137..1dc6c5b 100644 --- a/opsqueue/src/producer/client.rs +++ b/opsqueue/src/producer/client.rs @@ -45,6 +45,10 @@ impl Client { /// panics. /// /// Examples: `0.0.0.0:1312`, `my.opsqueue.instance.example.com`, `https://services.example.com/opsqueue` + /// + /// # Panics + /// + /// Panics if `host` includes a URL scheme other than `http` or `https`. #[must_use] pub fn new(host: &str) -> Self { let http_client = reqwest::Client::new(); @@ -66,6 +70,11 @@ impl Client { /// known to the server. /// /// This uses the `/producer/submissions/count` endpoint. + /// + /// # Errors + /// + /// Returns an error if the HTTP request fails, the server returns a non-success + /// status, or if the response body cannot be decoded. pub async fn count_submissions(&self) -> Result { (|| async { let base_url = &self.base_url; @@ -91,6 +100,11 @@ impl Client { /// Create a new submission for consumers to pick up. /// /// This uses the POST `/producer/submissions` endpoint. + /// + /// # Errors + /// + /// Returns an error if the request fails, the server returns a non-success status, + /// or the submission id in the response cannot be decoded. pub async fn insert_submission( &self, submission: &InsertSubmission, @@ -122,6 +136,11 @@ impl Client { /// /// Will return an error if the submission is already complete, failed, or /// cancelled, or if the submission could not be found. + /// + /// # Errors + /// + /// Returns an error if the submission was not found, was not cancellable, + /// or if the underlying HTTP/serialization layer fails. pub async fn cancel_submission( &self, submission_id: SubmissionId, @@ -166,8 +185,7 @@ impl Client { }) .retry(retry_policy()) .when(|e| match e { - L(_) => false, - R(L(_)) => false, + L(_) | R(L(_)) => false, R(R(client_err)) => client_err.is_ephemeral(), }) .notify(|err, dur| { @@ -179,6 +197,11 @@ impl Client { /// Get the status of an existing submission identified by its `submission_id`. /// /// This uses the GET `/producer/submissions` endpoint. + /// + /// # Errors + /// + /// Returns an error if the request fails, the server responds with a non-success + /// status, or the response payload cannot be decoded. pub async fn get_submission( &self, submission_id: SubmissionId, @@ -203,6 +226,12 @@ impl Client { .await } + /// Look up a submission id by prefix. + /// + /// # Errors + /// + /// Returns an error if the request fails, the server responds with a non-success + /// status, or the response payload cannot be decoded. pub async fn lookup_submission_id_by_prefix( &self, prefix: &str, @@ -229,6 +258,12 @@ impl Client { .await } + /// Look up submission ids by strategic metadata. + /// + /// # Errors + /// + /// Returns an error if too many submissions match, if the request fails, + /// if the server responds with an unexpected status, or if decoding fails. pub async fn lookup_submission_ids_by_strategic_metadata( &self, strategic_metadata: &StrategicMetadataMap, @@ -283,6 +318,10 @@ impl Client { /// prefixed with a "v", for example `v0.30.5`. /// /// Upon connection failure, this will not attempt to do any retrying. + /// + /// # Errors + /// + /// Returns an error if the request fails or if the server returns a non-success status. pub async fn server_version(&self) -> Result { let base_url = &self.base_url; let resp = self @@ -312,11 +351,10 @@ impl InternalProducerClientError { match self { // In the case of an unexpected HTTP status error, developer // intervention will be required. - Self::UnexpectedStatus(_) => false, + Self::UnexpectedStatus(_) | Self::ResponseDecodingError(_) => false, // In the case of an ungraceful restart, this case might theoretically trigger. // So even cleaner would be a tiny retry loop for this special case. // However, we certainly **do not** want to wait multiple minutes before returning. - Self::ResponseDecodingError(_) => false, // Reqwest doesn't make this very easy as it has a single error typed used for _everything_. Self::HTTPClientError(inner) => { // Failures in which a connection could not be established are ephemeral, @@ -408,11 +446,11 @@ mod tests { contents: vec![None, None, None], }, metadata: None, - strategic_metadata: Default::default(), + strategic_metadata: StrategicMetadataMap::default(), chunk_size: None, }; client - .insert_submission(&submission, &Default::default()) + .insert_submission(&submission, &std::collections::HashMap::default()) .await .expect("Should be OK"); @@ -422,15 +460,15 @@ mod tests { assert_eq!(count, 1); client - .insert_submission(&submission, &Default::default()) + .insert_submission(&submission, &std::collections::HashMap::default()) .await .expect("Should be OK"); client - .insert_submission(&submission, &Default::default()) + .insert_submission(&submission, &std::collections::HashMap::default()) .await .expect("Should be OK"); client - .insert_submission(&submission, &Default::default()) + .insert_submission(&submission, &std::collections::HashMap::default()) .await .expect("Should be OK"); @@ -451,11 +489,11 @@ mod tests { contents: vec![None, None, None], }, metadata: None, - strategic_metadata: Default::default(), + strategic_metadata: StrategicMetadataMap::default(), chunk_size: None, }; let submission_id = client - .insert_submission(&submission, &Default::default()) + .insert_submission(&submission, &std::collections::HashMap::default()) .await .expect("Should be OK"); diff --git a/opsqueue/src/producer/server.rs b/opsqueue/src/producer/server.rs index fb19203..7f14768 100644 --- a/opsqueue/src/producer/server.rs +++ b/opsqueue/src/producer/server.rs @@ -40,6 +40,12 @@ impl ServerState { max_submissions, } } + + /// Run a test-only producer HTTP server. + /// + /// # Panics + /// + /// Panics if binding the server socket fails or if serving fails unexpectedly. pub async fn serve_for_tests(self, server_addr: Box) { let app = Router::new().nest("/producer", self.build_router()); diff --git a/opsqueue/src/prometheus.rs b/opsqueue/src/prometheus.rs index 4ad4865..3f12719 100644 --- a/opsqueue/src/prometheus.rs +++ b/opsqueue/src/prometheus.rs @@ -44,6 +44,7 @@ pub const CONSUMER_FAIL_CHUNK_DURATION: &str = "consumer_fail_chunk_duration_sec pub const OPERATIONS_BACKLOG_GAUGE: &str = "operations_in_backlog_count"; +#[allow(clippy::too_many_lines)] pub fn describe_metrics() { describe_counter!( SUBMISSIONS_TOTAL_COUNTER, @@ -165,6 +166,11 @@ pub type PrometheusConfig = ( ); #[must_use] +/// Build Prometheus layer and recorder handle. +/// +/// # Panics +/// +/// Panics if Prometheus recorder installation fails. pub fn setup_prometheus() -> ( GenericMetricLayer<'static, PrometheusHandle, axum_prometheus::Handle>, PrometheusHandle, @@ -190,12 +196,18 @@ pub fn setup_prometheus() -> ( /// /// Adapted from #[must_use] +#[allow(clippy::cast_precision_loss)] pub fn time_delta_as_f64(td: chrono::TimeDelta) -> f64 { const NANOS_PER_SEC: f64 = 1_000_000_000.0; (td.num_seconds() as f64) + f64::from(td.subsec_nanos()) / NANOS_PER_SEC } /// Calculates the backlog-size metrics used for autoscaling. +/// +/// # Errors +/// +/// Returns an error if acquiring a DB connection or reading metrics fails. +#[allow(clippy::cast_precision_loss)] pub async fn calculate_scaling_metrics(db_pool: &DBPools) -> anyhow::Result<()> { let mut conn = db_pool.reader_conn().await?; let chunks_backlog_count: u64 = crate::common::chunk::db::count_chunks(&mut conn) diff --git a/opsqueue/src/server.rs b/opsqueue/src/server.rs index f11f2ac..d149da1 100644 --- a/opsqueue/src/server.rs +++ b/opsqueue/src/server.rs @@ -2,6 +2,7 @@ use std::{ any::Any, sync::{Arc, atomic::AtomicBool}, + mem, time::Duration, }; @@ -23,6 +24,11 @@ fn retry_policy() -> impl BackoffBuilder { } #[cfg(feature = "server-logic")] +/// Start serving producer and consumer endpoints. +/// +/// # Errors +/// +/// Returns an error if binding or serving the HTTP server fails. pub async fn serve_producer_and_consumer( config: &'static crate::config::Config, server_addr: &str, @@ -39,7 +45,7 @@ pub async fn serve_producer_and_consumer( config, pool.clone(), reservation_expiration, - cancellation_token.clone(), + cancellation_token, app_healthy_flag.clone(), prometheus_config.clone(), ); @@ -70,7 +76,7 @@ pub async fn serve_producer_and_consumer( .await.inspect_err(|_|{ // Drop the pipe after the server start retries have been exhausted. This ensures that the // parent process can safely block on reading from the pipe. - config.report_bound_port_pipe.take(); + mem::drop(config.report_bound_port_pipe.take()); }) } @@ -79,7 +85,7 @@ pub fn build_router( config: &'static crate::config::Config, pool: DBPools, reservation_expiration: Duration, - cancellation_token: CancellationToken, + cancellation_token: &CancellationToken, app_healthy_flag: Arc, prometheus_config: crate::prometheus::PrometheusConfig, ) -> Router<()> { @@ -149,6 +155,7 @@ pub async fn version_endpoint() -> String { crate::version_info() } +#[allow(clippy::needless_pass_by_value)] fn handle_panic(err: Box) -> Response { let details = if let Some(s) = err.downcast_ref::() { s.clone() diff --git a/opsqueue/src/tracing.rs b/opsqueue/src/tracing.rs index 9f8a78b..620c065 100644 --- a/opsqueue/src/tracing.rs +++ b/opsqueue/src/tracing.rs @@ -14,7 +14,7 @@ pub fn context_from_headers(headers: &http::HeaderMap) -> Context { #[must_use] pub fn context_to_headers(context: &Context) -> http::HeaderMap { let propagator = propagator(); - let mut headers = Default::default(); + let mut headers = http::HeaderMap::default(); propagator.inject_context(context, &mut HeaderInjector(&mut headers)); headers } From afe98b3da5f94f14213cdd0ea79869412d58494f Mon Sep 17 00:00:00 2001 From: Reinier Maas Date: Fri, 17 Jul 2026 16:00:29 +0200 Subject: [PATCH 04/10] AI: Resolve Clippy warnings in `opsqueue_python` --- libs/opsqueue_python/src/common.rs | 26 ++++++++ libs/opsqueue_python/src/consumer.rs | 26 +++++++- libs/opsqueue_python/src/producer.rs | 92 ++++++++++++++++++++++------ 3 files changed, 125 insertions(+), 19 deletions(-) diff --git a/libs/opsqueue_python/src/common.rs b/libs/opsqueue_python/src/common.rs index cefc6e1..65d8fb0 100644 --- a/libs/opsqueue_python/src/common.rs +++ b/libs/opsqueue_python/src/common.rs @@ -39,6 +39,7 @@ impl SubmissionId { Ok(SubmissionId { id }) } + #[allow(clippy::trivially_copy_pass_by_ref)] fn __repr__(&self) -> String { format!("SubmissionId(id={})", self.id) } @@ -73,6 +74,7 @@ impl ChunkIndex { Ok(ChunkIndex { id }) } + #[allow(clippy::trivially_copy_pass_by_ref)] fn __repr__(&self) -> String { format!("ChunkIndex(id={})", self.id) } @@ -218,6 +220,16 @@ pub struct Chunk { } impl Chunk { + /// Build a Python-facing chunk by combining DB metadata and chunk contents. + /// + /// # Panics + /// + /// Panics if the chunk content is stored in object storage while the submission + /// has no associated prefix. + /// + /// # Errors + /// + /// Returns an error if fetching chunk bytes from object storage fails. pub async fn from_internal( c: chunk::Chunk, s: submission::Submission, @@ -539,6 +551,12 @@ impl SubmissionNotCancellable { } } +/// Await a future while also reacting to Python interruption signals. +/// +/// # Errors +/// +/// Returns the underlying future error, or a fatal Python exception when +/// an interrupt signal is detected. pub async fn run_unless_interrupted( future: impl IntoFuture>, ) -> Result @@ -588,6 +606,10 @@ pub async fn check_signals_in_background() -> FatalPythonException { /// 'Tokio scheduler thread' is preemptive, /// the same problem now no longer occurs: /// Both schedulers are able to make forward progress (even on a 1-CPU machine). +/// +/// # Panics +/// +/// Panics if creating the Tokio runtime fails. #[must_use] pub fn start_runtime() -> Arc { let runtime = tokio::runtime::Builder::new_multi_thread() @@ -605,6 +627,10 @@ pub fn start_runtime() -> Arc { /// Internally acquires the GIL! /// /// c.f. +/// +/// # Panics +/// +/// Panics if formatting a Python traceback fails. pub fn format_pyerr(err: &PyErr) -> String { Python::attach(|py| { let msg: Option = (|| { diff --git a/libs/opsqueue_python/src/consumer.rs b/libs/opsqueue_python/src/consumer.rs index 14afbc7..c6f8c59 100644 --- a/libs/opsqueue_python/src/consumer.rs +++ b/libs/opsqueue_python/src/consumer.rs @@ -54,6 +54,10 @@ impl ConsumerClient { /// Supports the formats listed here: /// :param `object_store_options`: A list of key-value strings as extra options for the chosen object store. /// For example, for GCS, see + /// + /// # Errors + /// + /// Returns an error if object store client initialization fails. #[new] #[pyo3(signature = (address, object_store_url, object_store_options=vec![]))] pub fn new( @@ -87,6 +91,11 @@ impl ConsumerClient { } #[allow(clippy::type_complexity)] + /// Reserve up to `max` chunks from the queue. + /// + /// # Errors + /// + /// Returns an error if reservation fails or if chunk bytes cannot be retrieved. pub fn reserve_chunks( &self, py: Python<'_>, @@ -105,6 +114,11 @@ impl ConsumerClient { } #[pyo3(signature = (submission_id, submission_prefix, chunk_index, output_content))] + /// Complete a chunk and optionally upload output content to object storage. + /// + /// # Errors + /// + /// Returns an error if upload or completion request fails. pub fn complete_chunk( &self, py: Python<'_>, @@ -131,6 +145,11 @@ impl ConsumerClient { } #[pyo3(signature = (submission_id, submission_prefix, chunk_index, failure))] + /// Mark a chunk as failed. + /// + /// # Errors + /// + /// Returns an error if the failure report cannot be submitted. pub fn fail_chunk( &self, py: Python<'_>, @@ -218,7 +237,7 @@ impl ConsumerClient { }) } - #[allow(clippy::type_complexity)] + #[allow(clippy::type_complexity, clippy::needless_pass_by_value)] fn reserve_chunks_gilless( &self, max: usize, @@ -288,6 +307,11 @@ impl ConsumerClient { }) } + /// Mark a chunk as failed without interacting with Python's GIL. + /// + /// # Errors + /// + /// Returns an error if the failure report cannot be submitted. pub fn fail_chunk_gilless( &self, submission_id: SubmissionId, diff --git a/libs/opsqueue_python/src/producer.rs b/libs/opsqueue_python/src/producer.rs index d0a3a25..5bbd357 100644 --- a/libs/opsqueue_python/src/producer.rs +++ b/libs/opsqueue_python/src/producer.rs @@ -34,7 +34,7 @@ const SUBMISSION_POLLING_INTERVAL: Duration = Duration::from_secs(5); #[pyclass(from_py_object, module = "opsqueue")] #[derive(Debug, Clone)] pub struct ProducerClient { - producer_client: ActualClient, + client: ActualClient, object_store_client: opsqueue::object_store::ObjectStoreClient, runtime: Arc, } @@ -51,6 +51,10 @@ impl ProducerClient { /// Supports the formats listed here: /// :param `object_store_options`: A list of key-value strings as extra options for the chosen object store. /// For example, for GCS, see + /// + /// # Errors + /// + /// Returns an error if object store client initialization fails. #[new] #[pyo3(signature = (address, object_store_url, object_store_options=vec![]))] pub fn new( @@ -63,14 +67,14 @@ impl ProducerClient { opsqueue::version_info() ); let runtime = start_runtime(); - let producer_client = ActualClient::new(address); + let client = ActualClient::new(address); let object_store_client = opsqueue::object_store::ObjectStoreClient::new(object_store_url, object_store_options)?; tracing::info!("Opsqueue ProducerClient initialized"); Ok(ProducerClient { - producer_client, + client, object_store_client, runtime, }) @@ -80,19 +84,23 @@ impl ProducerClient { pub fn __repr__(&self) -> String { format!( "", - self.producer_client.base_url(), + self.client.base_url(), self.object_store_client.url() ) } /// Return the Opsqueue server's version information + /// + /// # Errors + /// + /// Returns an error if contacting the server fails. pub fn server_version( &self, py: Python<'_>, ) -> CPyResult> { py.detach(|| { self.block_unless_interrupted(async { - self.producer_client + self.client .server_version() .await .map_err(|e| CError(R(e))) @@ -103,13 +111,17 @@ impl ProducerClient { /// Counts the number of ongoing submissions in the queue. /// /// Completed and failed submissions are not included in the count. + /// + /// # Errors + /// + /// Returns an error if contacting the server fails. pub fn count_submissions( &self, py: Python<'_>, ) -> CPyResult> { py.detach(|| { self.block_unless_interrupted(async { - self.producer_client + self.client .count_submissions() .await .map_err(|e| CError(R(e))) @@ -121,6 +133,10 @@ impl ProducerClient { /// /// Will return an error if the submission is already complete, failed, or /// cancelled, or if the submission could not be found. + /// + /// # Errors + /// + /// Returns an error if the submission cannot be cancelled or if the request fails. #[allow(clippy::result_large_err, clippy::type_complexity)] pub fn cancel_submission( &self, @@ -137,7 +153,7 @@ impl ProducerClient { > { py.detach(|| { self.block_unless_interrupted(async { - self.producer_client + self.client .cancel_submission(id.into()) .await .map_err(|e| CError(R(e))) @@ -151,6 +167,10 @@ impl ProducerClient { /// when the submission was started/completed/failed, etc. /// /// This call does _not_ fetch the submission's chunk contents on its own. + /// + /// # Errors + /// + /// Returns an error if contacting the server fails. pub fn get_submission_status( &self, py: Python<'_>, @@ -159,7 +179,7 @@ impl ProducerClient { { py.detach(|| { self.block_unless_interrupted(async { - self.producer_client + self.client .get_submission(id.into()) .await .map_err(|e| CError(R(e))) @@ -172,6 +192,10 @@ impl ProducerClient { /// Attempts to find the submission ID if only the prefix of the submission /// (AKA the path at which the submission's chunks are stored in the object store) /// is known. + /// + /// # Errors + /// + /// Returns an error if contacting the server fails. pub fn lookup_submission_id_by_prefix( &self, py: Python<'_>, @@ -179,7 +203,7 @@ impl ProducerClient { ) -> CPyResult, E> { py.detach(|| { self.block_unless_interrupted(async { - self.producer_client + self.client .lookup_submission_id_by_prefix(prefix) .await .map(|opt| opt.map(Into::into)) @@ -190,6 +214,11 @@ impl ProducerClient { /// Attempts to find the IDs of submission matching ALL key-values pairs of /// the given strategic metadata. + /// + /// # Errors + /// + /// Returns an error if too many submissions match or if the request fails. + #[allow(clippy::needless_pass_by_value)] pub fn lookup_submission_ids_by_strategic_metadata( &self, py: Python<'_>, @@ -204,7 +233,7 @@ impl ProducerClient { > { py.detach(|| { self.block_unless_interrupted(async { - self.producer_client + self.client .lookup_submission_ids_by_strategic_metadata(&strategic_metadata) .await .map(|res| res.into_iter().map(Into::into).collect()) @@ -216,6 +245,10 @@ impl ProducerClient { /// Directly inserts a submission without sending the chunks to GCS /// (but immediately embedding them in the DB). /// NOTE: This does not support `StrategicMetadata` currently + /// + /// # Errors + /// + /// Returns an error if submission insertion fails. #[pyo3(signature = (chunk_contents, metadata=None, chunk_size=None, otel_trace_carrier=CarrierMap::default()))] pub fn insert_submission_direct( &self, @@ -225,11 +258,11 @@ impl ProducerClient { chunk_size: Option, otel_trace_carrier: CarrierMap, ) -> CPyResult> { - let strategic_metadata = Default::default(); + let strategic_metadata = std::collections::HashMap::default(); py.detach(|| { let submission = opsqueue::producer::InsertSubmission { - chunk_size: chunk_size.map(|n| chunk::ChunkSize(n as i64)), + chunk_size: chunk_size.map(|n| chunk::ChunkSize(n.cast_signed())), chunk_contents: ChunkContents::Direct { contents: chunk_contents, }, @@ -237,7 +270,7 @@ impl ProducerClient { strategic_metadata, }; self.block_unless_interrupted(async move { - self.producer_client + self.client .insert_submission(&submission, &otel_trace_carrier) .await .map_err(|e| R(e).into()) @@ -248,6 +281,11 @@ impl ProducerClient { #[pyo3(signature = (chunk_contents, metadata=None, strategic_metadata=None, chunk_size=None, otel_trace_carrier=CarrierMap::default()))] #[allow(clippy::type_complexity)] + /// Insert submission chunks via object storage and enqueue the submission. + /// + /// # Errors + /// + /// Returns an error if chunk upload or submission insertion fails. pub fn insert_submission_chunks( &self, py: Python<'_>, @@ -296,7 +334,7 @@ impl ProducerClient { metadata, strategic_metadata: strategic_metadata.unwrap_or_default(), }; - self.producer_client + self.client .insert_submission(&submission, &otel_trace_carrier) .await .map(|submission_id| { @@ -309,6 +347,11 @@ impl ProducerClient { } #[allow(clippy::result_large_err, clippy::type_complexity)] + /// Try streaming completed submission chunks without waiting. + /// + /// # Errors + /// + /// Returns an error if the submission is incomplete/failed or if the request fails. pub fn try_stream_completed_submission_chunks( &self, py: Python<'_>, @@ -338,6 +381,11 @@ impl ProducerClient { #[pyo3(signature = (chunk_contents, metadata=None, strategic_metadata=None, chunk_size=None, otel_trace_carrier=CarrierMap::default()))] #[allow(clippy::result_large_err, clippy::type_complexity)] + /// Submit chunks and then stream the completed output chunks. + /// + /// # Errors + /// + /// Returns an error if upload, submission creation, or streaming fails. pub fn run_submission_chunks( &self, py: Python<'_>, @@ -388,6 +436,10 @@ impl ProducerClient { /// to reduce the latency of tiny submissions. /// This interval is then doubled for each subsequent poll, /// until we check every few seconds. + /// + /// # Errors + /// + /// Returns an error if polling or output streaming fails. #[allow(clippy::result_large_err, clippy::type_complexity)] pub fn blocking_stream_completed_submission_chunks( &self, @@ -408,6 +460,11 @@ impl ProducerClient { }) } + /// Return an awaitable that resolves to an async iterator of output chunks. + /// + /// # Errors + /// + /// Returns a Python error if creating the awaitable fails. pub fn async_stream_completed_submission_chunks<'p>( &self, py: Python<'p>, @@ -487,7 +544,7 @@ impl ProducerClient { E![crate::errors::SubmissionFailed, InternalProducerClientError], > { match self - .producer_client + .client .get_submission(id.into()) .await .map_err(R)? @@ -498,8 +555,7 @@ impl ProducerClient { id.id ); let prefix = submission.prefix.unwrap_or_default(); - let py_chunks_iter = - PyChunksIter::new(self, prefix, submission.chunks_total.into()).await; + let py_chunks_iter = PyChunksIter::new(self, prefix, submission.chunks_total.into()); Ok(Some(py_chunks_iter)) } @@ -525,7 +581,7 @@ pub struct PyChunksIter { } impl PyChunksIter { - pub(crate) async fn new(client: &ProducerClient, prefix: String, chunks_total: u63) -> Self { + pub(crate) fn new(client: &ProducerClient, prefix: String, chunks_total: u63) -> Self { let stream = client .object_store_client .retrieve_chunks(prefix, chunks_total, ChunkType::Output) From c099a462dab0dcc7e050973e002656454df8ffd7 Mon Sep 17 00:00:00 2001 From: Reinier Maas Date: Fri, 17 Jul 2026 16:11:46 +0200 Subject: [PATCH 05/10] fixup! Run: `just clippy` --- libs/opsqueue_python/src/common.rs | 10 +++++++--- opsqueue/src/common/chunk.rs | 5 ++++- opsqueue/src/common/submission.rs | 10 ++++++++-- opsqueue/src/consumer/server/conn.rs | 5 +++-- opsqueue/src/consumer/strategy.rs | 2 +- opsqueue/src/db/conn.rs | 5 ++++- opsqueue/src/db/mod.rs | 3 ++- opsqueue/src/tracing.rs | 5 +++-- 8 files changed, 32 insertions(+), 13 deletions(-) diff --git a/libs/opsqueue_python/src/common.rs b/libs/opsqueue_python/src/common.rs index 65d8fb0..b0fbc0b 100644 --- a/libs/opsqueue_python/src/common.rs +++ b/libs/opsqueue_python/src/common.rs @@ -235,7 +235,9 @@ impl Chunk { s: submission::Submission, object_store_client: &ObjectStoreClient, ) -> Result { - let (content, prefix) = if let Some(bytes) = c.input_content { (bytes, None) } else { + let (content, prefix) = if let Some(bytes) = c.input_content { + (bytes, None) + } else { let prefix = s.prefix.unwrap(); tracing::debug!( "Fetching chunk content from object store: submission_id={}, prefix={}, chunk_index={}", @@ -366,7 +368,9 @@ pub enum SubmissionStatus { impl From for SubmissionStatus { fn from(value: opsqueue::common::submission::SubmissionStatus) -> Self { - use opsqueue::common::submission::SubmissionStatus::{InProgress, Completed, Failed, Cancelled}; + use opsqueue::common::submission::SubmissionStatus::{ + Cancelled, Completed, Failed, InProgress, + }; match value { InProgress(s) => SubmissionStatus::InProgress { submission: s.into(), @@ -525,7 +529,7 @@ pub enum SubmissionNotCancellable { impl From for SubmissionNotCancellable { fn from(value: opsqueue::common::errors::SubmissionNotCancellable) -> Self { - use opsqueue::common::errors::SubmissionNotCancellable::{Completed, Failed, Cancelled}; + use opsqueue::common::errors::SubmissionNotCancellable::{Cancelled, Completed, Failed}; match value { Completed(s) => SubmissionNotCancellable::Completed { submission: s.into(), diff --git a/opsqueue/src/common/chunk.rs b/opsqueue/src/common/chunk.rs index 80acdf0..d8dbd8a 100644 --- a/opsqueue/src/common/chunk.rs +++ b/opsqueue/src/common/chunk.rs @@ -224,7 +224,10 @@ impl Chunk { #[cfg(feature = "server-logic")] pub mod db { - use super::{ChunkIndex, Chunk, ChunkId, ChunkSize, SubmissionId, ChunkCompleted, DateTime, Utc, ChunkFailed, u63}; + use super::{ + Chunk, ChunkCompleted, ChunkFailed, ChunkId, ChunkIndex, ChunkSize, DateTime, SubmissionId, + Utc, u63, + }; use crate::common::errors::{ChunkNotFound, DatabaseError, E, SubmissionNotFound}; use crate::db::{Connection, True, WriterConnection}; use axum_prometheus::metrics::{counter, gauge}; diff --git a/opsqueue/src/common/submission.rs b/opsqueue/src/common/submission.rs index 9c063e0..3248cca 100644 --- a/opsqueue/src/common/submission.rs +++ b/opsqueue/src/common/submission.rs @@ -290,7 +290,11 @@ pub mod db { use chunk::ChunkSize; use sqlx::{QueryBuilder, Sqlite, query, query_scalar}; - use super::{chunk, E, SubmissionId, Submission, Chunk, Metadata, ChunkCount, SubmissionStatus, DateTime, Utc, SubmissionCompleted, ChunkIndex, SubmissionFailed, SubmissionCancelled, Duration}; + use super::{ + Chunk, ChunkCount, ChunkIndex, DateTime, Duration, E, Metadata, Submission, + SubmissionCancelled, SubmissionCompleted, SubmissionFailed, SubmissionId, SubmissionStatus, + Utc, chunk, + }; impl<'q> sqlx::Encode<'q, Sqlite> for SubmissionId { fn encode_by_ref( @@ -916,7 +920,9 @@ pub mod db { ) .fetch_optional(conn.get_inner()) .await?; - if submission_opt.is_none() { Err(E::R(SubmissionNotFound(id))) } else { + if submission_opt.is_none() { + Err(E::R(SubmissionNotFound(id))) + } else { counter!(crate::prometheus::SUBMISSIONS_CANCELLED_COUNTER).increment(1); histogram!(crate::prometheus::SUBMISSIONS_DURATION_CANCEL_HISTOGRAM).record( crate::prometheus::time_delta_as_f64(Utc::now() - id.timestamp()), diff --git a/opsqueue/src/consumer/server/conn.rs b/opsqueue/src/consumer/server/conn.rs index 82ee6df..42aa713 100644 --- a/opsqueue/src/consumer/server/conn.rs +++ b/opsqueue/src/consumer/server/conn.rs @@ -206,7 +206,7 @@ impl ConsumerConn { &mut self, msg: Envelope, ) -> Result<(), ConsumerConnError> { - use ClientToServerMessage::{WantToReserveChunks, CompleteChunk, FailChunk}; + use ClientToServerMessage::{CompleteChunk, FailChunk, WantToReserveChunks}; use SyncServerToClientResponse::ChunksReserved; let maybe_response = match msg.contents { WantToReserveChunks { max, strategy } => { @@ -367,7 +367,8 @@ impl ConsumerConnectionId { COLORS .get(i) - .map(|c| String::from(*c)).map_or_else(Self::uuid, ConsumerConnectionId) + .map(|c| String::from(*c)) + .map_or_else(Self::uuid, ConsumerConnectionId) } } diff --git a/opsqueue/src/consumer/strategy.rs b/opsqueue/src/consumer/strategy.rs index c510c39..a4d1145 100644 --- a/opsqueue/src/consumer/strategy.rs +++ b/opsqueue/src/consumer/strategy.rs @@ -38,7 +38,7 @@ impl Strategy { qb: &'a mut QueryBuilder<'a, Sqlite>, metastate: &MetaState, ) -> &'a mut QueryBuilder<'a, Sqlite> { - use Strategy::{Oldest, Newest, Random, PreferDistinct}; + use Strategy::{Newest, Oldest, PreferDistinct, Random}; match self { Oldest => qb.push("SELECT * FROM chunks ORDER BY submission_id ASC"), Newest => qb.push("SELECT * FROM chunks ORDER BY submission_id DESC"), diff --git a/opsqueue/src/db/conn.rs b/opsqueue/src/db/conn.rs index 29b5c96..e467eaa 100644 --- a/opsqueue/src/db/conn.rs +++ b/opsqueue/src/db/conn.rs @@ -4,7 +4,10 @@ use std::marker::PhantomData; use sqlx::{Sqlite, SqliteConnection, pool::PoolConnection}; -use super::{Connection, magic::{Bool, True, False}}; +use super::{ + Connection, + magic::{Bool, False, True}, +}; /// A connection to the database. /// diff --git a/opsqueue/src/db/mod.rs b/opsqueue/src/db/mod.rs index 5b04f0e..835d683 100644 --- a/opsqueue/src/db/mod.rs +++ b/opsqueue/src/db/mod.rs @@ -439,7 +439,8 @@ fn db_options(database_filename: &str) -> SqliteConnectOptions { async fn ensure_db_exists(database_filename: &str) { if Sqlite::database_exists(database_filename) .await - .unwrap_or(false) { + .unwrap_or(false) + { tracing::info!("Starting up using existing sqlite DB {}", database_filename); } else { tracing::info!("Creating backing sqlite DB {}", database_filename); diff --git a/opsqueue/src/tracing.rs b/opsqueue/src/tracing.rs index 620c065..0ba300a 100644 --- a/opsqueue/src/tracing.rs +++ b/opsqueue/src/tracing.rs @@ -38,8 +38,9 @@ pub fn context_to_json(context: &Context) -> String { #[must_use] pub fn json_to_context(json: &str) -> Context { let propagator = propagator(); - serde_json::from_str(json) - .map_or(Context::new(), |hashmap: CarrierMap| propagator.extract(&hashmap)) + serde_json::from_str(json).map_or(Context::new(), |hashmap: CarrierMap| { + propagator.extract(&hashmap) + }) } #[must_use] From 74fef75f72342c1e6b97b13a3a6127905acf4504 Mon Sep 17 00:00:00 2001 From: Reinier Maas Date: Fri, 17 Jul 2026 16:11:46 +0200 Subject: [PATCH 06/10] fixup! AI: Resolve Clippy warnings in `opsqueue_python` --- libs/opsqueue_python/src/producer.rs | 17 +++++------------ 1 file changed, 5 insertions(+), 12 deletions(-) diff --git a/libs/opsqueue_python/src/producer.rs b/libs/opsqueue_python/src/producer.rs index 5bbd357..e3167a9 100644 --- a/libs/opsqueue_python/src/producer.rs +++ b/libs/opsqueue_python/src/producer.rs @@ -100,10 +100,7 @@ impl ProducerClient { ) -> CPyResult> { py.detach(|| { self.block_unless_interrupted(async { - self.client - .server_version() - .await - .map_err(|e| CError(R(e))) + self.client.server_version().await.map_err(|e| CError(R(e))) }) }) } @@ -262,7 +259,7 @@ impl ProducerClient { py.detach(|| { let submission = opsqueue::producer::InsertSubmission { - chunk_size: chunk_size.map(|n| chunk::ChunkSize(n.cast_signed())), + chunk_size: chunk_size.map(|n| chunk::ChunkSize(n.cast_signed())), chunk_contents: ChunkContents::Direct { contents: chunk_contents, }, @@ -543,19 +540,15 @@ impl ProducerClient { Option, E![crate::errors::SubmissionFailed, InternalProducerClientError], > { - match self - .client - .get_submission(id.into()) - .await - .map_err(R)? - { + match self.client.get_submission(id.into()).await.map_err(R)? { Some(submission::SubmissionStatus::Completed(submission)) => { tracing::debug!( "Submission {} completed! Streaming result-chunks from object store", id.id ); let prefix = submission.prefix.unwrap_or_default(); - let py_chunks_iter = PyChunksIter::new(self, prefix, submission.chunks_total.into()); + let py_chunks_iter = + PyChunksIter::new(self, prefix, submission.chunks_total.into()); Ok(Some(py_chunks_iter)) } From ff344662e9be97472e24e22da797a6ad321ec76b Mon Sep 17 00:00:00 2001 From: Reinier Maas Date: Fri, 17 Jul 2026 16:11:46 +0200 Subject: [PATCH 07/10] fixup! AI: Resolve Clippy warnings --- opsqueue/src/consumer/dispatcher/mod.rs | 3 +-- opsqueue/src/consumer/strategy.rs | 4 ++-- opsqueue/src/server.rs | 2 +- 3 files changed, 4 insertions(+), 5 deletions(-) diff --git a/opsqueue/src/consumer/dispatcher/mod.rs b/opsqueue/src/consumer/dispatcher/mod.rs index d275314..8ba9b03 100644 --- a/opsqueue/src/consumer/dispatcher/mod.rs +++ b/opsqueue/src/consumer/dispatcher/mod.rs @@ -95,8 +95,7 @@ impl Dispatcher { stale_chunks_notifier: &UnboundedSender, ) -> Option { let chunk_id = ChunkId::from((chunk.submission_id, chunk.chunk_index)); - self - .reserver + self.reserver .try_reserve(chunk_id, chunk_id, stale_chunks_notifier) .map(|_| chunk) } diff --git a/opsqueue/src/consumer/strategy.rs b/opsqueue/src/consumer/strategy.rs index a4d1145..2923e08 100644 --- a/opsqueue/src/consumer/strategy.rs +++ b/opsqueue/src/consumer/strategy.rs @@ -751,7 +751,7 @@ pub mod test { let mut conn = db_pools.reader_conn().await.unwrap(); let mut query_builder = QueryBuilder::default(); let vals1: Vec = Strategy::Random - .build_query(&mut query_builder, &MetaState::default()) + .build_query(&mut query_builder, &MetaState::default()) .build_query_as() .fetch(conn.get_inner()) .try_collect() @@ -760,7 +760,7 @@ pub mod test { let mut query_builder = QueryBuilder::default(); let vals2: Vec = Strategy::Random - .build_query(&mut query_builder, &MetaState::default()) + .build_query(&mut query_builder, &MetaState::default()) .build_query_as() .fetch(conn.get_inner()) .try_collect() diff --git a/opsqueue/src/server.rs b/opsqueue/src/server.rs index d149da1..e42be26 100644 --- a/opsqueue/src/server.rs +++ b/opsqueue/src/server.rs @@ -1,8 +1,8 @@ //! Defines the HTTP endpoints that are used by both the `producer` and `consumer` APIs use std::{ any::Any, - sync::{Arc, atomic::AtomicBool}, mem, + sync::{Arc, atomic::AtomicBool}, time::Duration, }; From 7f7fb48878d7763d73bb794190d500d055673cac Mon Sep 17 00:00:00 2001 From: Reinier Maas Date: Fri, 17 Jul 2026 16:30:11 +0200 Subject: [PATCH 08/10] fixup! fixup! AI: Resolve Clippy warnings --- opsqueue/src/common/chunk.rs | 4 ++-- opsqueue/src/common/mod.rs | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/opsqueue/src/common/chunk.rs b/opsqueue/src/common/chunk.rs index d8dbd8a..e992b5b 100644 --- a/opsqueue/src/common/chunk.rs +++ b/opsqueue/src/common/chunk.rs @@ -42,8 +42,8 @@ impl<'a> serde::Deserialize<'a> for ChunkIndex { } /// Another name for `ChunkIndex`, indicating that we're dealing with the _total count_ of chunks. -/// i.e. when you have a value of type `ChunkCount`, there is a high likelyhood -/// that there are [`0..chunk_count`) (note the half-open range) chunks to select from. +/// i.e. when you have a value of type `ChunkCount`, there is a high likelihood +/// that there are `[0..chunk_count)` (note the half-open range) chunks to select from. pub type ChunkCount = ChunkIndex; /// The count of entries in each chunk. diff --git a/opsqueue/src/common/mod.rs b/opsqueue/src/common/mod.rs index 8c8636c..c685cd6 100644 --- a/opsqueue/src/common/mod.rs +++ b/opsqueue/src/common/mod.rs @@ -15,7 +15,7 @@ pub type MetaStateVal = i64; pub type StrategicMetadataMap = FxHashMap; /// Maximum number of submissions a lookup may return. -/// Guarantees: 0 < `MaxSubmissions` 1 < `i64::MAX`; +/// Guarantees: `0 < MaxSubmissions && MaxSubmissions + 1 < i64::MAX`; #[derive(Debug, Clone, Copy)] pub struct MaxSubmissions(NonZero); From 8d2e7e7fe64847c8eeb4d51bcbbc0e602b078e6c Mon Sep 17 00:00:00 2001 From: Reinier Maas Date: Fri, 17 Jul 2026 16:32:24 +0200 Subject: [PATCH 09/10] CI: Prevent typos using pre-commit The AI caught it this time, let's enforce it using something more stable. --- .pre-commit-config.yaml | 6 ++++++ default.nix | 1 + 2 files changed, 7 insertions(+) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 1a5d0a3..efdc078 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -13,6 +13,12 @@ repos: entry: trailing-whitespace-fixer language: system types: [text] + - id: typos + name: Typos + entry: typos --write-changes + language: system + types: [text] + exclude_types: [diff, json] - id: ruff-check name: Ruff linting entry: ruff check --fix diff --git a/default.nix b/default.nix index 2e33771..2da1a73 100644 --- a/default.nix +++ b/default.nix @@ -34,6 +34,7 @@ let pkgs.pre-commit pkgs.python3Packages.pre-commit-hooks pkgs.ruff + pkgs.typos # For compiling the Rust parts pkgs.rustToolchain From 5e3a10dd7e46eb5f10c429f4ddae8bef1ecd14d3 Mon Sep 17 00:00:00 2001 From: Reinier Maas Date: Fri, 17 Jul 2026 16:38:37 +0200 Subject: [PATCH 10/10] fixup! CI: Prevent typos using pre-commit --- .pre-commit-config.yaml | 1 + README.md | 2 +- doc/proposals/strategies_and_metadata.md | 10 +++++----- libs/opsqueue_python/python/opsqueue/exceptions.py | 6 +++--- libs/opsqueue_python/python/opsqueue/producer.py | 2 +- 5 files changed, 11 insertions(+), 10 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index efdc078..eb17abf 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -19,6 +19,7 @@ repos: language: system types: [text] exclude_types: [diff, json] + exclude: ^(libs/opsqueue_python/examples/capitalize_text/lipsum\.txt|libs/opsqueue_python/examples/capitalize_text/more_lipsum\.txt)$ - id: ruff-check name: Ruff linting entry: ruff check --fix diff --git a/README.md b/README.md index d372060..cc8a476 100644 --- a/README.md +++ b/README.md @@ -280,7 +280,7 @@ just test-integration` ``` This will set up the required steps below automatically. -Any arguments passsed are passed on to `pytest`, e.g.: +Any arguments passed are passed on to `pytest`, e.g.: ```bash just test-integration -vvvvvv -s -k diff --git a/doc/proposals/strategies_and_metadata.md b/doc/proposals/strategies_and_metadata.md index 31fed27..46d7c28 100644 --- a/doc/proposals/strategies_and_metadata.md +++ b/doc/proposals/strategies_and_metadata.md @@ -51,7 +51,7 @@ At some point, this will make the other operations, especially insertion, too sl In the rest of this document, the strategies and their potential implementation are each described in detail, and some tricks/ideas on making the right tradeoff w.r.t. what indexes to create while keeping the database manageable are given at the end. Of the above strategies, currently `Random`, `NewestFirst` and `OldestFirst` are implemented, exactly as described below. -The other strategies, most of which depend on the metadata of a particular submission in one way or another, are not implemented yet. Describing these composable strategies and a potential way to implement them with a balance between preformance and flexibility is the main goal of this proposal. +The other strategies, most of which depend on the metadata of a particular submission in one way or another, are not implemented yet. Describing these composable strategies and a potential way to implement them with a balance between performance and flexibility is the main goal of this proposal. ## Assumptions & Invariants @@ -84,10 +84,10 @@ Furthermore, it might be important to note: Opsqueue's consumer API consists of four main producures: - `reserve_chunks(max: PosInt, strategy: Strategy) -> Stream` -- `complete_chunk(SubmmissionId, ChunkIndex) -> ()` +- `complete_chunk(SubmissionId, ChunkIndex) -> ()` - `retry_or_fail_chunk(SubmissiionId, ChunkIndex) -> ()` -`complete_chunk` and `retry_or_faill_chunk` take a super brief write lock on the SQLite database to update the state of the particular chunk in its table, +`complete_chunk` and `retry_or_fail_chunk` take a super brief write lock on the SQLite database to update the state of the particular chunk in its table, either incrementing the number of retries, moving it to `chunks_failed` or `chunks_completed`. In very rare situations (once per submission), they do more work, by moving the submission to its result table as well (and for `retry_or_fail_chunk`, moving all remaining chunks of that submissionn as well). These two calls never block a consumer however; from the perspective of the consumer they are fire-and-forget. @@ -180,7 +180,7 @@ But we can normalize this further: 1. There are two kinds of metadata. 'I want access to extra submission-specific info inside the consumer implementation' which can just be a BLOB, and 'I want to customize the chhunk reservation strategy', hereafter called 'strategic metadata'. 2. Strategic metadata is relational in nature. Rather than an arbitrary BLOB, it is a `Map` for each submission, meaning we can store it as submission-key-value triples in a separate metadata table. -3. For extra efficiency, we can consider 'custom priority' yet separate from tthe other strategic metadata. This because we probably want to make a different choice w.r.t. providing an index there than for the other kinds of metadata. +3. For extra efficiency, we can consider 'custom priority' yet separate from the other strategic metadata. This because we probably want to make a different choice w.r.t. providing an index there than for the other kinds of metadata. 4. We don't have to support more than a single 'custom priority' field, because if people have a multi-tier priority system they can combine that into a single formula. The strategic metadata table would look as follows: @@ -221,7 +221,7 @@ because they can query the metadata table directly, and then join the chunks tab Combining it with the randomm order however requires either a full table scan of the chunks, or adding a join table `chunks_x_metadata` which will be quite large as it will contain the product of the number of chunks times the number of metadata keys. -We could mitigate the write amplification and larger database size somewhat by storing it in a differrent SQLite 'schema' (a different database file) that we don't persist and that isn't part of the backups, and recreating it on startup. See the 'cheaper indexes' section below for details. +We could mitigate the write amplification and larger database size somewhat by storing it in a different SQLite 'schema' (a different database file) that we don't persist and that isn't part of the backups, and recreating it on startup. See the 'cheaper indexes' section below for details. Once we have such index, we could use the 'cutting the deck' technique with it as well, which is really nice. diff --git a/libs/opsqueue_python/python/opsqueue/exceptions.py b/libs/opsqueue_python/python/opsqueue/exceptions.py index 615dc95..c946f33 100644 --- a/libs/opsqueue_python/python/opsqueue/exceptions.py +++ b/libs/opsqueue_python/python/opsqueue/exceptions.py @@ -198,7 +198,7 @@ class UnexpectedOpsqueueConsumerServerResponseError(OpsqueueInternalError): to a sync method something other than what it had expected. This may indicate that the client and server versions are not matching, - or that there is an implementatoin error in the Opsqueue code. + or that there is an implementation error in the Opsqueue code. """ pass @@ -232,7 +232,7 @@ class ChunkStorageError(OpsqueueInternalError): class InternalConsumerClientError(OpsqueueInternalError): """ - Raised for any othre kind of Consumer client exception + Raised for any other kind of Consumer client exception """ pass @@ -240,7 +240,7 @@ class InternalConsumerClientError(OpsqueueInternalError): class InternalProducerClientError(OpsqueueInternalError): """ - Raised for any othre kind of Producer client exception + Raised for any other kind of Producer client exception """ pass diff --git a/libs/opsqueue_python/python/opsqueue/producer.py b/libs/opsqueue_python/python/opsqueue/producer.py index d72f0cd..82a877e 100644 --- a/libs/opsqueue_python/python/opsqueue/producer.py +++ b/libs/opsqueue_python/python/opsqueue/producer.py @@ -359,7 +359,7 @@ def get_submission_status( def lookup_submission_id_by_prefix(self, prefix: str) -> SubmissionId | None: """ Attempts to find the submission ID if only the prefix of the submission - (AKA the path at which the submision's chunks are stored in the object store) + (AKA the path at which the submission's chunks are stored in the object store) is known. Returns `None` if no submission could be found.