Merge pull request #557 from PrimeIntellect-ai/release/0.3.4

JannikSt · web-flow · commit be374bbacc2a · 2025-06-23T18:25:00.000+10:00
Release/0.3.4
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -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.3"
+version = "0.3.4"
 edition = "2021"
 
 [workspace.features]
diff --git a/crates/orchestrator/src/main.rs b/crates/orchestrator/src/main.rs
@@ -126,51 +126,6 @@ struct Args {
     max_healthy_nodes_with_same_endpoint: u32,
 }
 
-async fn run_inactive_node_metric_migration(store_context: Arc<StoreContext>) -> Result<()> {
-    info!("Starting migration of inactive node metrics to new data model");
-
-    let node_addresses = match store_context.node_store.get_nodes().await {
-        Ok(nodes) => {
-            let addresses: Vec<_> = nodes.into_iter().map(|node| node.address).collect();
-            info!("Found {} nodes to migrate", addresses.len());
-            addresses
-        }
-        Err(e) => {
-            error!("Error getting all nodes for migration: {}", e);
-            return Ok(()); // Don't fail startup if migration can't get nodes
-        }
-    };
-
-    let mut migrated_count = 0;
-    let mut error_count = 0;
-
-    for node_address in node_addresses {
-        match store_context
-            .metrics_store
-            .migrate_node_metrics_if_needed(node_address)
-            .await
-        {
-            Ok(()) => {
-                migrated_count += 1;
-                if migrated_count % 100 == 0 {
-                    info!("Migrated {} nodes so far...", migrated_count);
-                }
-            }
-            Err(e) => {
-                error!("Error migrating metrics for node {}: {}", node_address, e);
-                error_count += 1;
-                // Continue with other nodes even if one fails
-            }
-        }
-    }
-
-    info!(
-        "Migration completed. Successfully migrated {} nodes, {} errors",
-        migrated_count, error_count
-    );
-    Ok(())
-}
-
 #[tokio::main]
 async fn main() -> Result<()> {
     let args = Args::parse();
@@ -221,17 +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()));
 
-    // Run one-time migration for inactive nodes
-    let migration_store_context = store_context.clone();
-    if matches!(server_mode, ServerMode::ProcessorOnly | ServerMode::Full) {
-        let migration_store_context = migration_store_context.clone();
-        tokio::spawn(async move {
-            if let Err(e) = run_inactive_node_metric_migration(migration_store_context).await {
-                error!("Failed to run inactive node metric migration: {}", e);
-            }
-        });
-    }
-
     let p2p_client = Arc::new(P2PClient::new(wallet.clone()).await.unwrap());
 
     let contracts = ContractBuilder::new(wallet.provider())
@@ -313,26 +257,6 @@ async fn main() -> Result<()> {
                     Some(webhook_plugins.clone()),
                 );
 
-                // Run groups index migration on startup
-                if matches!(server_mode, ServerMode::ProcessorOnly | ServerMode::Full) {
-                    match group_plugin.migrate_groups_index().await {
-                        Ok(count) => {
-                            if count > 0 {
-                                info!(
-                                    "Groups index migration completed: {} groups migrated",
-                                    count
-                                );
-                            } else {
-                                info!("Groups index migration: no groups to migrate");
-                            }
-                        }
-                        Err(e) => {
-                            error!("Groups index migration failed: {}", e);
-                            return Err(e);
-                        }
-                    }
-                }
-
                 let status_group_plugin = group_plugin.clone();
                 let group_plugin_for_server = group_plugin.clone();
                 node_groups_plugin = Some(Arc::new(group_plugin_for_server));
diff --git a/crates/orchestrator/src/plugins/node_groups/mod.rs b/crates/orchestrator/src/plugins/node_groups/mod.rs
@@ -1138,88 +1138,6 @@ impl NodeGroupsPlugin {
         Ok(mappings)
     }
 
-    /// Migrate existing group data to populate the groups index
-    /// This method should be called once after deploying the new groups index feature
-    pub async fn migrate_groups_index(&self) -> Result<usize, Error> {
-        debug!("Starting groups index migration");
-        let mut conn = self.store.client.get_multiplexed_async_connection().await?;
-
-        // Check if migration is needed by seeing if groups index is empty
-        let existing_groups_in_index: Vec<String> = conn.smembers(GROUPS_INDEX_KEY).await?;
-        if !existing_groups_in_index.is_empty() {
-            info!(
-                "Groups index already contains {} groups, migration appears completed",
-                existing_groups_in_index.len()
-            );
-            return Ok(existing_groups_in_index.len());
-        }
-
-        // Get all node-to-group mappings using the old method
-        let node_mappings: HashMap<String, String> = conn.hgetall(NODE_GROUP_MAP_KEY).await?;
-
-        if node_mappings.is_empty() {
-            info!("No existing groups found to migrate");
-            return Ok(0);
-        }
-
-        // Collect unique group IDs from node mappings
-        let existing_group_ids: HashSet<String> = node_mappings.values().cloned().collect();
-        info!(
-            "Found {} unique group IDs to migrate",
-            existing_group_ids.len()
-        );
-
-        // Verify these groups actually exist by checking their keys
-        let group_keys: Vec<String> = existing_group_ids
-            .iter()
-            .map(|id| Self::get_group_key(id))
-            .collect();
-
-        let group_values: Vec<Option<String>> = conn.mget(&group_keys).await?;
-
-        let mut valid_group_ids = Vec::new();
-        for (group_id, group_data) in existing_group_ids.iter().zip(group_values.iter()) {
-            if group_data.is_some() {
-                valid_group_ids.push(group_id.clone());
-            } else {
-                warn!(
-                    "Group {} exists in mappings but has no data, skipping",
-                    group_id
-                );
-            }
-        }
-
-        if valid_group_ids.is_empty() {
-            info!("No valid groups found to migrate");
-            return Ok(0);
-        }
-
-        // Add all valid group IDs to the groups index in a single operation
-        let _: () = conn.sadd(GROUPS_INDEX_KEY, &valid_group_ids).await?;
-
-        info!(
-            "Successfully migrated {} groups to groups index",
-            valid_group_ids.len()
-        );
-
-        // Verify the migration by checking the index
-        let migrated_count: usize = conn.scard(GROUPS_INDEX_KEY).await?;
-        if migrated_count != valid_group_ids.len() {
-            error!(
-                "Migration verification failed: expected {} groups in index, found {}",
-                valid_group_ids.len(),
-                migrated_count
-            );
-        } else {
-            info!(
-                "Migration verification successful: {} groups in index",
-                migrated_count
-            );
-        }
-
-        Ok(valid_group_ids.len())
-    }
-
     /// Get all groups assigned to a specific task
     /// Returns a list of group IDs that are currently working on the given task
     pub async fn get_groups_for_task(&self, task_id: &str) -> Result<Vec<String>, Error> {
diff --git a/crates/orchestrator/src/store/domains/metrics_store.rs b/crates/orchestrator/src/store/domains/metrics_store.rs
@@ -1,14 +1,13 @@
 use crate::store::core::RedisStore;
 use alloy::primitives::Address;
 use anyhow::{anyhow, Result};
-use log::{error, info};
+use log::error;
 use redis::AsyncCommands;
 use shared::models::metric::MetricEntry;
 use std::collections::HashMap;
 use std::sync::Arc;
 
 const ORCHESTRATOR_NODE_METRICS_STORE: &str = "orchestrator:node_metrics";
-const ORCHESTRATOR_METRICS_STORE: &str = "orchestrator:metrics";
 
 pub struct MetricsStore {
     redis: Arc<RedisStore>,
@@ -23,77 +22,6 @@ impl MetricsStore {
         label.replace(':', "")
     }
 
-    pub async fn migrate_node_metrics_if_needed(&self, node_address: Address) -> Result<()> {
-        let mut con = self.redis.client.get_multiplexed_async_connection().await?;
-        let new_key = format!("{}:{}", ORCHESTRATOR_NODE_METRICS_STORE, node_address);
-
-        // Check if the new node-centric key already exists
-        let exists: bool = con.exists(&new_key).await?;
-        if exists {
-            info!("Migration already complete for this node: {}", node_address);
-            // Migration already complete for this node
-            return Ok(());
-        }
-
-        // Perform the slow SCAN to find all metrics for this node in the old data structure
-        let pattern = format!("{}:*:*", ORCHESTRATOR_METRICS_STORE);
-        let mut iter: redis::AsyncIter<String> = con.scan_match(&pattern).await?;
-        let mut old_keys_to_migrate = Vec::new();
-
-        while let Some(key) = iter.next_item().await {
-            old_keys_to_migrate.push(key);
-        }
-        drop(iter);
-
-        // Collect all metrics for this node from the old structure
-        let mut node_metrics = HashMap::new();
-        let mut keys_to_clean = Vec::new();
-
-        for old_key in old_keys_to_migrate {
-            if let Some(value_str) = con
-                .hget::<_, _, Option<String>>(&old_key, node_address.to_string())
-                .await?
-            {
-                if let Ok(val) = value_str.parse::<f64>() {
-                    let parts: Vec<&str> = old_key.split(':').collect();
-                    if parts.len() >= 4 {
-                        let task_id = parts[2];
-                        let metric_name = parts[3];
-                        let new_metric_key = format!("{}:{}", task_id, metric_name);
-                        node_metrics.insert(new_metric_key, val);
-                        keys_to_clean.push(old_key);
-                    }
-                }
-            }
-        }
-
-        // If we have metrics for this node, perform the atomic migration
-        if !node_metrics.is_empty() {
-            // Use Redis MULTI/EXEC transaction for atomicity
-            let mut pipe = redis::pipe();
-            pipe.atomic();
-
-            // Set all metrics in the new node-centric key
-            for (metric_key, value) in &node_metrics {
-                pipe.hset(&new_key, metric_key, value);
-            }
-
-            // Clean up the old data structure by removing this node's fields
-            for old_key in &keys_to_clean {
-                pipe.hdel(old_key, node_address.to_string());
-            }
-
-            pipe.query_async::<()>(&mut con).await?;
-        }
-
-        // Always create the key to mark migration as complete, even if no metrics exist
-        // This prevents future migration attempts for nodes without data
-        let _: () = con.hset(&new_key, "_migrated", "true").await?;
-        let _: () = con.hdel(&new_key, "_migrated").await?;
-
-        Ok(())
-    }
-
     pub async fn store_metrics(
         &self,
         metrics: Option<Vec<MetricEntry>>,
