Merge #11: Library docs improvements

OpsBotPrime · OpsBotPrime · commit db579a67ad35 · 2025-09-25T10:43:03.000+02:00
Approved-by: rlycx
Priority: Normal
Auto-deploy: false
diff --git a/default.nix b/default.nix
@@ -12,6 +12,7 @@ let
       pytest
       pytest-random-order
       pytest-parallel
+      pytest-timeout
 
       # Repeated here so MyPy sees them:
       cbor2
diff --git a/libs/opsqueue_python/Cargo.toml b/libs/opsqueue_python/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "opsqueue_python"
-version = "0.30.5"
+version = "0.30.6"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
diff --git a/libs/opsqueue_python/src/common.rs b/libs/opsqueue_python/src/common.rs
@@ -232,7 +232,7 @@ impl Chunk {
 #[pymethods]
 impl Chunk {
     fn __repr__(&self) -> String {
-        format!("{:?}", self)
+        format!("{self:?}")
     }
 }
 
@@ -261,14 +261,14 @@ impl ChunkFailed {
 #[pymethods]
 impl ChunkFailed {
     fn __repr__(&self) -> String {
-        format!("{:?}", self)
+        format!("{self:?}")
     }
 }
 
 #[pymethods]
 impl Strategy {
     fn __repr__(&self) -> String {
-        format!("{:?}", self)
+        format!("{self:?}")
     }
 }
 
@@ -365,7 +365,7 @@ impl Submission {
 #[pymethods]
 impl SubmissionStatus {
     fn __repr__(&self) -> String {
-        format!("{:?}", self)
+        format!("{self:?}")
     }
 }
 
@@ -476,17 +476,17 @@ pub fn start_runtime() -> Arc<tokio::runtime::Runtime> {
 ///
 /// Internally acquires the GIL!
 ///
-/// c.f. https://pyo3.rs/main/doc/pyo3/types/trait.pytracebackmethods
+/// c.f. <https://pyo3.rs/main/doc/pyo3/types/trait.pytracebackmethods>
 pub fn format_pyerr(err: &PyErr) -> String {
     Python::with_gil(|py| {
         let msg: Option<String> = (|| {
             let traceback = err.traceback(py)?;
             let traceback_str = traceback
                 .format()
                 .expect("Tracebacks are always formattable");
-            let str = format!("{}{}", traceback_str, err);
+            let str = format!("{traceback_str}{err}");
             Some(str)
         })();
-        msg.unwrap_or_else(|| format!("{}", err))
+        msg.unwrap_or_else(|| format!("{err}"))
     })
 }
diff --git a/libs/opsqueue_python/src/consumer.rs b/libs/opsqueue_python/src/consumer.rs
@@ -51,9 +51,9 @@ impl ConsumerClient {
     /// :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: https://docs.rs/object_store/0.11.1/object_store/enum.ObjectStoreScheme.html#method.parse
+    ///   Supports the formats listed here: <https://docs.rs/object_store/0.11.1/object_store/enum.ObjectStoreScheme.html#method.parse>
     /// :param object_store_options: A list of key-value strings as extra options for the chosen object store.
-    ///        For example, for GCS, see https://docs.rs/object_store/0.11.2/object_store/gcp/enum.GoogleConfigKey.html#variants
+    ///        For example, for GCS, see <https://docs.rs/object_store/0.11.2/object_store/gcp/enum.GoogleConfigKey.html#variants>
     #[new]
     #[pyo3(signature = (address, object_store_url, object_store_options=vec![]))]
     pub fn new(
diff --git a/libs/opsqueue_python/src/errors.rs b/libs/opsqueue_python/src/errors.rs
@@ -38,7 +38,7 @@ import_exception!(opsqueue.exceptions, InternalProducerClientError);
 /// into PyErr, including those defined in other crates.
 ///
 /// This follows the 'newtype wrapper' approach from
-/// https://pyo3.rs/v0.22.5/function/error-handling#foreign-rust-error-types
+/// <https://pyo3.rs/v0.22.5/function/error-handling#foreign-rust-error-types>
 ///
 /// The 'C' stands for 'Convertible'.
 pub struct CError<T>(pub T);
@@ -52,15 +52,15 @@ impl<T> From<T> for CError<T> {
 /// into PyErr.
 ///
 /// This follows the 'newtype wrapper' approach from
-/// https://pyo3.rs/v0.22.5/function/error-handling#foreign-rust-error-types
+/// <https://pyo3.rs/v0.22.5/function/error-handling#foreign-rust-error-types>
 ///
 /// The 'C' stands for 'Convertible'.
 pub type CPyResult<T, E> = Result<T, CError<E>>;
 
 /// Indicates a 'fatal' PyErr: Any Python exception which is _not_ a subclass of `PyException`.
 ///
 /// These are known as 'fatal' exceptions in Python.
-/// c.f. https://docs.python.org/3/tutorial/errors.html#tut-userexceptions
+/// c.f. <https://docs.python.org/3/tutorial/errors.html#tut-userexceptions>
 ///
 /// We don't consume/wrap these errors but propagate them,
 /// allowing things like KeyboardInterrupt, SystemExit or MemoryError,
diff --git a/libs/opsqueue_python/src/lib.rs b/libs/opsqueue_python/src/lib.rs
@@ -37,6 +37,5 @@ fn opsqueue_internal(m: &Bound<'_, PyModule>) -> PyResult<()> {
     )?;
 
     // Top-level functions
-    // m.add_function(wrap_pyfunction!(sum_as_string, m)?)?;
     Ok(())
 }
diff --git a/libs/opsqueue_python/src/producer.rs b/libs/opsqueue_python/src/producer.rs
@@ -16,7 +16,7 @@ use opsqueue::{
 use opsqueue::{
     common::{chunk, submission, StrategicMetadataMap},
     object_store::{ChunkRetrievalError, ChunkType},
-    producer::common::ChunkContents,
+    producer::ChunkContents,
     tracing::CarrierMap,
     E,
 };
@@ -50,9 +50,9 @@ impl ProducerClient {
     /// :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: https://docs.rs/object_store/0.11.1/object_store/enum.ObjectStoreScheme.html#method.parse
+    ///   Supports the formats listed here: <https://docs.rs/object_store/0.11.1/object_store/enum.ObjectStoreScheme.html#method.parse>
     /// :param object_store_options: A list of key-value strings as extra options for the chosen object store.
-    ///        For example, for GCS, see https://docs.rs/object_store/0.11.2/object_store/gcp/enum.GoogleConfigKey.html#variants
+    ///        For example, for GCS, see <https://docs.rs/object_store/0.11.2/object_store/gcp/enum.GoogleConfigKey.html#variants>
     #[new]
     #[pyo3(signature = (address, object_store_url, object_store_options=vec![]))]
     pub fn new(
@@ -81,7 +81,7 @@ impl ProducerClient {
     pub fn __repr__(&self) -> String {
         format!(
             "<opsqueue_producer.ProducerClient(address={:?}, object_store_url={:?})>",
-            self.producer_client.endpoint_url,
+            self.producer_client.base_url(),
             self.object_store_client.url()
         )
     }
@@ -176,7 +176,7 @@ impl ProducerClient {
         let strategic_metadata = Default::default();
 
         py.allow_threads(|| {
-            let submission = opsqueue::producer::common::InsertSubmission {
+            let submission = opsqueue::producer::InsertSubmission {
                 chunk_size: chunk_size.map(|n| chunk::ChunkSize(n as i64)),
                 chunk_contents: ChunkContents::Direct {
                     contents: chunk_contents,
@@ -235,7 +235,7 @@ impl ProducerClient {
             tracing::debug!("Finished uploading to object store. {prefix} contains {chunk_count} chunks");
 
             self.block_unless_interrupted(async move {
-                let submission = opsqueue::producer::common::InsertSubmission {
+                let submission = opsqueue::producer::InsertSubmission {
                     chunk_size: chunk_size.map(chunk::ChunkSize),
                     chunk_contents: ChunkContents::SeeObjectStorage {
                         prefix: prefix.clone(),
diff --git a/opsqueue/Cargo.toml b/opsqueue/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "opsqueue"
-version = "0.30.5"
+version = "0.30.6"
 edition = "2021"
 description = "lightweight batch processing queue for heavy loads"
 repository = "https://github.com/channable/opsqueue"
diff --git a/opsqueue/app/main.rs b/opsqueue/app/main.rs
@@ -136,7 +136,6 @@ fn setup_tracing() -> OtelGuard {
                 .with_thread_ids(true)
                 .with_target(true),
         )
-        // .with(MetricsLayer::new(meter_provider.clone()))
         .with(tracing_opentelemetry::OpenTelemetryLayer::new(otel_tracer()))
         // While we don´t forward traces to Sentry, we do want info and above spans to show up as breadcrumbs
         // and error spans to show up as errors
diff --git a/opsqueue/src/common/chunk.rs b/opsqueue/src/common/chunk.rs
@@ -546,7 +546,6 @@ pub mod db {
     ) -> sqlx::Result<()> {
         const ROWS_PER_QUERY: usize = 1000;
 
-        // let start = std::time::Instant::now();
         let mut iter = chunks.iter().peekable();
         while iter.peek().is_some() {
             let query_chunks = iter.by_ref().take(ROWS_PER_QUERY);
@@ -721,7 +720,7 @@ pub mod test {
 
         let mut submission = Submission::new();
         submission.chunks_total = u63::new(1).into();
-        submission.id = Submission::generate_id();
+        submission.id = SubmissionId::new();
         let chunk = Chunk::new(
             submission.id,
             u63::new(0).into(),
diff --git a/opsqueue/src/common/submission.rs b/opsqueue/src/common/submission.rs
@@ -12,11 +12,45 @@ pub type Metadata = Vec<u8>;
 
 static ID_GENERATOR: snowflaked::sync::Generator = snowflaked::sync::Generator::new(0);
 
+/// Uniquely identifies a [`Submission`].
+///
+/// Submission IDs are snowflakes. These are 64-bit identifiers that includes
+/// creation time information and are sortable on that timestamp.
 #[derive(
     Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize,
 )]
 pub struct SubmissionId(u63);
 
+impl Default for SubmissionId {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl SubmissionId {
+    /// Generate a new [`SubmissionId`].
+    pub fn new() -> SubmissionId {
+        let inner: u64 = ID_GENERATOR.generate();
+        SubmissionId(u63::new(inner))
+    }
+    /// Access the [`std::time::SystemTime`] at which the ID was generated.
+    pub fn system_time(self) -> std::time::SystemTime {
+        use snowflaked::Snowflake;
+        let inner: u64 = self.0.into();
+
+        let unix_timestamp_ms = inner.timestamp();
+        let unix_timestamp = Duration::from_millis(unix_timestamp_ms);
+        ID_GENERATOR
+            .epoch()
+            .checked_add(unix_timestamp)
+            .expect("Invalid timestamp extracted from snowflake ID")
+    }
+    /// Get the time at which the submission ID was generated as a [`chrono::DateTime`].
+    pub fn timestamp(self) -> chrono::DateTime<Utc> {
+        self.system_time().into()
+    }
+}
+
 impl Display for SubmissionId {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         self.0.fmt(f)
@@ -71,24 +105,12 @@ impl From<&SubmissionId> for std::time::SystemTime {
     }
 }
 
-impl SubmissionId {
-    pub fn system_time(self) -> std::time::SystemTime {
-        use snowflaked::Snowflake;
-        let inner: u64 = self.0.into();
-
-        let unix_timestamp_ms = inner.timestamp();
-        let unix_timestamp = Duration::from_millis(unix_timestamp_ms);
-        ID_GENERATOR
-            .epoch()
-            .checked_add(unix_timestamp)
-            .expect("Invalid timestamp extracted from snowflake ID")
-    }
-
-    pub fn timestamp(self) -> chrono::DateTime<Utc> {
-        self.system_time().into()
-    }
-}
-
+/// A submission is a group of [`Chunk`]s.
+///
+/// This struct represents a submission that has not yet reached an end state.
+/// That is, they are neither completed nor failed.
+///
+/// Also see [`SubmissionCompleted`] and [`SubmissionFailed`].
 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 pub struct Submission {
     pub id: SubmissionId,
@@ -100,6 +122,11 @@ pub struct Submission {
     pub otel_trace_carrier: String,
 }
 
+/// A submission that has been completed successfully.
+///
+/// Once a submission is marked completed (which happens when all its chunks are
+/// marked as done), it gets moved to the `submissions_completed` table, and its
+/// old `submissions` record gets deleted.
 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 pub struct SubmissionCompleted {
     pub id: SubmissionId,
@@ -150,17 +177,12 @@ impl Submission {
         }
     }
 
-    pub fn generate_id() -> SubmissionId {
-        let inner: u64 = ID_GENERATOR.generate();
-        SubmissionId(u63::new(inner))
-    }
-
     pub fn from_vec(
         chunks: Vec<chunk::Content>,
         metadata: Option<Metadata>,
         chunk_size: ChunkSize,
     ) -> Option<(Submission, Vec<Chunk>)> {
-        let submission_id = Self::generate_id();
+        let submission_id = SubmissionId::new();
         let len = ChunkCount::new(u64::try_from(chunks.len()).ok()?).ok()?;
         let otel_trace_carrier = crate::tracing::current_context_to_json();
         let submission = Submission {
@@ -339,7 +361,7 @@ pub mod db {
         chunk_size: ChunkSize,
         mut conn: impl WriterConnection,
     ) -> Result<SubmissionId, DatabaseError> {
-        let submission_id = Submission::generate_id();
+        let submission_id = SubmissionId::new();
         let len = chunks_contents.len().try_into().expect("Vector length larger than u63 range. Unlikely because of RAM constraints but theoretically possible");
         let otel_trace_carrier = crate::tracing::current_context_to_json();
         let submission = Submission {
diff --git a/opsqueue/src/consumer/client.rs b/opsqueue/src/consumer/client.rs
@@ -268,8 +268,7 @@ impl Client {
                                 tracing::debug!("Opsqueue consumer client background task closing as WebSocket connection closed");
                                 break
                             } else if msg.is_ping() {
-                                tracing::debug!("Received Heartbeat, expect auto-pong");
-                                // let _ = ws_sink.lock().await.send(Message::pong("heartbeat")).await;
+                                tracing::debug!("Received Heartbeat, expecting auto-pong");
                             } else if msg.is_pong() {
                                 tracing::debug!("Received Pong reply to heartbeat, nice!");
                             } else if msg.is_binary() {
diff --git a/opsqueue/src/consumer/dispatcher/metastate.rs b/opsqueue/src/consumer/dispatcher/metastate.rs
@@ -12,15 +12,13 @@ impl MetaState {
     }
 
     pub fn increment(&self, key: &str, val: &MetaStateVal) {
-        // tracing::warn!("Incrementing {key}, {val:?}, me before: {self:?}");
         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) {
-        // tracing::warn!("Decrementing {key}, {val:?}, me before: {self:?}");
         let ripe_for_removal = {
             let meta_state_field = self
                 .0
diff --git a/opsqueue/src/consumer/dispatcher/reserver.rs b/opsqueue/src/consumer/dispatcher/reserver.rs
@@ -58,10 +58,6 @@ where
         }
     }
 
-    // pub fn metastate(&self) -> &MetaState {
-    //     &self.metastate
-    // }
-
     /// Attempts to reserve a particular key-val.
     ///
     /// Returns `None` if someone else currently is already reserving `key`.
diff --git a/opsqueue/src/consumer/server/conn.rs b/opsqueue/src/consumer/server/conn.rs
@@ -244,7 +244,6 @@ impl ConsumerConn {
             }
             CompleteChunk { id, output_content } => {
                 self.consumer_state.complete_chunk(id, output_content).await;
-                // .map_err(anyhow::Error::from)?;
                 None
             }
             FailChunk { id, failure } => {
@@ -329,7 +328,7 @@ impl ConsumerConnectionId {
     /// time. The colors list runs out after 24, and after that it'll fall back to UUIDs.
     pub fn color() -> ConsumerConnectionId {
         static mut INDEX: AtomicUsize = AtomicUsize::new(0);
-        const COLORS: [&'static str; 24] = [
+        const COLORS: [&str; 24] = [
             "red",
             "green",
             "blue",
@@ -364,7 +363,7 @@ impl ConsumerConnectionId {
             .get(i)
             .map(|c| String::from(*c))
             .map(ConsumerConnectionId)
-            .unwrap_or_else(|| Self::uuid())
+            .unwrap_or_else(Self::uuid)
     }
 }
 
diff --git a/opsqueue/src/consumer/server/state.rs b/opsqueue/src/consumer/server/state.rs
@@ -41,8 +41,6 @@ impl Drop for ConsumerState {
 impl ConsumerState {
     pub fn new(server_state: &Arc<ServerState>) -> Self {
         Self {
-            // pool: server_state.pool.read_pool.clone(),
-            // dispatcher: server_state.dispatcher.clone(),
             reservations: Arc::new(Mutex::new(HashSet::new())),
             server_state: server_state.clone(),
         }
diff --git a/opsqueue/src/consumer/strategy.rs b/opsqueue/src/consumer/strategy.rs
@@ -140,7 +140,7 @@ pub mod test {
                 let id = row.get::<i64, &str>("id");
                 let parent = row.get::<i64, &str>("parent");
                 let detail = row.get::<String, &str>("detail");
-                format!("{}, {}, {}", id, parent, detail)
+                format!("{id}, {parent}, {detail}")
             })
             .join("\n")
     }
diff --git a/opsqueue/src/object_store/mod.rs b/opsqueue/src/object_store/mod.rs
diff --git a/opsqueue/src/producer/client.rs b/opsqueue/src/producer/client.rs
diff --git a/opsqueue/src/producer/common.rs b/opsqueue/src/producer/common.rs
diff --git a/opsqueue/src/producer/mod.rs b/opsqueue/src/producer/mod.rs

Original file line number	Diff line number	Diff line change
`@@ -232,7 +232,7 @@ impl Chunk {`
`232`	`232`	`#[pymethods]`
`233`	`233`	`impl Chunk {`
`234`	`234`	`fn __repr__(&self) -> String {`
`235`		`- format!("{:?}", self)`
	`235`	`+ format!("{self:?}")`
`236`	`236`	`}`
`237`	`237`	`}`
`238`	`238`
`@@ -261,14 +261,14 @@ impl ChunkFailed {`
`261`	`261`	`#[pymethods]`
`262`	`262`	`impl ChunkFailed {`
`263`	`263`	`fn __repr__(&self) -> String {`
`264`		`- format!("{:?}", self)`
	`264`	`+ format!("{self:?}")`
`265`	`265`	`}`
`266`	`266`	`}`
`267`	`267`
`268`	`268`	`#[pymethods]`
`269`	`269`	`impl Strategy {`
`270`	`270`	`fn __repr__(&self) -> String {`
`271`		`- format!("{:?}", self)`
	`271`	`+ format!("{self:?}")`
`272`	`272`	`}`
`273`	`273`	`}`
`274`	`274`
`@@ -365,7 +365,7 @@ impl Submission {`
`365`	`365`	`#[pymethods]`
`366`	`366`	`impl SubmissionStatus {`
`367`	`367`	`fn __repr__(&self) -> String {`
`368`		`- format!("{:?}", self)`
	`368`	`+ format!("{self:?}")`
`369`	`369`	`}`
`370`	`370`	`}`
`371`	`371`
`@@ -476,17 +476,17 @@ pub fn start_runtime() -> Arc<tokio::runtime::Runtime> {`
`476`	`476`	`///`
`477`	`477`	`/// Internally acquires the GIL!`
`478`	`478`	`///`
`479`		`-/// c.f. https://pyo3.rs/main/doc/pyo3/types/trait.pytracebackmethods`
	`479`	`+/// c.f. <https://pyo3.rs/main/doc/pyo3/types/trait.pytracebackmethods>`
`480`	`480`	`pub fn format_pyerr(err: &PyErr) -> String {`
`481`	`481`	`Python::with_gil(\|py\| {`
`482`	`482`	`let msg: Option<String> = (\|\| {`
`483`	`483`	`let traceback = err.traceback(py)?;`
`484`	`484`	`let traceback_str = traceback`
`485`	`485`	`.format()`
`486`	`486`	`.expect("Tracebacks are always formattable");`
`487`		`- let str = format!("{}{}", traceback_str, err);`
	`487`	`+ let str = format!("{traceback_str}{err}");`
`488`	`488`	`Some(str)`
`489`	`489`	`})();`
`490`		`- msg.unwrap_or_else(\|\| format!("{}", err))`
	`490`	`+ msg.unwrap_or_else(\|\| format!("{err}"))`
`491`	`491`	`})`
`492`	`492`	`}`
Original file line number	Diff line number	Diff line change
`@@ -37,6 +37,5 @@ fn opsqueue_internal(m: &Bound<'_, PyModule>) -> PyResult<()> {`
`37`	`37`	`)?;`
`38`	`38`
`39`	`39`	`// Top-level functions`
`40`		`- // m.add_function(wrap_pyfunction!(sum_as_string, m)?)?;`
`41`	`40`	`Ok(())`
`42`	`41`	`}`
Original file line number	Diff line number	Diff line change
`@@ -136,7 +136,6 @@ fn setup_tracing() -> OtelGuard {`
`136`	`136`	`.with_thread_ids(true)`
`137`	`137`	`.with_target(true),`
`138`	`138`	`)`
`139`		`- // .with(MetricsLayer::new(meter_provider.clone()))`
`140`	`139`	`.with(tracing_opentelemetry::OpenTelemetryLayer::new(otel_tracer()))`
`141`	`140`	`// While we don´t forward traces to Sentry, we do want info and above spans to show up as breadcrumbs`
`142`	`141`	`// and error spans to show up as errors`