fix(multipart): Use correct discard reasons

CHANGELOG.md

@@ -6,6 +6,10 @@
 
 - Enable OTLP endpoints by default. ([#5951](https://github.com/getsentry/relay/pull/5951))
 
+**Bug Fixes**:
+
+- Emit more precise outcome discard reasons for the Playstation, Minidump, and Attachments endpoints. ([#5950](https://github.com/getsentry/relay/pull/5950))
+
 ## 26.4.2
 
 **Features**:

relay-server/src/endpoints/attachments.rs

@@ -5,6 +5,7 @@ use axum::routing::{MethodRouter, post};
 use multer::{Field, Multipart};
 use relay_config::Config;
 use relay_event_schema::protocol::EventId;
+use relay_quotas::DataCategory;
 use serde::Deserialize;
 use tower_http::limit::RequestBodyLimitLayer;
 
@@ -52,8 +53,12 @@ async fn multipart_to_envelope(
     )
     .await?;
 
-    let envelope =
-        common::managed_items_to_envelope(items, meta, state.outcome_aggregator(), path.event_id);
+    let envelope = items.map(|items, records| {
+        if items.iter().any(|i| i.creates_event()) {
+            records.modify_by(DataCategory::Error, 1);
+        }
+        Box::new(Envelope::from_request(Some(path.event_id), meta).with_items(items))
+    });
     Ok(envelope)
 }
 

relay-server/src/endpoints/common.rs

@@ -17,13 +17,12 @@ use serde::Deserialize;
 
 use crate::envelope::{
     AttachmentPlaceholder, AttachmentType, ContentType, Envelope, EnvelopeError, Item, ItemType,
-    ManagedItems,
+    Items,
 };
-use crate::extractors::RequestMeta;
 use crate::managed::{Managed, Rejected};
 use crate::service::ServiceState;
 use crate::services::buffer::{ProjectKeyPair, PushError};
-use crate::services::outcome::{DiscardItemType, DiscardReason, Outcome, TrackOutcome};
+use crate::services::outcome::{DiscardItemType, DiscardReason, Outcome};
 use crate::services::processor::{BucketSource, MetricData, ProcessMetrics};
 use crate::services::upload::{Create, Stream, Upload};
 use crate::statsd::{RelayCounters, RelayDistributions};
@@ -125,6 +124,29 @@ pub enum BadStoreRequest {
     ObjectstoreUploadFailed,
 }
 
+impl BadStoreRequest {
+    pub fn to_outcome(&self) -> Option<Outcome> {
+        let discard_reason = match self {
+            Self::InvalidCompressionContainer(_) => DiscardReason::InvalidCompression,
+            Self::InvalidMinidump => DiscardReason::InvalidMinidump,
+            #[cfg(sentry)]
+            Self::InvalidProsperodump => DiscardReason::InvalidProsperodump,
+            Self::MissingMinidump => DiscardReason::MissingMinidumpUpload,
+            #[cfg(sentry)]
+            Self::MissingProsperodump => DiscardReason::MissingProsperodumpUpload,
+            Self::Overflow(item_type) => DiscardReason::TooLarge(*item_type),
+            _ => DiscardReason::Internal,
+        };
+        Some(Outcome::Invalid(discard_reason))
+    }
+}
+
+impl From<Rejected<BadStoreRequest>> for BadStoreRequest {
+    fn from(rejected: Rejected<BadStoreRequest>) -> Self {
+        rejected.into_inner()
+    }
+}
+
 impl From<BytesRejection> for BadStoreRequest {
     fn from(value: BytesRejection) -> Self {
         match value {
@@ -292,7 +314,7 @@ pub fn event_id_from_formdata(data: &[u8]) -> Result<Option<EventId>, BadStoreRe
 ///
 /// Extracting the event id from chunked formdata fields on the Minidump endpoint (`sentry__1`,
 /// `sentry__2`, ...) is not supported. In this case, `None` is returned.
-pub fn event_id_from_items(items: &ManagedItems) -> Result<Option<EventId>, BadStoreRequest> {
+pub fn event_id_from_items(items: &Items) -> Result<Option<EventId>, BadStoreRequest> {
     if let Some(item) = items.iter().find(|item| item.ty() == &ItemType::Event)
         && let Some(event_id) = event_id_from_json(&item.payload())?
     {
@@ -575,27 +597,6 @@ where
     Some(item)
 }
 
-pub fn managed_items_to_envelope(
-    items: ManagedItems,
-    meta: RequestMeta,
-    outcome_aggregator: &Addr<TrackOutcome>,
-    event_id: EventId,
-) -> Managed<Box<Envelope>> {
-    let envelope = Envelope::from_request(Some(event_id), meta);
-    let mut envelope = Managed::from_envelope(envelope, outcome_aggregator.clone());
-    let mut has_event = false;
-    for item in items {
-        envelope.merge_with(item, |envelope, item, records| {
-            if !has_event && item.creates_event() {
-                records.modify_by(DataCategory::Error, 1);
-                has_event = true;
-            }
-            envelope.add_item(item);
-        });
-    }
-    envelope
-}
-
 #[derive(Debug)]
 pub struct TextResponse(pub Option<EventId>);
 

relay-server/src/endpoints/minidump.rs

@@ -24,7 +24,7 @@ use crate::endpoints::common::{self, BadStoreRequest, TextResponse, upload_to_ob
 use crate::envelope::ContentType::Minidump;
 use crate::envelope::{AttachmentType, Envelope, Item, ItemType};
 use crate::extractors::{RawContentType, RequestMeta};
-use crate::managed::Managed;
+use crate::managed::{Managed, ManagedResult};
 use crate::middlewares;
 use crate::service::ServiceState;
 use crate::services::outcome::{DiscardAttachmentType, DiscardItemType, DiscardReason, Outcome};
@@ -119,9 +119,8 @@ fn decode_minidump(minidump_data: Bytes, max_size: usize) -> Result<Bytes, BadSt
     match run_decoder(decoder) {
         Ok(decoded) => {
             if decoded.len() > max_size {
-                return Err(BadStoreRequest::Overflow(DiscardItemType::Attachment(
-                    DiscardAttachmentType::Minidump,
-                )));
+                let item_type = DiscardItemType::Attachment(DiscardAttachmentType::Minidump);
+                return Err(BadStoreRequest::Overflow(item_type));
             }
             Ok(Bytes::from(decoded))
         }
@@ -280,36 +279,37 @@ async fn multipart_to_envelope(
     )
     .await?;
 
-    let minidump_item = items
-        .iter_mut()
-        .find(|item| item.attachment_type() == Some(AttachmentType::Minidump))
-        .ok_or(BadStoreRequest::MissingMinidump)?;
+    let minidump_idx = items
+        .iter()
+        .position(|item| item.attachment_type() == Some(AttachmentType::Minidump))
+        .ok_or(BadStoreRequest::MissingMinidump)
+        .reject(&items)?;
 
     // Doing these operations does not make sense if we already streamed the minidump to objectstore.
-    if !minidump_item.is_attachment_ref() {
-        let payload = minidump_item.payload();
+    if !items[minidump_idx].is_attachment_ref() {
+        let payload = items[minidump_idx].payload();
         let payload = extract_embedded_minidump(payload.clone())
             .await?
             .unwrap_or(payload);
-        let payload = decode_minidump(payload, config.max_attachment_size())?;
+        let payload = decode_minidump(payload, config.max_attachment_size()).reject(&items)?;
 
-        minidump_item.modify(|inner, records| {
-            inner.set_payload(Minidump, payload);
-            records.lenient(DataCategory::Attachment);
-        });
-
-        validate_minidump(&minidump_item.payload())?;
+        validate_minidump(&payload).reject(&items)?;
 
-        minidump_item.modify(|inner, _| {
-            if let Some(minidump_filename) = inner.filename() {
-                inner.set_filename(remove_container_extension(minidump_filename).to_owned())
+        items.modify(|items, records| {
+            let minidump_item = &mut items[minidump_idx];
+            minidump_item.set_payload(Minidump, payload);
+            records.lenient(DataCategory::Attachment); // decoding the minidump changes its size
+            if let Some(minidump_filename) = minidump_item.filename() {
+                minidump_item.set_filename(remove_container_extension(minidump_filename).to_owned())
             }
         });
     }
 
     let event_id = common::event_id_from_items(&items)?.unwrap_or_else(EventId::new);
-    let envelope =
-        common::managed_items_to_envelope(items, meta, state.outcome_aggregator(), event_id);
+    let envelope = items.map(|items, records| {
+        records.modify_by(DataCategory::Error, 1);
+        Box::new(Envelope::from_request(Some(event_id), meta).with_items(items))
+    });
     Ok(envelope)
 }
 
@@ -396,7 +396,6 @@ async fn raw_minidump_to_envelope(
     item.set_filename(MINIDUMP_FILE_NAME);
     item.set_attachment_type(AttachmentType::Minidump);
     let mut item = Managed::with_meta_from_request_meta(&meta, state.outcome_aggregator(), item);
-
     if let Some(upload_context) = upload_context
         && matches!(upload_context.upload_minidumps, UploadDecision::Upload)
     {
@@ -415,20 +414,19 @@ async fn raw_minidump_to_envelope(
         let decoded_payload = decode_minidump(
             request.extract().await?,
             state.config().max_attachment_size(),
-        )?;
+        )
+        .reject(&item)?;
+        validate_minidump(&decoded_payload).reject(&item)?;
         item.modify(|inner, records| {
             inner.set_payload(Minidump, decoded_payload);
             records.lenient(DataCategory::Attachment); // decoding the minidump changes its size
         });
-        validate_minidump(&item.payload())?;
     };
 
     // Create an envelope with a random event id.
-    let envelope = Envelope::from_request(Some(EventId::new()), meta);
-    let mut envelope = Managed::from_envelope(envelope, state.outcome_aggregator().clone());
-    envelope.merge_with(item, |envelope, item, records| {
+    let envelope = item.map(|item, records| {
         records.modify_by(DataCategory::Error, 1);
-        envelope.add_item(item);
+        Box::new(Envelope::from_request(Some(EventId::new()), meta).with_items(vec![item]))
     });
     Ok(envelope)
 }

relay-server/src/endpoints/playstation.rs

@@ -189,16 +189,22 @@ async fn multipart_to_envelope(
     )
     .await?;
 
-    let prosperodump_item = items
-        .iter_mut()
-        .find(|item| item.attachment_type() == Some(AttachmentType::Prosperodump))
-        .ok_or(BadStoreRequest::MissingProsperodump)?;
-    prosperodump_item.modify(|inner, _| inner.set_payload(OctetStream, inner.payload()));
-    validate_prosperodump(&prosperodump_item.payload())?;
+    items.try_modify(|inner, _| -> Result<(), BadStoreRequest> {
+        let prosperodump = inner
+            .iter_mut()
+            .find(|item| item.attachment_type() == Some(AttachmentType::Prosperodump))
+            .ok_or(BadStoreRequest::MissingProsperodump)?;
+        let payload = prosperodump.payload();
+        validate_prosperodump(&payload)?;
+        prosperodump.set_payload(OctetStream, payload);
+        Ok(())
+    })?;
 
     let event_id = common::event_id_from_items(&items)?.unwrap_or_else(EventId::new);
-    let envelope =
-        common::managed_items_to_envelope(items, meta, state.outcome_aggregator(), event_id);
+    let envelope = items.map(|items, records| {
+        records.modify_by(DataCategory::Error, 1);
+        Box::new(Envelope::from_request(Some(event_id), meta).with_items(items))
+    });
     Ok(envelope)
 }
 

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::managed::Managed;
 use crate::statsd::RelayTimers;
 
 #[derive(Clone, Debug)]
@@ -770,7 +769,6 @@ impl Item {
     }
 }
 
-pub type ManagedItems = SmallVec<[Managed<Item>; 3]>;
 pub type Items = SmallVec<[Item; 3]>;
 pub type ItemIter<'a> = std::slice::Iter<'a, Item>;
 pub type ItemIterMut<'a> = std::slice::IterMut<'a, Item>;

relay-server/src/envelope/mod.rs

@@ -551,6 +551,14 @@ impl Envelope {
         self.items.push(item)
     }
 
+    /// Add new items and return `Self`.
+    pub fn with_items(mut self, items: impl IntoIterator<Item = Item>) -> Self {
+        for item in items {
+            self.items.push(item)
+        }
+        self
+    }
+
     /// Splits off the items from the envelope using provided predicates.
     ///
     /// First predicate is the additional condition on the count of found items by second

relay-server/src/managed/managed.rs

@@ -16,6 +16,7 @@ use relay_system::Addr;
 use smallvec::SmallVec;
 
 use crate::Envelope;
+use crate::endpoints::common::BadStoreRequest;
 use crate::extractors::RequestMeta;
 use crate::managed::{Counted, ManagedEnvelope, Quantities};
 use crate::services::outcome::{DiscardReason, Outcome, TrackOutcome};
@@ -88,6 +89,14 @@ impl OutcomeError for Infallible {
     }
 }
 
+impl OutcomeError for BadStoreRequest {
+    type Error = Self;
+
+    fn consume(self) -> (Option<Outcome>, Self) {
+        (self.to_outcome(), self)
+    }
+}
+
 /// A wrapper type which ensures outcomes have been emitted for an error.
 ///
 /// [`Managed`] wraps an error in [`Rejected`] once outcomes for have been emitted for the managed

relay-server/src/services/outcome.rs

@@ -377,6 +377,12 @@ pub enum DiscardReason {
     /// since it resolves the project first, and then checks for the valid project key.
     MultiProjectId,
 
+    /// (Relay) A request without a prosperodump was made to the playstation endpoint.
+    MissingProsperodumpUpload,
+
+    /// (Relay) The prosperodump submitted to the playstation endpoint was invalid.
+    InvalidProsperodump,
+
     /// (Relay) A minidump file was missing for the minidump endpoint.
     MissingMinidumpUpload,
 
@@ -509,6 +515,8 @@ impl DiscardReason {
             DiscardReason::DisallowedMethod => "disallowed_method",
             DiscardReason::ContentType => "content_type",
             DiscardReason::MultiProjectId => "multi_project_id",
+            DiscardReason::MissingProsperodumpUpload => "missing_prosperodump_upload",
+            DiscardReason::InvalidProsperodump => "invalid_prosperodump",
             DiscardReason::MissingMinidumpUpload => "missing_minidump_upload",
             DiscardReason::InvalidMinidump => "invalid_minidump",
             DiscardReason::SecurityReportType => "security_report_type",