fix(minidump): Streaming uploads from external relays

relay-server/src/endpoints/upload.rs

@@ -30,7 +30,9 @@
 #[cfg(feature = "processing")]
 use crate::services::objectstore;
 use crate::services::projects::cache::Project;
-use crate::services::upload::{self, ByteStream, SignedLocation};
+use crate::services::upload::{
+    self, ByteStream, Final, LocationQueryParams, Provisional, SignedLocation, UploadLength,
+};
 use crate::services::upstream::UpstreamRequestError;
 use crate::statsd::RelayCounters;
 use crate::utils::{ApiErrorResponse, MeteredStream};
@@ -76,7 +78,6 @@
         }
 
         let status = match self {
-            Error::Tus(tus::Error::DeferLengthNotAllowed) => StatusCode::FORBIDDEN,
             Error::Tus(_) => StatusCode::BAD_REQUEST,
             Error::Request(error) => return error.into_response(),
             Error::SendError(_) => StatusCode::INTERNAL_SERVER_ERROR,
@@ -129,7 +130,7 @@
     }
 }
 
-impl IntoResponse for SignedLocation {
+impl<L: UploadLength> IntoResponse for SignedLocation<L> {
     fn into_response(self) -> Response {
         let mut headers = tus::response_headers();
         match self.into_header_value() {
@@ -153,8 +154,7 @@
     check_kill_switch(&state)?;
 
     relay_log::trace!("Validating headers");
-    let upload_length = tus::validate_post_headers(&headers, meta.request_trust().is_trusted())
-        .map_err(Error::from)?;
+    let upload_length = tus::validate_post_headers(&headers).map_err(Error::from)?;
     let config = state.config();
 
     if upload_length.is_some_and(|len| len > config.max_upload_size()) {
@@ -177,6 +177,7 @@
     let scoping = check_request(&state, meta, upload_length, project).await?;
 
     // Unconditionally create the upload location:
+    relay_log::trace!("Creating upload location");
     let result = create(&state, scoping, upload_length).await;
     let location = result.inspect_err(|e| {
         relay_log::warn!(error = e as &dyn std::error::Error, "create failed");
@@ -195,7 +196,7 @@
     meta: RequestMeta,
     headers: HeaderMap,
     Path(upload::LocationPath { project_id, key }): Path<upload::LocationPath>,
-    Query(upload::LocationQueryParams { length, signature }): Query<upload::LocationQueryParams>,
+    Query(LocationQueryParams { length, signature }): Query<LocationQueryParams<Provisional>>,
     body: Body,
 ) -> axum::response::Result<impl IntoResponse> {
     check_kill_switch(&state)?;
@@ -220,15 +221,15 @@
         })?;
 
     relay_log::trace!("Checking request");
-    let scoping = check_request(&state, meta, length, project).await?;
+    let scoping = check_request(&state, meta, length.value(), project).await?;
 
     let stream = body
         .into_data_stream()
         .map(|result| result.map_err(io::Error::other))
         .boxed();
     let stream = MeteredStream::new(stream, "upload");
 
-    let (lower_bound, upper_bound) = match length {
+    let (lower_bound, upper_bound) = match length.value() {
         None => (1, config.max_upload_size()),
         Some(u) => (u, u),
     };
@@ -282,7 +283,7 @@
     state: &ServiceState,
     scoping: Scoping,
     upload_length: Option<usize>,
-) -> Result<SignedLocation, Error> {
+) -> Result<SignedLocation<Provisional>, Error> {
     let location = state
         .upload()
         .send(upload::Create {
@@ -297,9 +298,9 @@
 async fn upload(
     state: &ServiceState,
     scoping: Scoping,
-    location: SignedLocation,
+    location: SignedLocation<Provisional>,
     stream: BoundedStream<MeteredStream<ByteStream>>,
-) -> Result<SignedLocation, Error> {
+) -> Result<SignedLocation<Final>, Error> {
     let location = state
         .upload()
         .send(upload::Stream {

relay-server/src/processing/utils/attachments.rs

@@ -40,23 +40,22 @@ fn validate(item: &Item, config: &Config) -> Result<(), ProcessingError> {
 
     #[cfg(feature = "processing")]
     {
+        use crate::services::upload::{Final, SignedLocation};
+
         if !item.is_attachment_ref() {
             return Ok(());
         }
 
         let payload = item.payload();
         let payload: AttachmentPlaceholder =
             serde_json::from_slice(&payload).map_err(|_| ProcessingError::InvalidAttachmentRef)?;
-        let signed_location =
-            crate::services::upload::SignedLocation::try_from_str(payload.location)
-                .ok_or(ProcessingError::InvalidAttachmentRef)?;
+        let signed_location: SignedLocation<Final> = SignedLocation::try_from_str(payload.location)
+            .ok_or(ProcessingError::InvalidAttachmentRef)?;
         // NOTE: Using the received timestamp here breaks tests without a pop-relay.
         let location = signed_location
             .verify(chrono::Utc::now(), config)
             .map_err(|_| ProcessingError::InvalidAttachmentRef)?;
-        let signed_length = location
-            .length
-            .ok_or(ProcessingError::InvalidAttachmentRef)?;
+        let signed_length = location.length.into_inner();
 
         match item.attachment_body_size() == signed_length {
             true => Ok(()),

relay-server/src/services/store.rs

@@ -39,7 +39,7 @@ use crate::metrics::{ArrayEncoding, BucketEncoder, MetricOutcomes};
 use crate::service::ServiceError;
 use crate::services::global_config::GlobalConfigHandle;
 use crate::services::outcome::{DiscardReason, Outcome, TrackOutcome};
-use crate::services::upload::SignedLocation;
+use crate::services::upload::{Final, SignedLocation};
 use crate::statsd::{RelayCounters, RelayGauges, RelayTimers};
 use crate::utils::{self, FormDataIter};
 
@@ -1031,7 +1031,7 @@ impl StoreService {
         let payload = item.payload();
         let placeholder: AttachmentPlaceholder<'_> =
             serde_json::from_slice(&payload).map_err(|_| StoreError::InvalidAttachmentRef)?;
-        let location = SignedLocation::try_from_str(placeholder.location)
+        let location = SignedLocation::<Final>::try_from_str(placeholder.location)
             .ok_or(StoreError::InvalidAttachmentRef)?
             .verify(Utc::now(), &self.config)
             .map_err(|_| StoreError::InvalidAttachmentRef)?;