feat(upload): Revert use of multipart uploads

Cargo.lock

@@ -3366,7 +3366,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d76659c9da42fc50c394edda51e9d69dc6b3a6a770029e50f8e267c85cbfe69f"
 dependencies = [
  "async-compression",
- "base64",
  "bytes",
  "futures-util",
  "infer",

Cargo.toml

@@ -162,8 +162,8 @@ metrics = "0.24"
 metrics-exporter-dogstatsd = "0.9"
 num-traits = "0.2"
 num_cpus = "1"
-objectstore-client = { version = "0.2.1", features = ["multipart"] }
-objectstore-types = { version = "0.2.0" }
+objectstore-client = { version = "0.2" }
+objectstore-types = { version = "0.2" }
 opentelemetry-semantic-conventions = "0.31"
 opentelemetry-proto = { version = "0.31", default-features = false }
 papaya = "0.2"

relay-dynamic-config/src/feature.rs

@@ -92,11 +92,6 @@ pub enum Feature {
     /// Stream minidumps to objectstore.
     #[serde(rename = "projects:relay-minidump-uploads")]
     MinidumpUploads,
-    /// Use objectstore multipart for upload requests.
-    ///
-    /// See <https://getsentry.github.io/objectstore/rust/objectstore_service/multipart/>.
-    #[serde(rename = "projects:relay-upload-multipart")]
-    UploadMultipart,
     /// Split an NVIDIA GPU crash dump (`.nv-gpudmp`) off a minidump upload into its
     /// own event.
     #[serde(rename = "organizations:gpu-crash-symbolication")]

relay-server/src/endpoints/common.rs

@@ -596,7 +596,6 @@ where
             project: project.clone(),
             length: None,
             attachment_type: item.attachment_type(),
-            multipart: false,
         })
         .await
         .map_err(|_| BadStoreRequest::UploadFailed)?

relay-server/src/endpoints/upload.rs

@@ -185,18 +185,12 @@ async fn handle_post(
             StatusCode::SERVICE_UNAVAILABLE
         })?;
 
-    let multipart = match project.state() {
-        ProjectState::Enabled(p) => p.has_feature(Feature::UploadMultipart),
-        _ => false,
-    };
-
     relay_log::trace!("Checking request");
     let project_context = validate_and_limit(&state, meta, &headers, project).await?;
 
     // Unconditionally create the upload location:
     relay_log::trace!("Creating upload location");
-
-    let result = create(&state, project_context, &headers, multipart).await;
+    let result = create(&state, project_context, &headers).await;
     let location = result.inspect_err(|e| {
         relay_log::warn!(error = e as &dyn std::error::Error, "create failed");
     })?;
@@ -319,15 +313,13 @@ async fn create(
     state: &ServiceState,
     project: ProjectContext,
     headers: &tus::Headers,
-    multipart: bool,
 ) -> Result<SignedLocation<Provisional>, Error> {
     let location = state
         .upload()
         .send(upload::Create {
             project,
             length: headers.upload_length,
             attachment_type: headers.metadata.map(|m| m.attachment_type),
-            multipart,
         })
         .await??;
 

relay-server/src/services/objectstore.rs

@@ -2,21 +2,19 @@
 use std::array::TryFromSliceError;
 use std::borrow::Cow;
 use std::fmt;
-use std::num::{NonZeroU16, NonZeroUsize};
+use std::num::NonZeroU16;
 use std::sync::Arc;
 use std::time::Duration;
 
-use async_compression::tokio::bufread::ZstdEncoder;
 use bytes::Bytes;
 use futures::StreamExt;
 use http::StatusCode;
 use mime::Mime;
 use objectstore_client::{
-    Client, Compression, ExpirationPolicy, SecretKey as SigningKey, Session, TokenGenerator,
-    UploadId, Usecase,
+    Client, ExpirationPolicy, SecretKey as SigningKey, Session, TokenGenerator, Usecase,
 };
 
-use objectstore_types::multipart::InvalidUploadId;
+use objectstore_types::multipart::{InvalidUploadId, UploadId};
 use relay_base_schema::organization::OrganizationId;
 use relay_base_schema::project::ProjectId;
 use relay_config::ObjectstoreServiceConfig;
@@ -25,7 +23,6 @@ use relay_system::{
     Addr, AsyncResponse, FromMessage, Interface, LoadShed, NoResponse, Sender, SimpleService,
 };
 use sentry_protos::snuba::v1::{AnyValue, TraceItem, any_value};
-use tokio_util::io::{ReaderStream, StreamReader};
 
 use crate::constants::DEFAULT_ATTACHMENT_RETENTION;
 use crate::envelope::{ContentType, Item, ItemType};
@@ -38,22 +35,17 @@ use crate::services::store::{
 };
 use crate::services::upload::ByteStream;
 use crate::statsd::{RelayCounters, RelayTimers};
-use crate::utils::{
-    BoundedStream, MeteredStream, Rechunk, RetryableStream, TakeOnce, find_error_source,
-};
+use crate::utils::{BoundedStream, MeteredStream, RetryableStream, TakeOnce, find_error_source};
 
 use super::outcome::Outcome;
 
-/// Size of an individual request to objectstore.
-const CHUNK_SIZE: NonZeroUsize = NonZeroUsize::new(5 * 1024 * 1024).unwrap();
-
 /// Messages that the objectstore service can handle.
 pub enum Objectstore {
     Event(Managed<Box<StoreEvent>>),
     TraceAttachment(Managed<StoreTraceAttachment>),
     EventAttachment(Managed<StoreAttachment>),
     RawProfile(Managed<StoreRawProfile>),
-    Create(CreateMultipart, Sender<Result<UploadRef, Error>>),
+    Create(Create, Sender<Result<UploadRef, Error>>),
     Stream(Stream, Sender<Result<ObjectstoreKey, Error>>),
 }
 
@@ -140,7 +132,7 @@ impl MessageKind {
 }
 
 /// A request to create a new objectstore multipart upload.
-pub struct CreateMultipart {
+pub struct Create {
     /// The sentry org.
     pub organization_id: OrganizationId,
     /// The sentry project.
@@ -151,10 +143,10 @@ pub struct CreateMultipart {
     pub retention: u16,
 }
 
-impl FromMessage<CreateMultipart> for Objectstore {
+impl FromMessage<Create> for Objectstore {
     type Response = AsyncResponse<Result<UploadRef, Error>>;
 
-    fn from_message(message: CreateMultipart, sender: Sender<Result<UploadRef, Error>>) -> Self {
+    fn from_message(message: Create, sender: Sender<Result<UploadRef, Error>>) -> Self {
         Self::Create(message, sender)
     }
 }
@@ -342,7 +334,7 @@ pub struct UploadRef {
     /// They key of the file (chosen by relay).
     pub key: String,
     /// The ID of the multipart upload session (chosen by objectstore).
-    /// `None` if the upload is not multipart.
+    /// `None` if the upload is not a resumable session.
     pub upload_id: Option<UploadId>,
 }
 
@@ -821,31 +813,21 @@ impl ObjectstoreServiceInner {
         Ok(Some(stored_key))
     }
 
-    async fn handle_create(&self, create: CreateMultipart) -> Result<UploadRef, Error> {
-        let CreateMultipart {
+    async fn handle_create(&self, create: Create) -> Result<UploadRef, Error> {
+        let Create {
             organization_id,
             project_id,
             key,
-            retention,
+            retention: _,
         } = create;
-        let session = self.session(&self.event_attachments, organization_id, project_id)?;
+        let _session = self.session(&self.event_attachments, organization_id, project_id)?;
 
-        let multipart_upload = session
-            .initiate_multipart_upload()
-            .expiration_policy(ExpirationPolicy::TimeToLive(Duration::from_hours(
-                u64::from(retention) * 24,
-            )))
-            .key(&key)
-            .compression(Compression::Zstd) // make explicit because parts need to be manually compressed.
-            .send()
-            .await?;
-        debug_assert_eq!(&key, multipart_upload.key());
-
-        let upload_id = multipart_upload.upload_id();
+        // This is intentionally a stub. Once Objectstore implements resumable uploads,
+        // create an upload session here.
 
         Ok(UploadRef {
             key,
-            upload_id: Some(upload_id.clone()),
+            upload_id: None,
         })
     }
 
@@ -991,66 +973,17 @@ impl ObjectstoreServiceInner {
                 upload_ref,
                 retention,
             } => {
-                let UploadRef { key, upload_id } = upload_ref;
-                let Some(upload_id) = upload_id else {
-                    // No upload ID: simple upload in a single request.
-                    let request = session.put_stream(body.boxed()).key(key);
-                    let response = request
-                        .expiration_policy(ExpirationPolicy::TimeToLive(Duration::from_hours(
-                            u64::from(retention) * 24,
-                        )))
-                        .send()
-                        .await?;
-                    return Ok(ObjectstoreKey(response.key));
-                };
-
-                let multipart_upload =
-                    session.resume_multipart_upload(key, upload_id.to_string())?;
+                let UploadRef { key, upload_id: _ } = upload_ref;
 
-                let body = ReaderStream::new(ZstdEncoder::new(StreamReader::new(body)));
+                let request = session.put_stream(body.boxed()).key(key);
+                let response = request
+                    .expiration_policy(ExpirationPolicy::TimeToLive(Duration::from_hours(
+                        u64::from(retention) * 24,
+                    )))
+                    .send()
+                    .await?;
 
-                // Unfortunately, MinIO has the limitation that the length of a multipart request
-                // has to be known. Therefore, we need to materialize the stream into concrete
-                // chunks of bytes and send each chunk as an individual request.
-                let chunks = Rechunk::new(body, CHUNK_SIZE);
-                let mut body = chunks.enumerate();
-
-                let result = relay_statsd::metric!(
-                    timer(RelayTimers::AttachmentUploadDuration),
-                    type = kind.as_str(),
-                {
-                    let mut parts = vec![];
-                    // NOTE: Once every upload is a multipart upload, we can remove `RetryableStream`
-                    // because streams will never be effectively retried.
-                    while let Some((i, chunk)) = body.next().await {
-                        let chunk = chunk?;
-                        let part_number = u32::try_from(i + 1)
-                        .map_err(|_| objectstore_client::Error::InvalidPartNumber(u32::MAX))?;
-                        relay_log::trace!("Part number {part_number}");
-
-                        // NOTE: This is a retry loop within a retry loop (see caller of this function).
-                        // if we keep the Rechunked approach we might as well remove the outer loop for streaming uploads.
-                        let mut attempts = 0;
-                        let part = loop {
-                            let result = multipart_upload.put(chunk.clone(), part_number, None).await;
-                            attempts += 1;
-                            if attempts < self.max_attempts.get()
-                                && matches!(&result, Err(e) if is_retryable(e))
-                            {
-                                relay_log::trace!("Attempt {attempts}: Failed with {result:?}, retrying");
-                                tokio::time::sleep(self.retry_interval).await;
-                            } else {
-                                relay_log::trace!("Final attempt");
-                                break result;
-                            }
-                        };
-
-                        parts.push(part?);
-                    }
-                    multipart_upload.complete(parts).await?
-                });
-
-                Ok(ObjectstoreKey(result))
+                Ok(ObjectstoreKey(response.key))
             }
         }
     }

relay-server/src/services/upload.rs

@@ -138,8 +138,6 @@ pub struct Create {
     pub length: Option<usize>,
     /// The attachment type of the upload.
     pub attachment_type: Option<AttachmentType>,
-    /// Whether multipart uploads should be used for this upload.
-    pub multipart: bool,
 }
 
 /// The type used to stream a request body.
@@ -271,12 +269,10 @@ impl Service {
             project,
             length,
             attachment_type,
-            multipart,
         }: Create,
     ) -> Result<SignedLocation<Provisional>, Error> {
         match &self.backend {
             Backend::Upstream { addr } => {
-                let _ = multipart; // upstream will check feature flag again, no need to propagate.
                 let (request, rx) = UploadRequest::create(project, length, attachment_type);
                 addr.send(SendRequest(request));
                 let response = rx.await??;
@@ -297,13 +293,11 @@ impl Service {
                     ..
                 } = project.scoping;
 
-                let (key, upload_id) = match (multipart, length) {
-                    // We should only create a multipart upload in objectstore if it was requested,
-                    // and if the upload actually has data (multipart does not allow empty parts).
-                    (false, _) | (_, Some(0)) => (key, None),
+                let (key, upload_id) = match length {
+                    Some(0) => (key, None), // multipart does not allow empty uploads
                     _ => {
                         let UploadRef { key, upload_id } = addr
-                            .send(objectstore::CreateMultipart {
+                            .send(objectstore::Create {
                                 organization_id,
                                 project_id,
                                 key,

relay-server/src/utils/stream/mod.rs

@@ -1,13 +1,9 @@
 mod bounded;
 mod metered;
 mod peek;
-#[cfg(any(feature = "processing", test))]
-mod rechunked;
 mod retryable;
 
 pub use bounded::*;
 pub use metered::*;
 pub use peek::*;
-#[cfg(feature = "processing")]
-pub use rechunked::*;
 pub use retryable::*;