scope issue fixed, trackin range fixed, moved block to async closure

Author: Davda-James

src/builder/analyzer.rs

@@ -969,14 +969,41 @@ impl AnalyzerContext {
             op_futs.push(self.analyze_reactive_op(op_scope, reactive_op).await?);
         }
         let collector_len = op_scope.states.lock().unwrap().collectors.len();
+        let scope_qualifier = self.build_scope_qualifier(op_scope);
         let result_fut = async move {
             Ok(AnalyzedOpScope {
                 reactive_ops: try_join_all(op_futs).await?,
                 collector_len,
+                scope_qualifier,
             })
         };
         Ok(result_fut)
     }
+
+    fn build_scope_qualifier(&self, op_scope: &Arc<OpScope>) -> String {
+        let mut scope_names = Vec::new();
+        let mut current_scope = Some(op_scope.as_ref());
+
+        // Walk up the parent chain to collect scope names
+        while let Some(scope) = current_scope {
+            if let Some((parent, _)) = &scope.parent {
+                scope_names.push(scope.name.clone());
+                current_scope = Some(parent.as_ref());
+            } else {
+                break;
+            }
+        }
+
+        // Reverse to get the correct order (root to leaf)
+        scope_names.reverse();
+
+        // Build the qualifier string: "" for root, "name." for single level, "parent.child." for nested
+        if scope_names.is_empty() {
+            String::new()
+        } else {
+            format!("{}.", scope_names.join("."))
+        }
+    }
 }
 
 pub fn build_flow_instance_context(

src/builder/plan.rs

@@ -130,6 +130,7 @@ pub enum AnalyzedReactiveOp {
 pub struct AnalyzedOpScope {
     pub reactive_ops: Vec<AnalyzedReactiveOp>,
     pub collector_len: usize,
+    pub scope_qualifier: String,
 }
 
 pub struct ExecutionPlan {

src/execution/evaluator.rs

@@ -364,7 +364,8 @@ async fn evaluate_op_scope(
             AnalyzedReactiveOp::Transform(op) => {
                 // Track transform operation start
                 if let Some(ref op_stats) = operation_in_process_stats {
-                    let transform_key = format!("transform/{}", op.name);
+                    let transform_key =
+                        format!("transform/{}{}", op_scope.scope_qualifier, op.name);
                     op_stats.start_processing(&transform_key, 1);
                 }
 
@@ -400,7 +401,8 @@ async fn evaluate_op_scope(
 
                 // Track transform operation completion
                 if let Some(ref op_stats) = operation_in_process_stats {
-                    let transform_key = format!("transform/{}", op.name);
+                    let transform_key =
+                        format!("transform/{}{}", op_scope.scope_qualifier, op.name);
                     op_stats.finish_processing(&transform_key, 1);
                 }
 

src/execution/row_indexer.rs

@@ -375,16 +375,15 @@ impl<'a> RowIndexer<'a> {
                         })
                         .collect();
                     (!mutations_w_ctx.is_empty()).then(|| {
-                        // Track export operation start
-                        if let Some(ref op_stats) = self.operation_in_process_stats {
-                            let export_key = format!("export/{}", export_op_group.target_kind);
-                            op_stats.start_processing(&export_key, 1);
-                        }
-
                         let export_key = format!("export/{}", export_op_group.target_kind);
                         let operation_in_process_stats = self.operation_in_process_stats;
 
                         async move {
+                            // Track export operation start
+                            if let Some(ref op_stats) = operation_in_process_stats {
+                                op_stats.start_processing(&export_key, 1);
+                            }
+
                             let result = export_op_group
                                 .target_factory
                                 .apply_mutation(mutations_w_ctx)

src/execution/source_indexer.rs

@@ -338,10 +338,6 @@ impl SourceIndexingContext {
 
             // Track that we're starting to process this row
             update_stats.processing.start(1);
-            if let Some(ref op_stats) = operation_in_process_stats {
-                let import_key = format!("import/{}", import_op.name);
-                op_stats.start_processing(&import_key, 1);
-            }
 
             let eval_ctx = SourceRowEvaluationContext {
                 plan: &plan,
@@ -351,20 +347,26 @@ impl SourceIndexingContext {
                 import_op_idx: self.source_idx,
             };
             let process_time = chrono::Utc::now();
+            let operation_in_process_stats_cloned = operation_in_process_stats.clone();
             let row_indexer = row_indexer::RowIndexer::new(
                 &eval_ctx,
                 &self.setup_execution_ctx,
                 mode,
                 process_time,
                 &update_stats,
-                operation_in_process_stats.as_ref().map(|s| s.as_ref()),
+                operation_in_process_stats_cloned
+                    .as_ref()
+                    .map(|s| s.as_ref()),
                 &pool,
             )?;
 
             let source_data = row_input.data;
             let mut row_state_operator =
                 LocalSourceRowStateOperator::new(&row_input.key, &self.state, &update_stats);
             let mut ordinal_touched = false;
+
+            let operation_in_process_stats_for_async = operation_in_process_stats.clone();
+            let operation_name_for_async = operation_name.clone();
             let result = {
                 let row_state_operator = &mut row_state_operator;
                 let row_key = &row_input.key;
@@ -417,6 +419,9 @@ impl SourceIndexingContext {
                                 (ordinal, source_data.content_version_fp, value)
                             }
                             _ => {
+                                if let Some(ref op_stats) = operation_in_process_stats_for_async {
+                                    op_stats.start_processing(&operation_name_for_async, 1);
+                                }
                                 let data = import_op
                         .executor
                         .get_value(
@@ -433,6 +438,9 @@ impl SourceIndexingContext {
                             },
                         )
                         .await?;
+                                if let Some(ref op_stats) = operation_in_process_stats_for_async {
+                                    op_stats.finish_processing(&operation_name_for_async, 1);
+                                }
                                 (
                                     data.ordinal.ok_or_else(|| {
                                         anyhow::anyhow!("ordinal is not available")
@@ -496,9 +504,6 @@ impl SourceIndexingContext {
 
             // Track that we're finishing processing this row (regardless of success/failure)
             update_stats.processing.end(1);
-            if let Some(ref op_stats) = operation_in_process_stats {
-                op_stats.finish_processing(&operation_name, 1);
-            }
 
             result?;
             if let Some(ack_fn) = ack_fn {