From 2d6949ff2561a91a0e4d672d6c97c80b184de32b Mon Sep 17 00:00:00 2001 From: Joris Bayer Date: Fri, 26 Jun 2026 12:44:45 +0200 Subject: [PATCH 01/12] build(objectstore): Update library --- Cargo.lock | 82 +++++++++++++++++++++--------------------------------- 1 file changed, 32 insertions(+), 50 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index af2180e5df3..0bd1af91c21 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1415,7 +1415,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -2013,6 +2013,16 @@ version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "135b12329e5e3ce057a9f972339ea52bc954fe1e9358ef27f95e89716fbc5424" +[[package]] +name = "humantime-serde" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57a3db5ea5923d99402c94e9feb261dc5ee9b4efa158b0315f788cf549cc200c" +dependencies = [ + "humantime", + "serde", +] + [[package]] name = "hyper" version = "1.9.0" @@ -2312,7 +2322,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -3307,9 +3317,9 @@ dependencies = [ [[package]] name = "objectstore-client" -version = "0.1.8" +version = "0.1.12" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "96e5089ca231d450c233b98f2bed7ce3f4cb43bc512df5d638f8ce93adb5211d" +checksum = "2713028097b2f0f00696d7d389a877f7370c02813548124c4422941f669732f3" dependencies = [ "async-compression", "bytes", @@ -3320,22 +3330,24 @@ dependencies = [ "objectstore-types", "percent-encoding", "reqwest", - "sentry-core 0.48.1", + "sentry-core", "serde", "thiserror 2.0.18", "tokio", "tokio-util", "url", + "zstd-safe", ] [[package]] name = "objectstore-types" -version = "0.1.8" +version = "0.1.12" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "05c1eee221b1c0eb5cf6c6427b4cca12ec3a60de48e6f1be4c9b3f5fdaba5290" +checksum = "c83aedfcc050322caefbfdba0372518f316624995f04f97a1f00cafb08ce1d0c" dependencies = [ "http", "humantime", + "humantime-serde", "mediatype", "serde", "thiserror 2.0.18", @@ -4622,7 +4634,7 @@ dependencies = [ "relay-common", "relay-crash", "sentry", - "sentry-core 0.47.0", + "sentry-core", "serde", "tracing", "tracing-subscriber", @@ -5227,7 +5239,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -5382,7 +5394,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5b55fb86dfd3a2f5f76ea78310a88f96c4ea21a3031f8d212443d56123fd0521" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -5403,7 +5415,7 @@ dependencies = [ "reqwest", "sentry-backtrace", "sentry-contexts", - "sentry-core 0.47.0", + "sentry-core", "sentry-debug-images", "sentry-log", "sentry-panic", @@ -5421,7 +5433,7 @@ checksum = "46a8c2c1bd5c1f735e84f28b48e7d72efcaafc362b7541bc8253e60e8fcdffc6" dependencies = [ "backtrace", "regex", - "sentry-core 0.47.0", + "sentry-core", ] [[package]] @@ -5434,7 +5446,7 @@ dependencies = [ "libc", "os_info", "rustc_version", - "sentry-core 0.47.0", + "sentry-core", "uname", ] @@ -5451,19 +5463,6 @@ dependencies = [ "url", ] -[[package]] -name = "sentry-core" -version = "0.48.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d5f5abf20c42cb1593ec1638976e2647da55f79bccac956444c1707b6cce259a" -dependencies = [ - "rand 0.9.4", - "sentry-types 0.48.1", - "serde", - "serde_json", - "url", -] - [[package]] name = "sentry-debug-images" version = "0.47.0" @@ -5471,7 +5470,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dd9646a972b57896d4a92ed200cf76139f8e30b3cfd03b6662ae59926d26633c" dependencies = [ "findshlibs", - "sentry-core 0.47.0", + "sentry-core", ] [[package]] @@ -5498,7 +5497,7 @@ checksum = "235865e639f1d72414fa5d35374e105c292e2ee3a2f90919d961bbb486626ee6" dependencies = [ "bitflags", "log", - "sentry-core 0.47.0", + "sentry-core", ] [[package]] @@ -5508,7 +5507,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6127d3d304ba5ce0409401e85aae538e303a569f8dbb031bf64f9ba0f7174346" dependencies = [ "sentry-backtrace", - "sentry-core 0.47.0", + "sentry-core", ] [[package]] @@ -5531,7 +5530,7 @@ checksum = "61c5253dc4ad89863a866b93aeaaac1c9d60f2f774663b5024afe2d57e0a101c" dependencies = [ "http", "pin-project", - "sentry-core 0.47.0", + "sentry-core", "tower-layer", "tower-service", "url", @@ -5545,7 +5544,7 @@ checksum = "27701acc51e68db5281802b709010395bfcbcb128b1d0a4e5873680d3b47ff0c" dependencies = [ "bitflags", "sentry-backtrace", - "sentry-core 0.47.0", + "sentry-core", "tracing-core", "tracing-subscriber", ] @@ -5584,23 +5583,6 @@ dependencies = [ "uuid", ] -[[package]] -name = "sentry-types" -version = "0.48.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "82d8e81058ec155992191f61c7b29bfa7b2cf12012131e7cdc0678020898a7c9" -dependencies = [ - "debugid", - "hex", - "rand 0.9.4", - "serde", - "serde_json", - "thiserror 2.0.18", - "time", - "url", - "uuid", -] - [[package]] name = "sentry_protos" version = "0.7.1" @@ -6289,7 +6271,7 @@ dependencies = [ "getrandom 0.3.4", "once_cell", "rustix", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -7086,7 +7068,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.48.0", ] [[package]] From 14958016f6f67ed7c82cd70f1cd6d1db2e217269 Mon Sep 17 00:00:00 2001 From: Joris Bayer Date: Fri, 26 Jun 2026 16:19:29 +0200 Subject: [PATCH 02/12] ref: Create multipart upload --- Cargo.lock | 1 + Cargo.toml | 10 ++++- relay-server/src/services/objectstore.rs | 52 +++++++++++++++++++++++- relay-server/src/services/upload.rs | 19 ++++++++- 4 files changed, 77 insertions(+), 5 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 0bd1af91c21..c9d5902be77 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3322,6 +3322,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2713028097b2f0f00696d7d389a877f7370c02813548124c4422941f669732f3" dependencies = [ "async-compression", + "base64", "bytes", "futures-util", "infer", diff --git a/Cargo.toml b/Cargo.toml index 5932888d8d8..698ecaacf31 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -90,7 +90,13 @@ android_trace_log = { version = "0.3", features = ["serde"] } # Keep it pinned until it's possible to disable backtrace. anyhow = "=1.0.69" arc-swap = "1" -async-compression = { version = "0.4", features = ["tokio", "gzip", "brotli", "deflate", "zstd"] } +async-compression = { version = "0.4", features = [ + "tokio", + "gzip", + "brotli", + "deflate", + "zstd", +] } async-trait = "0.1" axum = "0.8" axum-extra = "0.12" @@ -153,7 +159,7 @@ metrics = "0.24" metrics-exporter-dogstatsd = "0.9" num-traits = "0.2" num_cpus = "1" -objectstore-client = "0.1" +objectstore-client = { version = "0.1", features = ["multipart"] } opentelemetry-semantic-conventions = "0.31" opentelemetry-proto = { version = "0.31", default-features = false } papaya = "0.2" diff --git a/relay-server/src/services/objectstore.rs b/relay-server/src/services/objectstore.rs index 0ca9ae7eacc..1af2c6d971e 100644 --- a/relay-server/src/services/objectstore.rs +++ b/relay-server/src/services/objectstore.rs @@ -9,7 +9,8 @@ use bytes::Bytes; use futures::StreamExt; use http::StatusCode; use objectstore_client::{ - Client, ExpirationPolicy, SecretKey as SigningKey, Session, TokenGenerator, Usecase, + Client, Compression, ExpirationPolicy, SecretKey as SigningKey, Session, TokenGenerator, + Usecase, }; use relay_base_schema::organization::OrganizationId; use relay_base_schema::project::ProjectId; @@ -43,6 +44,7 @@ pub enum Objectstore { TraceAttachment(Managed), EventAttachment(Managed), RawProfile(Managed), + Create(Create, Sender>), Stream(Stream, Sender>), } @@ -54,6 +56,7 @@ impl Objectstore { Self::EventAttachment(_) => MessageKind::EventAttachment, Self::RawProfile(_) => MessageKind::RawProfile, Self::Stream { .. } => MessageKind::Stream, + Self::Create { .. } => MessageKind::Create, } } @@ -68,6 +71,7 @@ impl Objectstore { Self::EventAttachment(_) => 1, Self::RawProfile(_) => 1, Self::Stream { .. } => 1, + Self::Create { .. } => 0, } } } @@ -114,6 +118,7 @@ enum MessageKind { TraceAttachment, RawProfile, Stream, + Create, } impl MessageKind { @@ -124,10 +129,26 @@ impl MessageKind { Self::TraceAttachment => "attachment_v2", Self::RawProfile => "profile_raw", Self::Stream => "stream", + Self::Create => "create", } } } +/// A request to create a new objectstore multipart upload. +pub struct Create { + pub organization_id: OrganizationId, + pub project_id: ProjectId, + pub key: String, +} + +impl FromMessage for Objectstore { + type Response = AsyncResponse>; + + fn from_message(message: Create, sender: Sender>) -> Self { + Self::Create(message, sender) + } +} + /// A stream that can be uploaded to objectstore. pub struct Stream { pub organization_id: OrganizationId, @@ -424,6 +445,9 @@ impl LoadShed for ObjectstoreService { Objectstore::Stream(_, sender) => { sender.send(Err(error)); } + Objectstore::Create(_, sender) => { + sender.send(Err(error)); + } } } } @@ -463,6 +487,13 @@ impl ObjectstoreServiceInner { Objectstore::RawProfile(profile) => { self.handle_raw_profile(profile).await; } + Objectstore::Create(create, sender) => { + let result = self.handle_create(create).await; + if let Err(error) = &result { + error.log(MessageKind::Create); + } + sender.send(result); + } Objectstore::Stream(stream, sender) => { let result = self.handle_stream(stream).await; if let Err(error) = &result { @@ -710,6 +741,25 @@ impl ObjectstoreServiceInner { Ok(Some(stored_key)) } + async fn handle_create(&self, create: Create) -> Result { + let Create { + organization_id, + project_id, + key, + } = create; + let session = self.session(&self.event_attachments, organization_id, project_id)?; + + let multipart_upload = session + .initiate_multipart_upload() + .key(&key) + .compression(Compression::Zstd) // make explicit because parts need to be manually compressed. + .send() + .await?; + debug_assert_eq!(&key, multipart_upload.key()); + + Ok(ObjectstoreKey(key)) + } + async fn handle_stream(&self, stream: Stream) -> Result { let Stream { organization_id, diff --git a/relay-server/src/services/upload.rs b/relay-server/src/services/upload.rs index 79f320b245c..0b7ddd1004a 100644 --- a/relay-server/src/services/upload.rs +++ b/relay-server/src/services/upload.rs @@ -272,9 +272,24 @@ impl Service { SignedLocation::try_from_response(response) } #[cfg(feature = "processing")] - Backend::Objectstore { addr: _, config } => { - // We can create & sign a location right here, no need to query the objectstore service. + Backend::Objectstore { addr, config } => { let key = Uuid::now_v7().as_simple().to_string(); + let Scoping { + organization_id, + project_id, + .. + } = project.scoping; + + let key = addr + .send(objectstore::Create { + organization_id, + project_id, + key, + }) + .await + .map_err(Error::ObjectstoreServiceUnavailable)?? + .into_inner(); + Location { project_id: project.scoping.project_id, key, From f74c1ca1525376f2ae08abe77405d7d3f1007139 Mon Sep 17 00:00:00 2001 From: Joris Bayer Date: Fri, 26 Jun 2026 17:07:06 +0200 Subject: [PATCH 03/12] wip --- relay-server/src/services/objectstore.rs | 146 ++++++++++++++--------- relay-server/src/services/upload.rs | 19 ++- 2 files changed, 106 insertions(+), 59 deletions(-) diff --git a/relay-server/src/services/objectstore.rs b/relay-server/src/services/objectstore.rs index 1af2c6d971e..28335c4c759 100644 --- a/relay-server/src/services/objectstore.rs +++ b/relay-server/src/services/objectstore.rs @@ -9,8 +9,8 @@ use bytes::Bytes; use futures::StreamExt; use http::StatusCode; use objectstore_client::{ - Client, Compression, ExpirationPolicy, SecretKey as SigningKey, Session, TokenGenerator, - Usecase, + Client, Compression, ExpirationPolicy, MultipartUpload, SecretKey as SigningKey, Session, + TokenGenerator, Usecase, }; use relay_base_schema::organization::OrganizationId; use relay_base_schema::project::ProjectId; @@ -44,7 +44,7 @@ pub enum Objectstore { TraceAttachment(Managed), EventAttachment(Managed), RawProfile(Managed), - Create(Create, Sender>), + Create(Create, Sender>), Stream(Stream, Sender>), } @@ -142,9 +142,9 @@ pub struct Create { } impl FromMessage for Objectstore { - type Response = AsyncResponse>; + type Response = AsyncResponse>; - fn from_message(message: Create, sender: Sender>) -> Self { + fn from_message(message: Create, sender: Sender>) -> Self { Self::Create(message, sender) } } @@ -153,7 +153,7 @@ impl FromMessage for Objectstore { pub struct Stream { pub organization_id: OrganizationId, pub project_id: ProjectId, - pub key: String, + pub upload_ref: UploadRef, pub stream: BoundedStream>, } @@ -321,6 +321,14 @@ impl ObjectstoreKey { } } +/// Identifier needed to resume an existing upload. +pub struct UploadRef { + /// They key of the file (chosen by relay). + pub key: String, + /// The ID of the multipart upload session (chosen by objectstore). + pub upload_id: String, +} + impl fmt::Display for ObjectstoreKey { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { self.0.fmt(f) @@ -741,7 +749,7 @@ impl ObjectstoreServiceInner { Ok(Some(stored_key)) } - async fn handle_create(&self, create: Create) -> Result { + async fn handle_create(&self, create: Create) -> Result { let Create { organization_id, project_id, @@ -756,24 +764,26 @@ impl ObjectstoreServiceInner { .send() .await?; debug_assert_eq!(&key, multipart_upload.key()); + let upload_id = multipart_upload.upload_id().to_string(); - Ok(ObjectstoreKey(key)) + Ok(UploadRef { key, upload_id }) } async fn handle_stream(&self, stream: Stream) -> Result { let Stream { organization_id, project_id, - key, + upload_ref, stream, } = stream; let session = self.session(&self.event_attachments, organization_id, project_id)?; + let UploadRef { key, upload_id } = upload_ref; + let multipart_upload = session.resume_multipart_upload(key, upload_id)?; + self.upload( MessageKind::Stream, - &session, - Some(key), - Body::Stream(TakeOnce::new(stream)), + BodyTarget::Stream(TakeOnce::new(stream), &multipart_upload), None, None, ) @@ -792,9 +802,11 @@ impl ObjectstoreServiceInner { let retention_hours = retention.checked_mul(24); self.upload( kind, - session, - key, - Body::Bytes(payload), + BodyTarget::Bytes { + body: payload, + session, + key, + }, retention_hours, content_type, ) @@ -804,16 +816,14 @@ impl ObjectstoreServiceInner { async fn upload( &self, kind: MessageKind, - session: &Session, - key: Option, - body: Body, + body: BodyTarget<'_>, retention_hours: Option, content_type: Option, ) -> Result { let mut attempts = 0; let timeout = match &body { - Body::Bytes(_) => self.timeout, - Body::Stream(_) => self.stream_timeout, + BodyTarget::Bytes { .. } => self.timeout, + BodyTarget::Stream { .. } => self.stream_timeout, }; let result = tokio::time::timeout(timeout, async { let mut result = None; @@ -823,15 +833,8 @@ impl ObjectstoreServiceInner { }; attempts += 1; result.replace( - self.attempt_upload( - kind, - session, - key.clone(), - body, - retention_hours, - content_type, - ) - .await, + self.attempt_upload(kind, body, retention_hours, content_type) + .await, ); if attempts < self.max_attempts.get() @@ -866,29 +869,39 @@ impl ObjectstoreServiceInner { async fn attempt_upload( &self, kind: MessageKind, - session: &Session, - key: Option, - body: BodyAttempt, + body: BodyAttempt<'_>, retention_hours: Option, content_type: Option, ) -> Result { let mut request = match body { - BodyAttempt::Bytes(bytes) => session.put(bytes), - BodyAttempt::Stream(stream) => session.put_stream(stream.boxed()), + BodyAttempt::Bytes { body, session, key } => { + let mut request = session.put(body); + if let Some(content_type) = content_type { + request = request.content_type(content_type.as_str()); + } + if let Some(retention_hours) = retention_hours { + request = request.expiration_policy(ExpirationPolicy::TimeToLive( + Duration::from_hours(retention_hours.into()), + )); + } + if let Some(key) = key { + request = request.key(key); + } + let response = request.send().await?; + } + BodyAttempt::Stream(stream, multipart_upload) => { + while let Some((i, chunk)) = stream.enumerate().next().await { + // TODO: are chunks adequate size? + let chunk = chunk?; + let part_number = u32::try_from(i)?; + let part = multipart_upload.put(chunk, part_number, None).await?; + } + let mut part = multipart_upload + .put_stream(stream.boxed(), 0, 0, None) // FIXME: set content length + .await?; + } }; - if let Some(content_type) = content_type { - request = request.content_type(content_type.as_str()); - } - if let Some(retention_hours) = retention_hours { - request = request.expiration_policy(ExpirationPolicy::TimeToLive( - Duration::from_hours(retention_hours.into()), - )); - } - if let Some(key) = key { - request = request.key(key); - } - let response = relay_statsd::metric!( timer(RelayTimers::AttachmentUploadDuration), type = kind.as_str(), @@ -919,16 +932,32 @@ impl ObjectstoreServiceInner { /// Common interface for calls to [`ObjectstoreServiceInner::upload`]. /// /// This type is shared across retries. -enum Body { - Bytes(Bytes), - Stream(TakeOnce>>), +enum BodyTarget<'a> { + Bytes { + body: Bytes, + session: &'a Session, + key: Option, + }, + Stream( + TakeOnce>>, + &'a MultipartUpload, + ), } -impl Body { +impl BodyTarget<'_> { fn try_clone(&self) -> Option { match self { - Self::Bytes(bytes) => Some(BodyAttempt::Bytes(bytes.clone())), - Self::Stream(stream) => RetryableStream::new(stream.clone()).map(BodyAttempt::Stream), + Self::Bytes { + body: bytes, + session, + key, + } => Some(BodyAttempt::Bytes { + body: bytes.clone(), + session, + key, + }), + Self::Stream(stream, multipart_upload) => RetryableStream::new(stream.clone()) + .map(|stream| BodyAttempt::Stream(stream, multipart_upload)), } } } @@ -936,9 +965,16 @@ impl Body { /// Common interface for calls to [`ObjectstoreServiceInner::attempt_upload`]. /// /// This type is instantiated for every retry. -enum BodyAttempt { - Bytes(Bytes), - Stream(RetryableStream>>), +enum BodyAttempt<'a> { + Bytes { + body: Bytes, + session: &'a Session, + key: Option, + }, + Stream( + RetryableStream>>, + &'a MultipartUpload, + ), } fn is_retryable(error: &objectstore_client::Error) -> bool { diff --git a/relay-server/src/services/upload.rs b/relay-server/src/services/upload.rs index 0b7ddd1004a..b2f6cd21267 100644 --- a/relay-server/src/services/upload.rs +++ b/relay-server/src/services/upload.rs @@ -273,6 +273,8 @@ impl Service { } #[cfg(feature = "processing")] Backend::Objectstore { addr, config } => { + use crate::services::objectstore::UploadRef; + let key = Uuid::now_v7().as_simple().to_string(); let Scoping { organization_id, @@ -280,20 +282,20 @@ impl Service { .. } = project.scoping; - let key = addr + let UploadRef { key, upload_id } = addr .send(objectstore::Create { organization_id, project_id, key, }) .await - .map_err(Error::ObjectstoreServiceUnavailable)?? - .into_inner(); + .map_err(Error::ObjectstoreServiceUnavailable)??; Location { project_id: project.scoping.project_id, key, length: Provisional(length), + upload_id, other: Default::default(), } .try_sign(config) @@ -318,10 +320,13 @@ impl Service { } #[cfg(feature = "processing")] Backend::Objectstore { addr, config } => { + use crate::services::objectstore::UploadRef; + let Location { project_id, key, length, + upload_id, other, } = location.verify(received, config)?; @@ -334,7 +339,7 @@ impl Service { .send(objectstore::Stream { organization_id: scoping.organization_id, project_id, - key, + upload_ref: UploadRef { key, upload_id }, stream, }) .await @@ -346,6 +351,7 @@ impl Service { project_id, key, length, + upload_id, other, } .try_sign(config) @@ -440,6 +446,8 @@ pub struct Location { pub key: String, /// Value of the `Upload-Length` header. `None` if `Upload-Defer-Length: 1`. pub length: L, + /// The ID of the upload session, as provided by objectstore. + pub upload_id: Option, pub other: UploadParams, } @@ -449,16 +457,19 @@ impl Location { project_id, key, length, + upload_id, other, } = self; #[derive(Debug, Serialize)] struct QueryParams<'a> { pub upload_length: Option, + pub upload_id: &'a String, #[serde(flatten)] pub other: &'a UploadParams, } let params = QueryParams { upload_length: length.value(), + upload_id, other, }; let query = serde_urlencoded::to_string(params)?; From 2712d83d2516c5f31c6762f785d70ac1ed034244 Mon Sep 17 00:00:00 2001 From: Joris Bayer Date: Tue, 30 Jun 2026 13:35:33 +0200 Subject: [PATCH 04/12] wip --- relay-server/src/services/objectstore.rs | 57 +++++++++++++++--------- 1 file changed, 35 insertions(+), 22 deletions(-) diff --git a/relay-server/src/services/objectstore.rs b/relay-server/src/services/objectstore.rs index 28335c4c759..4cd07956f6c 100644 --- a/relay-server/src/services/objectstore.rs +++ b/relay-server/src/services/objectstore.rs @@ -20,6 +20,7 @@ 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}; @@ -38,6 +39,11 @@ use crate::utils::{BoundedStream, MeteredStream, RetryableStream, TakeOnce, find use super::outcome::Outcome; +/// Maximum size of an individual request to objectstore. +/// +/// This only applies to streams. In-memory attachments are never chunked. +const CHUNK_SIZE: usize = 100 * 1024 * 1024; // 100 MiB + /// Messages that the objectstore service can handle. pub enum Objectstore { Envelope(StoreEnvelope), @@ -873,7 +879,7 @@ impl ObjectstoreServiceInner { retention_hours: Option, content_type: Option, ) -> Result { - let mut request = match body { + match body { BodyAttempt::Bytes { body, session, key } => { let mut request = session.put(body); if let Some(content_type) = content_type { @@ -887,30 +893,37 @@ impl ObjectstoreServiceInner { if let Some(key) = key { request = request.key(key); } - let response = request.send().await?; + let response = relay_statsd::metric!( + timer(RelayTimers::AttachmentUploadDuration), + type = kind.as_str(), + { + request.send().await? + }); + + Ok(ObjectstoreKey(response.key)) } BodyAttempt::Stream(stream, multipart_upload) => { - while let Some((i, chunk)) = stream.enumerate().next().await { - // TODO: are chunks adequate size? - let chunk = chunk?; - let part_number = u32::try_from(i)?; - let part = multipart_upload.put(chunk, part_number, None).await?; - } - let mut part = multipart_upload - .put_stream(stream.boxed(), 0, 0, None) // FIXME: set content length - .await?; - } - }; + let reader = StreamReader::new(stream); + let mut rechunked = ReaderStream::with_capacity(reader, CHUNK_SIZE).enumerate(); - let response = relay_statsd::metric!( - timer(RelayTimers::AttachmentUploadDuration), - type = kind.as_str(), - { - request.send().await - } - )?; + let result = relay_statsd::metric!( + timer(RelayTimers::AttachmentUploadDuration), + type = kind.as_str(), + { + let mut parts = vec![]; + while let Some((i, chunk)) = rechunked.next().await { + let chunk = chunk?; + let part_number = u32::try_from(i) + .map_err(|_| objectstore_client::Error::InvalidPartNumber(u32::MAX))?; + let part = multipart_upload.put(chunk, part_number, None).await?; + parts.push(part); + } + multipart_upload.complete(parts).await? + }); - Ok(ObjectstoreKey(response.key)) + Ok(ObjectstoreKey(result)) + } + } } fn session( @@ -954,7 +967,7 @@ impl BodyTarget<'_> { } => Some(BodyAttempt::Bytes { body: bytes.clone(), session, - key, + key: key.clone(), }), Self::Stream(stream, multipart_upload) => RetryableStream::new(stream.clone()) .map(|stream| BodyAttempt::Stream(stream, multipart_upload)), From 98118b554be42ba7d3ebb49afba8378e6852e166 Mon Sep 17 00:00:00 2001 From: Joris Bayer Date: Thu, 2 Jul 2026 14:28:46 +0200 Subject: [PATCH 05/12] sentry-conventions --- relay-conventions/sentry-conventions | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/relay-conventions/sentry-conventions b/relay-conventions/sentry-conventions index bb21c4edc34..989dc716d72 160000 --- a/relay-conventions/sentry-conventions +++ b/relay-conventions/sentry-conventions @@ -1 +1 @@ -Subproject commit bb21c4edc3450fa18de19d6916705d50551557c6 +Subproject commit 989dc716d72d64e8dd30a3085414e15d41d6fcf7 From 56d3bb6e98ac3d6b8db86743bc1d29c22591847c Mon Sep 17 00:00:00 2001 From: Joris Bayer Date: Thu, 2 Jul 2026 14:54:37 +0200 Subject: [PATCH 06/12] wip: objectstore --- relay-server/src/services/objectstore.rs | 63 ++++++++++++++---------- relay-server/src/services/upload.rs | 5 +- 2 files changed, 42 insertions(+), 26 deletions(-) diff --git a/relay-server/src/services/objectstore.rs b/relay-server/src/services/objectstore.rs index 5ae89d6bd42..116aa865a94 100644 --- a/relay-server/src/services/objectstore.rs +++ b/relay-server/src/services/objectstore.rs @@ -9,8 +9,8 @@ use bytes::Bytes; use futures::StreamExt; use http::StatusCode; use objectstore_client::{ - Client, Compression, ExpirationPolicy, MultipartUpload, SecretKey as SigningKey, Session, - TokenGenerator, Usecase, + Client, Compression, ExpirationPolicy, SecretKey as SigningKey, Session, TokenGenerator, + Usecase, }; use relay_base_schema::organization::OrganizationId; use relay_base_schema::project::ProjectId; @@ -328,6 +328,7 @@ impl ObjectstoreKey { } /// Identifier needed to resume an existing upload. +#[derive(Debug, Clone)] pub struct UploadRef { /// They key of the file (chosen by relay). pub key: String, @@ -782,12 +783,13 @@ impl ObjectstoreServiceInner { } = stream; let session = self.session(&self.event_attachments, organization_id, project_id)?; - let UploadRef { key, upload_id } = upload_ref; - let multipart_upload = session.resume_multipart_upload(key, upload_id)?; - self.upload( MessageKind::Stream, - BodyTarget::Stream(TakeOnce::new(stream), &multipart_upload), + &session, + BodyTarget::Stream { + body: TakeOnce::new(stream), + upload_ref, + }, None, None, ) @@ -806,6 +808,7 @@ impl ObjectstoreServiceInner { let retention_hours = retention.checked_mul(24); self.upload( kind, + session, BodyTarget::Bytes { body: payload, session, @@ -820,6 +823,7 @@ impl ObjectstoreServiceInner { async fn upload( &self, kind: MessageKind, + session: &Session, body: BodyTarget<'_>, retention_hours: Option, content_type: Option, @@ -837,7 +841,7 @@ impl ObjectstoreServiceInner { }; attempts += 1; result.replace( - self.attempt_upload(kind, body, retention_hours, content_type) + self.attempt_upload(kind, session, body, retention_hours, content_type) .await, ); @@ -873,12 +877,13 @@ impl ObjectstoreServiceInner { async fn attempt_upload( &self, kind: MessageKind, - body: BodyAttempt<'_>, + session: &Session, + body: BodyAttempt, retention_hours: Option, content_type: Option, ) -> Result { match body { - BodyAttempt::Bytes { body, session, key } => { + BodyAttempt::Bytes { body, key } => { let mut request = session.put(body); if let Some(content_type) = content_type { request = request.content_type(content_type.as_str()); @@ -900,8 +905,11 @@ impl ObjectstoreServiceInner { Ok(ObjectstoreKey(response.key)) } - BodyAttempt::Stream(stream, multipart_upload) => { - let reader = StreamReader::new(stream); + BodyAttempt::Stream { body, upload_ref } => { + let UploadRef { key, upload_id } = upload_ref; + let multipart_upload = session.resume_multipart_upload(key, upload_id)?; + + let reader = StreamReader::new(body); let mut rechunked = ReaderStream::with_capacity(reader, CHUNK_SIZE).enumerate(); let result = relay_statsd::metric!( @@ -949,10 +957,10 @@ enum BodyTarget<'a> { session: &'a Session, key: Option, }, - Stream( - TakeOnce>>, - &'a MultipartUpload, - ), + Stream { + body: TakeOnce>>, + upload_ref: UploadRef, + }, } impl BodyTarget<'_> { @@ -964,11 +972,14 @@ impl BodyTarget<'_> { key, } => Some(BodyAttempt::Bytes { body: bytes.clone(), - session, key: key.clone(), }), - Self::Stream(stream, multipart_upload) => RetryableStream::new(stream.clone()) - .map(|stream| BodyAttempt::Stream(stream, multipart_upload)), + Self::Stream { body, upload_ref } => { + RetryableStream::new(body.clone()).map(|body| BodyAttempt::Stream { + body, + upload_ref: upload_ref.clone(), + }) + } } } } @@ -976,16 +987,15 @@ impl BodyTarget<'_> { /// Common interface for calls to [`ObjectstoreServiceInner::attempt_upload`]. /// /// This type is instantiated for every retry. -enum BodyAttempt<'a> { +enum BodyAttempt { Bytes { body: Bytes, - session: &'a Session, key: Option, }, - Stream( - RetryableStream>>, - &'a MultipartUpload, - ), + Stream { + body: RetryableStream>>, + upload_ref: UploadRef, + }, } fn is_retryable(error: &objectstore_client::Error) -> bool { @@ -1064,7 +1074,10 @@ mod tests { .send(Stream { organization_id: OrganizationId::new(0), project_id: ProjectId::new(1), - key: "my_file".into(), + upload_ref: UploadRef { + key: "my_file".into(), + upload_id: "my_upload".into(), + }, stream, }) .await diff --git a/relay-server/src/services/upload.rs b/relay-server/src/services/upload.rs index b2f6cd21267..0180156dac1 100644 --- a/relay-server/src/services/upload.rs +++ b/relay-server/src/services/upload.rs @@ -295,7 +295,7 @@ impl Service { project_id: project.scoping.project_id, key, length: Provisional(length), - upload_id, + upload_id: Some(upload_id), other: Default::default(), } .try_sign(config) @@ -570,6 +570,7 @@ impl SignedLocation { pub fn from_parts( project_id: ProjectId, key: String, + upload_id: Option, length: L, signature: String, other: UploadParams, @@ -578,6 +579,7 @@ impl SignedLocation { location: Location { project_id, key, + upload_id, length, other, }, @@ -895,6 +897,7 @@ mod tests { project_id: ProjectId::new(42), key: "upload-key".to_owned(), length: Provisional(Some(123)), + upload_id: Some("my_upload".to_owned()), other: UploadParams::default(), } } From 8262bd904cc9b90cc6807699e7d00e0385b7e873 Mon Sep 17 00:00:00 2001 From: Joris Bayer Date: Fri, 3 Jul 2026 11:23:25 +0200 Subject: [PATCH 07/12] wip: location data --- relay-server/src/endpoints/upload.rs | 13 +- .../src/processing/utils/attachments.rs | 2 +- relay-server/src/services/objectstore.rs | 19 +-- relay-server/src/services/upload.rs | 133 ++++++++++-------- 4 files changed, 89 insertions(+), 78 deletions(-) diff --git a/relay-server/src/endpoints/upload.rs b/relay-server/src/endpoints/upload.rs index 021a7bf51c9..f78d785969c 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, Final, LocationData, LocationQueryParams, ProjectContext, Provisional, + SignedLocation, }; use crate::services::upstream::UpstreamRequestError; use crate::statsd::RelayCounters; @@ -133,7 +133,7 @@ impl IntoResponse for Error { } } -impl IntoResponse for SignedLocation { +impl IntoResponse for SignedLocation { fn into_response(self) -> Response { let mut headers = tus::response_headers(); match self.into_header_value() { @@ -203,7 +203,7 @@ async fn handle_patch( headers: HeaderMap, Path(upload::LocationPath { project_id, key }): Path, Query(LocationQueryParams { - upload_length, + location_data, upload_signature, other, }): Query>, @@ -214,8 +214,9 @@ async fn handle_patch( relay_log::trace!("Validating headers"); tus::validate_patch_headers(&headers).map_err(Error::from)?; + let upload_length = location_data.upload_length(); let location = - SignedLocation::from_parts(project_id, key, upload_length, upload_signature, other); + SignedLocation::from_parts(project_id, key, location_data, upload_signature, other); let config = state.config(); @@ -240,7 +241,7 @@ async fn handle_patch( .boxed(); let stream = MeteredStream::new(stream, "upload"); - let (lower_bound, upper_bound) = match upload_length.value() { + let (lower_bound, upper_bound) = match upload_length { None => (1, config.max_upload_size()), Some(u) => (u, u), }; diff --git a/relay-server/src/processing/utils/attachments.rs b/relay-server/src/processing/utils/attachments.rs index 89d3a5be63a..ff7a0ba0a2f 100644 --- a/relay-server/src/processing/utils/attachments.rs +++ b/relay-server/src/processing/utils/attachments.rs @@ -55,7 +55,7 @@ fn validate(item: &Item, config: &Config) -> Result<(), ProcessingError> { let location = signed_location .verify(chrono::Utc::now(), config) .map_err(|_| ProcessingError::InvalidAttachmentRef)?; - let signed_length = location.length.into_inner(); + let signed_length = location.location_data.length(); match item.attachment_body_size() == signed_length { true => Ok(()), diff --git a/relay-server/src/services/objectstore.rs b/relay-server/src/services/objectstore.rs index 116aa865a94..5950d873aba 100644 --- a/relay-server/src/services/objectstore.rs +++ b/relay-server/src/services/objectstore.rs @@ -809,11 +809,7 @@ impl ObjectstoreServiceInner { self.upload( kind, session, - BodyTarget::Bytes { - body: payload, - session, - key, - }, + BodyTarget::Bytes { body: payload, key }, retention_hours, content_type, ) @@ -824,7 +820,7 @@ impl ObjectstoreServiceInner { &self, kind: MessageKind, session: &Session, - body: BodyTarget<'_>, + body: BodyTarget, retention_hours: Option, content_type: Option, ) -> Result { @@ -951,10 +947,9 @@ impl ObjectstoreServiceInner { /// Common interface for calls to [`ObjectstoreServiceInner::upload`]. /// /// This type is shared across retries. -enum BodyTarget<'a> { +enum BodyTarget { Bytes { body: Bytes, - session: &'a Session, key: Option, }, Stream { @@ -963,14 +958,10 @@ enum BodyTarget<'a> { }, } -impl BodyTarget<'_> { +impl BodyTarget { fn try_clone(&self) -> Option { match self { - Self::Bytes { - body: bytes, - session, - key, - } => Some(BodyAttempt::Bytes { + Self::Bytes { body: bytes, key } => Some(BodyAttempt::Bytes { body: bytes.clone(), key: key.clone(), }), diff --git a/relay-server/src/services/upload.rs b/relay-server/src/services/upload.rs index 0180156dac1..473ed59c2eb 100644 --- a/relay-server/src/services/upload.rs +++ b/relay-server/src/services/upload.rs @@ -227,12 +227,12 @@ pub struct Service { } /// A response channel that emits a metric for each response. -pub struct InstrumentedSender { +pub struct InstrumentedSender { metric: RelayCounters, inner: Sender, Error>>, } -impl InstrumentedSender { +impl InstrumentedSender { fn send(self, result: Result, Error>) { let result_msg = match &result { Ok(_) => "success", @@ -294,8 +294,10 @@ impl Service { Location { project_id: project.scoping.project_id, key, - length: Provisional(length), - upload_id: Some(upload_id), + location_data: Provisional { + upload_length: length, + upload_id, + }, other: Default::default(), } .try_sign(config) @@ -325,33 +327,39 @@ impl Service { let Location { project_id, key, - length, - upload_id, + location_data, other, } = location.verify(received, config)?; let scoping = project.scoping; debug_assert_eq!(scoping.project_id, project_id); - debug_assert!(stream.length().is_none_or(|l| Some(l) == length.value())); + debug_assert!( + stream + .length() + .is_none_or(|l| Some(l) == location_data.upload_length()) + ); let byte_counter = stream.byte_counter(); let key = addr .send(objectstore::Stream { organization_id: scoping.organization_id, project_id, - upload_ref: UploadRef { key, upload_id }, + upload_ref: UploadRef { + key, + upload_id: location_data.upload_id, + }, stream, }) .await .map_err(Error::ObjectstoreServiceUnavailable)?? .into_inner(); - let length = Final(byte_counter.get()); Location { project_id, key, - length, - upload_id, + location_data: Final { + upload_length: byte_counter.get(), + }, other, } .try_sign(config) @@ -361,7 +369,7 @@ impl Service { async fn timeout(&self, future: F) -> Result, Error> where - L: UploadLength, + L: LocationData, F: IntoFuture, Error>>, { tokio::time::timeout(self.timeout, future).await? @@ -392,23 +400,30 @@ impl LoadShed for Service { } } -/// An interface for known or unknown upload lengths. +/// An interface to retrieve query parameters from different location types. /// /// This allows code sharing between [`Provisional`] and [`Final`] upload locations. -pub trait UploadLength: for<'de> Deserialize<'de> { - fn value(&self) -> Option; +pub trait LocationData: for<'de> Deserialize<'de> { + fn upload_length(&self) -> Option; + fn upload_id(&self) -> Option<&str>; } /// A provisional upload length which may or may not yet be known. /// /// See also [`Final`]. -#[derive(Debug, Clone, Copy, Deserialize)] -#[serde(transparent)] -pub struct Provisional(Option); +#[derive(Debug, Clone, Deserialize)] +pub struct Provisional { + upload_length: Option, + upload_id: String, +} -impl UploadLength for Provisional { - fn value(&self) -> Option { - self.0 +impl LocationData for Provisional { + fn upload_length(&self) -> Option { + self.upload_length + } + + fn upload_id(&self) -> Option<&str> { + Some(&self.upload_id) } } @@ -416,18 +431,24 @@ impl UploadLength for Provisional { /// /// See also [`Provisional`]. #[derive(Debug, Clone, Copy, Deserialize)] -pub struct Final(usize); +pub struct Final { + upload_length: usize, +} impl Final { /// Get the value. - pub fn into_inner(self) -> usize { - self.0 + pub fn length(&self) -> usize { + self.upload_length } } -impl UploadLength for Final { - fn value(&self) -> Option { - Some(self.0) +impl LocationData for Final { + fn upload_length(&self) -> Option { + Some(self.upload_length) + } + + fn upload_id(&self) -> Option<&str> { + None } } @@ -444,32 +465,30 @@ pub struct Location { pub project_id: ProjectId, /// Objectstore identifier. pub key: String, - /// Value of the `Upload-Length` header. `None` if `Upload-Defer-Length: 1`. - pub length: L, - /// The ID of the upload session, as provided by objectstore. - pub upload_id: Option, + /// Upload length and upload ID. + pub location_data: L, + /// Additional fields, retained for forward compatibility. pub other: UploadParams, } -impl Location { +impl Location { fn try_to_uri(&self) -> Result { let Location { project_id, key, - length, - upload_id, + location_data, other, } = self; #[derive(Debug, Serialize)] struct QueryParams<'a> { pub upload_length: Option, - pub upload_id: &'a String, + pub upload_id: Option<&'a str>, #[serde(flatten)] pub other: &'a UploadParams, } let params = QueryParams { - upload_length: length.value(), - upload_id, + upload_length: location_data.upload_length(), + upload_id: location_data.upload_id(), other, }; let query = serde_urlencoded::to_string(params)?; @@ -507,11 +526,10 @@ pub struct LocationPath { /// Query parameters for the upload endpoint. #[derive(Debug, Deserialize)] -#[serde(bound = "L: UploadLength")] -pub struct LocationQueryParams { - #[serde(alias = "length")] - pub upload_length: L, - #[serde(alias = "signature")] +#[serde(bound = "L: LocationData")] +pub struct LocationQueryParams { + #[serde(flatten)] + pub location_data: L, pub upload_signature: String, #[serde(flatten)] pub other: UploadParams, @@ -558,20 +576,19 @@ impl<'de> Deserialize<'de> for UploadParams { /// A verifiable [`Location`] signed by this Relay or an upstream Relay. #[derive(Debug)] -pub struct SignedLocation { +pub struct SignedLocation { location: Location, signature: Signature, } -impl SignedLocation { +impl SignedLocation { /// Creates an unverified location from path and query params. /// /// Call `verify` to make sure the signature is correct. pub fn from_parts( project_id: ProjectId, key: String, - upload_id: Option, - length: L, + location_data: L, signature: String, other: UploadParams, ) -> Self { @@ -579,8 +596,7 @@ impl SignedLocation { location: Location { project_id, key, - upload_id, - length, + location_data, other, }, signature: Signature(signature), @@ -644,7 +660,7 @@ impl SignedLocation { impl SignedLocation where - L: UploadLength, + L: LocationData, LocationQueryParams: for<'de> Deserialize<'de>, { fn try_from_response(response: Response) -> Result { @@ -681,7 +697,7 @@ where // Parse query parameters. let LocationQueryParams { - upload_length, + location_data: upload_length, upload_signature, other, } = serde_urlencoded::from_str(query).ok()?; @@ -896,8 +912,10 @@ mod tests { Location { project_id: ProjectId::new(42), key: "upload-key".to_owned(), - length: Provisional(Some(123)), - upload_id: Some("my_upload".to_owned()), + location_data: Provisional { + upload_length: Some(123), + upload_id: "my_upload".to_owned(), + }, other: UploadParams::default(), } } @@ -989,24 +1007,25 @@ mod tests { #[test] fn parse_location_incomplete() { - let url = "signature=foo"; + let url = "upload_signature=foo&upload_id=abc"; // Can only parse provisional: let provisional: LocationQueryParams = serde_urlencoded::from_str(url).unwrap(); - assert!(provisional.upload_length.0.is_none()); + assert!(provisional.location_data.upload_length.is_none()); assert!(serde_urlencoded::from_str::>(url).is_err()); } #[test] fn parse_location_complete() { - let json = r#"signature=foo&length=123"#; + let json = r#"upload_signature=foo&upload_length=123&upload_id=abc"#; let provisional: LocationQueryParams = serde_urlencoded::from_str(json).unwrap(); - assert_eq!(provisional.upload_length.0, Some(123)); + assert_eq!(provisional.location_data.upload_length, Some(123)); + assert_eq!(&provisional.location_data.upload_id, "abc"); let full: LocationQueryParams = serde_urlencoded::from_str(json).unwrap(); - assert_eq!(full.upload_length.0, 123); + assert_eq!(full.location_data.upload_length, 123); } #[test] From c7a4e3162eada0a0e4cb16cca6e193ff13f28283 Mon Sep 17 00:00:00 2001 From: Joris Bayer Date: Fri, 3 Jul 2026 11:40:22 +0200 Subject: [PATCH 08/12] wip: types --- relay-server/src/endpoints/upload.rs | 14 +++++--- relay-server/src/services/upload.rs | 49 +++++++++++++++++++++------- 2 files changed, 48 insertions(+), 15 deletions(-) diff --git a/relay-server/src/endpoints/upload.rs b/relay-server/src/endpoints/upload.rs index f78d785969c..07f3705b2db 100644 --- a/relay-server/src/endpoints/upload.rs +++ b/relay-server/src/endpoints/upload.rs @@ -203,7 +203,8 @@ async fn handle_patch( headers: HeaderMap, Path(upload::LocationPath { project_id, key }): Path, Query(LocationQueryParams { - location_data, + upload_length, + upload_id, upload_signature, other, }): Query>, @@ -214,9 +215,14 @@ async fn handle_patch( relay_log::trace!("Validating headers"); tus::validate_patch_headers(&headers).map_err(Error::from)?; - let upload_length = location_data.upload_length(); - let location = - SignedLocation::from_parts(project_id, key, location_data, upload_signature, other); + let location = SignedLocation::from_parts( + project_id, + key, + upload_length, + upload_id, + upload_signature, + other, + ); let config = state.config(); diff --git a/relay-server/src/services/upload.rs b/relay-server/src/services/upload.rs index 473ed59c2eb..0979493ad8d 100644 --- a/relay-server/src/services/upload.rs +++ b/relay-server/src/services/upload.rs @@ -404,6 +404,11 @@ impl LoadShed for Service { /// /// This allows code sharing between [`Provisional`] and [`Final`] upload locations. pub trait LocationData: for<'de> Deserialize<'de> { + type LengthType: for<'de> Deserialize<'de>; + type IdType: for<'de> Deserialize<'de>; + + fn from_parts(upload_length: Self::LengthType, upload_id: Self::IdType) -> Self; + fn upload_length(&self) -> Option; fn upload_id(&self) -> Option<&str>; } @@ -413,11 +418,21 @@ pub trait LocationData: for<'de> Deserialize<'de> { /// See also [`Final`]. #[derive(Debug, Clone, Deserialize)] pub struct Provisional { - upload_length: Option, - upload_id: String, + pub upload_length: Option, + pub upload_id: String, } impl LocationData for Provisional { + type LengthType = Option; + type IdType = String; + + fn from_parts(upload_length: Self::LengthType, upload_id: Self::IdType) -> Self { + Self { + upload_length, + upload_id, + } + } + fn upload_length(&self) -> Option { self.upload_length } @@ -443,6 +458,13 @@ impl Final { } impl LocationData for Final { + type LengthType = usize; + type IdType = (); + + fn from_parts(upload_length: usize, _upload_id: ()) -> Self { + Self { upload_length } + } + fn upload_length(&self) -> Option { Some(self.upload_length) } @@ -528,8 +550,8 @@ pub struct LocationPath { #[derive(Debug, Deserialize)] #[serde(bound = "L: LocationData")] pub struct LocationQueryParams { - #[serde(flatten)] - pub location_data: L, + pub upload_length: L::LengthType, + pub upload_id: L::IdType, pub upload_signature: String, #[serde(flatten)] pub other: UploadParams, @@ -588,7 +610,8 @@ impl SignedLocation { pub fn from_parts( project_id: ProjectId, key: String, - location_data: L, + upload_length: L::LengthType, + upload_id: L::IdType, signature: String, other: UploadParams, ) -> Self { @@ -596,7 +619,7 @@ impl SignedLocation { location: Location { project_id, key, - location_data, + location_data: LocationData::from_parts(upload_length, upload_id), other, }, signature: Signature(signature), @@ -697,7 +720,8 @@ where // Parse query parameters. let LocationQueryParams { - location_data: upload_length, + upload_length, + upload_id, upload_signature, other, } = serde_urlencoded::from_str(query).ok()?; @@ -706,6 +730,7 @@ where project_id, key, upload_length, + upload_id, upload_signature, other, )) @@ -1012,7 +1037,7 @@ mod tests { // Can only parse provisional: let provisional: LocationQueryParams = serde_urlencoded::from_str(url).unwrap(); - assert!(provisional.location_data.upload_length.is_none()); + assert!(provisional.upload_length.is_none()); assert!(serde_urlencoded::from_str::>(url).is_err()); } @@ -1022,10 +1047,12 @@ mod tests { let provisional: LocationQueryParams = serde_urlencoded::from_str(json).unwrap(); - assert_eq!(provisional.location_data.upload_length, Some(123)); - assert_eq!(&provisional.location_data.upload_id, "abc"); + assert_eq!(provisional.upload_length, Some(123)); + assert_eq!(&provisional.upload_id, "abc"); + + let json = r#"upload_signature=foo&upload_length=123"#; let full: LocationQueryParams = serde_urlencoded::from_str(json).unwrap(); - assert_eq!(full.location_data.upload_length, 123); + assert_eq!(full.upload_length, 123); } #[test] From 9851f5066f23dcef27be88d9934bc637baac50de Mon Sep 17 00:00:00 2001 From: Joris Bayer Date: Fri, 3 Jul 2026 11:57:04 +0200 Subject: [PATCH 09/12] fix --- relay-server/src/services/upload.rs | 46 ++++++++++++++++++++++++----- 1 file changed, 38 insertions(+), 8 deletions(-) diff --git a/relay-server/src/services/upload.rs b/relay-server/src/services/upload.rs index 0979493ad8d..595f28a13ad 100644 --- a/relay-server/src/services/upload.rs +++ b/relay-server/src/services/upload.rs @@ -459,9 +459,9 @@ impl Final { impl LocationData for Final { type LengthType = usize; - type IdType = (); + type IdType = Option<()>; // the option is here to be able to parse an URL without &upload_id. - fn from_parts(upload_length: usize, _upload_id: ()) -> Self { + fn from_parts(upload_length: usize, _upload_id: Option<()>) -> Self { Self { upload_length } } @@ -1057,17 +1057,27 @@ mod tests { #[test] fn parse_location_with_other() { - let json = - r#"upload_signature=foo&upload_length=123¬_an_upload_param=123&upload_type=bar"#; + let json = r#"upload_signature=foo&upload_id=abc¬_an_upload_param=123&upload_type=bar"#; let provisional: LocationQueryParams = serde_urlencoded::from_str(json).unwrap(); insta::assert_debug_snapshot!(provisional, @r#" LocationQueryParams { - upload_length: Provisional( - Some( - 123, - ), + upload_length: None, + upload_id: "abc", + upload_signature: "foo", + other: UploadParams( + { + "upload_type": "bar", + }, + ), + } + "#); + let full: LocationQueryParams = serde_urlencoded::from_str(json).unwrap(); + insta::assert_debug_snapshot!(full, @r#" + LocationQueryParams { + upload_length: Final( + 123, ), upload_signature: "foo", other: UploadParams( @@ -1077,6 +1087,26 @@ mod tests { ), } "#); + } + + #[test] + fn parse_final_location_with_other() { + let json = + r#"upload_signature=foo&upload_length=123¬_an_upload_param=123&upload_type=bar"#; + + let provisional: LocationQueryParams = serde_urlencoded::from_str(json).unwrap(); + insta::assert_debug_snapshot!(provisional, @r#" + LocationQueryParams { + upload_length: 123, + upload_id: None, + upload_signature: "foo", + other: UploadParams( + { + "upload_type": "bar", + }, + ), + } + "#); let full: LocationQueryParams = serde_urlencoded::from_str(json).unwrap(); insta::assert_debug_snapshot!(full, @r#" LocationQueryParams { From 7699058cf7e3986c24a4778a771bf423043ee08d Mon Sep 17 00:00:00 2001 From: Joris Bayer Date: Mon, 6 Jul 2026 09:30:42 +0200 Subject: [PATCH 10/12] zstd compression --- relay-server/src/services/objectstore.rs | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/relay-server/src/services/objectstore.rs b/relay-server/src/services/objectstore.rs index 5950d873aba..94773f209f3 100644 --- a/relay-server/src/services/objectstore.rs +++ b/relay-server/src/services/objectstore.rs @@ -5,6 +5,7 @@ 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; @@ -905,18 +906,19 @@ impl ObjectstoreServiceInner { let UploadRef { key, upload_id } = upload_ref; let multipart_upload = session.resume_multipart_upload(key, upload_id)?; - let reader = StreamReader::new(body); - let mut rechunked = ReaderStream::with_capacity(reader, CHUNK_SIZE).enumerate(); + let body = ReaderStream::new(ZstdEncoder::new(StreamReader::new(body))); + let mut body = body.enumerate(); let result = relay_statsd::metric!( timer(RelayTimers::AttachmentUploadDuration), type = kind.as_str(), { let mut parts = vec![]; - while let Some((i, chunk)) = rechunked.next().await { + while let Some((i, chunk)) = body.next().await { let chunk = chunk?; - let part_number = u32::try_from(i) + let part_number = u32::try_from(i + 1) .map_err(|_| objectstore_client::Error::InvalidPartNumber(u32::MAX))?; + dbg!(part_number, chunk.len()); let part = multipart_upload.put(chunk, part_number, None).await?; parts.push(part); } From 0463e92a11c58ce15a0989bc967a46fd7aeaca2b Mon Sep 17 00:00:00 2001 From: Joris Bayer Date: Mon, 6 Jul 2026 10:37:07 +0200 Subject: [PATCH 11/12] test: use final location --- tests/integration/conftest.py | 10 +++++++--- tests/integration/consts.py | 3 ++- tests/integration/test_minidump.py | 6 +++--- tests/integration/test_upload.py | 31 +++++++++++++++++++++++------- 4 files changed, 36 insertions(+), 14 deletions(-) diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 2855ccd073b..c04b822e94b 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -47,7 +47,11 @@ objectstore, ) -from .consts import DUMMY_UPLOAD_LOCATION, ZSTD_MAGIC_HEADER +from .consts import ( + DUMMY_UPLOAD_FINAL_LOCATION, + DUMMY_UPLOAD_PROVISIONAL_LOCATION, + ZSTD_MAGIC_HEADER, +) @pytest.fixture(scope="session") @@ -309,7 +313,7 @@ def create(**opts): return Response( "", status=201, - headers={"Location": DUMMY_UPLOAD_LOCATION}, + headers={"Location": DUMMY_UPLOAD_PROVISIONAL_LOCATION}, ) @mini_sentry.app.route("/api//upload//", methods=["PATCH"]) @@ -319,5 +323,5 @@ def upload(**opts): return Response( "", status=204, - headers={"Location": DUMMY_UPLOAD_LOCATION}, + headers={"Location": DUMMY_UPLOAD_FINAL_LOCATION}, ) diff --git a/tests/integration/consts.py b/tests/integration/consts.py index 91d60f244da..8e5a150fb8d 100644 --- a/tests/integration/consts.py +++ b/tests/integration/consts.py @@ -2,6 +2,7 @@ METRICS_EXTRACTION_MIN_SUPPORTED_VERSION = 4 DUMMY_UPLOAD_PATH = "/api/42/upload/019cdc82ed6c7761ba21fd34b86481c2/" -DUMMY_UPLOAD_LOCATION = f"{DUMMY_UPLOAD_PATH}?upload_length=11&upload_signature=z_fUMhT0EZqJz6OQtwGHqTlOOLPpTVpvPa-rYTg18FVWZM1OGny-LeVJB5H-sSR_5e--I1xt-FlCmRG2bsmcAQ.eyJ0IjoiMjAyNi0wMy0xMVQxMDo0ODoxMy45NDM1ODNaIn0" +DUMMY_UPLOAD_PROVISIONAL_LOCATION = f"{DUMMY_UPLOAD_PATH}?upload_length=11&upload_id=my_upload&upload_signature=z_fUMhT0EZqJz6OQtwGHqTlOOLPpTVpvPa-rYTg18FVWZM1OGny-LeVJB5H-sSR_5e--I1xt-FlCmRG2bsmcAQ.eyJ0IjoiMjAyNi0wMy0xMVQxMDo0ODoxMy45NDM1ODNaIn0" +DUMMY_UPLOAD_FINAL_LOCATION = f"{DUMMY_UPLOAD_PATH}?upload_length=11&upload_signature=z_fUMhT0EZqJz6OQtwGHqTlOOLPpTVpvPa-rYTg18FVWZM1OGny-LeVJB5H-sSR_5e--I1xt-FlCmRG2bsmcAQ.eyJ0IjoiMjAyNi0wMy0xMVQxMDo0ODoxMy45NDM1ODNaIn0" ZSTD_MAGIC_HEADER = b"\x28\xb5\x2f\xfd" diff --git a/tests/integration/test_minidump.py b/tests/integration/test_minidump.py index 9dda28003ea..d4219258348 100644 --- a/tests/integration/test_minidump.py +++ b/tests/integration/test_minidump.py @@ -13,9 +13,9 @@ from urllib3.filepost import encode_multipart_formdata from sentry_relay.consts import DataCategory +from tests.integration.consts import DUMMY_UPLOAD_FINAL_LOCATION from .asserts import matches_any, time_within_delta from .test_attachment_ref import upload_and_make_ref -from .consts import DUMMY_UPLOAD_LOCATION MINIDUMP_ATTACHMENT_NAME = "upload_file_minidump" EVENT_ATTACHMENT_NAME = "__sentry-event" @@ -1131,7 +1131,7 @@ def test_minidump_objectstore_uploads( logs.headers["content_type"] == "application/vnd.sentry.attachment-ref+json" ) assert json.loads(logs.payload.bytes) == { - "location": DUMMY_UPLOAD_LOCATION, + "location": DUMMY_UPLOAD_FINAL_LOCATION, } else: assert ( @@ -1146,7 +1146,7 @@ def test_minidump_objectstore_uploads( == "application/vnd.sentry.attachment-ref+json" ) assert json.loads(minidump.payload.bytes) == { - "location": DUMMY_UPLOAD_LOCATION, + "location": DUMMY_UPLOAD_FINAL_LOCATION, "content_type": "application/x-dmp", } else: diff --git a/tests/integration/test_upload.py b/tests/integration/test_upload.py index 3adc2a53e79..1153e62bda5 100644 --- a/tests/integration/test_upload.py +++ b/tests/integration/test_upload.py @@ -10,7 +10,11 @@ from enum import Enum import pytest -from .consts import DUMMY_UPLOAD_PATH, DUMMY_UPLOAD_LOCATION +from .consts import ( + DUMMY_UPLOAD_FINAL_LOCATION, + DUMMY_UPLOAD_PATH, + DUMMY_UPLOAD_PROVISIONAL_LOCATION, +) class FeatureState(Enum): @@ -89,7 +93,10 @@ def test_forward_patch( data = b"hello world" response = relay.patch( "%s&sentry_key=%s" - % (DUMMY_UPLOAD_LOCATION, mini_sentry.get_dsn_public_key(project_id)), + % ( + DUMMY_UPLOAD_PROVISIONAL_LOCATION, + mini_sentry.get_dsn_public_key(project_id), + ), headers={ "Tus-Resumable": "1.0.0", "Content-Type": "application/offset+octet-stream", @@ -121,7 +128,9 @@ def create(**opts): create_attempts += 1 if create_attempts == 1: return Response("", status=503) - return Response("", status=201, headers={"Location": DUMMY_UPLOAD_LOCATION}) + return Response( + "", status=201, headers={"Location": DUMMY_UPLOAD_PROVISIONAL_LOCATION} + ) project_id = 42 project_key = mini_sentry.get_dsn_public_key(project_id) @@ -273,7 +282,10 @@ def test_upload_body_size( data = "x" * size response = relay.patch( "%s&sentry_key=%s" - % (DUMMY_UPLOAD_LOCATION, mini_sentry.get_dsn_public_key(project_id)), + % ( + DUMMY_UPLOAD_PROVISIONAL_LOCATION, + mini_sentry.get_dsn_public_key(project_id), + ), headers={ "Tus-Resumable": "1.0.0", "Content-Type": "application/offset+octet-stream", @@ -346,7 +358,9 @@ def test_timeout( @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_FINAL_LOCATION} + ) project_id = 42 relay = relay( @@ -360,7 +374,10 @@ def slow_upload(**opts): data = b"hello world" response = relay.patch( "%s&sentry_key=%s" - % (DUMMY_UPLOAD_LOCATION, mini_sentry.get_dsn_public_key(project_id)), + % ( + DUMMY_UPLOAD_PROVISIONAL_LOCATION, + mini_sentry.get_dsn_public_key(project_id), + ), headers={ "Tus-Resumable": "1.0.0", "Upload-Offset": "0", @@ -538,7 +555,7 @@ def slow_upstream(**opts): def do_upload(): return relay.patch( - f"{DUMMY_UPLOAD_LOCATION}&sentry_key={project_key}", + f"{DUMMY_UPLOAD_PROVISIONAL_LOCATION}&sentry_key={project_key}", headers={ "Content-Length": str(len(data)), "Content-Type": "application/offset+octet-stream", From 4290652a0f63d1f3ffd5b32f0c1469d340c1328d Mon Sep 17 00:00:00 2001 From: Joris Bayer Date: Mon, 6 Jul 2026 14:41:30 +0200 Subject: [PATCH 12/12] lint --- relay-server/src/services/objectstore.rs | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/relay-server/src/services/objectstore.rs b/relay-server/src/services/objectstore.rs index 94773f209f3..98699b0a4f6 100644 --- a/relay-server/src/services/objectstore.rs +++ b/relay-server/src/services/objectstore.rs @@ -40,11 +40,6 @@ use crate::utils::{BoundedStream, MeteredStream, RetryableStream, TakeOnce, find use super::outcome::Outcome; -/// Maximum size of an individual request to objectstore. -/// -/// This only applies to streams. In-memory attachments are never chunked. -const CHUNK_SIZE: usize = 100 * 1024 * 1024; // 100 MiB - /// Messages that the objectstore service can handle. pub enum Objectstore { Envelope(StoreEnvelope), @@ -914,11 +909,13 @@ impl ObjectstoreServiceInner { type = kind.as_str(), { let mut parts = vec![]; + // FIXME: wait for objectstore to allow omitting the content length. + // Submitting every element of the stream as a separate HTTP request is not + // efficient. 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))?; - dbg!(part_number, chunk.len()); let part = multipart_upload.put(chunk, part_number, None).await?; parts.push(part); }