Check timings

Author: gorbak25

crates/corro-agent/src/agent/handlers.rs

@@ -681,6 +681,7 @@ impl HandleChangesState {
         target_size: usize,
         reason: &str,
     ) -> Option<usize> {
+        let t1 = Instant::now();
         // Must complete previous task before spawning a new one
         if self.processing_task.is_some() {
             return None;
@@ -716,6 +717,8 @@ impl HandleChangesState {
         counter!("corro.agent.changes.batch.spawned").increment(1);
         gauge!("corro.agent.changes.batch_size").set(self.current_batch_size as f64);
 
+        info!("drain_and_spawn took: {}", t1.elapsed());
+
         Some(batch_cost)
     }
 

crates/corro-agent/src/agent/util.rs

@@ -820,6 +820,10 @@ pub async fn process_multiple_changes(
         warn!("process_multiple_changes: removing duplicates took too long - {elapsed:?}");
     }
 
+    // Deduplicate
+    let phase1_t = start.elapsed();
+    let t2 = Instant::now();
+
     let mut conn = agent.pool().write_normal().await?;
 
     let changesets = block_in_place(|| {
@@ -1094,6 +1098,10 @@ pub async fn process_multiple_changes(
         Ok::<_, ChangeError>(changesets)
     })?;
 
+    // block_in place db + bookie stuff
+    let phase2_t = t2.elapsed();
+    let t3 = Instant::now();
+
     let mut change_chunk_size = 0;
 
     for (_actor_id, changeset, db_version, _src) in changesets {
@@ -1106,6 +1114,20 @@ pub async fn process_multiple_changes(
         .record(start.elapsed());
     histogram!("corro.agent.changes.processing.chunk_size").record(change_chunk_size as f64);
 
+    // Broadcast changes
+    let phase3_t = t3.elapsed();
+
+    let total = phase1_t + phase2_t + phase3_t;
+    info!(
+        "preprocess: {} ({} of total), db_stuff: {} ({} of total), match_changes: {} ({} of total)",
+        phase1_t.as_secs_f64(),
+        phase1_t.as_secs_f64() / total.as_secs_f64(),
+        phase2_t.as_secs_f64(),
+        phase2_t.as_secs_f64() / total.as_secs_f64(),
+        phase3_t.as_secs_f64(),
+        phase3_t.as_secs_f64() / total.as_secs_f64()
+    );
+
     Ok(())
 }