make max queue len configurable

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

@@ -731,6 +731,7 @@ pub async fn handle_changes(
     mut tripwire: Tripwire,
 ) {
     let max_changes_chunk: usize = agent.config().perf.apply_queue_len;
+    let max_queue_len: usize = agent.config().perf.processing_queue_len;
     let mut queue: VecDeque<(ChangeV1, ChangeSource, Instant)> = VecDeque::new();
     let mut buf = vec![];
     let mut buf_cost = 0;
@@ -742,7 +743,6 @@ 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();
@@ -863,7 +863,7 @@ pub async fn handle_changes(
         }
 
         // drop items when the queue is full.
-        if queue.len() > MAX_QUEUE_LEN {
+        if queue.len() > max_queue_len {
             warn!(
                 "dropping changes from {} because changes queue is full",
                 change.actor_id

crates/corro-types/src/config.rs

@@ -14,6 +14,9 @@ const fn default_apply_queue() -> usize {
 const fn default_wal_threshold() -> usize {
     10
 }
+const fn default_processing_queue() -> usize {
+    100000
+}
 
 /// Used for the apply channel
 const fn default_huge_channel() -> usize {
@@ -187,6 +190,8 @@ pub struct PerfConfig {
     pub apply_queue_len: usize,
     #[serde(default = "default_wal_threshold")]
     pub wal_threshold_gb: usize,
+    #[serde(default = "default_processing_queue")]
+    pub processing_queue_len: usize,
 }
 
 impl Default for PerfConfig {
@@ -204,6 +209,7 @@ impl Default for PerfConfig {
             apply_queue_timeout: default_apply_timeout(),
             apply_queue_len: default_apply_queue(),
             wal_threshold_gb: default_wal_threshold(),
+            processing_queue_len: default_processing_queue(),
         }
     }
 }