feat: Data Submission Worker Integration.

CHANGELOG.md

@@ -6,6 +6,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## Added
 
+- creation of data submission jobs.
 - Function to calculate the kzg proof of x_0.
 - Tests for updating the state.
 - Function to update the state and publish blob on ethereum in state update job.

crates/orchestrator/src/database/mod.rs

@@ -29,7 +29,7 @@ pub trait Database: Send + Sync {
     async fn update_job(&self, job: &JobItem) -> Result<()>;
     async fn update_job_status(&self, job: &JobItem, new_status: JobStatus) -> Result<()>;
     async fn update_metadata(&self, job: &JobItem, metadata: HashMap<String, String>) -> Result<()>;
-    async fn get_latest_job_by_type_and_internal_id(&self, job_type: JobType) -> Result<Option<JobItem>>;
+    async fn get_latest_job_by_type(&self, job_type: JobType) -> Result<Option<JobItem>>;
     async fn get_jobs_without_successor(
         &self,
         job_a_type: JobType,
@@ -43,6 +43,9 @@ pub trait Database: Send + Sync {
         job_status: JobStatus,
         internal_id: String,
     ) -> Result<Vec<JobItem>>;
+
+    // TODO: can be extendible to support multiple status.
+    async fn get_jobs_by_status(&self, status: JobStatus) -> Result<Vec<JobItem>>;
 }
 
 pub trait DatabaseConfig {

crates/orchestrator/src/database/mongodb/mod.rs

@@ -112,7 +112,7 @@ impl Database for MongoDb {
         Ok(())
     }
 
-    async fn get_latest_job_by_type_and_internal_id(&self, job_type: JobType) -> Result<Option<JobItem>> {
+    async fn get_latest_job_by_type(&self, job_type: JobType) -> Result<Option<JobItem>> {
         let filter = doc! {
             "job_type": mongodb::bson::to_bson(&job_type)?,
         };
@@ -290,4 +290,29 @@ impl Database for MongoDb {
 
         Ok(results)
     }
+
+    async fn get_jobs_by_status(&self, job_status: JobStatus) -> Result<Vec<JobItem>> {
+        let filter = doc! {
+            "job_status": bson::to_bson(&job_status)?
+        };
+
+        let mut jobs = self
+            .get_job_collection()
+            .find(filter, None)
+            .await
+            .expect("Failed to fetch jobs by given job type and status");
+
+        let mut results = Vec::new();
+
+        while let Some(result) = jobs.next().await {
+            match result {
+                Ok(job_item) => {
+                    results.push(job_item);
+                }
+                Err(e) => return Err(e.into()),
+            }
+        }
+
+        Ok(results)
+    }
 }

crates/orchestrator/src/jobs/mod.rs

@@ -80,7 +80,7 @@ pub async fn process_job(id: Uuid) -> Result<()> {
     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(_) => {
+        JobStatus::Created | JobStatus::VerificationFailed => {
             log::info!("Processing job with id {:?}", id);
         }
         _ => {
@@ -134,8 +134,9 @@ pub async fn verify_job(id: Uuid) -> Result<()> {
         JobVerificationStatus::Verified => {
             config.database().update_job_status(&job, JobStatus::Completed).await?;
         }
-        JobVerificationStatus::Rejected(e) => {
-            config.database().update_job_status(&job, JobStatus::VerificationFailed(e)).await?;
+        JobVerificationStatus::Rejected(_) => {
+            // TODO: change '_' to 'e' and add error 'e' to metadata of job status.
+            config.database().update_job_status(&job, JobStatus::VerificationFailed).await?;
 
             // retry job processing if we haven't exceeded the max limit
             let process_attempts = get_u64_from_metadata(&job.metadata, JOB_PROCESS_ATTEMPT_METADATA_KEY)?;

crates/orchestrator/src/jobs/types.rs

@@ -98,7 +98,7 @@ pub enum JobStatus {
     /// The job was processed but the was unable to be verified under the given time
     VerificationTimeout,
     /// The job failed processing
-    VerificationFailed(String),
+    VerificationFailed,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]

crates/orchestrator/src/main.rs

@@ -2,6 +2,7 @@ use dotenvy::dotenv;
 use orchestrator::config::config;
 use orchestrator::queue::init_consumers;
 use orchestrator::routes::app_router;
+use orchestrator::workers::data_submission::DataSubmissionWorker;
 use orchestrator::workers::proof_registration::ProofRegistrationWorker;
 use orchestrator::workers::proving::ProvingWorker;
 use orchestrator::workers::snos::SnosWorker;
@@ -33,14 +34,15 @@ async fn main() {
     tokio::spawn(start_cron(Box::new(ProvingWorker), 60));
     tokio::spawn(start_cron(Box::new(ProofRegistrationWorker), 60));
     tokio::spawn(start_cron(Box::new(UpdateStateWorker), 60));
+    tokio::spawn(start_cron(Box::new(DataSubmissionWorker), 60));
 
     tracing::info!("Listening on http://{}", address);
     axum::serve(listener, app).await.expect("Failed to start axum server");
 }
 
 async fn start_cron(worker: Box<dyn Worker>, interval: u64) {
     loop {
-        worker.run_worker().await.expect("Error in running the worker.");
+        worker.run_worker_if_enabled().await.expect("Error in running the worker.");
         tokio::time::sleep(tokio::time::Duration::from_secs(interval)).await;
     }
 }

crates/orchestrator/src/tests/workers/snos/mod.rs

@@ -30,13 +30,13 @@ async fn test_snos_worker(#[case] db_val: bool) -> Result<(), Box<dyn Error>> {
 
     // Mocking db function expectations
     if !db_val {
-        db.expect_get_last_successful_job_by_type().times(1).with(eq(JobType::SnosRun)).returning(|_| Ok(None));
+        db.expect_get_latest_job_by_type().times(1).with(eq(JobType::SnosRun)).returning(|_| Ok(None));
         start_job_index = 1;
         block = 5;
     } else {
         let uuid_temp = Uuid::new_v4();
 
-        db.expect_get_last_successful_job_by_type()
+        db.expect_get_latest_job_by_type()
             .with(eq(JobType::SnosRun))
             .returning(move |_| Ok(Some(get_job_item_mock_by_id("1".to_string(), uuid_temp))));
         block = 6;

crates/orchestrator/src/workers/data_submission.rs

@@ -0,0 +1,48 @@
+use crate::config::config;
+use crate::jobs::create_job;
+use crate::jobs::types::JobType;
+use crate::workers::Worker;
+use async_trait::async_trait;
+use std::collections::HashMap;
+use std::error::Error;
+
+pub struct DataSubmissionWorker;
+
+#[async_trait]
+impl Worker for DataSubmissionWorker {
+    // 0. All ids are assumed to be block numbers.
+    // 1. Fetch the latest completed Proving job.
+    // 2. Fetch the latest DA job creation.
+    // 3. Create jobs from after the lastest DA job already created till latest completed proving job.
+    async fn run_worker(&self) -> Result<(), Box<dyn Error>> {
+        let config = config().await;
+
+        // provides latest completed proof creation job id
+        let latest_proven_job_id = config
+            .database()
+            .get_last_successful_job_by_type(JobType::ProofCreation)
+            .await
+            .unwrap()
+            .map(|item| item.internal_id)
+            .unwrap_or("0".to_string());
+
+        // provides latest triggered data submission job id
+        let latest_data_submission_job_id = config
+            .database()
+            .get_latest_job_by_type(JobType::DataSubmission)
+            .await
+            .unwrap()
+            .map(|item| item.internal_id)
+            .unwrap_or("0".to_string());
+
+        let latest_data_submission_id: u64 = latest_data_submission_job_id.parse()?;
+        let latest_proven_id: u64 = latest_proven_job_id.parse()?;
+
+        // creating data submission jobs for latest blocks without pre-running data submission jobs jobs don't yet exist.
+        for x in latest_data_submission_id + 1..latest_proven_id + 1 {
+            create_job(JobType::DataSubmission, x.to_string(), HashMap::new()).await?;
+        }
+
+        Ok(())
+    }
+}

crates/orchestrator/src/workers/mod.rs

@@ -1,13 +1,39 @@
-use std::error::Error;
-
+use crate::{config::config, jobs::types::JobStatus};
 use async_trait::async_trait;
+use std::error::Error;
 
+pub mod data_submission;
 pub mod proof_registration;
 pub mod proving;
 pub mod snos;
 pub mod update_state;
 
 #[async_trait]
 pub trait Worker: Send + Sync {
+    async fn run_worker_if_enabled(&self) -> Result<(), Box<dyn Error>> {
+        if !self.is_worker_enabled().await? {
+            return Ok(());
+        }
+        self.run_worker().await
+    }
+
     async fn run_worker(&self) -> Result<(), Box<dyn Error>>;
+
+    // TODO: Assumption : False Negative
+    // we are assuming that the worker will spawn only 1 job for a block and no two jobs will ever exist
+    // for a single block, the code might fail to work as expected if this happens.
+
+    // Checks if any of the jobs have failed
+    // Haults any new job creation till all the count of failed jobs is not Zero.
+    async fn is_worker_enabled(&self) -> Result<bool, Box<dyn Error>> {
+        let config = config().await;
+
+        let failed_da_jobs = config.database().get_jobs_by_status(JobStatus::VerificationFailed).await?;
+
+        if !failed_da_jobs.is_empty() {
+            return Ok(false);
+        }
+
+        Ok(true)
+    }
 }

crates/orchestrator/src/workers/snos.rs

@@ -22,7 +22,7 @@ impl Worker for SnosWorker {
         let latest_block_number = provider.block_number().await?;
         let latest_block_processed_data = config
             .database()
-            .get_last_successful_job_by_type(JobType::SnosRun)
+            .get_latest_job_by_type(JobType::SnosRun)
             .await
             .unwrap()
             .map(|item| item.internal_id)