feat(consumer): add a DLQ-by-age load-shedding lever

rust_snuba/benches/processors.rs

@@ -104,6 +104,7 @@ fn create_factory(
         join_timeout_ms: None,
         health_check: "arroyo".to_string(),
         use_row_binary: false,
+        dlq_configured: false,
     };
     Box::new(factory)
 }

rust_snuba/src/consumer.rs

@@ -193,6 +193,10 @@ pub fn consumer_impl(
     // DLQ policy applies only if we are not skipping writes, otherwise we don't want to be
     // writing to the DLQ topics in prod.
 
+    // Whether a DLQ topic is configured. The DLQ-by-age strategy relies on the
+    // DLQ policy, so it is only wired into the factory when this is true.
+    let dlq_configured = consumer_config.dlq_topic.is_some();
+
     let dlq_policy = consumer_config.dlq_topic.map(|dlq_topic_config| {
         let producer = KafkaProducer::new(KafkaConfig::new_producer_config(
             vec![],
@@ -278,6 +282,7 @@ pub fn consumer_impl(
         join_timeout_ms,
         health_check: health_check.to_string(),
         use_row_binary,
+        dlq_configured,
     };
 
     let processor = StreamProcessor::with_kafka(config, factory, topic, dlq_policy);

rust_snuba/src/factory_v2.rs

@@ -27,6 +27,7 @@ use crate::strategies::accountant::RecordCogs;
 
 use crate::strategies::clickhouse::writer_v2::{ClickhouseWriterStep, InsertFormat};
 use crate::strategies::commit_log::ProduceCommitLog;
+use crate::strategies::dlq_by_age::DlqByAge;
 use crate::strategies::healthcheck::HealthCheck as SnubaHealthCheck;
 use crate::strategies::join_timeout::SetJoinTimeout;
 use crate::strategies::processor::{
@@ -62,6 +63,10 @@ pub struct ConsumerStrategyFactoryV2 {
     pub join_timeout_ms: Option<u64>,
     pub health_check: String,
     pub use_row_binary: bool,
+    // Whether this consumer has a DLQ topic configured. The DLQ-by-age strategy
+    // is only inserted when true, since without a DLQ policy an InvalidMessage
+    // is logged and silently dropped by arroyo (losing data).
+    pub dlq_configured: bool,
 }
 
 impl ProcessingStrategyFactory<KafkaPayload> for ConsumerStrategyFactoryV2 {
@@ -288,6 +293,16 @@ impl ProcessingStrategyFactory<KafkaPayload> for ConsumerStrategyFactoryV2 {
             Some(Duration::from_millis(self.join_timeout_ms.unwrap_or(0))),
         );
 
+        // Shed a configurable proportion of too-old messages to the DLQ so a
+        // backlogged consumer can catch up to fresh data. Only wired when a DLQ
+        // is configured (see DlqByAge docs / the `dlq_configured` field);
+        // disabled per-storage by default via sentry-options.
+        let next_step: Box<dyn ProcessingStrategy<KafkaPayload>> = if self.dlq_configured {
+            Box::new(DlqByAge::new(next_step, self.storage_config.name.clone()))
+        } else {
+            Box::new(next_step)
+        };
+
         if let Some(path) = &self.health_check_file {
             {
                 if self.health_check == "snuba" {