ref(processing): Use new transactions processor

CHANGELOG.md

@@ -1,6 +1,6 @@
 # Changelog
 
 ## Unreleased
 
 **Bug Fixes**:
 
@@ -28,6 +28,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/managed.rs

@@ -87,6 +87,11 @@ impl OutcomeError for Infallible {
 pub struct Rejected<T>(T);
 
 impl<T> Rejected<T> {
+    /// Returns a reference to the underlying error.
+    pub fn inner(&self) -> &T {
+        &self.0
+    }
+
     /// Extracts the underlying error.
     pub fn into_inner(self) -> T {
         self.0

relay-server/src/processing/common.rs

@@ -8,12 +8,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)
@@ -59,4 +61,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,15 @@ 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));
+        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 +74,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 +89,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 +101,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 +120,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.