Test tweaks

Author: gorbak25

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

@@ -642,7 +642,7 @@ impl HandleChangesState {
         max_queue_len: usize,
     ) -> Self {
         Self {
-            queue: VecDeque::new(),
+            queue: VecDeque::with_capacity(max_queue_len),
             buf_cost: 0,
             current_batch_size: min_batch_size,
             processing_task: None,
@@ -687,7 +687,7 @@ impl HandleChangesState {
             return None;
         }
 
-        let mut batch = Vec::new();
+        let mut batch = Vec::with_capacity(self.current_batch_size);
         let mut batch_cost = 0;
 
         while let Some((change, src, queued_at)) = self.queue.pop_front() {

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

@@ -1104,11 +1104,15 @@ pub async fn process_multiple_changes(
 
     let mut change_chunk_size = 0;
 
-    for (_actor_id, changeset, db_version, _src) in changesets {
+    for (_actor_id, changeset, _db_version, _src) in &changesets {
         change_chunk_size += changeset.len();
-        match_changes(agent.subs_manager(), &changeset, db_version);
-        match_changes(agent.updates_manager(), &changeset, db_version);
     }
+    tokio::spawn(async move {
+        for (_actor_id, changeset, db_version, _src) in changesets {
+            match_changes(agent.subs_manager(), &changeset, db_version);
+            match_changes(agent.updates_manager(), &changeset, db_version);
+        }
+    });
 
     histogram!("corro.agent.changes.processing.time.seconds", "source" => "remote")
         .record(start.elapsed());