fix: handle RouteHandle::submit returning SubmitOutcome

frontend/Cargo.toml

@@ -17,7 +17,7 @@ async-trait = "0.1"
 axum = { version = "0.8", features = ["http2"] }
 tower-http = { version = "0.6", features = ["cors", "compression-full"] }
 polytope-edr = { git = "https://github.com/ecmwf/polytope-edr", branch = "master" }
-bits = { git = "https://github.com/ecmwf/bits-broker.git", rev = "d6ceded4324652739f5e6a79c14bf72550e3eecf", features = ["nats"] }
+bits = { git = "https://github.com/ecmwf/bits-broker.git", rev = "9eb29779c65fb09d867a543f77525fc160b7d4f0", features = ["nats"] }
 bytes = "1"
 clap = { version = "4", features = ["derive"] }
 futures = "0.3"

frontend/src/api/v1.rs

@@ -8,7 +8,7 @@ use axum::{
     http::{HeaderMap, StatusCode, header},
     response::{IntoResponse, Response},
 };
-use bits::{Job, JobResult, PollOutcome};
+use bits::{Job, JobResult, PollOutcome, SubmitOutcome};
 use bytes::BytesMut;
 use futures::TryStreamExt;
 use serde::{Deserialize, Serialize};
@@ -101,7 +101,23 @@ pub async fn submit_request(
     super::set_job_mock_time_metadata(&mut job, mock_time_extensions.mock_time.as_ref());
 
     let submitted_request = job.request.clone();
-    let handle = route_handle.submit(job);
+    let handle = match route_handle.submit(job) {
+        SubmitOutcome::Accepted(handle) => handle,
+        SubmitOutcome::Overloaded => {
+            tracing::warn!(
+                "event.name" = "api.job.rejected",
+                outcome = "overloaded",
+                collection = %collection,
+                reason = "broker_overloaded",
+                "job rejected"
+            );
+            return super::overloaded_response(json!({
+                "status": "failed",
+                "message": "broker at capacity",
+                "retryable": true,
+            }));
+        }
+    };
     super::audit_mock_job_submission(
         mock_audit.as_ref().map(|Extension(audit)| audit),
         &handle.id,

frontend/src/api/v2.rs

@@ -8,7 +8,7 @@ use axum::{
     http::{HeaderMap, StatusCode, header},
     response::{IntoResponse, Response},
 };
-use bits::{Job, JobResult, PollOutcome};
+use bits::{Job, JobResult, PollOutcome, SubmitOutcome};
 use serde_json::{Value, json};
 
 use crate::auth::{AuthUser, MockRolesAudit};
@@ -82,7 +82,21 @@ pub async fn submit_collection(
         "client IP candidate headers present"
     );
     let submitted_request = job.request.clone();
-    let id = route_handle.submit(job).id;
+    let id = match route_handle.submit(job) {
+        SubmitOutcome::Accepted(handle) => handle.id,
+        SubmitOutcome::Overloaded => {
+            tracing::warn!(
+                "event.name" = "api.job.rejected",
+                outcome = "overloaded",
+                collection = %collection,
+                reason = "broker_overloaded",
+                "job rejected"
+            );
+            return super::overloaded_response(
+                json!({"error": "broker at capacity", "retryable": true}),
+            );
+        }
+    };
     if let Some(Extension(user)) = auth_user.as_ref() {
         tracing::info!("event.name" = "api.job.submitted", outcome = "success", job.id = %id, "enduser.id" = %user.username, "enduser.realm" = %user.realm, polytope.request = %polytope_observability::request(&submitted_request), "job submitted");
     } else {

frontend/src/lib.rs

@@ -194,7 +194,14 @@ impl polytope_edr::RequestSubmitter for BitsSubmitter {
                         ))
                     })?
                     .clone();
-                route_handle.submit(job)
+                match route_handle.submit(job) {
+                    bits::SubmitOutcome::Accepted(handle) => handle,
+                    bits::SubmitOutcome::Overloaded => {
+                        return Err(polytope_edr::SubmitError::Upstream(
+                            "broker at capacity".to_string(),
+                        ));
+                    }
+                }
             };
             Ok(polytope_edr::SubmitResponse {
                 id: handle.id.clone(),