ref(upload): Explicitly reject items when an objectstore upload fails

relay-server/src/endpoints/common.rs

@@ -19,7 +19,7 @@ use crate::envelope::{
     AttachmentPlaceholder, AttachmentType, ContentType, Envelope, EnvelopeError, Item, ItemType,
     Items,
 };
-use crate::managed::{Managed, Rejected};
+use crate::managed::{Managed, ManagedResult, Rejected};
 use crate::service::ServiceState;
 use crate::services::buffer::{ProjectKeyPair, PushError};
 use crate::services::outcome::{DiscardItemType, DiscardReason, Outcome};
@@ -543,7 +543,36 @@ pub async fn upload_to_objectstore<S, E>(
     scoping: Scoping,
     upload: &Addr<Upload>,
     referrer: &'static str,
-) -> Option<Managed<Item>>
+) -> Result<Managed<Item>, Rejected<()>>
+where
+    S: futures::Stream<Item = Result<Bytes, E>> + Send + 'static,
+    E: Into<Box<dyn std::error::Error + Send + Sync>> + 'static,
+{
+    let res = upload_to_objectstore_inner(
+        stream,
+        content_type,
+        &mut item,
+        config,
+        scoping,
+        upload,
+        referrer,
+    )
+    .await;
+    match res {
+        Some(()) => Ok(item),
+        None => Err(Outcome::Invalid(DiscardReason::Internal)).reject(&item),
+    }
+}
+
+async fn upload_to_objectstore_inner<S, E>(
+    stream: S,
+    content_type: Option<String>,
+    item: &mut Managed<Item>,
+    config: &Config,
+    scoping: Scoping,
+    upload: &Addr<Upload>,
+    referrer: &'static str,
+) -> Option<()>
 where
     S: futures::Stream<Item = Result<Bytes, E>> + Send + 'static,
     E: Into<Box<dyn std::error::Error + Send + Sync>> + 'static,
@@ -597,7 +626,7 @@ where
         inner.set_attachment_length(byte_counter.get());
         records.lenient(DataCategory::Attachment); // item was empty before
     });
-    Some(item)
+    Some(())
 }
 
 #[derive(Debug)]

relay-server/src/endpoints/minidump.rs

@@ -245,7 +245,8 @@ impl<'a> AttachmentStrategy for MinidumpAttachmentStrategy<'a> {
                     upload_context.upload,
                     "minidump",
                 )
-                .await)
+                .await
+                .ok())
             }
             UploadDecision::Inline => read_inline(field, item).await,
             UploadDecision::Drop(limits) => {
@@ -419,7 +420,7 @@ async fn raw_minidump_to_envelope(
             "minidump",
         )
         .await
-        .ok_or(BadStoreRequest::ObjectstoreUploadFailed)?;
+        .map_err(|_| BadStoreRequest::ObjectstoreUploadFailed)?;
     } else {
         let minidump_data = request.extract().await?;
         item.try_modify(|inner, records| -> Result<(), BadStoreRequest> {

relay-server/src/endpoints/playstation.rs

@@ -149,7 +149,8 @@ impl<'a> AttachmentStrategy for PlaystationAttachmentStrategy<'a> {
                     upload_context.upload,
                     "playstation",
                 )
-                .await)
+                .await
+                .ok())
             }
             _ => match utils::read_bytes_into_item(field, item, config).await {
                 // Don't bubble up errors caused by large attachments, skip over them and continue

tests/integration/test_minidump.py

@@ -1,6 +1,7 @@
 import os
 from unittest import mock
 
+from flask import Response
 import msgpack
 
 import json
@@ -984,6 +985,66 @@ def test_minidump_objectstore_uploads(
         assert minidump.payload.bytes == minidump_content
 
 
+def test_minidump_objectstore_errors(
+    mini_sentry,
+    relay,
+):
+    project_id = 42
+    minidump_content = b"MDMP content"
+    log_content = b"Some log file content"
+
+    project_config = mini_sentry.add_full_project_config(project_id)
+    project_config["config"].setdefault("features", []).append(
+        "projects:relay-minidump-attachment-uploads"
+    )
+
+    @mini_sentry.app.route("/api/<project>/upload/", methods=["POST"])
+    def create(**opts):
+
+        return Response(
+            "Hell no",
+            status=400,
+        )
+
+    relay = relay(
+        mini_sentry,
+        options={
+            "outcomes": {
+                "emit_outcomes": True,
+                "batch_size": 1,
+                "batch_interval": 1,
+            }
+        },
+    )
+
+    relay.send_minidump(
+        project_id=project_id,
+        files=[
+            (MINIDUMP_ATTACHMENT_NAME, "minidump.dmp", minidump_content),
+            ("logs", "log.txt", log_content),
+        ],
+    )
+
+    mini_sentry.captured_envelopes.get()
+
+    assert mini_sentry.get_aggregated_outcomes() == [
+        {
+            "category": DataCategory.ATTACHMENT,
+            "outcome": 3,  # invalid
+            "project_id": 42,
+            "reason": "internal",
+            "quantity": 1,
+        },
+        {
+            "category": DataCategory.ATTACHMENT_ITEM,
+            "outcome": 3,  # invalid
+            "project_id": 42,
+            "reason": "internal",
+            "quantity": 1,
+        },
+    ]
+
+
 def test_size_limits_multipart_chunked(mini_sentry, relay):
 
     project_id = 42