ref(processor): Pass transactions through new pipeline

relay-server/src/envelope/mod.rs

@@ -196,6 +196,16 @@ impl<M> EnvelopeHeaders<M> {
         }
     }
 
+    /// Overrides the dynamic sampling context in envelope headers.
+    pub fn set_dsc(&mut self, dsc: DynamicSamplingContext) {
+        self.trace = Some(ErrorBoundary::Ok(dsc));
+    }
+
+    /// Removes the dynamic sampling context from envelope headers.
+    pub fn remove_dsc(&mut self) {
+        self.trace = None;
+    }
+
     /// Returns the timestamp when the event has been sent, according to the SDK.
     pub fn sent_at(&self) -> Option<DateTime<Utc>> {
         self.sent_at
@@ -439,15 +449,15 @@ impl Envelope {
         self.headers.dsc()
     }
 
-    /// Overrides the dynamic sampling context in envelope headers.
-    pub fn set_dsc(&mut self, dsc: DynamicSamplingContext) {
-        self.headers.trace = Some(ErrorBoundary::Ok(dsc));
-    }
+    // /// Overrides the dynamic sampling context in envelope headers.
+    // pub fn set_dsc(&mut self, dsc: DynamicSamplingContext) {
+    //     self.headers.set_dsc(dsc);
+    // }
 
-    /// Removes the dynamic sampling context from envelope headers.
-    pub fn remove_dsc(&mut self) {
-        self.headers.trace = None;
-    }
+    // /// Removes the dynamic sampling context from envelope headers.
+    // pub fn remove_dsc(&mut self) {
+    //     self.headers.remove_dsc();
+    // }
 
     /// Features required to process this envelope.
     pub fn required_features(&self) -> &[Feature] {
@@ -1221,7 +1231,7 @@ mod tests {
             );
             *Envelope::parse_bytes(bytes).unwrap()
         };
-        envelope.set_dsc(dsc.clone());
+        envelope.headers.set_dsc(dsc.clone());
 
         assert_eq!(
             envelope.dsc().unwrap().transaction.as_ref().unwrap(),

relay-server/src/managed/counted.rs

@@ -1,3 +1,5 @@
+use std::collections::BTreeMap;
+
 use relay_event_schema::protocol::{
     OurLog, SessionAggregateItem, SessionAggregates, SessionUpdate, Span, SpanV2, TraceMetric,
 };
@@ -29,6 +31,15 @@ impl Counted for () {
     }
 }
 
+impl<T: Counted> Counted for Option<T> {
+    fn quantities(&self) -> Quantities {
+        match self {
+            Some(inner) => inner.quantities(),
+            None => Quantities::new(),
+        }
+    }
+}
+
 impl Counted for Item {
     fn quantities(&self) -> Quantities {
         self.quantities()
@@ -175,3 +186,18 @@ where
         self.as_ref().quantities()
     }
 }
+
+impl<T> Counted for Vec<T>
+where
+    T: Counted,
+{
+    fn quantities(&self) -> Quantities {
+        let mut quantities = BTreeMap::new();
+        for element in self {
+            for (category, size) in element.quantities() {
+                *quantities.entry(category).or_default() += size;
+            }
+        }
+        quantities.into_iter().collect()
+    }
+}

relay-server/src/processing/mod.rs

@@ -27,6 +27,7 @@ pub mod logs;
 pub mod sessions;
 pub mod spans;
 pub mod trace_metrics;
+pub mod transactions;
 pub mod utils;
 
 /// A processor, for an arbitrary unit of work extracted from an envelope.
@@ -47,7 +48,9 @@ pub trait Processor {
     /// Extracts a [`Self::UnitOfWork`] from a [`ManagedEnvelope`].
     ///
     /// This is infallible, if a processor wants to report an error,
-    /// it should return a [`Self::UnitOfWork`] which later, can produce an error when being
+    /// it should return a [`Self::UnitOfWork`] which later, can produce an error when being processed.
+    ///
+    /// Returns `None` if nothing in the envelope concerns this processor.
     fn prepare_envelope(&self, envelope: &mut ManagedEnvelope)
     -> Option<Managed<Self::UnitOfWork>>;
 

relay-server/src/processing/spans/mod.rs

@@ -22,7 +22,7 @@ mod filter;
 mod integrations;
 mod process;
 #[cfg(feature = "processing")]
-mod store;
+pub mod store;
 mod validate;
 
 type Result<T, E = Error> = std::result::Result<T, E>;