Add delegation support

Author: SemMulder

Cargo.lock

@@ -7,12 +7,12 @@ repository = "https://github.com/channable/opsqueue"
 license = "MIT"
 
 [lib]
-name="opsqueue"
-path="src/lib.rs"
+name = "opsqueue"
+path = "src/lib.rs"
 
 [[bin]]
-name="opsqueue"
-path="app/main.rs"
+name = "opsqueue"
+path = "app/main.rs"
 required-features = ["server-logic"]
 
 [dependencies]
@@ -80,6 +80,8 @@ workspace = true
 
 [dev-dependencies]
 insta.workspace = true
+wiremock = "0.6.5"
+tower = "0.5.3"
 
 [features]
 # Dependencies only in use by the server-logic:
@@ -92,8 +94,9 @@ server-logic = [
     "dep:tower-http",
     "dep:axum-prometheus",
     "dep:sentry",
-    "dep:sentry-tracing"
-    ]
+    "dep:sentry-tracing",
+    "dep:reqwest",
+]
 # Dependencies only in use by the client libraries:
 client-logic = [
     "dep:reqwest",

opsqueue/Cargo.toml

@@ -0,0 +1 @@
+DROP TABLE submissions_external_task;

opsqueue/migrations/20260811150000_submissions_external_task.down.sql

@@ -0,0 +1,6 @@
+CREATE TABLE submissions_external_task
+(
+    submission_id    BIGINT NOT NULL UNIQUE,
+    task_id          TEXT   NOT NULL UNIQUE,
+    last_status_sent TEXT
+);

opsqueue/migrations/20260811150000_submissions_external_task.up.sql

@@ -300,18 +300,21 @@ pub mod db {
         chunk_id: ChunkId,
         output_content: Option<Vec<u8>>,
         mut conn: impl WriterConnection,
-    ) -> Result<(), E<DatabaseError, SubmissionNotFound>> {
-        let chunks_moved = conn
+    ) -> Result<bool, E<DatabaseError, SubmissionNotFound>> {
+        let (chunks_moved, completed_submission) = conn
             .transaction(move |mut tx| {
                 Box::pin(async move {
                     let chunks_moved =
                         complete_chunk_raw(chunk_id, output_content, &mut tx).await?;
+
+                    let mut completed_submission = false;
                     if chunks_moved {
-                        crate::common::submission::db::maybe_complete_submission(
-                            chunk_id.submission_id,
-                            &mut tx,
-                        )
-                        .await?;
+                        completed_submission =
+                            crate::common::submission::db::maybe_complete_submission(
+                                chunk_id.submission_id,
+                                &mut tx,
+                            )
+                            .await?;
                     } else {
                         tracing::warn!(
                             "Could not complete chunk {:?} because it was either: \
@@ -320,15 +323,18 @@ pub mod db {
                         );
                     }
 
-                    Result::<bool, E<DatabaseError, SubmissionNotFound>>::Ok(chunks_moved)
+                    Result::<(bool, bool), E<DatabaseError, SubmissionNotFound>>::Ok((
+                        chunks_moved,
+                        completed_submission,
+                    ))
                 })
             })
             .await?;
 
         if chunks_moved {
             counter!(crate::prometheus::CHUNKS_COMPLETED_COUNTER).increment(1);
         }
-        Ok(())
+        Ok(completed_submission)
     }
 
     /// This function MUST be called inside a transaction.
@@ -657,8 +663,8 @@ pub mod db {
             submission_id,
             submission_id,
         )
-        .execute(conn.get_inner())
-        .await?;
+            .execute(conn.get_inner())
+            .await?;
         Ok(())
     }
 
@@ -687,8 +693,8 @@ pub mod db {
             submission_id,
             submission_id,
         )
-        .execute(conn.get_inner())
-        .await?;
+            .execute(conn.get_inner())
+            .await?;
 
         counter!(crate::prometheus::CHUNKS_SKIPPED_COUNTER).increment(query_res.rows_affected());
         Ok(())
@@ -962,10 +968,10 @@ pub mod test {
             .expect("insertion failed");
 
         let res = complete_chunk(chunk_id, None, &mut conn).await;
-        assert_matches!(res, Ok(()));
+        assert_matches!(res, Ok(false));
 
         let res = complete_chunk(chunk_id, None, &mut conn).await;
-        assert_matches!(res, Ok(()));
+        assert_matches!(res, Ok(false));
     }
 
     #[sqlx::test(migrator = "crate::MIGRATOR")]