ref(processing): Use new transactions processor

CHANGELOG.md

@@ -17,6 +17,7 @@
 **Internal**:
 
 - Derive the rate limiting decision in Relay from consumed quota. ([#5390](https://github.com/getsentry/relay/pull/5390))
+- Use new processor architecture to process transactions. ([#5379](https://github.com/getsentry/relay/pull/5379))
 
 ## 25.11.0
 

relay-server/src/managed/counted.rs

@@ -61,6 +61,10 @@ impl Counted for Box<Envelope> {
 
         let data = [
             (DataCategory::Attachment, summary.attachment_quantity),
+            (
+                DataCategory::AttachmentItem,
+                summary.attachment_item_quantity,
+            ),
             (DataCategory::Profile, summary.profile_quantity),
             (DataCategory::ProfileIndexed, summary.profile_quantity),
             (DataCategory::Span, summary.span_quantity),

relay-server/src/metrics/rate_limits.rs

@@ -187,11 +187,13 @@ impl<Q: AsRef<Vec<Quota>>> MetricsLimiter<Q> {
         metric_outcomes: &MetricOutcomes,
     ) -> bool {
         for category in [DataCategory::Transaction, DataCategory::Span] {
-            let active_rate_limits =
-                rate_limits.check_with_quotas(self.quotas.as_ref(), self.scoping.item(category));
+            let active_rate_limits = rate_limits.check_with_quotas(
+                dbg!(self.quotas.as_ref()),
+                self.scoping.item(dbg!(category)),
+            );
 
             // If a rate limit is active, discard relevant buckets.
-            if let Some(limit) = active_rate_limits.longest() {
+            if let Some(limit) = dbg!(active_rate_limits).longest() {
                 self.drop_with_outcome(
                     category,
                     Outcome::RateLimited(limit.reason_code.clone()),

relay-server/src/processing/common.rs

@@ -6,12 +6,14 @@ use crate::processing::logs::LogsProcessor;
 use crate::processing::sessions::SessionsProcessor;
 use crate::processing::spans::SpansProcessor;
 use crate::processing::trace_metrics::TraceMetricsProcessor;
+use crate::processing::transactions::TransactionProcessor;
 use crate::processing::{Forward, Processor};
 
 macro_rules! outputs {
     ($($variant:ident => $ty:ty,)*) => {
         /// All known [`Processor`] outputs.
         #[derive(Debug)]
+        #[allow(clippy::large_enum_variant)]
         pub enum Outputs {
             $(
                 $variant(<$ty as Processor>::Output)
@@ -57,4 +59,5 @@ outputs!(
     TraceMetrics => TraceMetricsProcessor,
     Spans => SpansProcessor,
     Sessions => SessionsProcessor,
+    Transactions => TransactionProcessor,
 );

relay-server/src/processing/mod.rs

@@ -78,7 +78,10 @@ pub struct Context<'a> {
     ///
     /// The caller needs to ensure the rate limits are not yet expired.
     pub rate_limits: &'a RateLimits,
-    /// Reservoir counters for "get more samples" functionality.
+
+    /// Counters used for getting more samples for a project on-demand.
+    ///
+    /// Reservoir counters are a legacy feature and will be removed in the near future.
     pub reservoir_counters: &'a ReservoirCounters,
 }
 

relay-server/src/processing/transactions/extraction.rs

@@ -57,12 +57,16 @@ pub fn extract_metrics(
         metrics_extracted,
         spans_extracted,
     } = ctx;
+    // TODO(follow-up): this function should always extract metrics. Dynamic sampling should validate
+    // the full metrics extraction config and skip sampling if it is incomplete.
 
     if metrics_extracted {
-        return Ok(EventMetricsExtracted(metrics_extracted));
+        debug_assert!(false, "metrics extraction called twice");
+        return Ok(EventMetricsExtracted(true));
     }
     let Some(event) = event.value_mut() else {
-        return Ok(EventMetricsExtracted(metrics_extracted));
+        // Nothing to extract, but metrics extraction was called.
+        return Ok(EventMetricsExtracted(true));
     };
 
     // NOTE: This function requires a `metric_extraction` in the project config. Legacy configs
@@ -71,7 +75,7 @@ pub fn extract_metrics(
     let combined_config = {
         let config = match &ctx.project_info.config.metric_extraction {
             ErrorBoundary::Ok(config) if config.is_supported() => config,
-            _ => return Ok(EventMetricsExtracted(metrics_extracted)),
+            _ => return Ok(EventMetricsExtracted(false)),
         };
         let global_config = match &ctx.global_config.metric_extraction {
             ErrorBoundary::Ok(global_config) => global_config,
@@ -86,7 +90,7 @@ pub fn extract_metrics(
                     // If there's an error with global metrics extraction, it is safe to assume that this
                     // Relay instance is not up-to-date, and we should skip extraction.
                     relay_log::debug!("Failed to parse global extraction config: {e}");
-                    return Ok(EventMetricsExtracted(metrics_extracted));
+                    return Ok(EventMetricsExtracted(false));
                 }
             }
         };
@@ -98,11 +102,11 @@ pub fn extract_metrics(
         Some(ErrorBoundary::Ok(tx_config)) => tx_config,
         Some(ErrorBoundary::Err(e)) => {
             relay_log::debug!("Failed to parse legacy transaction metrics config: {e}");
-            return Ok(EventMetricsExtracted(metrics_extracted));
+            return Ok(EventMetricsExtracted(false));
         }
         None => {
             relay_log::debug!("Legacy transaction metrics config is missing");
-            return Ok(EventMetricsExtracted(metrics_extracted));
+            return Ok(EventMetricsExtracted(false));
         }
     };
 
@@ -117,7 +121,7 @@ pub fn extract_metrics(
                 }
             });
 
-        return Ok(EventMetricsExtracted(metrics_extracted));
+        return Ok(EventMetricsExtracted(false));
     }
 
     // If spans were already extracted for an event, we rely on span processing to extract metrics.