Feat: Retry endpoint added

CHANGELOG.md

@@ -6,6 +6,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## Added
 
+- Added retry job endpoint for failed jobs
 - readme: setup instructions added
 - Added : Grafana dashboard
 - tests: http_client tests added
@@ -50,6 +51,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## Changed
 
+- verify_job now handles VerificationTimeout status
 - refactor: Readme and .env.example
 - refactor: http_mock version updated
 - refactor: prover-services renamed to prover-clients

crates/orchestrator/src/jobs/mod.rs

@@ -209,8 +209,8 @@ pub async fn process_job(id: Uuid, config: Arc<Config>) -> Result<(), JobError>
     match job.status {
         // we only want to process jobs that are in the created or verification failed state.
         // verification failed state means that the previous processing failed and we want to retry
-        JobStatus::Created | JobStatus::VerificationFailed => {
-            tracing::info!(job_id = ?id, status = ?job.status, "Job status is Created or VerificationFailed, proceeding with processing");
+        JobStatus::Created | JobStatus::VerificationFailed | JobStatus::RetryAttempt => {
+            tracing::info!(job_id = ?id, status = ?job.status, "Job status is Created or VerificationFailed or RetryAttempt, proceeding with processing");
         }
         _ => {
             tracing::warn!(job_id = ?id, status = ?job.status, "Job status is Invalid. Cannot process.");
@@ -333,8 +333,8 @@ pub async fn verify_job(id: Uuid, config: Arc<Config>) -> Result<(), JobError> {
     tracing::Span::current().record("internal_id", job.internal_id.clone());
 
     match job.status {
-        JobStatus::PendingVerification => {
-            tracing::debug!(job_id = ?id, "Job status is PendingVerification, proceeding with verification");
+        JobStatus::PendingVerification | JobStatus::VerificationTimeout => {
+            tracing::info!(job_id = ?id, status = ?job.status, "Job status is PendingVerification or VerificationTimeout, proceeding with verification");
         }
         _ => {
             tracing::error!(job_id = ?id, status = ?job.status, "Invalid job status for verification");

crates/orchestrator/src/jobs/types.rs

@@ -103,6 +103,8 @@ pub enum JobStatus {
     VerificationFailed,
     /// The job failed completing
     Failed,
+    /// The job is being retried
+    RetryAttempt,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]

crates/orchestrator/src/routes/job_routes.rs

@@ -10,6 +10,7 @@ use uuid::Uuid;
 
 use super::ApiResponse;
 use crate::config::Config;
+use crate::jobs::types::{JobItemUpdates, JobStatus};
 use crate::jobs::{process_job, verify_job, JobError};
 use crate::metrics::ORCHESTRATOR_METRICS;
 
@@ -73,6 +74,65 @@ async fn handle_verify_job_request(
         }
     }
 }
+
+async fn handle_retry_job_request(
+    Path(JobId { id }): Path<JobId>,
+    State(config): State<Arc<Config>>,
+) -> impl IntoResponse {
+    let job_id = match Uuid::parse_str(&id) {
+        Ok(id) => id,
+        Err(_) => {
+            return ApiResponse::<JobApiResponse>::error((JobError::InvalidId { id }).to_string()).into_response();
+        }
+    };
+
+    // Get the job and verify it's in a failed state
+    let job = match config.database().get_job_by_id(job_id).await {
+        Ok(Some(job)) => job,
+        Ok(None) => {
+            return ApiResponse::<JobApiResponse>::error(JobError::JobNotFound { id: job_id }.to_string())
+                .into_response();
+        }
+        Err(e) => {
+            return ApiResponse::<JobApiResponse>::error(e.to_string()).into_response();
+        }
+    };
+
+    // Check if job is in a failed state
+    if job.status != JobStatus::Failed {
+        return ApiResponse::<JobApiResponse>::error(format!(
+            "Job {} cannot be retried: current status is {:?}",
+            id, job.status
+        ))
+        .into_response();
+    }
+
+    // Update the job status to RetryAttempt
+    match config.database().update_job(&job, JobItemUpdates::new().update_status(JobStatus::RetryAttempt).build()).await
+    {
+        Ok(_) => {
+            // Process the job after successful status update
+            match process_job(job_id, config).await {
+                Ok(_) => {
+                    let response =
+                        JobApiResponse { job_id: job_id.to_string(), status: "retry_processing".to_string() };
+                    ApiResponse::success(response).into_response()
+                }
+                Err(e) => {
+                    ORCHESTRATOR_METRICS
+                        .failed_job_operations
+                        .add(1.0, &[KeyValue::new("operation_type", "retry_job")]);
+                    ApiResponse::<JobApiResponse>::error(e.to_string()).into_response()
+                }
+            }
+        }
+        Err(e) => {
+            ORCHESTRATOR_METRICS.failed_job_operations.add(1.0, &[KeyValue::new("operation_type", "retry_job")]);
+            ApiResponse::<JobApiResponse>::error(e.to_string()).into_response()
+        }
+    }
+}
+
 pub fn job_router(config: Arc<Config>) -> Router {
     Router::new().nest("/jobs", trigger_router(config.clone()))
 }
@@ -81,5 +141,6 @@ fn trigger_router(config: Arc<Config>) -> Router {
     Router::new()
         .route("/:id/process", get(handle_process_job_request))
         .route("/:id/verify", get(handle_verify_job_request))
+        .route("/:id/retry", get(handle_retry_job_request))
         .with_state(config)
 }

crates/orchestrator/src/tests/server/job_routes.rs

@@ -119,6 +119,44 @@ async fn test_trigger_verify_job(#[future] setup_trigger: (SocketAddr, Arc<Confi
     }
 }
 
+#[tokio::test]
+#[rstest]
+async fn test_trigger_retry_job(#[future] setup_trigger: (SocketAddr, Arc<Config>)) {
+    let (addr, config) = setup_trigger.await;
+
+    let job_type = JobType::DataSubmission;
+
+    // Create a failed job
+    let job_item = build_job_item(job_type.clone(), JobStatus::Failed, 1);
+    let mut job_handler = MockJob::new();
+
+    // We expect process_job to be called since retry triggers processing
+    job_handler.expect_process_job().times(1).returning(move |_, _| Ok("0xbeef".to_string()));
+    job_handler.expect_verification_polling_delay_seconds().return_const(1u64);
+
+    config.database().create_job(job_item.clone()).await.unwrap();
+    let job_id = job_item.clone().id;
+
+    let job_handler: Arc<Box<dyn Job>> = Arc::new(Box::new(job_handler));
+    let ctx = mock_factory::get_job_handler_context();
+    ctx.expect().times(1).with(eq(job_type)).returning(move |_| Arc::clone(&job_handler));
+
+    let client = hyper::Client::new();
+    let response = client
+        .request(Request::builder().uri(format!("http://{}/jobs/{}/retry", addr, job_id)).body(Body::empty()).unwrap())
+        .await
+        .unwrap();
+
+    // assertions
+    if let Some(job_fetched) = config.database().get_job_by_id(job_id).await.unwrap() {
+        assert_eq!(response.status(), 200);
+        assert_eq!(job_fetched.id, job_item.id);
+        assert_eq!(job_fetched.status, JobStatus::PendingVerification);
+    } else {
+        panic!("Could not get job from database")
+    }
+}
+
 #[rstest]
 #[tokio::test]
 async fn test_init_consumer() {