Add endpoint to cancel a submission

Author: jerbaroo

libs/opsqueue_python/python/opsqueue/producer.py

@@ -314,6 +314,15 @@ def count_submissions(self) -> int:
         """
         return self.inner.count_submissions()  # type: ignore[no-any-return]
 
+    def cancel_submission(self, submission_id: SubmissionId) -> None:
+        """
+        Cancel a specific submission if it's in progress.
+
+        Raises:
+        - `InternalProducerClientError` if there is a low-level internal error.
+        """
+        return self.inner.cancel_submission(submission_id)
+
     def get_submission_status(
         self, submission_id: SubmissionId
     ) -> SubmissionStatus | None:

libs/opsqueue_python/src/producer.rs

@@ -118,6 +118,25 @@ impl ProducerClient {
         })
     }
 
+    /// TODO docstring
+    pub fn cancel_submission(
+        &self,
+        py: Python<'_>,
+        id: SubmissionId,
+    ) -> CPyResult<(), E<FatalPythonException, InternalProducerClientError>>
+    {
+        py.allow_threads(|| {
+            self.block_unless_interrupted(async {
+                self.producer_client
+                    .cancel_submission(id.into())
+                    .await
+                    .map_err(|e| CError(R(e)))
+            })
+            // .map(|opt| opt.map(Into::into))
+            // .map_err(|e| ProducerClientError::new_err(e.to_string()))
+        })
+    }
+
     /// 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,

opsqueue/src/common/submission.rs

@@ -668,9 +668,9 @@ pub mod db {
             "
     INSERT INTO submissions_cancelled
     (id, chunks_total, prefix, metadata, cancelled_at, chunks_done)
-    SELECT id, chunks_total, prefix, metadata, julianday($1), chunks_done FROM submissions WHERE id = $3;
+    SELECT id, chunks_total, prefix, metadata, julianday($1), chunks_done FROM submissions WHERE id = $2;
 
-    DELETE FROM submissions WHERE id = $4 RETURNING *;
+    DELETE FROM submissions WHERE id = $3 RETURNING *;
     ",
             now,
             id,

opsqueue/src/producer/client.rs

@@ -101,6 +101,29 @@ impl Client {
         })
         .retry(retry_policy())
         .when(InternalProducerClientError::is_ephemeral)
+        .notify(|err, dur| {
+           tracing::debug!("retrying error {err:?} with sleeping {dur:?}");}) .await
+    }
+
+    /// TODO docstring
+    pub async fn cancel_submission(
+        &self,
+        submission_id: SubmissionId,
+    ) -> Result<(), InternalProducerClientError> {
+        (|| async {
+            let base_url = &self.base_url;
+            self
+                .http_client
+                .post(format!("{base_url}/submissions/cancel/{submission_id}"))
+                .send()
+                .await?
+                .error_for_status()?;
+            // let bytes = resp.bytes().await?;
+            // let body = serde_json::from_slice(&bytes)?;
+            Ok(())
+        })
+        .retry(retry_policy())
+        .when(InternalProducerClientError::is_ephemeral)
         .notify(|err, dur| {
             tracing::debug!("retrying error {err:?} with sleeping {dur:?}");
         })

opsqueue/src/producer/server.rs

@@ -46,6 +46,10 @@ impl ServerState {
     pub fn build_router(self) -> Router<()> {
         Router::new()
             .route("/submissions", post(insert_submission))
+            .route(
+                "/submissions/cancel/{submission_id}",
+                post(cancel_submission),
+            )
             .route(
                 "/submissions/count_completed",
                 get(submissions_count_completed),
@@ -85,6 +89,15 @@ where
     }
 }
 
+async fn cancel_submission(
+    State(state): State<ServerState>,
+    Path(submission_id): Path<SubmissionId>,
+) -> Result<(), ServerError> {
+    let mut conn = state.pool.writer_conn().await?;
+    submission::db::cancel_submission(submission_id, &mut conn).await?;
+    Ok(())
+}
+
 async fn submission_status(
     State(state): State<ServerState>,
     Path(submission_id): Path<SubmissionId>,