align toploc api with recent adjustments (#377)

Author: JannikSt

crates/validator/src/validators/synthetic_data/mod.rs

@@ -732,7 +732,6 @@ impl SyntheticDataValidator {
 
         Ok(())
     }
-
     pub async fn process_group_status_check(&self, group: ToplocGroup) -> Result<(), Error> {
         let toploc_config = self
             .find_matching_toploc_config(&group.prefix)
@@ -745,39 +744,56 @@ impl SyntheticDataValidator {
             .get_group_file_validation_status(&group.group_file_name)
             .await?;
 
-        info!("Group {} toploc status: {:?}", group.group_id, status);
-
         if let Some(metrics) = &self.metrics {
             metrics.record_group_validation_status(&group.group_id, &status.status.to_string());
         }
 
+        // Calculate total claimed units
+        let mut total_claimed_units: U256 = U256::from(0);
+        for work_key in &group.sorted_work_keys {
+            if let Some(work_info) = self.get_work_info_from_redis(work_key).await? {
+                total_claimed_units += work_info.work_units;
+            }
+        }
+
+        // Log basic info for all cases
+        info!(
+            "Group {} ({}) - Status: {:?}, Claimed: {}, Toploc: {} flops (input: {} flops)",
+            group.group_id,
+            group.group_file_name,
+            status.status,
+            total_claimed_units,
+            status.output_flops,
+            status.input_flops
+        );
+
+        // Handle rejection case
         if status.status == ValidationResult::Reject {
             let indices = status.failing_indices;
-            let mut work_keys_to_invalidate: Vec<String> = Vec::new();
-            for index in indices {
-                let work_key = group.sorted_work_keys[index as usize].clone();
-                work_keys_to_invalidate.push(work_key);
-            }
+            let work_keys_to_invalidate: Vec<String> = indices
+                .iter()
+                .map(|&idx| group.sorted_work_keys[idx as usize].clone())
+                .collect();
+
+            warn!(
+                "Group {} rejected - Invalidating keys: {:?}",
+                group.group_id, work_keys_to_invalidate
+            );
 
             for work_key in work_keys_to_invalidate {
                 self.invalidate_work(&work_key).await?;
             }
         }
 
-        let mut total_claimed_units: U256 = U256::from(0);
-        for work_key in &group.sorted_work_keys {
-            let work_info = self.get_work_info_from_redis(work_key).await?;
-            if let Some(work_info) = work_info {
-                total_claimed_units += work_info.work_units;
-            }
+        // Handle mismatched units case
+        if total_claimed_units != U256::from(status.output_flops as u64) {
+            warn!(
+                "Group {} units mismatch - Claimed: {}, Toploc: {}, Keys: {:?}",
+                group.group_id, total_claimed_units, status.output_flops, group.sorted_work_keys
+            );
         }
-        let toploc_units = status.flops;
-        info!(
-            "Group {} total claimed units: {:?} - toploc units: {:?}",
-            group.group_id, total_claimed_units, toploc_units
-        );
-        // TODO: We need to invalidate work that has claimed too many units
 
+        // Update validation status for all work keys
         for work_key in &group.sorted_work_keys {
             self.update_work_validation_status(work_key, &status.status)
                 .await?;
@@ -833,7 +849,7 @@ impl SyntheticDataValidator {
                 {
                     error!("Failed to process group task: {}", e);
                 }
-                info!(
+                debug!(
                     "waiting before next task: {}",
                     validator_clone_group_trigger.grace_interval
                 );

crates/validator/src/validators/synthetic_data/toploc.rs

@@ -23,7 +23,8 @@ pub struct Toploc {
 #[derive(Debug, PartialEq, Serialize, Deserialize)]
 pub struct GroupValidationResult {
     pub status: ValidationResult,
-    pub flops: f64,
+    pub input_flops: f64,
+    pub output_flops: f64,
     // This tells us which node(s) in a group actually failed the toploc validation
     pub failing_indices: Vec<i64>,
 }
@@ -86,7 +87,7 @@ impl Toploc {
             "{}/validate/{}",
             self.config.server_url, processed_file_name
         );
-        info!(
+        debug!(
             "Triggering remote toploc validation for {} {}",
             file_name, validate_url
         );
@@ -266,10 +267,15 @@ impl Toploc {
                                 _ => ValidationResult::Unknown,
                             };
 
-                            let flops = status_json
-                                .get("flops")
+                            let input_flops = status_json
+                                .get("input_flops")
+                                .and_then(|f| f.as_f64())
+                                .unwrap_or(0.0);
+                            let output_flops = status_json
+                                .get("output_flops")
                                 .and_then(|f| f.as_f64())
                                 .unwrap_or(0.0);
+
                             let failing_indices = status_json
                                 .get("failing_indices")
                                 .and_then(|f| f.as_array())
@@ -280,7 +286,8 @@ impl Toploc {
 
                             Ok(GroupValidationResult {
                                 status: validation_result,
-                                flops,
+                                input_flops,
+                                output_flops,
                                 failing_indices,
                             })
                         }
@@ -529,7 +536,7 @@ mod tests {
         let _status_mock = server
             .mock("GET", "/statusgroup/test-group.parquet")
             .with_status(200)
-            .with_body(r#"{"status": "accept", "flops": 12345.67, "failing_indices": []}"#)
+            .with_body(r#"{"status": "accept", "input_flops": 12345.67, "output_flops": 12345.67, "failing_indices": []}"#)
             .create();
 
         let config = ToplocConfig {
@@ -546,7 +553,8 @@ mod tests {
         assert!(result.is_ok());
         let group_result = result.unwrap();
         assert_eq!(group_result.status, ValidationResult::Accept);
-        assert_eq!(group_result.flops, 12345.67);
+        assert_eq!(group_result.input_flops, 12345.67);
+        assert_eq!(group_result.output_flops, 12345.67);
         assert!(group_result.failing_indices.is_empty());
         Ok(())
     }
@@ -558,7 +566,7 @@ mod tests {
         let _status_mock = server
             .mock("GET", "/statusgroup/test-group.parquet")
             .with_status(200)
-            .with_body(r#"{"status": "reject", "flops": 0.0, "failing_indices": [1, 3, 5]}"#)
+            .with_body(r#"{"status": "reject", "input_flops": 0.0, "output_flops": 0.0, "failing_indices": [1, 3, 5]}"#)
             .create();
 
         let config = ToplocConfig {
@@ -575,7 +583,8 @@ mod tests {
         assert!(result.is_ok());
         let group_result = result.unwrap();
         assert_eq!(group_result.status, ValidationResult::Reject);
-        assert_eq!(group_result.flops, 0.0);
+        assert_eq!(group_result.input_flops, 0.0);
+        assert_eq!(group_result.output_flops, 0.0);
         assert_eq!(group_result.failing_indices, vec![1, 3, 5]);
         Ok(())
     }