Toploc Integration for groups & Core storage rewrite (#353)

* implement group validation calls for toploc

* large scale storage rewrite, introduce mock storage provider for better testing

* update deployment, load toploc configs from env

* setup toploc group e2e tests

Author: JannikSt

Cargo.lock

@@ -87,7 +87,7 @@ watch-check:
 
 watch-validator:
 	set -a; source ${ENV_FILE}; set +a; \
-	cargo watch -w crates/validator/src -x "run --bin validator -- --validator-key $${PRIVATE_KEY_VALIDATOR} --rpc-url $${RPC_URL} --pool-id $${WORKER_COMPUTE_POOL_ID} --toploc-configs '$${TOPLOC_CONFIGS:-[]}' --bucket-name $${BUCKET_NAME} -l $${LOG_LEVEL:-info} --toploc-grace-interval $${TOPLOC_GRACE_INTERVAL:-30}"
+	cargo watch -w crates/validator/src -x "run --bin validator -- --validator-key $${PRIVATE_KEY_VALIDATOR} --rpc-url $${RPC_URL} --pool-id $${WORKER_COMPUTE_POOL_ID} --bucket-name $${BUCKET_NAME} -l $${LOG_LEVEL:-info} --toploc-grace-interval $${TOPLOC_GRACE_INTERVAL:-30}"
 
 watch-orchestrator:
 	set -a; source ${ENV_FILE}; set +a; \

Makefile

@@ -4,10 +4,7 @@ use actix_web::{
     HttpRequest, HttpResponse, Scope,
 };
 use redis::{Commands, RedisResult};
-use shared::{
-    models::storage::RequestUploadRequest,
-    utils::google_cloud::{generate_mapping_file, generate_upload_signed_url},
-};
+use shared::models::storage::RequestUploadRequest;
 use std::time::Duration;
 
 const MAX_FILE_SIZE: u64 = 100 * 1024 * 1024;
@@ -192,31 +189,16 @@ async fn request_upload(
         sha256
     );
 
-    // Get credentials from app state
-    let credentials = match &app_state.s3_credentials {
-        Some(creds) => creds,
-        None => {
-            return HttpResponse::InternalServerError().json(serde_json::json!({
-                "success": false,
-                "error": "Storage credentials not configured"
-            }))
-        }
-    };
-
     log::info!(
-        "Generating mapping file for sha256: {} to file: {} in bucket: {}",
+        "Generating mapping file for sha256: {} to file: {}",
         sha256,
         file_name,
-        app_state.bucket_name.clone().unwrap_or_default()
     );
 
-    if let Err(e) = generate_mapping_file(
-        app_state.bucket_name.clone().unwrap().as_str(),
-        credentials,
-        sha256,
-        &file_name,
-    )
-    .await
+    if let Err(e) = app_state
+        .storage_provider
+        .generate_mapping_file(sha256, &file_name)
+        .await
     {
         log::error!("Failed to generate mapping file: {}", e);
         return HttpResponse::InternalServerError().json(serde_json::json!({
@@ -231,15 +213,15 @@ async fn request_upload(
     );
 
     // Generate signed upload URL
-    match generate_upload_signed_url(
-        app_state.bucket_name.clone().unwrap().as_str(),
-        &file_name,
-        credentials,
-        Some(file_type.to_string()),
-        Duration::from_secs(3600), // 1 hour expiry
-        Some(*file_size),
-    )
-    .await
+    match app_state
+        .storage_provider
+        .generate_upload_signed_url(
+            &file_name,
+            Some(file_type.to_string()),
+            Duration::from_secs(3600), // 1 hour expiry
+            Some(*file_size),
+        )
+        .await
     {
         Ok(signed_url) => {
             // Increment rate limit counter after successful URL generation
@@ -347,11 +329,6 @@ mod tests {
 
         assert!(task_store.get_task(&task.id.to_string()).is_some());
 
-        if app_state.s3_credentials.is_none() {
-            println!("S3 credentials not configured");
-            return;
-        }
-
         let app =
             test::init_service(App::new().app_data(app_state.clone()).service(
                 web::scope("/storage").route("/request-upload", post().to(request_upload)),
@@ -384,11 +361,6 @@ mod tests {
     async fn test_request_upload_invalid_task() {
         let app_state = create_test_app_state().await;
 
-        if app_state.s3_credentials.is_none() {
-            println!("S3 credentials not configured");
-            return;
-        }
-
         let app =
             test::init_service(App::new().app_data(app_state.clone()).service(
                 web::scope("/storage").route("/request-upload", post().to(request_upload)),
@@ -474,11 +446,6 @@ mod tests {
 
         assert!(task_store.get_task(&task.id.to_string()).is_some());
 
-        if app_state.s3_credentials.is_none() {
-            println!("S3 credentials not configured");
-            return;
-        }
-
         let app =
             test::init_service(App::new().app_data(app_state.clone()).service(
                 web::scope("/storage").route("/request-upload", post().to(request_upload)),
@@ -607,11 +574,6 @@ mod tests {
 
         assert!(task_store.get_task(&task.id.to_string()).is_some());
 
-        if app_state.s3_credentials.is_none() {
-            println!("S3 credentials not configured");
-            return;
-        }
-
         let app =
             test::init_service(App::new().app_data(app_state.clone()).service(
                 web::scope("/storage").route("/request-upload", post().to(request_upload)),

crates/orchestrator/src/api/routes/storage.rs

@@ -16,15 +16,15 @@ use log::info;
 use serde_json::json;
 use shared::security::api_key_middleware::ApiKeyMiddleware;
 use shared::security::auth_signature_middleware::{ValidateSignature, ValidatorState};
+use shared::utils::StorageProvider;
 use shared::web3::contracts::core::builder::Contracts;
 use shared::web3::wallet::Wallet;
 use std::sync::Arc;
 
 pub struct AppState {
     pub store_context: Arc<StoreContext>,
     pub wallet: Arc<Wallet>,
-    pub s3_credentials: Option<String>,
-    pub bucket_name: Option<String>,
+    pub storage_provider: Arc<dyn StorageProvider>,
     pub heartbeats: Arc<LoopHeartbeats>,
     pub redis_store: Arc<RedisStore>,
     pub hourly_upload_limit: i64,
@@ -41,8 +41,7 @@ pub async fn start_server(
     store_context: Arc<StoreContext>,
     wallet: Arc<Wallet>,
     admin_api_key: String,
-    s3_credentials: Option<String>,
-    bucket_name: Option<String>,
+    storage_provider: Arc<dyn StorageProvider>,
     heartbeats: Arc<LoopHeartbeats>,
     redis_store: Arc<RedisStore>,
     hourly_upload_limit: i64,
@@ -56,8 +55,7 @@ pub async fn start_server(
     let app_state = Data::new(AppState {
         store_context,
         wallet,
-        s3_credentials,
-        bucket_name,
+        storage_provider,
         heartbeats,
         redis_store,
         hourly_upload_limit,

crates/orchestrator/src/api/server.rs

@@ -17,6 +17,8 @@ use url::Url;
 
 #[cfg(test)]
 pub async fn create_test_app_state() -> Data<AppState> {
+    use shared::utils::MockStorageProvider;
+
     use crate::{scheduler::Scheduler, utils::loop_heartbeats::LoopHeartbeats, ServerMode};
 
     let store = Arc::new(RedisStore::new_test());
@@ -35,8 +37,9 @@ pub async fn create_test_app_state() -> Data<AppState> {
     let store_context = Arc::new(StoreContext::new(store.clone()));
     let mode = ServerMode::Full;
     let scheduler = Scheduler::new(store_context.clone(), vec![]);
-    let s3_credentials = std::env::var("S3_CREDENTIALS").ok();
-    let bucket_name = std::env::var("BUCKET_NAME").ok();
+
+    let mock_storage = MockStorageProvider::new();
+    let storage_provider = Arc::new(mock_storage);
 
     Data::new(AppState {
         store_context: store_context.clone(),
@@ -49,8 +52,7 @@ pub async fn create_test_app_state() -> Data<AppState> {
             )
             .unwrap(),
         ),
-        s3_credentials,
-        bucket_name,
+        storage_provider,
         heartbeats: Arc::new(LoopHeartbeats::new(&mode)),
         hourly_upload_limit: 12,
         redis_store: store.clone(),
@@ -61,6 +63,8 @@ pub async fn create_test_app_state() -> Data<AppState> {
 
 #[cfg(test)]
 pub async fn create_test_app_state_with_nodegroups() -> Data<AppState> {
+    use shared::utils::MockStorageProvider;
+
     use crate::{
         plugins::node_groups::{NodeGroupConfiguration, NodeGroupsPlugin},
         scheduler::Scheduler,
@@ -84,8 +88,6 @@ pub async fn create_test_app_state_with_nodegroups() -> Data<AppState> {
     let store_context = Arc::new(StoreContext::new(store.clone()));
     let mode = ServerMode::Full;
     let scheduler = Scheduler::new(store_context.clone(), vec![]);
-    let s3_credentials = std::env::var("S3_CREDENTIALS").ok();
-    let bucket_name = std::env::var("BUCKET_NAME").ok();
 
     let config = NodeGroupConfiguration {
         name: "test-config".to_string(),
@@ -100,6 +102,9 @@ pub async fn create_test_app_state_with_nodegroups() -> Data<AppState> {
         store_context.clone(),
     )));
 
+    let mock_storage = MockStorageProvider::new();
+    let storage_provider = Arc::new(mock_storage);
+
     Data::new(AppState {
         store_context: store_context.clone(),
         contracts: None,
@@ -111,8 +116,7 @@ pub async fn create_test_app_state_with_nodegroups() -> Data<AppState> {
             )
             .unwrap(),
         ),
-        s3_credentials,
-        bucket_name,
+        storage_provider,
         heartbeats: Arc::new(LoopHeartbeats::new(&mode)),
         hourly_upload_limit: 12,
         redis_store: store.clone(),

crates/orchestrator/src/api/tests/helper.rs

@@ -28,6 +28,7 @@ use plugins::node_groups::NodeGroupsPlugin;
 use plugins::webhook::WebhookPlugin;
 use plugins::SchedulerPlugin;
 use plugins::StatusUpdatePlugin;
+use shared::utils::google_cloud::GcsStorageProvider;
 use shared::web3::contracts::core::builder::ContractBuilder;
 use shared::web3::contracts::structs::compute_pool::PoolStatus;
 use shared::web3::wallet::Wallet;
@@ -286,6 +287,12 @@ async fn main() -> Result<()> {
     let server_store_context = store_context.clone();
 
     let s3_credentials = std::env::var("S3_CREDENTIALS").ok();
+
+    let gcs_storage = GcsStorageProvider::new(&args.bucket_name.unwrap(), &s3_credentials.unwrap())
+        .await
+        .unwrap();
+    let storage_provider = Arc::new(gcs_storage);
+
     // Always start server regardless of mode
     tokio::select! {
         res = start_server(
@@ -294,8 +301,7 @@ async fn main() -> Result<()> {
             server_store_context.clone(),
             server_wallet,
             args.admin_api_key,
-            s3_credentials,
-            args.bucket_name,
+            storage_provider,
             heartbeats.clone(),
             store.clone(),
             args.hourly_s3_upload_limit,

crates/orchestrator/src/main.rs

@@ -31,3 +31,4 @@ rand = "0.9.0"
 google-cloud-storage = "0.24.0"
 base64 = "0.22.1"
 chrono = { workspace = true, features = ["serde"] }
+async-trait = "0.1.88"