Merge pull request #569 from PrimeIntellect-ai/release/0.3.6

Release/0.3.6

Author: JannikSt

Cargo.lock

@@ -40,7 +40,7 @@ iroh = "0.34.1"
 rand_v8 = { package = "rand", version = "0.8.5", features = ["std"] }
 rand_core_v6 = { package = "rand_core", version = "0.6.4", features = ["std"] }
 [workspace.package]
-version = "0.3.5"
+version = "0.3.6"
 edition = "2021"
 
 [workspace.features]

Cargo.toml

@@ -41,7 +41,6 @@ use shared::web3::contracts::core::builder::ContractBuilder;
 use shared::web3::contracts::structs::compute_pool::PoolStatus;
 use shared::web3::wallet::Wallet;
 use std::sync::Arc;
-use std::time::Duration;
 use tokio::task::JoinSet;
 use url::Url;
 
@@ -177,32 +176,6 @@ async fn main() -> Result<()> {
     let store = Arc::new(RedisStore::new(&args.redis_store_url));
     let store_context = Arc::new(StoreContext::new(store.clone()));
 
-    if server_mode == ServerMode::ProcessorOnly || server_mode == ServerMode::Full {
-        info!("Migrating node store to hash");
-        if let Err(e) = store_context.node_store.migrate_json_to_hash().await {
-            error!("Error migrating node store to hash: {}", e);
-        }
-        info!("Node store migrated to hash");
-    } else {
-        // Wait for all nodes to be migrated to hash format
-        loop {
-            match store_context.node_store.count_non_hash_format_nodes().await {
-                Ok(0) => {
-                    info!("All nodes are in hash format");
-                    break;
-                }
-                Ok(count) => {
-                    info!("Waiting for {} nodes to be migrated to hash format", count);
-                    tokio::time::sleep(Duration::from_secs(1)).await;
-                }
-                Err(e) => {
-                    error!("Error counting non-hash format nodes: {}", e);
-                    tokio::time::sleep(Duration::from_secs(1)).await;
-                }
-            }
-        }
-    }
-
     let p2p_client = Arc::new(P2PClient::new(wallet.clone()).await.unwrap());
 
     let contracts = ContractBuilder::new(wallet.provider())

crates/orchestrator/src/main.rs

@@ -375,109 +375,6 @@ impl NodeStore {
         let _: () = pipe.query_async(&mut con).await?;
         Ok(())
     }
-    /// Count how many nodes are already in hash format
-    pub async fn count_non_hash_format_nodes(&self) -> Result<usize> {
-        let mut con = self.redis.client.get_multiplexed_async_connection().await?;
-        let addresses: Vec<String> = con.smembers(ORCHESTRATOR_NODE_INDEX).await?;
-
-        let mut non_hash_count = 0;
-
-        // Process in batches for better performance
-        const BATCH_SIZE: usize = 100;
-
-        for chunk in addresses.chunks(BATCH_SIZE) {
-            // Check types of all keys in this batch
-            let mut type_pipe = redis::pipe();
-            let keys: Vec<String> = chunk
-                .iter()
-                .map(|addr| format!("{}:{}", ORCHESTRATOR_BASE_KEY, addr))
-                .collect();
-
-            for key in &keys {
-                type_pipe.cmd("TYPE").arg(key);
-            }
-
-            let types: Vec<String> = type_pipe.query_async(&mut con).await?;
-
-            // Count non-hash format keys (including "string" and "none" types)
-            for key_type in types.iter() {
-                if key_type != "hash" && key_type != "none" {
-                    non_hash_count += 1;
-                }
-            }
-        }
-
-        Ok(non_hash_count)
-    }
-
-    /// One-time migration from JSON to hash format
-    /// Run this once to convert all existing nodes
-    pub async fn migrate_json_to_hash(&self) -> Result<(usize, usize)> {
-        let mut con = self.redis.client.get_multiplexed_async_connection().await?;
-        let addresses: Vec<String> = con.smembers(ORCHESTRATOR_NODE_INDEX).await?;
-
-        let mut migrated = 0;
-
-        // Get count of nodes not yet in hash format
-        let non_hash_count = self.count_non_hash_format_nodes().await?;
-
-        // Process in batches for better performance
-        const BATCH_SIZE: usize = 100;
-
-        for chunk in addresses.chunks(BATCH_SIZE) {
-            // First, check types of all keys in this batch
-            let mut type_pipe = redis::pipe();
-            let keys: Vec<String> = chunk
-                .iter()
-                .map(|addr| format!("{}:{}", ORCHESTRATOR_BASE_KEY, addr))
-                .collect();
-
-            for key in &keys {
-                type_pipe.cmd("TYPE").arg(key);
-            }
-
-            let types: Vec<String> = type_pipe.query_async(&mut con).await?;
-
-            // Build migration pipeline for string keys
-            let mut migration_pipe = redis::pipe();
-            migration_pipe.atomic();
-            let mut batch_migrated = 0;
-
-            for (i, key_type) in types.iter().enumerate() {
-                if key_type == "string" {
-                    let key = &keys[i];
-
-                    // Get the JSON value
-                    let value: Option<String> = con.get(key).await?;
-
-                    if let Some(json_str) = value {
-                        if let Ok(node) = serde_json::from_str::<OrchestratorNode>(&json_str) {
-                            let fields = Self::node_to_hash_fields(&node)?;
-
-                            // Delete old string key and create new hash atomically
-                            migration_pipe.del(key);
-                            migration_pipe.hset_multiple(key, &fields);
-
-                            batch_migrated += 1;
-                        }
-                    }
-                }
-            }
-
-            // Execute migration for this batch
-            if batch_migrated > 0 {
-                let _: () = migration_pipe.query_async(&mut con).await?;
-                migrated += batch_migrated;
-            }
-        }
-
-        let already_hash = addresses.len() - non_hash_count;
-        info!(
-            "Migration complete: {} nodes migrated, {} already in hash format",
-            migrated, already_hash
-        );
-        Ok((migrated, already_hash))
-    }
 }
 
 #[cfg(test)]