refactor: job isolation done #204

Mohiiit · 2025-01-22T06:14:51Z

files and information needed by each job should be provided by the worker of the respective job, no inter-job dependency

apoorvsadana · 2025-01-22T09:59:59Z

crates/orchestrator/src/constants.rs

@@ -2,3 +2,10 @@ pub const BLOB_DATA_FILE_NAME: &str = "blob_data.txt";
 pub const SNOS_OUTPUT_FILE_NAME: &str = "snos_output.json";
 pub const PROGRAM_OUTPUT_FILE_NAME: &str = "program_output.txt";
 pub const CAIRO_PIE_FILE_NAME: &str = "cairo_pie.zip";
+
+pub const JOB_METADATA_CAIRO_PIE_PATH: &str = "cairo_pie_path";


we've a lot of metadata constants here. should we reuse this?

apoorvsadana · 2025-01-22T10:11:00Z

crates/orchestrator/src/jobs/da_job/mod.rs

@@ -133,7 +133,11 @@ impl Job for DaJob {
        // data transformation on the data
        tracing::trace!(job_id = ?job.id, "Applied FFT transformation");

-        store_blob_data(transformed_data.clone(), block_no, config.clone()).await?;
+        let blob_data_path = format!("{}/{}", block_no, BLOB_DATA_FILE_NAME);


let's change to worker specifying full path as discussed

apoorvsadana · 2025-01-22T10:23:10Z

crates/orchestrator/src/jobs/snos_job/mod.rs

@@ -196,17 +218,17 @@ impl SnosJob {
            SnosError::SnosOutputUnstorable { internal_id: internal_id.to_string(), message: e.to_string() }
        })?;

+        let program_output_key = format!("{block_number}/{PROGRAM_OUTPUT_FILE_NAME}");


will come from metadata now?

apoorvsadana · 2025-01-22T10:36:23Z

crates/orchestrator/src/jobs/state_update_job/mod.rs


-        let program_output = storage_client.get_data(&key).await.map_err(|e| JobError::Other(OtherError(e)))?;
+        // Get the array of program output paths from metadata
+        let program_paths: Vec<String> = serde_json::from_str(


pass path directly

as discussed, should we use config structs for metadata fields

apoorvsadana · 2025-01-24T07:37:57Z

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

-                ORCHESTRATOR_METRICS.db_calls_response_time.record(duration.as_secs_f64(), &attributes);
-                Ok(Some(job))
+                // Add debug logging to see the raw document
+                tracing::info!(raw_document = ?doc, "Raw document from MongoDB");


Suggested change

tracing::info!(raw_document = ?doc, "Raw document from MongoDB");

tracing::debug!(raw_document = ?doc, "Raw document from MongoDB");

apoorvsadana · 2025-01-24T07:41:42Z

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

+                tracing::info!(raw_document = ?doc, "Raw document from MongoDB");
+
+                // Try to deserialize and log any errors
+                match mongodb::bson::from_document::<JobItem>(doc.clone()) {


we're cloning here so that we can log in error later? if yes, should we explain that's why we're cloning it?

apoorvsadana · 2025-01-24T07:50:37Z

crates/orchestrator/src/jobs/metadata.rs

this is also types to be precise. maybe we should add this to types.rs OR create a types folder with jobs.rs and metadata.rs?

apoorvsadana · 2025-01-24T07:57:08Z

crates/orchestrator/src/jobs/metadata.rs

+pub struct ProvingMetadata {
+    // Required fields
+    pub block_number: u64,
+    pub cairo_pie_path: Option<String>,


if it's required it should not be optional right?
why does proving job care about block_number?
cross_verify can be renamed to ensure_on_chain_registration or something more understandable
let's delete verification_key_path

make an input field which is

enum Input {
Pie()...
}

can remove snos_fact and club it with ensure_on_chain_registration using Optional

also add comments to explain what are inputs in metadata and what is added by jobs on the fly

apoorvsadana · 2025-01-24T08:06:14Z

crates/orchestrator/src/jobs/metadata.rs

+
+    // State tracking
+    pub last_failed_block_no: Option<u64>,
+    pub tx_hashes: Vec<String>, // key: attempt_no, value: comma-separated tx hashes


Suggested change

pub tx_hashes: Vec<String>, // key: attempt_no, value: comma-separated tx hashes

pub tx_hashes: Vec<String>

apoorvsadana · 2025-01-24T08:37:06Z

crates/orchestrator/src/jobs/state_update_job/mod.rs

-            block_numbers = block_numbers.into_iter().filter(|&block| block >= last_failed_block).collect::<Vec<u64>>();
-        }
+        // Filter block numbers if there was a previous failure
+        let block_numbers = if let JobSpecificMetadata::StateUpdate(state_metadata) = &job.metadata.specific {


we can move this part outside

apoorvsadana · 2025-01-24T08:39:05Z

crates/orchestrator/src/jobs/state_update_job/mod.rs

-                .update_state_for_block(config.clone(), *block_no, snos, nonce)
-                .await
-                .map_err(|e| {
+            let snos = self.fetch_snos_for_block(i, config.clone(), job).await?;


get the snos path here itsefl

apoorvsadana · 2025-01-24T08:39:08Z

crates/orchestrator/src/jobs/state_update_job/mod.rs

-                .await
-                .map_err(|e| {
+            let snos = self.fetch_snos_for_block(i, config.clone(), job).await?;
+            let txn_hash = match self.update_state_for_block(config.clone(), *block_no, i, snos, nonce, job).await {


apoorvsadana · 2025-01-24T08:39:20Z

crates/orchestrator/src/jobs/state_update_job/mod.rs

-                .unwrap();
+
+                    // Update metadata in a single place
+                    if let JobSpecificMetadata::StateUpdate(ref mut state_metadata) = job.metadata.specific {


here as well

apoorvsadana · 2025-01-24T08:41:13Z

crates/orchestrator/src/jobs/state_update_job/utils.rs

+// Fetching the blob data (stored in remote storage during DA job) for a particular block
+// pub async fn fetch_program_data_for_block(block_number: u64, config: Arc<Config>, job: &JobItem)
+// -> color_eyre::Result<Vec<[u8; 32]>> {     let storage_client = config.storage();
+//     let key = block_number.to_string() + "/" + PROGRAM_OUTPUT_FILE_NAME;
+//     let blob_data = storage_client.get_data(&key).await?;
+//     let transformed_blob_vec_u8 = bytes_to_vec_u8(blob_data.as_ref())?;
+//     Ok(transformed_blob_vec_u8)
+// }


apoorvsadana · 2025-01-24T08:42:58Z

crates/orchestrator/src/workers/data_submission_worker.rs

+            let proving_metadata = match &proving_job.metadata.specific {
+                JobSpecificMetadata::Proving(metadata) => metadata,
+                _ => {
+                    tracing::error!(
+                        job_id = %proving_job.internal_id,
+                        "Invalid metadata type for proving job"
+                    );
+                    continue;
+                }
+            };


can just use internal id here

apoorvsadana · 2025-01-24T08:54:37Z

crates/orchestrator/src/workers/proving.rs

-                Ok(_) => tracing::info!(block_id = %job.internal_id, "Successfully created new proving job"),
+        for snos_job in successful_snos_jobs {
+            // Extract SNOS metadata
+            let snos_metadata = match &snos_job.metadata.specific {


can try some TryInto implementation to make it cleaner

apoorvsadana · 2025-01-24T08:57:25Z

crates/orchestrator/src/workers/update_state.rs

@@ -96,26 +94,73 @@ impl Worker for UpdateStateWorker {
                }
            }
            None => {
-                if blocks_to_process[0] != 0 {
+                if blocks_to_process[0] != 0 && blocks_to_process[0] != 1 {


need to remove this

Mohiiit added 2 commits January 22, 2025 11:43

refactor: job isolation done

e703bff

fix: tests fixed

ff84440

apoorvsadana reviewed Jan 22, 2025

View reviewed changes

Mohiiit added 2 commits January 23, 2025 16:03

feat: metadata struct introduced

decc1ad

refactor: metadata introduced to each job, happy flow working

d11c246

apoorvsadana reviewed Jan 24, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

refactor: job isolation done #204

refactor: job isolation done #204

Mohiiit commented Jan 22, 2025

apoorvsadana Jan 22, 2025

apoorvsadana Jan 22, 2025

apoorvsadana Jan 22, 2025

apoorvsadana Jan 22, 2025

apoorvsadana Jan 24, 2025

apoorvsadana Jan 24, 2025

apoorvsadana Jan 24, 2025

apoorvsadana Jan 24, 2025

apoorvsadana Jan 24, 2025

apoorvsadana Jan 24, 2025

apoorvsadana Jan 24, 2025

apoorvsadana Jan 24, 2025

apoorvsadana Jan 24, 2025

apoorvsadana Jan 24, 2025

apoorvsadana Jan 24, 2025

apoorvsadana Jan 24, 2025

apoorvsadana Jan 24, 2025

apoorvsadana Jan 24, 2025

apoorvsadana Jan 24, 2025

apoorvsadana Jan 24, 2025

	tracing::info!(raw_document = ?doc, "Raw document from MongoDB");
	tracing::debug!(raw_document = ?doc, "Raw document from MongoDB");

	pub tx_hashes: Vec<String>, // key: attempt_no, value: comma-separated tx hashes
	pub tx_hashes: Vec<String>

refactor: job isolation done #204

Are you sure you want to change the base?

refactor: job isolation done #204

Conversation

Mohiiit commented Jan 22, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment