From ded4a6ce41199e664286a39c0c0a6e9f23ea59b3 Mon Sep 17 00:00:00 2001 From: Evan Hicks Date: Mon, 31 Aug 2026 14:20:36 -0400 Subject: [PATCH 1/3] feat: Add a per topic pending/processing limit There is now a config option that can be specified in each topic config, allowing each topic to have a maximum number of tasks in the processing/pending state. This allows a taskbroker that processes multiple topics to limit each topic individually, without impacting the throughput of other topics. --- src/config/kafka.rs | 8 + src/config/mod.rs | 6 + src/config/store.rs | 6 + src/kafka/activation_writer.rs | 403 +++++++++++++++++++++++++++++---- src/store/adapters/postgres.rs | 6 +- src/store/tests.rs | 90 +++++++- src/store/traits.rs | 4 +- 7 files changed, 475 insertions(+), 48 deletions(-) diff --git a/src/config/kafka.rs b/src/config/kafka.rs index c0939bf7..522bc207 100644 --- a/src/config/kafka.rs +++ b/src/config/kafka.rs @@ -33,6 +33,14 @@ pub struct TopicConfig { /// Falls back to the global `kafka_auto_offset_reset` when unset. #[serde(default)] pub auto_offset_reset: Option, + /// The topic-specific maximum number of processing records that can be + /// in the ActivationStore. + #[serde(default)] + pub max_processing_activations: Option, + /// The topic-specific maximum number of pending activations that can be + /// in the ActivationStore. + #[serde(default)] + pub max_pending_activations: Option, } /// Configuration for a Kafka cluster. diff --git a/src/config/mod.rs b/src/config/mod.rs index 1b401b8c..11df7567 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -625,6 +625,8 @@ impl Config { session_timeout_ms: None, auto_commit_interval_ms: None, auto_offset_reset: None, + max_processing_activations: None, + max_pending_activations: None, }, ); assert!(prev.is_none(), "internal: duplicate topic '{topic_name}'"); @@ -643,6 +645,8 @@ impl Config { session_timeout_ms: None, auto_commit_interval_ms: None, auto_offset_reset: None, + max_processing_activations: None, + max_pending_activations: None, }, ); if prev.is_some() { @@ -675,6 +679,8 @@ impl Config { session_timeout_ms: None, auto_commit_interval_ms: None, auto_offset_reset: None, + max_processing_activations: None, + max_pending_activations: None, }); } } diff --git a/src/config/store.rs b/src/config/store.rs index bef22143..182a3b80 100644 --- a/src/config/store.rs +++ b/src/config/store.rs @@ -2,6 +2,7 @@ use std::time::Duration; use anyhow::Result; use serde::{Deserialize, Serialize}; +use std::collections::HashMap; use tracing::warn; use crate::config::Config; @@ -172,6 +173,10 @@ pub struct StoreConfig { /// in the ActivationStore (sqlite) pub max_processing_count: usize, + /// A dictionary of topic-specific maximum number of processing records that can be + /// in the ActivationStore + pub max_processing_count_per_topic: HashMap, + /// The maximum number of times a task can be reset from /// processing back to pending. When this limit is reached, /// the activation will be discarded/deadlettered. @@ -204,6 +209,7 @@ impl Default for StoreConfig { max_pending_count: 2048, max_delay_count: 8192, max_processing_count: 2048, + max_processing_count_per_topic: HashMap::new(), max_processing_attempts: 5, processing_deadline_grace_sec: 3, contention_drain_age_sec: 60, diff --git a/src/kafka/activation_writer.rs b/src/kafka/activation_writer.rs index 4040f74a..9a5b89ac 100644 --- a/src/kafka/activation_writer.rs +++ b/src/kafka/activation_writer.rs @@ -21,7 +21,9 @@ pub struct ActivationWriterConfig { pub topic: String, pub max_buf_len: usize, pub max_pending_activations: usize, + pub max_pending_activations_per_topic: Option, pub max_processing_activations: usize, + pub max_processing_activations_per_topic: Option, pub max_delay_activations: usize, pub db_max_size: Option, pub write_failure_backoff_ms: u64, @@ -31,12 +33,18 @@ impl ActivationWriterConfig { /// Convert from application configuration into ActivationWriter config for a /// single consumed topic. pub fn from_topic(config: &Config, topic: &str) -> Self { + let topic_config = config + .kafka_topics + .get(topic) + .unwrap_or_else(|| panic!("unknown topic '{topic}' - was config validated?")); Self { topic: topic.to_owned(), db_max_size: config.store.max_size, max_buf_len: config.store.insert_batch_max_length, max_pending_activations: config.store.max_pending_count, + max_pending_activations_per_topic: topic_config.max_pending_activations, max_processing_activations: config.store.max_processing_count, + max_processing_activations_per_topic: topic_config.max_processing_activations, max_delay_activations: config.store.max_delay_count, write_failure_backoff_ms: config.store.insert_failure_backoff_ms, } @@ -59,30 +67,27 @@ impl ActivationWriter { } } -impl Reducer for ActivationWriter { - type Input = Vec; - - type Output = (); - - async fn reduce(&mut self, batch: Self::Input) -> Result<(), anyhow::Error> { - assert!(self.batch.is_none()); - self.batch = Some(batch); - Ok(()) - } - - #[instrument(skip_all)] - async fn flush(&mut self) -> Result, anyhow::Error> { - let Some(ref batch) = self.batch else { - return Ok(None); - }; - - // If batch is empty (all tasks were forwarded), just mark as complete - if batch.is_empty() { - self.batch.take(); - return Ok(Some(())); +impl ActivationWriter { + async fn check_backpressure( + &self, + batch: &[Activation], + ) -> Result, anyhow::Error> { + if self.config.max_processing_activations_per_topic.is_some() + || self.config.max_pending_activations_per_topic.is_some() + { + let reason = self + .check_backpressure_for_topic(batch, Some(&self.config.topic)) + .await?; + return Ok(reason); } + self.check_backpressure_for_topic(batch, None).await + } - // Check if writing the batch would exceed the limits + async fn check_backpressure_for_topic( + &self, + batch: &[Activation], + topic: Option<&str>, + ) -> Result, anyhow::Error> { let DepthCounts { pending, delay, @@ -90,14 +95,20 @@ impl Reducer for ActivationWriter { processing, } = self .store - .count_depths() + .count_depths(topic) .await .expect("Error communicating with activation store"); - let exceeded_pending_limit = pending + batch.len() > self.config.max_pending_activations; + let exceeded_pending_limit = match self.config.max_pending_activations_per_topic { + Some(limit) => pending + batch.len() > limit, + None => pending + batch.len() > self.config.max_pending_activations, + }; let exceeded_delay_limit = delay + batch.len() > self.config.max_delay_activations; - let exceeded_processing_limit = - processing + claimed >= self.config.max_processing_activations; + let exceeded_processing_limit = match self.config.max_processing_activations_per_topic { + Some(limit) => processing + claimed >= limit, + None => processing + claimed >= self.config.max_processing_activations, + }; + let exceeded_db_size = if let Some(db_max_size) = self.config.db_max_size { self.store .db_size() @@ -107,7 +118,6 @@ impl Reducer for ActivationWriter { } else { false }; - // Check if the entire batch is either pending or delay let has_delay = batch .iter() @@ -122,20 +132,55 @@ impl Reducer for ActivationWriter { // a. There are delay activations in the batch, OR // b. The pending limit is also exceeded // 3. The pending limit is exceeded AND there are pending activations - if exceeded_processing_limit - || exceeded_db_size - || exceeded_delay_limit && (has_delay || exceeded_pending_limit) - || exceeded_pending_limit && has_pending - { - let reason = if exceeded_processing_limit { - "processing_limit" - } else if exceeded_delay_limit { - "delay_limit" - } else if exceeded_db_size { - "db_size_limit" - } else { - "pending_limit" - }; + let prefix = topic.unwrap_or("global"); + let reason: Option = match ( + exceeded_processing_limit, + exceeded_db_size, + exceeded_pending_limit, + has_pending, + exceeded_delay_limit, + has_delay, + ) { + (true, _, _, _, _, _) => Some(format!("{prefix}.processing_limit")), + (_, true, _, _, _, _) => Some(format!("{prefix}.db_size_limit")), + (_, _, true, true, _, _) => Some(format!("{prefix}.pending_limit")), + (_, _, _, _, true, true) => Some(format!("{prefix}.delay_limit")), + (_, _, _, _, _, _) => None, + }; + + Ok(reason) + } +} + +impl Reducer for ActivationWriter { + type Input = Vec; + + type Output = (); + + async fn reduce(&mut self, batch: Self::Input) -> Result<(), anyhow::Error> { + assert!(self.batch.is_none()); + self.batch = Some(batch); + Ok(()) + } + + #[instrument(skip_all)] + async fn flush(&mut self) -> Result, anyhow::Error> { + if self.batch.is_none() { + return Ok(None); + } + + // If batch is empty (all tasks were forwarded), just mark as complete + if self.batch.as_ref().is_some_and(|batch| batch.is_empty()) { + self.batch.take(); + return Ok(Some(())); + } + + let backpressure_reason = { + let batch = self.batch.as_ref().unwrap(); + self.check_backpressure(batch).await? + }; + + if let Some(reason) = backpressure_reason { metrics::counter!( "consumer.inflight_activation_writer.backpressure", "topic" => self.config.topic.clone(), @@ -146,7 +191,10 @@ impl Reducer for ActivationWriter { } let write_to_store_start = Instant::now(); - let res = self.store.store(batch).await; + let res = { + let batch = self.batch.as_ref().unwrap(); + self.store.store(batch).await + }; match res { Ok(entries) => { @@ -220,6 +268,8 @@ mod tests { }; use super::{ActivationWriter, ActivationWriterConfig, Reducer}; + use crate::config::DEFAULT_TOPIC; + use crate::store::types::TopicPartition; #[tokio::test] #[rstest] @@ -232,7 +282,9 @@ mod tests { db_max_size: None, max_buf_len: 100, max_pending_activations: 10, + max_pending_activations_per_topic: None, max_processing_activations: 10, + max_processing_activations_per_topic: None, max_delay_activations: 10, write_failure_backoff_ms: 4000, }; @@ -279,7 +331,9 @@ mod tests { db_max_size: None, max_buf_len: 100, max_pending_activations: 10, + max_pending_activations_per_topic: None, max_processing_activations: 10, + max_processing_activations_per_topic: None, max_delay_activations: 10, write_failure_backoff_ms: 4000, }; @@ -315,7 +369,9 @@ mod tests { db_max_size: None, max_buf_len: 100, max_pending_activations: 0, + max_pending_activations_per_topic: None, max_processing_activations: 10, + max_processing_activations_per_topic: None, max_delay_activations: 10, write_failure_backoff_ms: 4000, }; @@ -356,7 +412,9 @@ mod tests { db_max_size: None, max_buf_len: 100, max_pending_activations: 0, + max_pending_activations_per_topic: None, max_processing_activations: 10, + max_processing_activations_per_topic: None, max_delay_activations: 0, write_failure_backoff_ms: 4000, }; @@ -406,7 +464,9 @@ mod tests { db_max_size: None, max_buf_len: 100, max_pending_activations: 10, + max_pending_activations_per_topic: None, max_processing_activations: 10, + max_processing_activations_per_topic: None, max_delay_activations: 0, write_failure_backoff_ms: 4000, }; @@ -454,7 +514,9 @@ mod tests { db_max_size: None, max_buf_len: 100, max_pending_activations: 10, + max_pending_activations_per_topic: None, max_processing_activations: 1, + max_processing_activations_per_topic: None, max_delay_activations: 0, write_failure_backoff_ms: 4000, }; @@ -513,6 +575,259 @@ mod tests { // writer.store.remove_db().await.unwrap(); } + #[tokio::test] + #[rstest] + #[case::sqlite("sqlite")] + #[case::postgres("postgres")] + async fn test_writer_backpressure_processing_limit_reached_for_topic(#[case] adapter: &str) { + let store = create_test_store(adapter).await; + let writer_config = ActivationWriterConfig { + topic: "taskworker".to_string(), + db_max_size: None, + max_buf_len: 100, + max_pending_activations: 10, + max_pending_activations_per_topic: None, + max_processing_activations: 10, + max_processing_activations_per_topic: Some(1), + max_delay_activations: 0, + write_failure_backoff_ms: 4000, + }; + + let received_at = DateTime::from_timestamp_nanos(0); + let namespace = generate_unique_namespace(); + + let existing_activation = ActivationBuilder::new() + .id("existing") + .taskname("existing_task") + .namespace(&namespace) + .received_at(received_at) + .status(ActivationStatus::Processing) + .build(TaskActivationBuilder::new()); + + store.store(&[existing_activation]).await.unwrap(); + + let mut writer = ActivationWriter::new(store.clone(), writer_config); + let batch = vec![ + ActivationBuilder::new() + .id("0") + .taskname("pending_task") + .namespace(&namespace) + .received_at(received_at) + .build(TaskActivationBuilder::new()), + ActivationBuilder::new() + .id("1") + .taskname("delay_task") + .namespace(&namespace) + .received_at(received_at) + .build(TaskActivationBuilder::new()), + ]; + + writer.reduce(batch).await.unwrap(); + let flush_result = writer.flush().await.unwrap(); + assert!(flush_result.is_none()); + + let count_pending = writer.store.count_pending_activations().await.unwrap(); + assert_eq!(count_pending, 0); + let count_delay = writer + .store + .count_by_status(ActivationStatus::Delay) + .await + .unwrap(); + assert_eq!(count_delay, 0); + let count_processing = writer + .store + .count_by_status(ActivationStatus::Processing) + .await + .unwrap(); + // Only the existing processing activation should remain, new ones should be blocked + assert_eq!(count_processing, 1); + // TODO: Because the store and the writer both access the DB, both need to be cleaned up. + // Uncomment this when we figure out how to do that cleanly. + // writer.store.remove_db().await.unwrap(); + } + + #[tokio::test] + #[rstest] + #[case::sqlite("sqlite")] + #[case::postgres("postgres")] + async fn test_writer_backpressure_pending_limit_reached_for_topic(#[case] adapter: &str) { + let store = create_test_store(adapter).await; + let writer_config = ActivationWriterConfig { + topic: "taskworker".to_string(), + db_max_size: None, + max_buf_len: 100, + max_pending_activations: 10, + max_pending_activations_per_topic: Some(1), + max_processing_activations: 10, + max_processing_activations_per_topic: None, + max_delay_activations: 0, + write_failure_backoff_ms: 4000, + }; + + let received_at = DateTime::from_timestamp_nanos(0); + let namespace = generate_unique_namespace(); + + let existing_activation = ActivationBuilder::new() + .id("existing") + .taskname("existing_task") + .namespace(&namespace) + .received_at(received_at) + .build(TaskActivationBuilder::new()); + + store.store(&[existing_activation]).await.unwrap(); + + let mut writer = ActivationWriter::new(store.clone(), writer_config); + let batch = vec![ + ActivationBuilder::new() + .id("0") + .taskname("pending_task") + .namespace(&namespace) + .received_at(received_at) + .build(TaskActivationBuilder::new()), + ActivationBuilder::new() + .id("1") + .taskname("delay_task") + .namespace(&namespace) + .received_at(received_at) + .build(TaskActivationBuilder::new()), + ]; + + writer.reduce(batch).await.unwrap(); + let flush_result = writer.flush().await.unwrap(); + assert!(flush_result.is_none()); + + let count_pending = writer.store.count_pending_activations().await.unwrap(); + assert_eq!(count_pending, 1); + let count_delay = writer + .store + .count_by_status(ActivationStatus::Delay) + .await + .unwrap(); + assert_eq!(count_delay, 0); + let count_processing = writer + .store + .count_by_status(ActivationStatus::Processing) + .await + .unwrap(); + // Only the existing processing activation should remain, new ones should be blocked + assert_eq!(count_processing, 0); + // TODO: Because the store and the writer both access the DB, both need to be cleaned up. + // Uncomment this when we figure out how to do that cleanly. + // writer.store.remove_db().await.unwrap(); + } + + #[tokio::test] + #[rstest] + // #[case::sqlite("sqlite")] NOTE: SQLite doesn't implement per topic filtering + #[case::postgres("postgres")] + async fn test_writer_backpressure_processing_limit_reached_for_one_topic_but_not_another( + #[case] adapter: &str, + ) { + let store = create_test_store(adapter).await; + store.assign_partitions( + &mut vec![ + TopicPartition::new("taskworker-2", 0), + TopicPartition::new(DEFAULT_TOPIC, 0), + ] + .into_iter(), + ); + + let received_at = DateTime::from_timestamp_nanos(0); + let namespace = generate_unique_namespace(); + + // Another topic has processing activations, but it should not block the default topic. + let existing_activation1 = ActivationBuilder::new() + .id("existing") + .topic("taskworker-2") + .taskname("existing_task") + .namespace(&namespace) + .received_at(received_at) + .status(ActivationStatus::Processing) + .build(TaskActivationBuilder::new()); + let existing_activation2 = ActivationBuilder::new() + .id("existing2") + .topic("taskworker-2") + .taskname("existing_task") + .namespace(&namespace) + .received_at(received_at) + .status(ActivationStatus::Processing) + .build(TaskActivationBuilder::new()); + + store + .store(&[existing_activation1, existing_activation2]) + .await + .unwrap(); + + // The writer on the other topic should block since its processing limit is reached. + let other_writer_config = ActivationWriterConfig { + topic: "taskworker-2".to_string(), + db_max_size: None, + max_buf_len: 100, + max_pending_activations: 10, + max_pending_activations_per_topic: None, + max_processing_activations: 10, + max_processing_activations_per_topic: Some(1), + max_delay_activations: 0, + write_failure_backoff_ms: 4000, + }; + let mut other_writer = ActivationWriter::new(store.clone(), other_writer_config); + let batch = vec![ + ActivationBuilder::new() + .id("0") + .taskname("pending_task") + .topic("taskworker-2") + .namespace(&namespace) + .received_at(received_at) + .build(TaskActivationBuilder::new()), + ]; + + other_writer.reduce(batch).await.unwrap(); + let flush_result = other_writer.flush().await.unwrap(); + assert!(flush_result.is_none()); + + let count_pending = store.count_pending_activations().await.unwrap(); + assert_eq!(count_pending, 0); + let count_processing = store + .count_by_status(ActivationStatus::Processing) + .await + .unwrap(); + assert_eq!(count_processing, 2); + + // The writer on the default topic should not block since its processing limit is not reached. + let writer_config = ActivationWriterConfig { + topic: DEFAULT_TOPIC.to_string(), + db_max_size: None, + max_buf_len: 100, + max_pending_activations: 10, + max_pending_activations_per_topic: None, + max_processing_activations: 10, + max_processing_activations_per_topic: Some(1), + max_delay_activations: 0, + write_failure_backoff_ms: 4000, + }; + let mut writer = ActivationWriter::new(store.clone(), writer_config); + let batch = vec![ + ActivationBuilder::new() + .id("0") + .taskname("pending_task") + .topic(DEFAULT_TOPIC) + .namespace(&namespace) + .received_at(received_at) + .build(TaskActivationBuilder::new()), + ]; + + writer.reduce(batch).await.unwrap(); + let flush_result = writer.flush().await.unwrap(); + assert!(flush_result.is_some()); + + let count_pending = store.count_pending_activations().await.unwrap(); + assert_eq!(count_pending, 1); + + // TODO: Because the store and the writer both access the DB, both need to be cleaned up. + // Uncomment this when we figure out how to do that cleanly. + // writer.store.remove_db().await.unwrap(); + } + #[tokio::test] #[rstest] #[case::sqlite("sqlite")] @@ -525,7 +840,9 @@ mod tests { db_max_size: Some(50_000), max_buf_len: 100, max_pending_activations: 5000, + max_pending_activations_per_topic: None, max_processing_activations: 5000, + max_processing_activations_per_topic: None, max_delay_activations: 0, write_failure_backoff_ms: 4000, }; @@ -557,7 +874,9 @@ mod tests { db_max_size: None, max_buf_len: 100, max_pending_activations: 10, + max_pending_activations_per_topic: None, max_processing_activations: 10, + max_processing_activations_per_topic: None, max_delay_activations: 10, write_failure_backoff_ms: 4000, }; diff --git a/src/store/adapters/postgres.rs b/src/store/adapters/postgres.rs index c5e2bdbc..2a0c96eb 100644 --- a/src/store/adapters/postgres.rs +++ b/src/store/adapters/postgres.rs @@ -796,7 +796,7 @@ impl ActivationStore for PostgresStore { #[instrument(skip_all)] #[framed] - async fn count_depths(&self) -> Result { + async fn count_depths(&self, topic: Option<&str>) -> Result { retry_query(&self.config.retry, "count_depths", || async { // Notice that statuses are embedded into the query for simplicity - if the enum is every changed, this must change too! let mut query_builder = QueryBuilder::new( @@ -808,6 +808,10 @@ impl ActivationStore for PostgresStore { ); self.add_partition_condition(&mut query_builder, true); + if let Some(topic) = topic { + query_builder.push(" AND topic = "); + query_builder.push_bind(topic); + } let row: (i64, i64, i64, i64) = query_builder .build_query_as() diff --git a/src/store/tests.rs b/src/store/tests.rs index 07b1d0fb..a1abec18 100644 --- a/src/store/tests.rs +++ b/src/store/tests.rs @@ -91,6 +91,13 @@ async fn test_store(#[case] adapter: &str) { #[case::postgres("postgres")] async fn test_count_depths(#[case] adapter: &str) { let store = create_test_store(adapter).await; + store.assign_partitions( + &mut vec![ + TopicPartition::new("taskworker-2", 0), + TopicPartition::new(DEFAULT_TOPIC, 0), + ] + .into_iter(), + ); // Check counts for an empty store let pending = store @@ -106,14 +113,17 @@ async fn test_count_depths(#[case] adapter: &str) { .await .unwrap(); - let depths = store.count_depths().await.unwrap(); + let depths = store.count_depths(None).await.unwrap(); assert_eq!(depths.pending, pending); assert_eq!(depths.delay, delay); assert_eq!(depths.processing, processing); + assert_eq!(depths.claimed, 0); // Check counts for a store with four activations with varying statuses - let batch = make_activations(4); + let mut batch = make_activations(4); + batch[0].topic = "taskworker-2".into(); + batch[1].topic = "taskworker-2".into(); assert!(store.store(&batch).await.is_ok()); store @@ -142,14 +152,88 @@ async fn test_count_depths(#[case] adapter: &str) { .await .unwrap(); - let depths = store.count_depths().await.unwrap(); + let depths = store.count_depths(None).await.unwrap(); assert_eq!(depths.pending, pending, "pending"); assert_eq!(depths.delay, delay, "delay"); assert_eq!(depths.processing, processing, "processing"); + assert_eq!(depths.claimed, 0, "claimed"); assert_eq!(pending, 1); assert_eq!(delay, 1); assert_eq!(processing, 1); +} + +#[tokio::test] +#[rstest] +#[case::postgres("postgres")] +async fn test_count_depths_with_topic_filter(#[case] adapter: &str) { + let store = create_test_store(adapter).await; + store.assign_partitions( + &mut vec![ + TopicPartition::new("taskworker-2", 0), + TopicPartition::new(DEFAULT_TOPIC, 0), + ] + .into_iter(), + ); + + // Check counts for a store with activations with varying statuses + let mut batch = make_activations(5); + batch[0].topic = "taskworker-2".into(); + batch[1].topic = "taskworker-2".into(); + assert!(store.store(&batch).await.is_ok()); + + store + .set_status("id_0", ActivationStatus::Processing, None, None) + .await + .unwrap(); + store + .set_status("id_1", ActivationStatus::Delay, None, None) + .await + .unwrap(); + store + .set_status("id_2", ActivationStatus::Complete, None, None) + .await + .unwrap(); + store + .set_status("id_3", ActivationStatus::Claimed, None, None) + .await + .unwrap(); + + let pending = store + .count_by_status(ActivationStatus::Pending) + .await + .unwrap(); + let delay = store + .count_by_status(ActivationStatus::Delay) + .await + .unwrap(); + let processing = store + .count_by_status(ActivationStatus::Processing) + .await + .unwrap(); + let claimed = store + .count_by_status(ActivationStatus::Claimed) + .await + .unwrap(); + + let depths = store.count_depths(None).await.unwrap(); + + assert_eq!(depths.pending, pending, "pending"); + assert_eq!(depths.delay, delay, "delay"); + assert_eq!(depths.processing, processing, "processing"); + assert_eq!(depths.claimed, claimed, "claimed"); + assert_eq!(pending, 1); + assert_eq!(delay, 1); + assert_eq!(processing, 1); + assert_eq!(claimed, 1); + + let topic_depths = store.count_depths(Some("taskworker-2")).await.unwrap(); + + // These activations were processing/delayed + assert_eq!(topic_depths.pending, 0, "pending"); + assert_eq!(topic_depths.delay, 1, "delay"); + assert_eq!(topic_depths.processing, 1, "processing"); + assert_eq!(topic_depths.claimed, 0, "claimed"); store.remove_db().await.unwrap(); } diff --git a/src/store/traits.rs b/src/store/traits.rs index 4fbe56c6..267dc419 100644 --- a/src/store/traits.rs +++ b/src/store/traits.rs @@ -125,7 +125,7 @@ pub trait ActivationStore: Send + Sync { /// Queue depths for pending, delay, and processing (writer backpressure and upkeep gauges). /// Default implementation uses separate calls, but stores may override with a single query. - async fn count_depths(&self) -> Result { + async fn count_depths(&self, _topic: Option<&str>) -> Result { let (pending, delay, claimed, processing) = join!( self.count_by_status(ActivationStatus::Pending), self.count_by_status(ActivationStatus::Delay), @@ -147,7 +147,7 @@ pub trait ActivationStore: Send + Sync { async fn count_depths_per_partition( &self, ) -> Result, Error> { - let total = self.count_depths().await?; + let total = self.count_depths(None).await?; Ok(HashMap::from([( TopicPartition::new(crate::config::DEFAULT_TOPIC, -1), total, From d9dc9ef0e7ad6f9519145a93ab2cb42bd8038cc1 Mon Sep 17 00:00:00 2001 From: Evan Hicks Date: Mon, 31 Aug 2026 15:13:40 -0400 Subject: [PATCH 2/3] remove unused config --- src/config/store.rs | 5 ----- 1 file changed, 5 deletions(-) diff --git a/src/config/store.rs b/src/config/store.rs index 182a3b80..7082a27e 100644 --- a/src/config/store.rs +++ b/src/config/store.rs @@ -173,10 +173,6 @@ pub struct StoreConfig { /// in the ActivationStore (sqlite) pub max_processing_count: usize, - /// A dictionary of topic-specific maximum number of processing records that can be - /// in the ActivationStore - pub max_processing_count_per_topic: HashMap, - /// The maximum number of times a task can be reset from /// processing back to pending. When this limit is reached, /// the activation will be discarded/deadlettered. @@ -209,7 +205,6 @@ impl Default for StoreConfig { max_pending_count: 2048, max_delay_count: 8192, max_processing_count: 2048, - max_processing_count_per_topic: HashMap::new(), max_processing_attempts: 5, processing_deadline_grace_sec: 3, contention_drain_age_sec: 60, From 76bc0355a3f88153c89d4872220f638dc4581244 Mon Sep 17 00:00:00 2001 From: Evan Hicks Date: Mon, 31 Aug 2026 15:16:35 -0400 Subject: [PATCH 3/3] import --- src/config/store.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/src/config/store.rs b/src/config/store.rs index 7082a27e..bef22143 100644 --- a/src/config/store.rs +++ b/src/config/store.rs @@ -2,7 +2,6 @@ use std::time::Duration; use anyhow::Result; use serde::{Deserialize, Serialize}; -use std::collections::HashMap; use tracing::warn; use crate::config::Config;