Merge #220: Remove / fix leftover TODO comments

Approved-by: Qqwy
Priority: Normal
Auto-deploy: false

Author: OpsBotPrime

libs/opsqueue_python/opsqueue_python.nix

@@ -19,30 +19,9 @@ buildPythonPackage rec {
     name = "opsqueue_python";
     src = ../../.;
 
-    # TODO: I couldn't get it to work with the filter enabled.
-    # We're now copying slightly too much to the Nix store.
-    # Re-enable properly.
-
-    # srcWhitelist = [
-    #   "Cargo.toml"
-    #   "Cargo.lock"
-    #   "libs/opsqueue_python"
-    #   "libs/opsqueue_python/Cargo.toml"
-    #   "libs/opsqueue_python/pyproject.toml"
-    #   "libs/opsqueue_python/src(/.*)?"
-    #   "libs/opsqueue_python/python(/.*)?"
-
-    #   # Opsqueue is a dependency, so that needs to be included too
-    #   "opsqueue"
-    #   "opsqueue/Cargo.toml"
-    #   "opsqueue/.cargo(/.*)?"
-    #   "opsqueue/build\.rs"
-    #   "opsqueue/opsqueue_example_database_schema\.db"
-    #   # "opsqueue/app(/.*)?"
-    #   "opsqueue/migrations(/.*)?"
-    #   "opsqueue/src(/.*)?"
-    # ];
-
+    # We're copying slightly too much to the Nix store here,
+    # but using the more granular file filter was very error-prone.
+    # This is one thing that could be improved a little in the future.
     srcGlobalWhitelist = [
       ".py"
       ".pyi"

libs/opsqueue_python/python/opsqueue/exceptions.py

@@ -14,9 +14,6 @@ class SubmissionFailedError(Exception):
 
     This means that part of the submission is missing and therefore
     it can never be 'completed' anymore.
-
-    TODO include chunk error in error message,
-    and get at info about failed chunk.
     """
 
     def __init__(
@@ -86,7 +83,7 @@ class ChunkCountIsZeroError(IncorrectUsageError):
     """
     Raised when making an empty submission.
 
-    TODO it is likely that empty submissoins should not be a user error
+    TODO it is likely that empty submissions should not be a user error
     but rather be handled transparently by Opsqueue.
     """
 

libs/opsqueue_python/python/opsqueue/producer.py

@@ -142,7 +142,6 @@ def insert_submission(
 
         Raises:
         - `ChunkSizeIsZeroError` if passing an incorrect chunk size of zero;
-        - `ChunkCountIsZeroError` if passing an empty list of operations;
         - `InternalProducerClientError` if there is a low-level internal error.
         """
         return self.insert_submission_chunks(
@@ -164,7 +163,8 @@ def blocking_stream_completed_submission(
 
         Raises:
         - `InternalProducerClientError` if there is a low-level internal error.
-        - TODO special exception for when the submission fails.
+        - `SubmissionFailedError` if the submission failed permanently
+          (after retrying a consumer kept failing on one of the chunks)
         """
         return _unchunk_iterator(
             self.blocking_stream_completed_submission_chunks(submission_id),
@@ -205,9 +205,9 @@ def run_submission_chunks(
         (If opsqueue or the object storage cannot be reached, exceptions will also be raised).
 
         Raises:
-        - `ChunkCountIsZeroError` if passing an empty list of operations;
         - `InternalProducerClientError` if there is a low-level internal error.
-        - TODO special exception for when the submission fails.
+        - `SubmissionFailedError` if the submission failed permanently
+          (after retrying a consumer kept failing on one of the chunks)
         """
         submission_id = self.insert_submission_chunks(
             chunk_contents,
@@ -225,7 +225,10 @@ async def async_run_submission_chunks(
         strategic_metadata: None | dict[str, str | int] = None,
         chunk_size: None | int = None,
     ) -> AsyncIterator[bytes]:
-        # TODO: the insertion is not async yet.
+        # NOTE: the insertion is not currently async.
+        # Why? Simplicity. This is unlikely to be the bottleneck
+        # for most async apps.
+        # If it does cause a problem in the future this can be revisited
         submission_id = self.insert_submission_chunks(
             chunk_contents,
             metadata=metadata,
@@ -248,7 +251,6 @@ def insert_submission_chunks(
         returning an ID you can use to track the submission's progress afterwards.
 
         Raises:
-        - `ChunkCountIsZeroError` if passing an empty list of operations;
         - `InternalProducerClientError` if there is a low-level internal error.
         """
         otel_trace_carrier = tracing.current_opentelemetry_tracecontext_to_carrier()
@@ -270,7 +272,8 @@ def blocking_stream_completed_submission_chunks(
 
         Raises:
         - `InternalProducerClientError` if there is a low-level internal error.
-        - TODO special exception for when the submission fails.
+        - `SubmissionFailedError` if the submission failed permanently
+          (after retrying a consumer kept failing on one of the chunks)
         """
         return self.inner.blocking_stream_completed_submission_chunks(submission_id)  # type: ignore[no-any-return]
 

libs/opsqueue_python/src/consumer.rs

@@ -230,8 +230,6 @@ impl ConsumerClient {
             E<ChunkRetrievalError, E<InternalConsumerClientError, IncorrectUsage<LimitIsZero>>>,
         >,
     > {
-        // TODO: Currently we do short-polling here if there are no chunks available.
-        // This is quite suboptimal; long-polling would be much nicer.
         const POLL_INTERVAL: Duration = Duration::from_millis(500);
         loop {
             let res = self.block_unless_interrupted(async {

libs/opsqueue_python/src/errors.rs

@@ -6,8 +6,7 @@ use opsqueue::common::chunk::ChunkId;
 use opsqueue::common::errors::{
     ChunkNotFound, IncorrectUsage, SubmissionNotFound, UnexpectedOpsqueueConsumerServerResponse, E,
 };
-use opsqueue::common::NonZeroIsZero;
-use pyo3::exceptions::{PyBaseException, PyException};
+use pyo3::exceptions::PyBaseException;
 use pyo3::{import_exception, Bound, PyErr, Python};
 
 use crate::common::{ChunkIndex, SubmissionId};
@@ -20,7 +19,6 @@ import_exception!(opsqueue.exceptions, IncorrectUsageError);
 import_exception!(opsqueue.exceptions, TryFromIntError);
 import_exception!(opsqueue.exceptions, ChunkNotFoundError);
 import_exception!(opsqueue.exceptions, SubmissionNotFoundError);
-import_exception!(opsqueue.exceptions, ChunkCountIsZeroError);
 import_exception!(opsqueue.exceptions, NewObjectStoreClientError);
 import_exception!(opsqueue.exceptions, SubmissionNotCompletedYetError);
 
@@ -119,12 +117,6 @@ impl From<CError<opsqueue::object_store::ChunkStorageError>> for PyErr {
     }
 }
 
-impl From<CError<NonZeroIsZero<opsqueue::common::chunk::ChunkIndex>>> for PyErr {
-    fn from(value: CError<NonZeroIsZero<opsqueue::common::chunk::ChunkIndex>>) -> Self {
-        ChunkCountIsZeroError::new_err(value.0.to_string())
-    }
-}
-
 impl<T: Error> From<CError<IncorrectUsage<T>>> for PyErr {
     fn from(value: CError<IncorrectUsage<T>>) -> Self {
         IncorrectUsageError::new_err(value.0.to_string())
@@ -180,14 +172,6 @@ impl From<CError<opsqueue::object_store::NewObjectStoreClientError>> for PyErr {
     }
 }
 
-// TODO: Only temporary. We want to get rid of all usage of anyhow
-// in the boundary to PyO3
-impl From<CError<anyhow::Error>> for PyErr {
-    fn from(value: CError<anyhow::Error>) -> Self {
-        PyException::new_err(value.0.to_string())
-    }
-}
-
 impl From<CError<UnexpectedOpsqueueConsumerServerResponse>> for PyErr {
     fn from(value: CError<UnexpectedOpsqueueConsumerServerResponse>) -> Self {
         UnexpectedOpsqueueConsumerServerResponseError::new_err(value.0.to_string())

libs/opsqueue_python/src/producer.rs

@@ -8,21 +8,18 @@ use pyo3::{
 };
 
 use futures::{stream::BoxStream, StreamExt, TryStreamExt};
+use opsqueue::{
+    common::errors::E::{self, L, R},
+    object_store::{ChunksStorageError, NewObjectStoreClientError},
+    producer::client::{Client as ActualClient, InternalProducerClientError},
+};
 use opsqueue::{
     common::{chunk, submission, StrategicMetadataMap},
     object_store::{ChunkRetrievalError, ChunkType},
     producer::common::ChunkContents,
     tracing::CarrierMap,
     E,
 };
-use opsqueue::{
-    common::{
-        errors::E::{self, L, R},
-        NonZeroIsZero,
-    },
-    object_store::{ChunksStorageError, NewObjectStoreClientError},
-    producer::client::{Client as ActualClient, InternalProducerClientError},
-};
 use ux_serde::u63;
 
 use crate::{
@@ -163,6 +160,9 @@ 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
     #[pyo3(signature = (chunk_contents, metadata=None, chunk_size=None, otel_trace_carrier=CarrierMap::default()))]
     pub fn insert_submission_direct(
         &self,
@@ -172,7 +172,6 @@ impl ProducerClient {
         chunk_size: Option<u64>,
         otel_trace_carrier: CarrierMap,
     ) -> CPyResult<SubmissionId, E<FatalPythonException, InternalProducerClientError>> {
-        // TODO
         let strategic_metadata = Default::default();
 
         py.allow_threads(|| {
@@ -208,7 +207,6 @@ impl ProducerClient {
         SubmissionId,
         E![
             FatalPythonException,
-            NonZeroIsZero<chunk::ChunkIndex>,
             ChunksStorageError,
             InternalProducerClientError,
         ],
@@ -227,7 +225,7 @@ impl ProducerClient {
                     self.object_store_client
                         .store_chunks(&prefix, ChunkType::Input, stream)
                         .await
-                        .map_err(|e| CError(R(R(L(e)))))
+                        .map_err(|e| CError(R(L(e))))
                 })
             })?;
             let chunk_count = chunk::ChunkIndex::from(chunk_count);
@@ -250,7 +248,7 @@ impl ProducerClient {
                         tracing::debug!("Submitting finished; Submission ID {submission_id} assigned to subfolder {prefix}");
                         submission_id.into()
                     })
-                    .map_err(|e| R(R(R(e))).into())
+                    .map_err(|e| R(R(e)).into())
             })
         })
     }
@@ -269,7 +267,6 @@ impl ProducerClient {
             InternalProducerClientError,
         ],
     > {
-        // TODO: Use CPyResult instead
         py.allow_threads(|| {
             self.block_unless_interrupted(async move {
                 match self
@@ -299,7 +296,6 @@ impl ProducerClient {
         E![
             FatalPythonException,
             errors::SubmissionFailed,
-            NonZeroIsZero<chunk::ChunkIndex>,
             ChunksStorageError,
             InternalProducerClientError,
         ],
@@ -325,7 +321,7 @@ impl ProducerClient {
                 CError(match e {
                     L(e) => L(e),
                     R(L(e)) => R(L(e)),
-                    R(R(e)) => R(R(R(R(e)))),
+                    R(R(e)) => R(R(R(e))),
                 })
             })?;
         Ok(res)

libs/opsqueue_python/tests/test_rountrip.py

@@ -47,6 +47,26 @@ def run_consumer() -> None:
         assert res == sum(range(1, 101))
 
 
+def test_empty_submission(
+    opsqueue: OpsqueueProcess, any_consumer_strategy: Strategy
+) -> None:
+    """
+    Empty submissions ought to be supported without problems.
+    Opsqueue should immediately consider these 'completed'
+    and no errors should be thrown.
+    """
+    producer_client = ProducerClient(
+        f"localhost:{opsqueue.port}", "file:///tmp/opsqueue/test_empty_submission"
+    )
+
+    input_iter: list[int] = []
+    output_iter: Iterator[int] = producer_client.run_submission(
+        input_iter, chunk_size=20
+    )
+    res = sum(output_iter)
+    assert res == 0
+
+
 def test_roundtrip_explicit_serialization_format(
     opsqueue: OpsqueueProcess,
     any_consumer_strategy: Strategy,

opsqueue/src/common/chunk.rs

@@ -5,7 +5,6 @@ use ux_serde::u63;
 
 use super::errors::TryFromIntError;
 use super::submission::SubmissionId;
-use super::MayBeZero;
 
 /// Index of this particular chunk in a submission.
 #[derive(
@@ -64,12 +63,6 @@ impl ChunkIndex {
     }
 }
 
-impl MayBeZero for ChunkIndex {
-    fn is_zero(&self) -> bool {
-        self.0 == u63::new(0)
-    }
-}
-
 impl PartialEq<u64> for ChunkIndex {
     fn eq(&self, other: &u64) -> bool {
         u64::from(*self) == *other

opsqueue/src/common/mod.rs

@@ -1,60 +1,9 @@
-use std::fmt::Debug;
-
 use rustc_hash::FxHashMap;
 
 pub mod chunk;
 pub mod errors;
 pub mod submission;
 
-pub trait MayBeZero {
-    fn is_zero(&self) -> bool;
-}
-
-#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, serde::Serialize)]
-pub struct NonZero<T>(T);
-
-impl<'de, T: MayBeZero + Debug + serde::Deserialize<'de>> serde::Deserialize<'de> for NonZero<T> {
-    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
-    where
-        D: serde::Deserializer<'de>,
-    {
-        let inner = T::deserialize(deserializer)?;
-        NonZero::try_from(inner).map_err(|e| {
-            serde::de::Error::invalid_value(
-                serde::de::Unexpected::Other(&format!("zero (0)-value {e:?}")),
-                &&*format!("a non-zero value of type{}", std::any::type_name::<T>()),
-            )
-        })
-    }
-}
-
-impl<T> NonZero<T> {
-    pub fn inner(&self) -> &T {
-        &self.0
-    }
-    pub fn into_inner(self) -> T {
-        self.0
-    }
-}
-
-#[derive(thiserror::Error, Debug)]
-#[error("Given a zero-value where a non-zero value is expected: {0:?}")]
-pub struct NonZeroIsZero<T>(T);
-
-impl<T: MayBeZero> NonZero<T> {
-    /// builds a new NonZero from a given value.
-    ///
-    /// Nicer would be to implement TryFrom but we cannot
-    /// because of the blanket implementation `TryFrom<T, U: Into<T>> ...` :-(
-    pub fn try_from(value: T) -> Result<Self, NonZeroIsZero<T>> {
-        if value.is_zero() {
-            Err(NonZeroIsZero(value))
-        } else {
-            Ok(Self(value))
-        }
-    }
-}
-
 /// 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.
 ///