From df72a6fb450d84a8f2f192a08a4f6910f97d55ba Mon Sep 17 00:00:00 2001 From: Enoch Tang Date: Mon, 9 Nov 2026 16:39:58 -0500 Subject: [PATCH] fix(config): Reject pending and delay limits below the insert batch length The activation writer only stores a batch when pending + batch.len() is within the limit, and it never splits a batch. If max_pending_count (or a topic's max_pending_activations, or max_delay_count) is below insert_batch_max_length, a full batch can never be written, even with an empty store. The writer retries it forever and the consumer stops. Low traffic hides this because batches flush on the timer before they fill. Check this at startup in normalize_and_validate. The global pending limit is only checked when at least one consumed topic has no limit of its own. Produce-only topics are skipped. Co-Authored-By: Claude Opus 5.5 --- src/config/mod.rs | 216 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 216 insertions(+) diff --git a/src/config/mod.rs b/src/config/mod.rs index 501759b3..af0ef231 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -836,6 +836,56 @@ impl Config { self.normalize_applications(); self.validate_applications()?; + self.validate_store_limits()?; + + Ok(()) + } + + /// The activation writer only stores a batch when `pending + batch.len()` + /// is within the pending limit. If a limit is below `insert_batch_max_length`, + /// a full batch can't be stored. The writer will keep retrying and the consumer + /// stops making progress. + fn validate_store_limits(&self) -> Result<()> { + let batch_len = self.store.insert_batch_max_length; + + // Each consumed topic's writer uses either that topic's + // max_pending_activations or, if it has none, store.max_pending_count. + // Only check the global limit if at least one topic falls back to it. + let mut uses_global_pending_limit = false; + + for (topic_name, topic_config) in &self.kafka_topics { + if topic_config.produce_only { + continue; + } + + match topic_config.max_pending_activations { + Some(limit) if limit < batch_len => { + return Err(anyhow!( + "topic '{topic_name}' max_pending_activations ({limit}) must be at \ + least store.insert_batch_max_length ({batch_len}), or a full batch \ + can never be written" + )); + } + Some(_) => {} + None => uses_global_pending_limit = true, + } + } + + if uses_global_pending_limit && self.store.max_pending_count < batch_len { + return Err(anyhow!( + "store.max_pending_count ({}) must be at least store.insert_batch_max_length \ + ({batch_len}), or a full batch can never be written", + self.store.max_pending_count + )); + } + + if self.store.max_delay_count < batch_len { + return Err(anyhow!( + "store.max_delay_count ({}) must be at least store.insert_batch_max_length \ + ({batch_len}), or a full batch of delayed tasks can never be written", + self.store.max_delay_count + )); + } Ok(()) } @@ -2348,4 +2398,170 @@ kafka_clusters: Ok(()) }); } + + /// Load a config from a YAML file through `from_args`, so deprecated + /// mapping and normalization run the same way they do at startup. + fn config_from_yaml(yaml: &str) -> anyhow::Result { + let mut result = None; + Jail::expect_with(|jail| { + jail.create_file("config.yaml", yaml)?; + let args = Args { + run: Run::Broker, + config: Some("config.yaml".to_owned()), + }; + result = Some(Config::from_args(&args)); + Ok(()) + }); + result.unwrap() + } + + /// Two consumed topics plus produce-only retry and DLQ topics. `store`, + /// `a` and `b` are extra fields for the store and the `a` and `b` topics. + fn two_topic_yaml(store: &str, a: &str, b: &str) -> String { + format!( + r#" +kafka_deadletter_topic: dlq +kafka_retry_topic: retry + +store: +{store} + +kafka_topics: + a: + cluster: my-cluster + consumer_group: taskbroker-a +{a} + b: + cluster: my-cluster + consumer_group: taskbroker-b +{b} + retry: + cluster: my-cluster + consumer_group: taskbroker-retry + produce_only: true + dlq: + cluster: my-cluster + consumer_group: taskbroker-dlq + produce_only: true + max_pending_activations: 1 + +kafka_clusters: + my-cluster: + address: 10.0.0.1:9092 +"# + ) + } + + #[test] + fn test_store_limits_reject_pending_below_batch_len() { + // The us launchpad case: pending lowered to 128, batch length left at + // the default of 256. A full batch could never be written. + let err = config_from_yaml( + r#" +store: + max_pending_count: 128 +"#, + ) + .unwrap_err(); + + assert!( + err.to_string().contains("store.max_pending_count (128)"), + "{err}" + ); + } + + #[test] + fn test_store_limits_allow_pending_equal_to_batch_len() { + // 0 pending + a full batch of 128 is not over a limit of 128. + let config = config_from_yaml( + r#" +store: + max_pending_count: 128 + insert_batch_max_length: 128 +"#, + ) + .unwrap(); + + assert_eq!(config.store.max_pending_count, 128); + assert_eq!(config.store.insert_batch_max_length, 128); + } + + #[test] + fn test_store_limits_apply_to_deprecated_batch_len() { + // The deprecated field maps onto insert_batch_max_length before + // validation, so it is checked too. + let err = config_from_yaml( + r#" +db_insert_batch_max_len: 4096 +"#, + ) + .unwrap_err(); + + assert!( + err.to_string().contains("store.max_pending_count (2048)"), + "{err}" + ); + } + + #[test] + fn test_store_limits_reject_delay_below_batch_len() { + let err = config_from_yaml( + r#" +store: + max_delay_count: 100 + insert_batch_max_length: 200 +"#, + ) + .unwrap_err(); + + assert!( + err.to_string().contains("store.max_delay_count (100)"), + "{err}" + ); + } + + #[test] + fn test_store_limits_reject_topic_pending_below_batch_len() { + let yaml = two_topic_yaml( + " insert_batch_max_length: 256", + " max_pending_activations: 100", + "", + ); + let err = config_from_yaml(&yaml).unwrap_err(); + + assert!( + err.to_string() + .contains("topic 'a' max_pending_activations (100)"), + "{err}" + ); + } + + #[test] + fn test_store_limits_skip_global_pending_when_every_topic_overrides() { + // Both consumed topics set their own limit, so the low global limit is + // never used. The produce-only DLQ's tiny limit is also ignored. + let yaml = two_topic_yaml( + " max_pending_count: 1\n insert_batch_max_length: 256", + " max_pending_activations: 256", + " max_pending_activations: 512", + ); + + assert!(config_from_yaml(&yaml).is_ok()); + } + + #[test] + fn test_store_limits_check_global_pending_when_a_topic_has_no_override() { + // Topic b has no override, so it falls back to the global limit. + let yaml = two_topic_yaml( + " max_pending_count: 1\n insert_batch_max_length: 256", + " max_pending_activations: 256", + "", + ); + let err = config_from_yaml(&yaml).unwrap_err(); + + assert!( + err.to_string().contains("store.max_pending_count (1)"), + "{err}" + ); + } }