[In progress] Test : e2e test

.env.test

@@ -15,7 +15,8 @@ AWS_DEFAULT_REGION="localhost"
 
 MADARA_RPC_URL="http://localhost:3000"
 ETHEREUM_RPC_URL="http://localhost:3001"
-MEMORY_PAGES_CONTRACT_ADDRESS="0x000000000000000000000000000000000001dead"
+ETHEREUM_MAINNET_RPC_URL="https://mainnet.infura.io/v3/bf9e41563a6a45e28eb60382d85ef3c9" # for forking during tests
+MEMORY_PAGES_CONTRACT_ADDRESS="0x47312450B3Ac8b5b8e247a6bB6d523e7605bDb60"
 PRIVATE_KEY="0xdead"
 # Private key of Test wallet provided by Anvil
 ETHEREUM_PRIVATE_KEY="0xac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80"

Cargo.toml

@@ -24,7 +24,16 @@ authors = ["Apoorv Sadana <@apoorvsadana>"]
 [workspace.dependencies]
 num = { version = "0.4.1" }
 async-trait = { version = "0.1.77" }
-alloy = { version = "0.2.1", features = ["full"] }
+alloy = { version = "0.2.1", features = ["full",
+  "node-bindings",
+  "rpc-types-debug",
+  "rpc-types-trace",
+  "json-rpc",
+  "rpc-client",] }
+aws-config = { version = "1.1.7", features = ["behavior-version-latest"] }
+aws-sdk-s3 = { version = "1.38.0", features = ["behavior-version-latest"] }
+aws-sdk-eventbridge = { version = "1.41.0", features = ["behavior-version-latest"] }
+aws-sdk-sqs = "1.36.0"
 axum = { version = "0.7.4" }
 axum-macros = "0.4.1"
 bincode = "1.3.3"

crates/orchestrator/Cargo.toml

@@ -17,9 +17,9 @@ arc-swap = { workspace = true }
 assert_matches = "1.5.0"
 async-std = "1.12.0"
 async-trait = { workspace = true }
-aws-config = { version = "1.1.7", features = ["behavior-version-latest"] }
-aws-sdk-s3 = { version = "1.38.0", features = ["behavior-version-latest"] }
-aws-sdk-sqs = "1.36.0"
+aws-config = { workspace = true }
+aws-sdk-s3 = { workspace = true }
+aws-sdk-sqs = { workspace = true }
 axum = { workspace = true, features = ["macros"] }
 axum-macros = { workspace = true }
 bincode = { workspace = true }

crates/orchestrator/src/config.rs

@@ -1,6 +1,6 @@
 use std::sync::Arc;
 
-use crate::data_storage::aws_s3::config::{AWSS3Config, AWSS3ConfigType};
+use crate::data_storage::aws_s3::config::{AWSS3Config, AWSS3ConfigType, S3LocalStackConfig};
 use crate::data_storage::aws_s3::AWSS3;
 use crate::data_storage::{DataStorage, DataStorageConfig};
 use arc_swap::{ArcSwap, Guard};
@@ -180,6 +180,9 @@ pub async fn build_settlement_client(
 pub async fn build_storage_client() -> Box<dyn DataStorage + Send + Sync> {
     match get_env_var_or_panic("DATA_STORAGE").as_str() {
         "s3" => Box::new(AWSS3::new(AWSS3ConfigType::WithoutEndpoint(AWSS3Config::new_from_env())).await),
+        "s3_localstack" => {
+            Box::new(AWSS3::new(AWSS3ConfigType::WithEndpoint(S3LocalStackConfig::new_from_env())).await)
+        }
         _ => panic!("Unsupported Storage Client"),
     }
 }

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

@@ -275,6 +275,7 @@ impl StateUpdateJob {
         } else if snos.use_kzg_da == Felt252::ONE {
             let blob_data = fetch_blob_data_for_block(block_no).await.map_err(|e| JobError::Other(OtherError(e)))?;
 
+            // TODO : Fetch Program Output from data storage client
             // Fetching nonce before the transaction is run
             // Sending update_state transaction from the settlement client
             settlement_client

crates/orchestrator/src/queue/job_queue.rs

@@ -1,10 +1,11 @@
 use std::future::Future;
+use std::str::FromStr;
 use std::time::Duration;
 
 use color_eyre::eyre::Context;
 use color_eyre::Result as EyreResult;
 use omniqueue::{Delivery, QueueError};
-use serde::{Deserialize, Serialize};
+use serde::{Deserialize, Deserializer, Serialize};
 use tokio::time::sleep;
 use tracing::log;
 use uuid::Uuid;
@@ -57,11 +58,42 @@ pub enum WorkerTriggerType {
     UpdateState,
 }
 
-#[derive(Debug, Serialize, Deserialize, Clone)]
+#[derive(Debug, Serialize, Clone)]
 pub struct WorkerTriggerMessage {
-    pub(crate) worker: WorkerTriggerType,
+    pub worker: WorkerTriggerType,
 }
 
+impl FromStr for WorkerTriggerType {
+    type Err = String;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s {
+            "Proving" => Ok(WorkerTriggerType::Proving),
+            "Snos" => Ok(WorkerTriggerType::Snos),
+            "ProofRegistration" => Ok(WorkerTriggerType::ProofRegistration),
+            "DataSubmission" => Ok(WorkerTriggerType::DataSubmission),
+            "UpdateState" => Ok(WorkerTriggerType::UpdateState),
+            _ => Err(format!("Unknown WorkerTriggerType: {}", s)),
+        }
+    }
+}
+
+impl<'de> Deserialize<'de> for WorkerTriggerMessage {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let s = String::deserialize(deserializer)?;
+        let s = s.trim_start_matches('{').trim_end_matches('}');
+        let parts: Vec<&str> = s.split(':').collect();
+        if parts.len() != 2 || parts[0] != "worker" {
+            return Err(serde::de::Error::custom("Invalid format"));
+        }
+        Ok(WorkerTriggerMessage { worker: WorkerTriggerType::from_str(parts[1]).map_err(serde::de::Error::custom)? })
+    }
+}
+
+#[derive(Debug)]
 enum DeliveryReturnType {
     Message(Delivery),
     NoMessage,

crates/orchestrator/src/tests/config.rs

@@ -152,8 +152,6 @@ impl TestConfigBuilder {
             self.storage.unwrap(),
         );
 
-        drop_database().await.unwrap();
-
         config_force_init(config).await;
 
         server

crates/orchestrator/src/workers/mod.rs

@@ -1,13 +1,26 @@
 use crate::{config::config, jobs::types::JobStatus};
 use async_trait::async_trait;
 use std::error::Error;
+use thiserror::Error;
 
 pub mod data_submission_worker;
 pub mod proof_registration;
 pub mod proving;
 pub mod snos;
 pub mod update_state;
 
+#[derive(Error, Debug)]
+pub enum WorkerError {
+    #[error("Worker execution failed: {0}")]
+    ExecutionError(String),
+
+    #[error("JSON RPC error: {0}")]
+    JsonRpcError(String),
+
+    #[error("Other error: {0}")]
+    Other(Box<dyn Error + Send + Sync>),
+}
+
 #[async_trait]
 pub trait Worker: Send + Sync {
     async fn run_worker_if_enabled(&self) -> Result<(), Box<dyn Error>> {

crates/prover-services/gps-fact-checker/Cargo.toml

@@ -5,13 +5,13 @@ edition.workspace = true
 
 [dependencies]
 alloy = { workspace = true, features = [
-  "sol-types",
-  "json",
-  "contract",
-  "providers",
-  "rpc-client",
-  "transport-http",
-  "reqwest",
+    "sol-types",
+    "json",
+    "contract",
+    "providers",
+    "rpc-client",
+    "transport-http",
+    "reqwest",
 ] }
 async-trait.workspace = true
 cairo-vm.workspace = true
@@ -20,3 +20,6 @@ starknet.workspace = true
 thiserror.workspace = true
 url.workspace = true
 utils.workspace = true
+tokio.workspace = true
+rstest.workspace = true
+dotenvy.workspace = true