drop items when processing queue is full

superfly · Aug 16, 2024 · 181f7cf · 181f7cf
1 parent 9a91edd
commit 181f7cf
Showing 1 changed file with 10 additions and 0 deletions.
diff --git a/crates/corro-agent/src/agent/handlers.rs b/crates/corro-agent/src/agent/handlers.rs
@@ -719,6 +719,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();
@@ -838,6 +839,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)) {