drop items when processing queue is full

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

@@ -742,6 +742,7 @@ pub async fn handle_changes(
         agent.config().perf.apply_queue_timeout as u64,
     ));
 
+    const MAX_QUEUE_LEN: usize = 10000;
     const MAX_SEEN_CACHE_LEN: usize = 10000;
     const KEEP_SEEN_CACHE_SIZE: usize = 1000;
     let mut seen: IndexMap<_, RangeInclusiveSet<CrsqlSeq>> = IndexMap::new();
@@ -861,6 +862,15 @@ pub async fn handle_changes(
             continue;
         }
 
+        // drop items when the queue is full.
+        if queue.len() > MAX_QUEUE_LEN {
+            warn!(
+                "dropping changes from {} because changes queue is full",
+                change.actor_id
+            );
+            continue;
+        }
+
         if let Some(mut seqs) = change.seqs().cloned() {
             let v = *change.versions().start();
             if let Some(seen_seqs) = seen.get(&(change.actor_id, v)) {