diff --git a/CHANGELOG.md b/CHANGELOG.md index 3e189794024..96875e646b7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ - No longer write the deprecated `sentry.transaction` and `db.system` attributes. ([#6237](https://github.com/getsentry/relay/pull/6237), [#6238](https://github.com/getsentry/relay/pull/6238)) - Allow additional exceptions in minidump and apple crash report events. ([#6241](https://github.com/getsentry/relay/pull/6241)) +**Internal**: + +- Support resumable uploads (feature-flagged). ([#6203](https://github.com/getsentry/relay/pull/6203)) + ## 26.7.0 **Features**: diff --git a/relay-server/src/endpoints/common.rs b/relay-server/src/endpoints/common.rs index 7c82aad262e..b358e0f5993 100644 --- a/relay-server/src/endpoints/common.rs +++ b/relay-server/src/endpoints/common.rs @@ -24,7 +24,7 @@ use crate::service::ServiceState; use crate::services::buffer::{ProjectKeyPair, PushError}; use crate::services::outcome::{DiscardItemType, DiscardReason, Outcome}; use crate::services::processor::{BucketSource, MetricData, ProcessMetrics}; -use crate::services::upload::{Create, ProjectContext, Stream, Upload}; +use crate::services::upload::{Create, ProjectContext, Stream, StreamResult, Upload}; use crate::statsd::{RelayCounters, RelayDistributions}; use crate::utils::{ self, ApiErrorResponse, BoundedStream, FormDataIter, MeteredStream, find_error_source, @@ -613,12 +613,16 @@ where received: Utc::now(), project, location, + offset: 0, stream, }) .await .ok()?; - let location = result + let StreamResult { + location, + offset: _, + } = result .inspect_err(|e| { relay_log::warn!( error = e as &dyn std::error::Error, diff --git a/relay-server/src/endpoints/upload.rs b/relay-server/src/endpoints/upload.rs index 5686ee0a5f8..909102c175e 100644 --- a/relay-server/src/endpoints/upload.rs +++ b/relay-server/src/endpoints/upload.rs @@ -31,8 +31,8 @@ use crate::services::objectstore; use crate::services::projects::cache::Project; use crate::services::projects::project::ProjectState; use crate::services::upload::{ - self, ByteStream, Final, LocationQueryParams, ProjectContext, Provisional, SignedLocation, - UploadLength, + self, ByteStream, LocationQueryParams, ProjectContext, Provisional, SignedLocation, + StreamResult, UploadLength, }; use crate::services::upstream::UpstreamRequestError; use crate::statsd::RelayCounters; @@ -56,6 +56,9 @@ enum Error { #[error("TUS protocol error: {0}")] Tus(#[from] tus::Error), + #[error("Invalid Upload-Offset {0} for Upload-Length {1}")] + InvalidOffset(usize, usize), + #[error("request error: {0}")] Request(#[from] BadStoreRequest), @@ -80,6 +83,7 @@ impl IntoResponse for Error { let status = match self { Error::Tus(_) => StatusCode::BAD_REQUEST, + Error::InvalidOffset(_, _) => StatusCode::BAD_REQUEST, Error::Request(error) => return error.into_response(), Error::SendError(_) => StatusCode::INTERNAL_SERVER_ERROR, Error::Upload(error) => match error { @@ -100,19 +104,30 @@ impl IntoResponse for Error { Some(status) => status, None => StatusCode::INTERNAL_SERVER_ERROR, }, - upload::Error::InvalidLocation(_) | upload::Error::SigningFailed => { + upload::Error::InvalidFromUpstream { .. } | upload::Error::SigningFailed => { StatusCode::INTERNAL_SERVER_ERROR } + upload::Error::InvalidFromClient { .. } => StatusCode::BAD_REQUEST, #[cfg(feature = "processing")] upload::Error::InvalidUploadId(_) => StatusCode::BAD_REQUEST, + upload::Error::OffsetWithoutLength => StatusCode::BAD_REQUEST, upload::Error::SerializeFailed(_) => StatusCode::INTERNAL_SERVER_ERROR, upload::Error::InvalidSignature(_) => StatusCode::BAD_REQUEST, upload::Error::ObjectstoreServiceUnavailable(_) => StatusCode::SERVICE_UNAVAILABLE, #[cfg(feature = "processing")] upload::Error::Objectstore(service_error) => match service_error.kind { objectstore::ErrorKind::InvalidScoping => StatusCode::INTERNAL_SERVER_ERROR, + objectstore::ErrorKind::InvalidOffset { .. } => StatusCode::BAD_REQUEST, + objectstore::ErrorKind::OffsetWithoutUploadId => StatusCode::BAD_REQUEST, + objectstore::ErrorKind::InvalidUploadLength { .. } => StatusCode::BAD_REQUEST, + objectstore::ErrorKind::RequestTooSmall { .. } => StatusCode::BAD_REQUEST, + objectstore::ErrorKind::UnalignedBody { .. } => StatusCode::BAD_REQUEST, + objectstore::ErrorKind::UnknownKey { .. } => StatusCode::NOT_FOUND, objectstore::ErrorKind::Timeout(_) => StatusCode::GATEWAY_TIMEOUT, objectstore::ErrorKind::LoadShed => StatusCode::SERVICE_UNAVAILABLE, + objectstore::ErrorKind::CompressionFailed(_) => { + StatusCode::INTERNAL_SERVER_ERROR + } objectstore::ErrorKind::UploadFailed(error) => match error { objectstore_client::Error::Io(error) if is_upload_length_error(&error) => { StatusCode::BAD_REQUEST @@ -184,10 +199,12 @@ async fn handle_post( StatusCode::SERVICE_UNAVAILABLE })?; - let multipart = match project.state() { - ProjectState::Enabled(p) => p.has_feature(Feature::UploadMultipart), - _ => false, - }; + // We don't enable multipart for `Upload-Defer-Length: 1`, or if the feature flag is disabled. + let multipart = headers.upload_length.is_some() + && 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?; @@ -224,7 +241,7 @@ async fn handle_patch( check_kill_switch(&state)?; relay_log::trace!("Validating headers"); - tus::validate_patch_headers(&headers).map_err(Error::from)?; + let offset = tus::validate_patch_headers(&headers).map_err(Error::from)?; let location = SignedLocation::from_parts( project_id, @@ -259,35 +276,35 @@ async fn handle_patch( let stream = MeteredStream::new(stream, "upload"); let (lower_bound, upper_bound) = match upload_length.value() { - None => (1, config.max_upload_size()), - Some(u) => (u, u), + None => (0, config.max_upload_size()), + Some(u) => { + let remaining_bytes = u + .checked_sub(offset) + .ok_or(Error::InvalidOffset(offset, u))?; + (0, remaining_bytes) + } }; let stream = BoundedStream::new(stream, lower_bound, upper_bound); - let byte_counter = stream.byte_counter(); relay_log::trace!("Uploading"); - let result = upload(&state, project_context, location, stream).await; - let location = result.inspect_err(|e| { + let result = upload(&state, project_context, location, offset, stream).await; + let StreamResult { location, offset } = result.inspect_err(|e| { relay_log::warn!(error = e as &dyn std::error::Error, "upload failed"); })?; - let upload_offset = byte_counter.get(); - let mut response = NoContent.into_response(); // Not required by TUS, but we respond with the location header: - response.headers_mut().insert( - header::LOCATION, - location - .into_header_value() - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?, - ); + let location = location + .into_header_value() + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + response.headers_mut().insert(header::LOCATION, location); response .headers_mut() .insert(tus::TUS_RESUMABLE, tus::TUS_VERSION); response .headers_mut() - .insert(tus::UPLOAD_OFFSET, upload_offset.into()); + .insert(tus::UPLOAD_OFFSET, offset.into()); Ok(response) } @@ -337,19 +354,21 @@ async fn upload( state: &ServiceState, project: ProjectContext, location: SignedLocation, + offset: usize, stream: BoundedStream>, -) -> Result, Error> { - let location = state +) -> Result { + let res = state .upload() .send(upload::Stream { received: Utc::now(), project, location, + offset, stream, }) .await??; - Ok(location) + Ok(res) } /// Check request by converting it into a pseudo-envelope. diff --git a/relay-server/src/processing/utils/attachments.rs b/relay-server/src/processing/utils/attachments.rs index 4b8bc0af450..cc3fac0f090 100644 --- a/relay-server/src/processing/utils/attachments.rs +++ b/relay-server/src/processing/utils/attachments.rs @@ -52,6 +52,13 @@ fn validate(item: &Item, config: &Config) -> Result<(), ProcessingError> { serde_json::from_slice(&payload).map_err(|_| ProcessingError::InvalidAttachmentRef)?; let signed_location: SignedLocation = SignedLocation::try_from_str(payload.location) .ok_or(ProcessingError::InvalidAttachmentRef)?; + + if signed_location.upload_id().is_some() { + // `SignedLocation` should never have an upload ID. + // NOTE: we could encode this into the `Final` type. + return Err(ProcessingError::InvalidAttachmentRef); + } + // NOTE: Using the received timestamp here breaks tests without a pop-relay. let location = signed_location .verify(chrono::Utc::now(), config) diff --git a/relay-server/src/services/objectstore.rs b/relay-server/src/services/objectstore.rs index 60886dfa567..451e08c4704 100644 --- a/relay-server/src/services/objectstore.rs +++ b/relay-server/src/services/objectstore.rs @@ -1,20 +1,19 @@ //! Objectstore service for uploading attachments. use std::array::TryFromSliceError; use std::fmt; +use std::io::Write; use std::num::{NonZeroU16, NonZeroUsize}; use std::sync::Arc; use std::time::Duration; -use async_compression::tokio::bufread::ZstdEncoder; use bytes::Bytes; use futures::StreamExt; use http::StatusCode; use objectstore_client::{ - Client, Compression, ExpirationPolicy, SecretKey as SigningKey, Session, TokenGenerator, - UploadId, Usecase, + Client, CompletePart, Compression, ExpirationPolicy, MultipartUpload, PartInfo, + SecretKey as SigningKey, Session, TokenGenerator, UploadId, Usecase, }; -use objectstore_types::multipart::InvalidUploadId; use relay_base_schema::organization::OrganizationId; use relay_base_schema::project::ProjectId; use relay_config::ObjectstoreServiceConfig; @@ -23,7 +22,6 @@ use relay_system::{ Addr, AsyncResponse, FromMessage, Interface, LoadShed, NoResponse, Sender, SimpleService, }; use sentry_protos::snuba::v1::TraceItem; -use tokio_util::io::{ReaderStream, StreamReader}; use crate::constants::DEFAULT_ATTACHMENT_RETENTION; use crate::envelope::{ContentType, Item, ItemType}; @@ -36,13 +34,19 @@ 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, + BoundedStream, ByteCounter, MeteredStream, Rechunk, 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(); +/// The minimum distance between two `Upload-Offset`s in bytes. +/// +/// Every `Upload-Offset` must be 0 modulo `UPLOAD_GRANULARITY`. +const UPLOAD_GRANULARITY: NonZeroUsize = NonZeroUsize::new(1024 * 1024).unwrap(); // 1 MiB + +/// Every part except the last needs to be at least this size. +const MULTIPART_MIN_PART_SIZE: usize = 5 * 1024 * 1024; // 5 MiB /// Messages that the objectstore service can handle. pub enum Objectstore { @@ -51,7 +55,7 @@ pub enum Objectstore { EventAttachment(Managed), RawProfile(Managed), Create(CreateMultipart, Sender>), - Stream(Stream, Sender>), + Stream(Stream, Sender>), } impl Objectstore { @@ -154,18 +158,32 @@ impl FromMessage for Objectstore { } } +/// Whether the current request contains the full upload. +#[derive(Clone)] +pub enum StreamMode { + /// The current request contains the full upload. + Oneshot(ByteCounter), + /// The current request might be partial and requires multipart upload under the hood. + Multipart { + upload_id: String, + offset: usize, + length: usize, + }, +} + /// A stream that can be uploaded to objectstore. pub struct Stream { pub organization_id: OrganizationId, pub project_id: ProjectId, - pub upload_ref: UploadRef, + pub key: String, + pub mode: StreamMode, pub stream: BoundedStream>, } impl FromMessage for Objectstore { - type Response = AsyncResponse>; + type Response = AsyncResponse>; - fn from_message(message: Stream, sender: Sender>) -> Self { + fn from_message(message: Stream, sender: Sender>) -> Self { Self::Stream(message, sender) } } @@ -276,10 +294,28 @@ impl> From for Error { pub enum ErrorKind { #[error("invalid scoping")] InvalidScoping, + #[error( + "invalid upload offset {offset}: must be aligned to the upload granularity {granularity}" + )] + InvalidOffset { offset: usize, granularity: usize }, + #[error("offset without upload ID")] + OffsetWithoutUploadId, + #[error("invalid Upload-Length {length}: server has {server_offset} bytes")] + InvalidUploadLength { length: usize, server_offset: usize }, + #[error("request too small: compressed body is {size}, need at least {min_size}")] + RequestTooSmall { size: usize, min_size: usize }, + #[error( + "the body of a non-final request must be a multiple of the upload granularity {granularity}, found {trailing} trailing bytes" + )] + UnalignedBody { trailing: usize, granularity: usize }, + #[error("unknown key {0}")] + UnknownKey(String), #[error("timeout: {0}")] Timeout(#[from] tokio::time::error::Elapsed), #[error("load shed")] LoadShed, + #[error("compression failed: {0}")] + CompressionFailed(#[source] std::io::Error), #[error("upload failed: {0}")] UploadFailed(#[from] objectstore_client::Error), #[error("UUID conversion failed: {0}")] @@ -290,8 +326,15 @@ impl ErrorKind { fn as_str(&self) -> &'static str { match self { Self::InvalidScoping => "invalid_scoping", + Self::InvalidOffset { .. } => "invalid_offset", + Self::OffsetWithoutUploadId => "offset_without_upload_id", + Self::InvalidUploadLength { .. } => "invalid_upload_length", + Self::RequestTooSmall { .. } => "request_too_small", + Self::UnalignedBody { .. } => "unaligned_body", + Self::UnknownKey { .. } => "invalid_length", Self::Timeout(_) => "timeout", Self::LoadShed => "load_shed", + Self::CompressionFailed(_) => "compression_failed", Self::UploadFailed(_) => "upload_failed", Self::Uuid(_) => "uuid", } @@ -299,6 +342,9 @@ impl ErrorKind { fn is_client_error(&self) -> bool { match self { + ErrorKind::InvalidOffset { .. } + | ErrorKind::InvalidUploadLength { .. } + | ErrorKind::UnalignedBody { .. } => true, ErrorKind::UploadFailed(objectstore_client::Error::Reqwest(error)) => { find_error_source(error, is_user_error).is_some() } @@ -334,17 +380,8 @@ pub struct UploadRef { /// The ID of the multipart upload session (chosen by objectstore). /// `None` if the upload is not multipart. pub upload_id: Option, -} - -impl UploadRef { - /// Validates the upload ID and returns a new upload reference. - pub fn new(key: String, upload_id: Option) -> Result { - let upload_id = match upload_id { - Some(s) => Some(UploadId::new(s)?), - None => None, - }; - Ok(Self { key, upload_id }) - } + /// The byte offset from which to resume the upload. + pub offset: usize, } impl fmt::Display for ObjectstoreKey { @@ -572,8 +609,8 @@ impl ObjectstoreServiceInner { .await; match result { - Ok(stored_key) => { - attachment.modify(|a, _| a.set_stored_key(stored_key.into_inner())); + Ok(UploadRef { key, .. }) => { + attachment.modify(|a, _| a.set_stored_key(key)); } Err(error) => { error.log(MessageKind::Event); @@ -626,11 +663,9 @@ impl ObjectstoreServiceInner { }; match upload_result { - Ok(stored_key) => { + Ok(UploadRef { key, .. }) => { attachment.modify(|attachment, _| { - attachment - .attachment - .set_stored_key(stored_key.into_inner()); + attachment.attachment.set_stored_key(key); }); self.store.send(attachment); } @@ -683,7 +718,7 @@ impl ObjectstoreServiceInner { #[cfg(debug_assertions)] let original_key = key.clone(); - let _stored_key = self + let _upload_ref = self .upload_bytes( MessageKind::TraceAttachment, &session, @@ -696,7 +731,7 @@ impl ObjectstoreServiceInner { .reject(&trace_item)?; #[cfg(debug_assertions)] - debug_assert_eq!(_stored_key.into_inner(), original_key); + debug_assert_eq!(_upload_ref.key, original_key); } // Only after successful upload forward the attachment to the store. @@ -718,12 +753,12 @@ impl ObjectstoreServiceInner { .try_upload_raw_profile(payload, content_type, store_message.retention_days, scoping) .await { - Ok(Some(stored_id)) => { + Ok(Some(UploadRef { key, .. })) => { store_message.modify(|message, _| { message.attachments.push(ProfileAttachment { name, content_type, - stored_id, + stored_id: ObjectstoreKey(key), }) }); } @@ -744,7 +779,7 @@ impl ObjectstoreServiceInner { content_type: ContentType, retention: u16, scoping: Scoping, - ) -> Result, Error> { + ) -> Result, Error> { if payload.is_empty() { return Ok(None); } @@ -754,7 +789,7 @@ impl ObjectstoreServiceInner { .for_project(scoping.organization_id.value(), scoping.project_id.value()) .session(&self.objectstore_client)?; - let stored_key = self + let upload_ref = self .upload_bytes( MessageKind::RawProfile, &session, @@ -765,7 +800,7 @@ impl ObjectstoreServiceInner { ) .await?; - Ok(Some(stored_key)) + Ok(Some(upload_ref)) } async fn handle_create(&self, create: CreateMultipart) -> Result { @@ -789,16 +824,19 @@ impl ObjectstoreServiceInner { Ok(UploadRef { key, upload_id: Some(upload_id.clone()), + offset: 0, }) } - async fn handle_stream(&self, stream: Stream) -> Result { + async fn handle_stream(&self, stream: Stream) -> Result { let Stream { organization_id, project_id, - upload_ref, + key, + mode, stream, } = stream; + let session = self.session(&self.event_attachments, organization_id, project_id)?; self.upload( @@ -806,7 +844,8 @@ impl ObjectstoreServiceInner { &session, Upload::Stream { body: TakeOnce::new(stream), - upload_ref, + key, + mode, }, ) .await @@ -820,7 +859,7 @@ impl ObjectstoreServiceInner { retention: u16, key: Option, content_type: Option, - ) -> Result { + ) -> Result { let retention_hours = retention.checked_mul(24); self.upload( kind, @@ -840,7 +879,7 @@ impl ObjectstoreServiceInner { kind: MessageKind, session: &Session, body: Upload, - ) -> Result { + ) -> Result { let mut attempts = 0; let timeout = match &body { Upload::Bytes { .. } => self.timeout, @@ -889,7 +928,7 @@ impl ObjectstoreServiceInner { kind: MessageKind, session: &Session, body: UploadAttempt, - ) -> Result { + ) -> Result { match body { UploadAttempt::Bytes { body, @@ -897,6 +936,7 @@ impl ObjectstoreServiceInner { retention_hours, content_type, } => { + let offset = body.len(); let mut request = session.put(body); if let Some(content_type) = content_type { request = request.content_type(content_type.as_str()); @@ -916,68 +956,197 @@ impl ObjectstoreServiceInner { request.send().await? }); - Ok(ObjectstoreKey(response.key)) + Ok(UploadRef { + key: response.key, + upload_id: None, + offset, + }) } - UploadAttempt::Stream { body, upload_ref } => { - 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.send().await?; - return Ok(ObjectstoreKey(response.key)); - }; - - let multipart_upload = - session.resume_multipart_upload(key, upload_id.to_string())?; + UploadAttempt::Stream { body, key, mode } => { + let original_key = key.clone(); + match mode { + StreamMode::Oneshot(byte_counter) => { + let request = session.put_stream(body.boxed()).key(key); + let response = request.send().await?; + Ok(UploadRef { + key: response.key, + upload_id: None, + offset: byte_counter.get(), + }) + } + StreamMode::Multipart { + upload_id, + offset, + length, + } => { + let granularity = UPLOAD_GRANULARITY.get(); + + if (offset % granularity) != 0 { + return Err(AttemptUploadError::InvalidOffset { + offset, + granularity, + }); + } + + let mut multipart = session.resume_multipart_upload(key, upload_id)?; + let mut final_upload_id = Some(multipart.upload_id().clone()); + + relay_statsd::metric!( + timer(RelayTimers::AttachmentUploadDuration), + type = kind.as_str(), + { + // Ensure that the stream ends at the granularity boundary. + // NOTE: Once every upload is a multipart upload, we can remove `RetryableStream` + // because streams will never be effectively retried. + let mut body = Rechunk::new(body, UPLOAD_GRANULARITY); + + // 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. + // + // Every chunk needs to be > 5 MB because that's what multipart requires. + let mut flushed_parts = vec![]; + let mut previous_buffer = (offset, vec![]); + let mut compressor = zstd::stream::write::Encoder::new(vec![], 0).map_err(AttemptUploadError::Zstd)?; + let mut new_offset = offset; + let mut is_closing = false; + let mut misalignment = 0; + while let Some(bytes) = body.next().await { + let bytes = bytes.map_err(objectstore_client::Error::Io)?; + debug_assert!(bytes.len() <= granularity); + + // Only accept full grains, unless the upload is done. + is_closing = new_offset + bytes.len() == length; + if is_closing || bytes.len() == granularity { + compressor.write_all(&bytes).map_err(AttemptUploadError::Zstd)?; + new_offset += bytes.len(); + + // If the current compressed part is large enough, flush the previous one. + // We keep the current one around just in case we need to join it with the last part. + // (see concatenation after while loop). + if is_closing || compressor.get_ref().len() > MULTIPART_MIN_PART_SIZE { + let mut new_compressor = zstd::stream::write::Encoder::new(vec![], 0).map_err(AttemptUploadError::Zstd)?; + + // Replace previous buffer with current compressor content. + std::mem::swap(&mut compressor, &mut new_compressor); + let compressed = new_compressor.finish().map_err(AttemptUploadError::Zstd)?; + let mut current_buffer = (new_offset, compressed); + std::mem::swap(&mut current_buffer, &mut previous_buffer); + + // Flush previous buffer. + self.try_flush(current_buffer, &mut multipart, &mut flushed_parts).await?; + } + } else { + // We're at the end of a non-final request, but it is not + // aligned with granularity: return an error. + misalignment = bytes.len(); + } + + if is_closing { + // This should already be guaranteed by the BoundedStream. + #[cfg(debug_assertions)] + debug_assert!(body.next().await.is_none()); + break; + } + } - let body = ReaderStream::new(ZstdEncoder::new(StreamReader::new(body))); + // Remaining bytes are now < 5 MiB. Concatenate it with the previous part + // so we reach the limit. + let (_, mut remaining) = previous_buffer; + if !compressor.get_ref().is_empty() { + let remaining2 = compressor.finish().map_err(AttemptUploadError::Zstd)?; + relay_log::trace!("Combining {} previous bytes with {} remaining", remaining.len(), remaining2.len()); + remaining.extend(remaining2); + } + if is_closing || remaining.len() >= MULTIPART_MIN_PART_SIZE { + self.try_flush((new_offset, remaining), &mut multipart, &mut flushed_parts).await?; + } else if !remaining.is_empty() && misalignment == 0 { + // This can happen when a non-closing request body compresses so well that it + // fits under the S3 limit. In this case the client is forced + // to query the `Upload-Offset` again (see INGEST-1025) and retry with a larger chunk. + return Err(AttemptUploadError::RequestTooSmall {size: remaining.len(), min_size: MULTIPART_MIN_PART_SIZE}); + } - // 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(); + // A non-final request did not end at the granularity boundary. + // All full grains were stored, but signal an error to the client. + if misalignment > 0 { + return Err(AttemptUploadError::UnalignedBody { trailing: misalignment, granularity }); + } - 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; + if is_closing { + relay_log::trace!("Completing multipart upload"); + let parts = multipart.list_parts().await?; + + // Verify that the last written part number is the highest part number in the list. + // Otherwise clients could fake a small `Upload-Length` for a large upload. + let max_part_number = parts.iter().map(|p|p.part_number.get()).max().unwrap_or(0) as usize; + if max_part_number != new_offset.div_ceil(granularity) { + return Err(AttemptUploadError::InvalidUploadLength { + length, server_offset: max_part_number * granularity + }) + } + + let parts = parts.into_iter().map(|item| { + let PartInfo{ part_number, etag, .. } = item; + CompletePart { part_number, etag } + }); + + let final_key = multipart.complete(parts).await?; + final_upload_id = None; + debug_assert_eq!(&original_key, &final_key); } - }; - parts.push(part?); + Ok(UploadRef { + key: original_key, + upload_id: final_upload_id, + offset: new_offset, + }) + }) } - multipart_upload.complete(parts).await? - }); - - Ok(ObjectstoreKey(result)) + } } } } + async fn try_flush( + &self, + buffer: (usize, Vec), + multipart: &mut MultipartUpload, + parts: &mut Vec, + ) -> Result<(), AttemptUploadError> { + let (end_offset, bytes) = buffer; + relay_log::trace!("Trying to flush {} bytes", bytes.len()); + if bytes.is_empty() { + return Ok(()); + } + let bytes = Bytes::from(bytes); + let part_number = end_offset.div_ceil(UPLOAD_GRANULARITY.get()); + let part_number = u32::try_from(part_number) + .map_err(|_| objectstore_client::Error::InvalidPartNumber(u32::MAX))?; + + // 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 + .put(bytes.clone(), part_number, None) + .await + .map_err(AttemptUploadError::Objectstore); + attempts += 1; + if attempts < self.max_attempts.get() && matches!(&result, Err(e) if is_retryable(e)) { + relay_log::trace!("Multipart attempt {attempts}: Failed with {result:?}, retrying"); + tokio::time::sleep(self.retry_interval).await; + } else { + relay_log::trace!("Final attempt"); + break result; + } + }?; + + parts.push(part); + + Ok(()) + } + fn session( &self, usecase: &Usecase, @@ -994,6 +1163,57 @@ impl ObjectstoreServiceInner { } } +#[derive(Debug, thiserror::Error)] +enum AttemptUploadError { + #[error(transparent)] + Objectstore(#[from] objectstore_client::Error), + #[error("zstd error: {0}")] + Zstd(#[source] std::io::Error), + #[error("invalid Upload-Offset {offset}: must be a multiple of {granularity}")] + InvalidOffset { offset: usize, granularity: usize }, + #[error("invalid Upload-Length {length}: server has {server_offset} bytes")] + InvalidUploadLength { length: usize, server_offset: usize }, + #[error("Request too small: compressed body is {size}, need at least {min_size}")] + RequestTooSmall { size: usize, min_size: usize }, + #[error( + "the body of a non-final request must be a multiple of the upload granularity {granularity}, found {trailing} trailing bytes" + )] + UnalignedBody { trailing: usize, granularity: usize }, +} + +impl From for ErrorKind { + fn from(value: AttemptUploadError) -> Self { + match value { + AttemptUploadError::Objectstore(error) => ErrorKind::UploadFailed(error), + AttemptUploadError::Zstd(error) => ErrorKind::CompressionFailed(error), + AttemptUploadError::InvalidOffset { + offset, + granularity, + } => ErrorKind::InvalidOffset { + offset, + granularity, + }, + AttemptUploadError::InvalidUploadLength { + length, + server_offset, + } => ErrorKind::InvalidUploadLength { + length, + server_offset, + }, + AttemptUploadError::RequestTooSmall { size, min_size } => { + ErrorKind::RequestTooSmall { size, min_size } + } + AttemptUploadError::UnalignedBody { + trailing, + granularity, + } => ErrorKind::UnalignedBody { + trailing, + granularity, + }, + } + } +} + /// Common interface for calls to [`ObjectstoreServiceInner::upload`]. /// /// This type is shared across retries. @@ -1006,7 +1226,8 @@ enum Upload { }, Stream { body: TakeOnce>>, - upload_ref: UploadRef, + key: String, + mode: StreamMode, }, } @@ -1024,10 +1245,11 @@ impl Upload { retention_hours: *retention_hours, content_type: *content_type, }), - Self::Stream { body, upload_ref } => { + Self::Stream { body, key, mode } => { RetryableStream::new(body.clone()).map(|body| UploadAttempt::Stream { body, - upload_ref: upload_ref.clone(), + key: key.clone(), + mode: mode.clone(), }) } } @@ -1046,13 +1268,14 @@ enum UploadAttempt { }, Stream { body: RetryableStream>>, - upload_ref: UploadRef, + key: String, + mode: StreamMode, }, } -fn is_retryable(error: &objectstore_client::Error) -> bool { +fn is_retryable(error: &AttemptUploadError) -> bool { match error { - objectstore_client::Error::Reqwest(error) => { + AttemptUploadError::Objectstore(objectstore_client::Error::Reqwest(error)) => { error.is_connect() || error.is_timeout() || matches!( @@ -1101,6 +1324,7 @@ fn drop_attachments(event: &mut Managed>) { #[cfg(test)] mod tests { + use relay_event_schema::protocol::EventId; use relay_quotas::DataCategory; use relay_system::Service; @@ -1122,15 +1346,14 @@ mod tests { usize::MAX, ); + let byte_counter = stream.byte_counter(); let result = addr .send(Stream { organization_id: OrganizationId::new(0), project_id: ProjectId::new(1), - upload_ref: UploadRef { - key: "my_file".to_owned(), - upload_id: Some(UploadId::new("my_upload".to_owned()).unwrap()), - }, stream, + key: "my_key".to_owned(), + mode: StreamMode::Oneshot(byte_counter), }) .await .unwrap(); diff --git a/relay-server/src/services/upload.rs b/relay-server/src/services/upload.rs index 872a22fe32b..0f049e085e4 100644 --- a/relay-server/src/services/upload.rs +++ b/relay-server/src/services/upload.rs @@ -58,11 +58,15 @@ pub enum Error { Timeout(#[from] tokio::time::error::Elapsed), #[error("error response from upstream: {0}")] Upstream(#[source] reqwest::Error), - #[error("upstream provided invalid location: {0:?}")] - InvalidLocation(Option), + #[error("upstream provided invalid data for {0}: {1:?}")] + InvalidFromUpstream(&'static str, Option), + #[error("invalid input: {0}")] + InvalidFromClient(&'static str), #[cfg(feature = "processing")] #[error(transparent)] InvalidUploadId(#[from] objectstore_types::multipart::InvalidUploadId), + #[error("Upload-Offset > 0 is currently unsupported with Defer-Length: 1")] + OffsetWithoutLength, #[error("serializing location failed: {0}")] SerializeFailed(#[from] serde_urlencoded::ser::Error), #[error("failed to sign location")] @@ -87,9 +91,11 @@ impl Error { Error::UpstreamRequest(_) => "upstream_request", Error::Timeout(_) => "timeout", Error::Upstream(_) => "upstream_response", - Error::InvalidLocation(_) => "invalid_location", + Error::InvalidFromClient { .. } => "invalid_from_client", + Error::InvalidFromUpstream { .. } => "invalid_from_upstream", #[cfg(feature = "processing")] Error::InvalidUploadId(_) => "invalid_upload_id", + Error::OffsetWithoutLength => "offset_with_defer_length", Error::SigningFailed => "signing_failed", Error::SerializeFailed(_) => "serialize_failed", Error::InvalidSignature(_) => "invalid_signature", @@ -107,12 +113,12 @@ pub enum Upload { /// Creates an upload resource. /// /// Returns the trusted identifier of the upload. - Create(Create, InstrumentedSender), + Create(Create, InstrumentedSender>), /// Upload a stream of bytes for a given location. /// - /// The service also returns the signed location. This is redundant, but creates a simpler - /// flow for the caller side. - Upload(Stream, InstrumentedSender), + /// The service also returns the signed location and the current Upload-Offset stored on the + /// server. + Stream(Box, InstrumentedSender), } impl Interface for Upload {} @@ -151,10 +157,39 @@ pub struct Stream { pub project: ProjectContext, /// The location to upload to. pub location: SignedLocation, + /// The offset from which to resume the upload. + pub offset: usize, /// The body to be uploaded to objectstore, with length validation. pub stream: BoundedStream>, } +/// The result of a [`Stream`] operation. +pub struct StreamResult { + /// The signed location of the upload. + /// + /// This is "final" because we have either a pre-commited `Upload-Length` or + /// a oneshot upload. + pub location: SignedLocation, + /// The byte offset stored on the server after the operation. + pub offset: usize, +} + +impl StreamResult { + fn try_from_response(response: Response) -> Result { + let offset = response + .headers() + .get(tus::UPLOAD_OFFSET) + .ok_or(Error::InvalidFromUpstream(tus::UPLOAD_OFFSET, None))? + .to_str() + .map_err(|_| Error::InvalidFromUpstream(tus::UPLOAD_OFFSET, None))? + .parse() + .map_err(|_| Error::InvalidFromUpstream(tus::UPLOAD_OFFSET, None))?; + let location = SignedLocation::try_from_response(response)?; + + Ok(Self { location, offset }) + } +} + impl FromMessage for Upload { type Response = AsyncResponse, Error>>; @@ -173,11 +208,11 @@ impl FromMessage for Upload { } impl FromMessage for Upload { - type Response = AsyncResponse, Error>>; + type Response = AsyncResponse>; - fn from_message(message: Stream, sender: Sender, Error>>) -> Self { - Self::Upload( - message, + fn from_message(message: Stream, sender: Sender>) -> Self { + Self::Stream( + Box::new(message), InstrumentedSender { metric: RelayCounters::UploadUpload, inner: sender, @@ -234,13 +269,13 @@ pub struct Service { } /// A response channel that emits a metric for each response. -pub struct InstrumentedSender { +pub struct InstrumentedSender { metric: RelayCounters, - inner: Sender, Error>>, + inner: Sender>, } -impl InstrumentedSender { - fn send(self, result: Result, Error>) { +impl InstrumentedSender { + fn send(self, result: Result) { let result_msg = match &result { Ok(_) => "success", Err(e) => e.variant(), @@ -300,7 +335,11 @@ impl Service { // and if the upload actually has data (multipart does not allow empty parts). (false, _) | (_, Some(0)) => (key, None), _ => { - let UploadRef { key, upload_id } = addr + let UploadRef { + key, + upload_id, + offset: _, + } = addr .send(objectstore::CreateMultipart { organization_id, project_id, @@ -326,24 +365,26 @@ impl Service { } } - async fn upload(&self, stream: Stream) -> Result, Error> { + async fn upload(&self, stream: Stream) -> Result { let Stream { #[cfg_attr(not(feature = "processing"), expect(unused))] received, project, location, + offset, stream, } = stream; match &self.backend { Backend::Upstream { addr } => { - let (request, rx) = UploadRequest::upload(project, location.try_to_uri()?, stream); + let (request, rx) = + UploadRequest::upload(project, location.try_to_uri()?, offset, stream); addr.send(SendRequest(request)); let response = rx.await??; - SignedLocation::try_from_response(response) + StreamResult::try_from_response(response) } #[cfg(feature = "processing")] Backend::Objectstore { addr, config } => { - use crate::services::objectstore::UploadRef; + use crate::services::objectstore::{StreamMode, UploadRef}; let Location { project_id, @@ -355,38 +396,68 @@ impl Service { let scoping = project.scoping; debug_assert_eq!(scoping.project_id, project_id); - debug_assert!(stream.length().is_none_or(|l| Some(l) == length.value())); - let byte_counter = stream.byte_counter(); - let upload_ref = UploadRef::new(key, upload_id)?; - let key = addr + let mode = match (length.value(), upload_id) { + (Some(length), Some(upload_id)) => StreamMode::Multipart { + upload_id, + offset, + length, + }, + (_, None) => { + if offset != 0 { + return Err(Error::OffsetWithoutLength); + } + StreamMode::Oneshot(stream.byte_counter()) + } + (None, Some(_)) => { + // NOTE: this restriction could be encoded into the Location type. + return Err(Error::InvalidFromClient( + "upload_id without `Upload-Length`", + )); + } + }; + let upload_ref = addr .send(objectstore::Stream { organization_id: scoping.organization_id, project_id, - upload_ref, + stream, + key, + mode: mode.clone(), }) .await - .map_err(Error::ObjectstoreServiceUnavailable)?? - .into_inner(); - let length = Final(byte_counter.get()); + .map_err(Error::ObjectstoreServiceUnavailable)??; - Location { - project_id, + let UploadRef { key, - length, - upload_id: None, - other, - } - .try_sign(config) + upload_id, + offset, + } = upload_ref; + + // For oneshot requests, update the length: + let length = match mode { + StreamMode::Oneshot(byte_counter) => byte_counter.get(), + StreamMode::Multipart { length, .. } => length, + }; + + Ok(StreamResult { + location: Location { + project_id, + key, + length: Final(length), + upload_id: upload_id.map(|id| id.to_string()), + other, + } + .try_sign(config)?, + offset, + }) } } } - async fn timeout(&self, future: F) -> Result, Error> + async fn timeout(&self, future: F) -> Result where - L: UploadLength, - F: IntoFuture, Error>>, + F: IntoFuture>, { tokio::time::timeout(self.timeout, future).await? } @@ -400,8 +471,8 @@ impl SimpleService for Service { Upload::Create(create, sender) => { sender.send(self.timeout(self.create(create)).await); } - Upload::Upload(stream, sender) => { - sender.send(self.timeout(self.upload(stream)).await); + Upload::Stream(stream, sender) => { + sender.send(self.timeout(self.upload(*stream)).await); } } } @@ -411,7 +482,7 @@ impl LoadShed for Service { fn handle_loadshed(&self, message: Upload) { match message { Upload::Create(_, tx) => tx.send(Err(Error::LoadShed)), - Upload::Upload(_, tx) => tx.send(Err(Error::LoadShed)), + Upload::Stream(_, tx) => tx.send(Err(Error::LoadShed)), } } } @@ -617,6 +688,11 @@ impl SignedLocation { HeaderValue::from_str(&self.try_to_uri()?).map_err(Error::Internal) } + /// Returns the multipart upload ID, if this location refers to a multipart upload. + pub fn upload_id(&self) -> Option<&str> { + self.location.upload_id.as_deref() + } + fn try_to_uri(&self) -> Result { let Self { location, @@ -660,11 +736,12 @@ where let header = response .headers() .get(hyper::header::LOCATION) - .ok_or(Error::InvalidLocation(None))?; + .ok_or(Error::InvalidFromUpstream("location", None))?; let uri = header .to_str() - .map_err(|_| Error::InvalidLocation(Some(header.clone())))?; - Self::try_from_str(uri).ok_or(Error::InvalidLocation(Some(header.clone()))) + .map_err(|_| Error::InvalidFromUpstream("location", Some(header.clone())))?; + Self::try_from_str(uri) + .ok_or(Error::InvalidFromUpstream("location", Some(header.clone()))) } Err(e) => Err(Error::Upstream(e)), } @@ -712,6 +789,7 @@ enum RequestKind { }, Upload { uri: String, + offset: usize, stream: TakeOnce>>, encoding: HttpEncoding, }, @@ -751,6 +829,7 @@ impl UploadRequest { fn upload( project: ProjectContext, uri: String, + offset: usize, stream: BoundedStream>, ) -> ( Self, @@ -762,6 +841,7 @@ impl UploadRequest { project, kind: RequestKind::Upload { uri, + offset, stream: TakeOnce::new(stream), encoding: HttpEncoding::Zstd, // just a default, will be overwritten by .configure() }, @@ -844,6 +924,7 @@ impl UpstreamRequest for UploadRequest { } RequestKind::Upload { uri: _, + offset, stream, encoding, } => { @@ -851,7 +932,7 @@ impl UpstreamRequest for UploadRequest { relay_log::error!("upload request stream was already consumed"); return Err(HttpError::Misconfigured); }; - tus::add_upload_headers(builder); + tus::add_upload_headers(builder, *offset); let body = encode_body(body, *encoding); builder.content_encoding(*encoding); diff --git a/relay-server/src/statsd.rs b/relay-server/src/statsd.rs index 5e5ad83e6d2..1e7273ed2a5 100644 --- a/relay-server/src/statsd.rs +++ b/relay-server/src/statsd.rs @@ -981,7 +981,7 @@ pub enum RelayCounters { /// This metric is tagged with: /// - `result`: `success` or the failure reason. UploadCreate, - /// The number of times an upload location is created through the upload service. + /// The number of times bytes are streamed through the upload service. /// /// This metric is tagged with: /// - `result`: `success` or the failure reason. diff --git a/relay-server/src/utils/tus.rs b/relay-server/src/utils/tus.rs index 1fc89bd83b6..97631e3668b 100644 --- a/relay-server/src/utils/tus.rs +++ b/relay-server/src/utils/tus.rs @@ -30,8 +30,8 @@ pub enum Error { upload_defer_length: Option, }, /// The `Upload-Offset` header is missing or invalid - #[error("expected Upload-Offset: 0, got: {0:?}")] - UploadOffset(Option), + #[error("expected Upload-Offset >= 0")] + UploadOffset, /// The `Content-Type` header is not what TUS expects. #[error("expected Content-Type: {expected}, got: {received}")] ContentType { @@ -144,7 +144,9 @@ pub fn validate_post_headers(headers: &HeaderMap) -> Result { } /// Validates TUS protocol headers and returns the expected upload length. -pub fn validate_patch_headers(headers: &HeaderMap) -> Result<(), Error> { +/// +/// Returns the offset from which the upload is resumed. +pub fn validate_patch_headers(headers: &HeaderMap) -> Result { let tus_version = headers.get(TUS_RESUMABLE); if tus_version != Some(&TUS_VERSION) { return Err(Error::Version( @@ -163,14 +165,9 @@ pub fn validate_patch_headers(headers: &HeaderMap) -> Result<(), Error> { }); } - let upload_offset: usize = - parse_header(headers, UPLOAD_OFFSET).ok_or(Error::UploadOffset(None))?; - if upload_offset != 0 { - // Only allow full uploads for now. - return Err(Error::UploadOffset(Some(upload_offset))); - } + let upload_offset: usize = parse_header(headers, UPLOAD_OFFSET).ok_or(Error::UploadOffset)?; - Ok(()) + Ok(upload_offset) } /// Prepares the required TUS request headers for upstream requests. @@ -196,10 +193,10 @@ pub fn add_creation_headers( } /// Prepares the required TUS request headers for upstream requests. -pub fn add_upload_headers(builder: &mut RequestBuilder) { +pub fn add_upload_headers(builder: &mut RequestBuilder, offset: usize) { builder.header(TUS_RESUMABLE, TUS_VERSION); builder.header(http::header::CONTENT_TYPE, EXPECTED_CONTENT_TYPE); - builder.header(UPLOAD_OFFSET, "0"); // always zero until we implement retries / chunking + builder.header(UPLOAD_OFFSET, offset.to_string()); } /// Prepares the required TUS response headers. @@ -479,17 +476,17 @@ mod tests { headers.insert(TUS_RESUMABLE, HeaderValue::from_static("1.0.0")); headers.insert(http::header::CONTENT_TYPE, EXPECTED_CONTENT_TYPE); let result = validate_patch_headers(&headers); - assert!(matches!(result, Err(Error::UploadOffset(None)))); + assert!(matches!(result, Err(Error::UploadOffset))); } #[test] - fn test_validate_patch_headers_nonzero_upload_offset() { + fn test_validate_patch_headers_noninteger_upload_offset() { let mut headers = HeaderMap::new(); headers.insert(TUS_RESUMABLE, HeaderValue::from_static("1.0.0")); headers.insert(http::header::CONTENT_TYPE, EXPECTED_CONTENT_TYPE); - headers.insert(UPLOAD_OFFSET, HeaderValue::from_static("512")); + headers.insert(UPLOAD_OFFSET, HeaderValue::from_static("adsf")); let result = validate_patch_headers(&headers); - assert!(matches!(result, Err(Error::UploadOffset(Some(512))))); + assert!(matches!(result, Err(Error::UploadOffset))); } #[test] diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 78e33844f7d..75288ff8918 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -307,7 +307,6 @@ def dummy_upload(mini_sentry): # noqa @mini_sentry.app.route("/api//upload/", methods=["POST"]) def create(**opts): - return Response( "", status=201, @@ -316,10 +315,14 @@ def create(**opts): @mini_sentry.app.route("/api//upload//", methods=["PATCH"]) def upload(**opts): - assert request.headers["Content-Encoding"] == "zstd" - assert request.data.startswith(ZSTD_MAGIC_HEADER) + if request.data: + assert request.headers["Content-Encoding"] == "zstd" + assert request.data.startswith(ZSTD_MAGIC_HEADER) return Response( "", status=204, - headers={"Location": DUMMY_UPLOAD_LOCATION}, + headers={ + "Location": DUMMY_UPLOAD_LOCATION, + "Upload-Offset": len(request.data), + }, ) diff --git a/tests/integration/test_minidump.py b/tests/integration/test_minidump.py index 2b0a646f55a..f7badb5f233 100644 --- a/tests/integration/test_minidump.py +++ b/tests/integration/test_minidump.py @@ -492,20 +492,21 @@ def test_minidump_invalid_nested_formdata(mini_sentry, relay): @pytest.mark.parametrize( - "rate_limit,minidump_filename,use_objectstore,stream_upload", + "rate_limit,minidump_filename,use_objectstore,stream_upload,multipart", [ - (None, "minidump.dmp", True, False), - (None, "minidump.dmp", False, False), - ("attachment", "minidump.dmp", True, False), - ("attachment", "minidump.dmp", False, False), - ("transaction", "minidump.dmp", False, False), - (None, "minidump.dmp.gz", False, False), - (None, "minidump.dmp.xz", False, False), - (None, "minidump.dmp.bz2", False, False), - (None, "minidump.dmp.bz2", True, False), - (None, "minidump.dmp.zst", False, False), - (None, "minidump.dmp.zst", True, False), - (None, "minidump.dmp.zst", True, True), + (None, "minidump.dmp", True, False, False), + (None, "minidump.dmp", False, False, False), + ("attachment", "minidump.dmp", True, False, False), + ("attachment", "minidump.dmp", False, False, False), + ("transaction", "minidump.dmp", False, False, False), + (None, "minidump.dmp.gz", False, False, False), + (None, "minidump.dmp.xz", False, False, False), + (None, "minidump.dmp.bz2", False, False, False), + (None, "minidump.dmp.bz2", True, False, False), + (None, "minidump.dmp.zst", False, False, False), + (None, "minidump.dmp.zst", True, False, False), + (None, "minidump.dmp.zst", True, True, False), + (None, "minidump.dmp.zst", True, True, True), ], ) def test_minidump_with_processing( @@ -518,6 +519,7 @@ def test_minidump_with_processing( use_objectstore, objectstore, stream_upload, + multipart, ): dmp_path = os.path.join(os.path.dirname(__file__), "fixtures/native/minidump.dmp") with open(dmp_path, "rb") as f: @@ -542,6 +544,10 @@ def test_minidump_with_processing( project_config["config"].setdefault("features", []).append( "projects:relay-minidump-uploads" ) + if multipart: + project_config["config"].setdefault("features", []).append( + "projects:relay-upload-multipart" + ) options = { "processing": { @@ -1056,7 +1062,7 @@ def test_size_limits(mini_sentry, relay, limit, expected_status_code): ], ids=["no-option", "no-minidumps", "stream-minidumps"], ) -def test_minidump_objectstore_uploads( +def test_minidump_streaming_uploads( mini_sentry, relay, dummy_upload, diff --git a/tests/integration/test_upload.py b/tests/integration/test_upload.py index 359ed7985c0..70e5f0a28ad 100644 --- a/tests/integration/test_upload.py +++ b/tests/integration/test_upload.py @@ -2,20 +2,29 @@ Tests for the TUS upload endpoint (/api/{project_id}/upload/). """ +import re import time import uuid from concurrent.futures import ThreadPoolExecutor, as_completed from flask import Response +from objectstore_client.multipart import MultipartUpload import pytest +import random from sentry_relay.auth import SecretKey +from .asserts import matches, matches_any from .consts import ( DUMMY_UPLOAD_PATH, DUMMY_UPLOAD_LOCATION, ) +LOCATION_REGEX = re.compile(r"/api/\d+/upload/(\w+)/") +LOCATION_REGEX_WITH_UPLOAD_ID = re.compile( + r"/api/\d+/upload/(\w+)/?.*upload_id=([\w-]+)" +) + @pytest.fixture def project_config(mini_sentry): @@ -222,8 +231,8 @@ def test_upload_missing_upload_length(mini_sentry, relay, dummy_upload, project_ [ pytest.param( 10, - 400, - "stream shorter than lower bound: received 10 < 11", + 204, + None, id="smaller_than_announced", ), pytest.param( @@ -271,9 +280,10 @@ def test_upload_body_size( ) assert response.status_code == expected_status_code - assert response.text == expected_error or any( - expected_error in source for source in response.json()["causes"] - ), response.json() + if expected_error: + assert response.text == expected_error or any( + expected_error in source for source in response.json()["causes"] + ), response.json() @pytest.mark.parametrize("data_category", ["attachment", "attachment_item"]) @@ -331,10 +341,16 @@ def test_timeout( """Ensure that the general HTTP timeout does not affect the upload endpoint""" mini_sentry.allow_chunked = True + data = b"hello world" + @mini_sentry.app.route(DUMMY_UPLOAD_PATH, methods=["PATCH"]) def slow_upload(**opts): time.sleep(2) - return Response("", status=204, headers={"Location": DUMMY_UPLOAD_LOCATION}) + return Response( + "", + status=204, + headers={"Location": DUMMY_UPLOAD_LOCATION, "Upload-Offset": len(data)}, + ) project_id = 42 relay = relay( @@ -345,7 +361,6 @@ def slow_upload(**opts): }, ) - data = b"hello world" response = relay.patch( "%s&sentry_key=%s" % ( @@ -424,9 +439,8 @@ def test_create_processing( assert response.headers["Upload-Offset"] == str(len(data)), response.headers -@pytest.mark.parametrize("length", [9, 11]) def test_processing_invalid_length( - mini_sentry, relay, relay_with_processing, project_config, length + mini_sentry, relay, relay_with_processing, project_config ): mini_sentry.fail_on_relay_error = False project_id = 42 @@ -447,8 +461,8 @@ def test_processing_invalid_length( assert response.headers["Tus-Resumable"] == "1.0.0" assert "Upload-Offset" not in response.headers - # Use the location to send a PATCH request that is too long // too short - data = length * b"X" + # Use the location to send a PATCH request that is too long + data = 11 * b"X" response = relay.patch( f"{response.headers['Location']}&sentry_key={project_key}", headers={ @@ -585,14 +599,14 @@ def test_objectstore_retries( } ) - location = f"/api/{project_id}/upload/019cdc82ed6c7761ba21fd34b86481c2/" - sep = "?" + location = ( + f"/api/{project_id}/upload/019cdc82ed6c7761ba21fd34b86481c2/?upload_length=11" + ) if with_multipart: - location += "?upload_id=my_upload_id" - sep = "&" + location += "&upload_id=my_upload_id" signature = SecretKey.parse(relay.secret_key).sign(location.encode()) signed_location = ( - f"{location}{sep}sentry_key={project_key}&upload_signature={signature}" + f"{location}&sentry_key={project_key}&upload_signature={signature}" ) data = b"hello world" @@ -653,8 +667,23 @@ def multipart_upload(**opts): assert response.status_code == 504 -def upload_something(relay, project_id, project_key): - data = b"hello world" +@pytest.mark.parametrize("feature_flag", [False, True]) +def test_upload_offset( + mini_sentry, relay_with_processing, project_config, objectstore, feature_flag +): + mini_sentry.fail_on_relay_error = False + project_id = 42 + config = mini_sentry.add_full_project_config(project_id)["config"] + features = config.setdefault("features", []) + if feature_flag: + features.append("projects:relay-upload-multipart") + project_key = mini_sentry.get_dsn_public_key(project_id) + + relay = relay_with_processing() + objectstore = objectstore("attachments", project_id) + + part = random.randbytes(13 * 1024 * 1024) + data = 3 * part response = relay.post( f"/api/{project_id}/upload/?sentry_key={project_key}", headers={ @@ -663,18 +692,152 @@ def upload_something(relay, project_id, project_key): "Upload-Length": str(len(data)), }, ) - assert response.status_code == 201, response.json() + assert response.status_code == 201 - return relay.patch( + # Upload only part of the bytes + split = len(data) // 3 + data1 = data[:split] + data2 = data[split:] + + response = relay.patch( f"{response.headers['Location']}&sentry_key={project_key}", headers={ - "Content-Length": str(len(data)), + "Content-Length": str(len(data1)), "Content-Type": "application/offset+octet-stream", "Tus-Resumable": "1.0.0", "Upload-Offset": "0", }, - data=data, + data=data1, + ) + assert response.status_code == 204, response.text + # Relay acknowledges exactly the bytes it durably stored: + assert response.headers["Upload-Offset"] == str(len(data1)) + + if feature_flag: + key, upload_id = LOCATION_REGEX_WITH_UPLOAD_ID.match( + response.headers["Location"] + ).groups() + part1, part2 = MultipartUpload(objectstore, key, upload_id).list_parts() + assert vars(part1) == { + "part_number": 6, + "etag": matches_any(), + "last_modified": matches_any(), + "size": matches(lambda x: 6e6 < x < 7e6), + } + assert vars(part2) == { + "part_number": 13, + "etag": matches_any(), + "last_modified": matches_any(), + "size": matches(lambda x: 7e6 < x < 8e6), + } + else: + (key,) = LOCATION_REGEX.match(response.headers["Location"]).groups() + # The file is already there + assert objectstore.get(key).payload.read() == data1 + + # Try to upload more: + response = relay.patch( + f"{response.headers['Location']}&sentry_key={project_key}", + headers={ + "Content-Length": str(len(data2)), + "Content-Type": "application/offset+octet-stream", + "Tus-Resumable": "1.0.0", + "Upload-Offset": str(len(data1)), + }, + data=data2, + ) + if feature_flag: + assert response.status_code == 204 + assert response.headers["Upload-Offset"] == str(len(data)) + (key,) = LOCATION_REGEX.match(response.headers["Location"]).groups() + assert objectstore.get(key).payload.read() == data + else: + assert response.status_code == 400 + assert response.json() == { + "causes": [ + "Upload-Offset > 0 is currently unsupported with Defer-Length: 1", + ], + "detail": "upload error: Upload-Offset > 0 is currently unsupported with Defer-Length: 1", + } + + +def test_upload_unaligned_body_rejected( + mini_sentry, relay_with_processing, objectstore +): + """A non-final PATCH body that is not a multiple of the upload granularity is rejected. + + Silently dropping the sub-grain tail would return a 204 whose ``Upload-Offset`` + excludes it; for bodies smaller than one grain that means acknowledging zero + progress, which sends offset-driven clients into an endless retry loop. + + Full grains received before the unaligned tail are still persisted where they + form a valid part, so the client can resume from the last durable offset. + """ + project_id = 42 + config = mini_sentry.add_full_project_config(project_id)["config"] + config.setdefault("features", []).append("projects:relay-upload-multipart") + project_key = mini_sentry.get_dsn_public_key(project_id) + + relay = relay_with_processing() + objectstore = objectstore("attachments", project_id) + + data = random.randbytes(10 * 1024 * 1024) + response = relay.post( + f"/api/{project_id}/upload/?sentry_key={project_key}", + headers={ + "Content-Length": "0", + "Tus-Resumable": "1.0.0", + "Upload-Length": str(len(data)), + }, + ) + assert response.status_code == 201 + location = f"{response.headers['Location']}&sentry_key={project_key}" + + # Sub-grain body (512 KiB) that does not complete the upload: + response = relay.patch( + location, + headers={ + "Content-Length": str(512 * 1024), + "Content-Type": "application/offset+octet-stream", + "Tus-Resumable": "1.0.0", + "Upload-Offset": "0", + }, + data=data[: 512 * 1024], + ) + assert response.status_code == 400, response.text + assert "upload granularity" in response.text + + # Grain-aligned body with a sub-grain tail (6.5 MiB) is rejected as well: + response = relay.patch( + location, + headers={ + "Content-Length": str(13 * 512 * 1024), + "Content-Type": "application/offset+octet-stream", + "Tus-Resumable": "1.0.0", + "Upload-Offset": "0", + }, + data=data[: 13 * 512 * 1024], + ) + assert response.status_code == 400, response.text + assert "upload granularity" in response.text + + # The first 6 MiB of the rejected request were still persisted (they formed a + # valid part), so the upload can be resumed and completed from that offset: + resume_offset = 6 * 1024 * 1024 + response = relay.patch( + location, + headers={ + "Content-Length": str(len(data) - resume_offset), + "Content-Type": "application/offset+octet-stream", + "Tus-Resumable": "1.0.0", + "Upload-Offset": str(resume_offset), + }, + data=data[resume_offset:], ) + assert response.status_code == 204, response.text + assert response.headers["Upload-Offset"] == str(len(data)) + (key,) = LOCATION_REGEX.match(response.headers["Location"]).groups() + assert objectstore.get(key).payload.read() == data @pytest.mark.parametrize( @@ -739,3 +902,27 @@ def test_upload_minidump_opt_in( ) else: assert mini_sentry.captured_outcomes.empty() + + +def upload_something(relay, project_id, project_key): + data = b"hello world" + response = relay.post( + f"/api/{project_id}/upload/?sentry_key={project_key}", + headers={ + "Content-Length": "0", + "Tus-Resumable": "1.0.0", + "Upload-Length": str(len(data)), + }, + ) + assert response.status_code == 201, response.json() + + return relay.patch( + f"{response.headers['Location']}&sentry_key={project_key}", + headers={ + "Content-Length": str(len(data)), + "Content-Type": "application/offset+octet-stream", + "Tus-Resumable": "1.0.0", + "Upload-Offset": "0", + }, + data=data, + )