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
@@ -441,12 +451,12 @@ impl Envelope {
 
     /// Overrides the dynamic sampling context in envelope headers.
     pub fn set_dsc(&mut self, dsc: DynamicSamplingContext) {
-        self.headers.trace = Some(ErrorBoundary::Ok(dsc));
+        self.headers.set_dsc(dsc);
     }
 
     /// Removes the dynamic sampling context from envelope headers.
     pub fn remove_dsc(&mut self) {
-        self.headers.trace = None;
+        self.headers.remove_dsc();
     }
 
     /// Features required to process this envelope.
@@ -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,27 @@ impl Counted for () {
     }
 }
 
+impl<T: Counted> Counted for Option<T> {
+    fn quantities(&self) -> Quantities {
+        match self {
+            Some(inner) => inner.quantities(),
+            None => Quantities::new(),
+        }
+    }
+}
+
+impl<T, S> Counted for (T, S)
+where
+    T: Counted,
+    S: Counted,
+{
+    fn quantities(&self) -> Quantities {
+        let mut v = self.0.quantities();
+        v.extend(self.1.quantities());
+        v
+    }
+}
+
 impl Counted for Item {
     fn quantities(&self) -> Quantities {
         self.quantities()
@@ -175,3 +198,27 @@ where
         self.as_ref().quantities()
     }
 }
+
+impl<T: Counted> Counted for Vec<T> {
+    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()
+    }
+}
+
+impl<T: Counted, const N: usize> Counted for SmallVec<[T; N]> {
+    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/managed/managed.rs

@@ -15,7 +15,9 @@ use relay_system::Addr;
 use smallvec::SmallVec;
 
 use crate::Envelope;
+use crate::envelope::Item;
 use crate::managed::{Counted, ManagedEnvelope, Quantities};
+use crate::processing::CountRateLimited;
 use crate::services::outcome::{DiscardReason, Outcome, TrackOutcome};
 use crate::services::processor::ProcessingError;
 
@@ -152,6 +154,21 @@ impl<T: Counted> Managed<T> {
         Managed::from_parts(other, Arc::clone(&self.meta))
     }
 
+    /// Merge the current managed item with another managed item.
+    ///
+    /// The merged tuple uses the meta of `self`. It is the responsibility of the caller to make sure
+    /// that this matches the `meta` of `other`.
+    pub fn merge<S>(self, other: Managed<S>) -> Managed<(T, S)>
+    where
+        S: Counted,
+    {
+        let (this, meta) = self.destructure();
+        let (other, other_meta) = other.destructure();
+        debug_assert!(Arc::ptr_eq(&meta, &other_meta));
+
+        Managed::from_parts((this, other), meta)
+    }
+
     /// Original received timestamp.
     pub fn received_at(&self) -> DateTime<Utc> {
         self.meta.received_at
@@ -162,6 +179,13 @@ impl<T: Counted> Managed<T> {
         self.meta.scoping
     }
 
+    /// Get the address of the outcome aggregator.
+    ///
+    /// NOTE: This should not be exposed, only here for the transition period.
+    pub fn outcome_addr(&self) -> &Addr<TrackOutcome> {
+        &self.meta.outcome_aggregator
+    }
+
     /// Splits [`Self`] into two other [`Managed`] items.
     ///
     /// The two resulting managed instances together are expected to have the same outcomes as the original instance..
@@ -560,6 +584,13 @@ impl<T: Counted + fmt::Debug> fmt::Debug for Managed<T> {
     }
 }
 
+impl<T: Counted> Managed<Option<T>> {
+    pub fn transpose(self) -> Option<Managed<T>> {
+        let (o, meta) = self.destructure();
+        o.map(|t| Managed::from_parts(t, meta))
+    }
+}
+
 impl From<Managed<Box<Envelope>>> for ManagedEnvelope {
     fn from(value: Managed<Box<Envelope>>) -> Self {
         let (value, meta) = value.destructure();

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>;