fix(outcomes): Account for rejected envelope items

CHANGELOG.md

@@ -15,6 +15,10 @@
 
 - Make `spool.envelopes.max_unspool_envelopes_per_second` hot-reloadable. ([#6452](https://github.com/getsentry/relay/pull/6452))
 
+**Internal**:
+
+- Correct rate-limit outcome accounting for rejected envelope items, including V1 user reports (`user_report`). ([#6526](https://github.com/getsentry/relay/pull/6526))
+
 ## 26.10.0
 
 **Breaking Changes**:

relay-server/src/endpoints/common.rs

@@ -465,7 +465,6 @@ pub async fn handle_managed_envelope(
         .project_cache_handle()
         .get(project_key)
         .check_envelope(&mut envelope)
-        .await
         .map_err(|err| err.map(BadStoreRequest::EventRejected))?;
 
     if envelope.is_empty() {

relay-server/src/endpoints/upload.rs

@@ -511,7 +511,6 @@ async fn validate_and_limit(
     let mut envelope = Managed::from_envelope(envelope, state.outcome_aggregator().clone());
     let rate_limits = project
         .check_envelope(&mut envelope)
-        .await
         .map_err(|err| err.map(BadStoreRequest::EventRejected).into_inner())?
         .propagatable();
 
@@ -557,7 +556,6 @@ async fn validate(
 
     let _ = project
         .check_envelope(&mut envelope)
-        .await
         .map_err(|err| err.map(BadStoreRequest::EventRejected).into_inner())?;
 
     // We are not really processing an envelope here, only keep the updated scoping:

relay-server/src/envelope/item.rs

@@ -12,7 +12,6 @@ use smallvec::{SmallVec, smallvec};
 
 use crate::envelope::{AttachmentType, ContentType, EnvelopeError};
 use crate::integrations::{Integration, LogsIntegration, SpansIntegration};
-use crate::statsd::RelayTimers;
 use crate::utils::DebugBytes;
 
 #[derive(Clone)]
@@ -108,10 +107,13 @@ impl Item {
                 smallvec![
                     (DataCategory::Transaction, item_count),
                     (DataCategory::TransactionIndexed, item_count),
-                    (DataCategory::Span, item_count + self.span_count() as usize),
+                    (
+                        DataCategory::Span,
+                        item_count + self.span_count().unwrap_or(0) as usize,
+                    ),
                     (
                         DataCategory::SpanIndexed,
-                        item_count + self.span_count() as usize,
+                        item_count + self.span_count().unwrap_or(0) as usize,
                     ),
                 ]
             }
@@ -219,38 +221,22 @@ impl Item {
 
     /// Returns the number of spans in the `event.spans` array.
     ///
-    /// Should always be 0 except for transaction items.
+    /// Only transaction items are expected to carry this header. Returns `None` if the header is
+    /// absent or invalid.
     ///
     /// When a transaction is dropped before spans were extracted from a transaction,
     /// this number is used to emit correct outcomes for the spans category.
     ///
     /// This number does *not* count the transaction itself.
-    pub fn span_count(&self) -> u32 {
-        self.headers.get(ItemHeaderKey::SpanCount).unwrap_or(0)
+    pub fn span_count(&self) -> Option<u32> {
+        self.headers.get(ItemHeaderKey::SpanCount)
     }
 
     /// Sets the number of spans in the transaction payload.
     pub fn set_span_count(&mut self, value: Option<usize>) {
         self.headers.set_or_remove(ItemHeaderKey::SpanCount, value);
     }
 
-    /// Sets the `span_count` item header by shallow parsing the event.
-    ///
-    /// Returns the recomputed count.
-    fn refresh_span_count(&mut self) -> usize {
-        let count = self.parse_span_count();
-        self.set_span_count(count);
-        count.unwrap_or(0)
-    }
-
-    /// Returns the `span_count`` header, and computes it if it has not yet been set.
-    pub fn ensure_span_count(&mut self) -> usize {
-        match self.headers.get(ItemHeaderKey::SpanCount) {
-            Some(count) => count,
-            None => self.refresh_span_count(),
-        }
-    }
-
     /// Returns the content type of this item's payload.
     pub fn content_type(&self) -> Option<ContentType> {
         self.headers.get(ItemHeaderKey::ContentType)
@@ -723,23 +709,6 @@ impl Item {
     pub fn is_container(&self) -> bool {
         self.content_type().is_some_and(ContentType::is_container)
     }
-
-    fn parse_span_count(&self) -> Option<usize> {
-        #[derive(Debug, serde::Deserialize)]
-        struct PartialEvent {
-            spans: crate::utils::SeqCount,
-        }
-
-        if self.headers.ty != ItemType::Transaction {
-            return None;
-        }
-
-        let event = relay_statsd::metric!(timer(RelayTimers::CheckNestedSpans), {
-            serde_json::from_slice::<PartialEvent>(&self.payload()).ok()?
-        });
-
-        Some(event.spans.0)
-    }
 }
 
 impl fmt::Debug for Item {

relay-server/src/envelope/mod.rs

@@ -542,6 +542,11 @@ impl Envelope {
         self.items.drain_filter(|item| cond(item)).collect()
     }
 
+    /// Replaces the item buffer, returning the previous buffer without copying its contents.
+    pub fn replace_items(&mut self, items: Items) -> Items {
+        std::mem::replace(&mut self.items, items)
+    }
+
     /// Adds a new item to this envelope.
     pub fn add_item(&mut self, item: Item) {
         self.items.push(item)

relay-server/src/managed/counted.rs

@@ -90,19 +90,16 @@ impl Counted for EnvelopeSummary {
     fn quantities(&self) -> Quantities {
         let mut quantities = Quantities::new();
 
-        if let Some(category) = self.event_category {
+        if let Some(category) = self.event.category() {
             quantities.push((category, 1));
             if let Some(category) = category.index_category() {
                 quantities.push((category, 1));
             }
         }
 
         let data = [
-            (DataCategory::Attachment, self.attachment_quantities.bytes()),
-            (
-                DataCategory::AttachmentItem,
-                self.attachment_quantities.count(),
-            ),
+            (DataCategory::Attachment, self.attachment_bytes),
+            (DataCategory::AttachmentItem, self.attachment_count),
             (DataCategory::Profile, self.profile_quantity.total),
             (DataCategory::ProfileBackend, self.profile_quantity.backend),
             (DataCategory::ProfileUi, self.profile_quantity.ui),

relay-server/src/managed/envelope.rs

@@ -6,7 +6,7 @@ use chrono::{DateTime, Utc};
 use relay_quotas::{DataCategory, Scoping};
 use relay_system::Addr;
 
-use crate::envelope::{Envelope, Item};
+use crate::envelope::Envelope;
 use crate::extractors::RequestMeta;
 use crate::managed::Counted as _;
 use crate::services::outcome::{DiscardReason, Outcome, TrackOutcome};
@@ -42,17 +42,6 @@ impl Handling {
     }
 }
 
-/// Represents the decision on whether or not to keep an envelope item.
-#[derive(Debug, Clone)]
-pub enum ItemAction {
-    /// Keep the item.
-    Keep,
-    /// Drop the item and log an outcome for it.
-    Drop(Outcome),
-    /// Drop the item without logging an outcome.
-    DropSilently,
-}
-
 #[derive(Debug)]
 struct EnvelopeContext {
     summary: EnvelopeSummary,
@@ -143,51 +132,12 @@ impl ManagedEnvelope {
         self
     }
 
-    /// Retains or drops items based on the [`ItemAction`].
-    ///
-    ///
-    /// This method operates in place and preserves the order of the retained items.
-    pub fn retain_items<F>(&mut self, mut f: F)
-    where
-        F: FnMut(&mut Item) -> ItemAction,
-    {
-        let mut outcomes = Vec::new();
-        self.envelope.retain_items(|item| match f(item) {
-            ItemAction::Keep => true,
-            ItemAction::DropSilently => false,
-            ItemAction::Drop(outcome) => {
-                for (category, quantity) in item.quantities() {
-                    outcomes.push((outcome.clone(), category, quantity));
-                }
-
-                false
-            }
-        });
-        for (outcome, category, quantity) in outcomes {
-            self.track_outcome(outcome, category, quantity);
-        }
-        // TODO: once `update` is private, it should be called here.
-    }
-
     /// Re-scopes this context to the given scoping.
     pub fn scope(&mut self, scoping: Scoping) -> &mut Self {
         self.context.scoping = scoping;
         self
     }
 
-    /// Removes event item(s) and logs an outcome.
-    ///
-    /// Note: This function relies on the envelope summary being correct.
-    pub fn reject_event(&mut self, outcome: Outcome) {
-        if let Some(event_category) = self.event_category() {
-            self.envelope.retain_items(|item| !item.creates_event());
-            if let Some(indexed) = event_category.index_category() {
-                self.track_outcome(outcome.clone(), indexed, 1);
-            }
-            self.track_outcome(outcome, event_category, 1);
-        }
-    }
-
     /// Records an outcome scoped to this envelope's context.
     ///
     /// This managed envelope should be updated using [`update`](Self::update) soon after this
@@ -215,11 +165,6 @@ impl ManagedEnvelope {
         }
     }
 
-    /// Returns the data category of the event item in the envelope.
-    fn event_category(&self) -> Option<DataCategory> {
-        self.context.summary.event_category
-    }
-
     /// Records rejection outcomes for all items stored in this context.
     ///
     /// This does not send outcomes for empty envelopes or request-only contexts.
@@ -238,7 +183,7 @@ impl ManagedEnvelope {
 
                 relay_log::error!(
                     tags.project_key = self.scoping().project_key.to_string(),
-                    tags.has_attachments = summary.attachment_quantities.bytes() > 0,
+                    tags.has_attachments = summary.attachment_bytes > 0,
                     tags.has_sessions = summary.session_quantity > 0,
                     tags.has_profiles = summary.profile_quantity.total > 0,
                     tags.has_transactions = summary.secondary_transaction_quantity > 0,
@@ -247,7 +192,7 @@ impl ManagedEnvelope {
                     tags.has_user_reports = summary.user_report_quantity > 0,
                     tags.has_trace_metrics = summary.trace_metric_quantity > 0,
                     tags.has_checkins = summary.monitor_quantity > 0,
-                    tags.event_category = ?summary.event_category,
+                    tags.event_category = ?summary.event.category(),
                     cached_summary = ?summary,
                     recomputed_summary = ?EnvelopeSummary::compute(self.envelope()),
                     "dropped envelope: {outcome}"

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

@@ -92,15 +92,10 @@ impl Processor for TransactionProcessor {
     fn prepare_envelope(&self, envelope: &mut ManagedEnvelope) -> Option<Managed<Self::Input>> {
         let headers = envelope.envelope().headers().clone();
 
-        let mut event = envelope
+        let event = envelope
             .envelope_mut()
             .take_item_by(|item| matches!(*item.ty(), ItemType::Transaction))?;
 
-        // Count number of spans by shallow-parsing the event.
-        // Needed for accounting but not in prod, because the event is immediately parsed afterwards.
-        #[cfg(debug_assertions)]
-        event.ensure_span_count();
-
         let attachments = envelope
             .envelope_mut()
             .take_items_by(|item| matches!(*item.ty(), ItemType::Attachment))