From 0b76b66efb22a95813d817bc5422f7e2aaa06f83 Mon Sep 17 00:00:00 2001 From: "Daniel Szoke (via Pi Coding Agent)" Date: Tue, 1 Sep 2026 14:08:43 +0200 Subject: [PATCH] feat(core): Make transport channel capacity configurable The transport channel previously had a hardcoded capacity of 30, which can saturate in high-throughput scenarios, causing envelopes to be dropped. Add a `transport_channel_capacity` option to `ClientOptions`, propagated to transports via `TransportOptions`. When unset, each transport uses its own default (currently 30, subject to change). A capacity of `0` is clamped to `1` with a debug message. Closes [#994](https://github.com/getsentry/sentry-rust/issues/994) Closes [RUST-149](https://linear.app/getsentry/issue/RUST-149/make-transport-channel-capacity-configurable) --- CHANGELOG.md | 1 + sentry-core/src/client/mod.rs | 2 + sentry-core/src/clientoptions.rs | 60 +++++++++++++++++++++++++++ sentry-core/src/transport/options.rs | 13 ++++++ sentry/src/transports/curl.rs | 4 +- sentry/src/transports/mod.rs | 6 +++ sentry/src/transports/reqwest.rs | 4 +- sentry/src/transports/thread.rs | 54 +++++++++++++++++++++++- sentry/src/transports/tokio_thread.rs | 58 +++++++++++++++++++++++++- sentry/src/transports/ureq.rs | 4 +- 10 files changed, 201 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4fa7a45c8..a446c0501 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ ### Features +- Added a `transport_channel_capacity` option to [`ClientOptions`](https://docs.rs/sentry-core/0.49.3/sentry_core/struct.ClientOptions.html), making the transport's channel capacity configurable. When unset, each transport uses its own default (currently `30`), which is subject to change ([#1311](https://github.com/getsentry/sentry-rust/issues/1311)). - The Tower integration's [`SentryHttpLayer`](https://docs.rs/sentry-tower/0.49.3/sentry_tower/struct.SentryHttpLayer.html) now records the [`http.response.status_code`](https://getsentry.github.io/sentry-conventions/attributes/http/) attribute on transactions ([#1253](https://github.com/getsentry/sentry-rust/pull/1253)). ### Deprecations diff --git a/sentry-core/src/client/mod.rs b/sentry-core/src/client/mod.rs index c406b7ecd..a1c01033c 100644 --- a/sentry-core/src/client/mod.rs +++ b/sentry-core/src/client/mod.rs @@ -631,6 +631,7 @@ fn build_envelope_sender(client_options: &ClientOptions) -> EnvelopeSender { http_proxy, https_proxy, accept_invalid_certs, + transport_channel_capacity, .. } = client_options; @@ -643,6 +644,7 @@ fn build_envelope_sender(client_options: &ClientOptions) -> EnvelopeSender { https_proxy: https_proxy.clone(), accept_invalid_certs: *accept_invalid_certs, client_report_recorder, + transport_channel_capacity: *transport_channel_capacity, }; transport_factory.create_transport_with_options(options) diff --git a/sentry-core/src/clientoptions.rs b/sentry-core/src/clientoptions.rs index 1dada3ab9..2228b1b01 100644 --- a/sentry-core/src/clientoptions.rs +++ b/sentry-core/src/clientoptions.rs @@ -1,5 +1,6 @@ use std::borrow::Cow; use std::fmt; +use std::num::NonZeroUsize; use std::sync::Arc; use std::time::Duration; @@ -245,6 +246,19 @@ pub struct ClientOptions { /// /// See [`max_request_body_size`](method@ClientOptions::max_request_body_size) for details. pub max_request_body_size: MaxRequestBodySize, + /// The maximum number of commands the transport channel can queue. + /// + /// The channel primarily carries envelopes, which are sent to Sentry on a background thread. + /// If the channel is full — for example, in high-throughput scenarios — new envelopes are + /// dropped and recorded as queue-overflow client reports, so increasing this value trades + /// memory usage for reliability. Control commands, such as flushing and shutdown, also count + /// against this capacity. + /// + /// If left unset, each transport uses its own default. The current default is `30` for all + /// built-in transports, but this is subject to change. + /// + /// See [`transport_channel_capacity`](method@ClientOptions::transport_channel_capacity). + pub transport_channel_capacity: Option, /// Deprecated. Setting this to `false` only disables automatic log capture by the /// log-capturing integrations (`log` and `tracing` with the `logs` feature); it does not /// disable logs captured manually via [`Hub::capture_log`](crate::Hub::capture_log) and the @@ -680,6 +694,27 @@ impl ClientOptions { } } + /// Sets the + /// [transport channel capacity](field@ClientOptions::transport_channel_capacity). + /// + /// The smallest usable channel capacity is `1`. If `0` is passed, the capacity is clamped + /// to `1` and a debug message is emitted. + #[inline] + pub fn transport_channel_capacity(self, transport_channel_capacity: usize) -> Self { + #[cfg_attr(not(feature = "client"), expect(clippy::unnecessary_lazy_evaluations))] + let transport_channel_capacity = NonZeroUsize::new(transport_channel_capacity) + .unwrap_or_else(|| { + #[cfg(feature = "client")] + sentry_debug!("cannot set transport channel capacity to 0; clamping to 1"); + NonZeroUsize::MIN + }) + .into(); + Self { + transport_channel_capacity, + ..self + } + } + /// Deprecated. Setting [`enable_logs`](field@ClientOptions::enable_logs) to `false` only /// disables automatic log capture by the log-capturing integrations (`log` and `tracing` /// with the `logs` feature); it does not disable logs captured manually via @@ -825,6 +860,10 @@ impl fmt::Debug for ClientOptions { .field("http_proxy", &self.http_proxy) .field("https_proxy", &self.https_proxy) .field("shutdown_timeout", &self.shutdown_timeout) + .field( + "transport_channel_capacity", + &self.transport_channel_capacity, + ) .field("accept_invalid_certs", &self.accept_invalid_certs) .field("auto_session_tracking", &self.auto_session_tracking) .field("session_mode", &self.session_mode) @@ -877,6 +916,7 @@ impl Default for ClientOptions { session_mode: SessionMode::Application, user_agent: Cow::Borrowed(USER_AGENT), max_request_body_size: MaxRequestBodySize::Medium, + transport_channel_capacity: None, #[expect(deprecated, reason = "still need to set deprecated fields")] enable_logs: true, before_send_log: None, @@ -902,3 +942,23 @@ impl From for ClientOptions { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn transport_channel_capacity_stores_value() { + let options = ClientOptions::new().transport_channel_capacity(42); + assert_eq!( + options.transport_channel_capacity, + Some(NonZeroUsize::new(42).unwrap()) + ); + } + + #[test] + fn transport_channel_capacity_clamps_zero() { + let options = ClientOptions::new().transport_channel_capacity(0); + assert_eq!(options.transport_channel_capacity, Some(NonZeroUsize::MIN)); + } +} diff --git a/sentry-core/src/transport/options.rs b/sentry-core/src/transport/options.rs index f6f2b51b8..27e44bd40 100644 --- a/sentry-core/src/transport/options.rs +++ b/sentry-core/src/transport/options.rs @@ -1,6 +1,7 @@ //! Includes the [`TransportOptions`] struct. use std::borrow::Cow; +use std::num::NonZeroUsize; use sentry_types::Dsn; @@ -26,6 +27,14 @@ pub struct TransportOptions { /// A handle for recording lost Sentry data. #[cfg(feature = "client")] pub client_report_recorder: ClientReportRecorder, + /// The maximum number of commands the transport channel can queue. + /// + /// The channel primarily carries envelopes; control commands, such as flushing and shutdown, + /// also count against this capacity. + /// + /// If `None`, the transport uses its own default. The current default is `30` for all + /// built-in transports, but this is subject to change. + pub transport_channel_capacity: Option, } impl TransportOptions { @@ -43,6 +52,7 @@ impl TransportOptions { https_proxy, accept_invalid_certs, user_agent, + transport_channel_capacity, .. } = options; @@ -54,6 +64,7 @@ impl TransportOptions { accept_invalid_certs: *accept_invalid_certs, #[cfg(feature = "client")] client_report_recorder: ClientReportRecorder::new_no_op(), + transport_channel_capacity: *transport_channel_capacity, }) } @@ -73,6 +84,7 @@ impl TransportOptions { accept_invalid_certs, #[cfg(feature = "client")] client_report_recorder: _, + transport_channel_capacity, } = self; let dsn = Some(dsn); @@ -83,6 +95,7 @@ impl TransportOptions { http_proxy, https_proxy, accept_invalid_certs, + transport_channel_capacity, ..Default::default() } } diff --git a/sentry/src/transports/curl.rs b/sentry/src/transports/curl.rs index 61452fefa..4c39ff010 100644 --- a/sentry/src/transports/curl.rs +++ b/sentry/src/transports/curl.rs @@ -7,7 +7,7 @@ use sentry_core::TransportOptions; use super::{ thread::{TransportThread, TransportThreadOptions}, - RateLimiter, HTTP_PAYLOAD_TOO_LARGE, HTTP_PAYLOAD_TOO_LARGE_MESSAGE, + RateLimiter, DEFAULT_CHANNEL_CAPACITY, HTTP_PAYLOAD_TOO_LARGE, HTTP_PAYLOAD_TOO_LARGE_MESSAGE, }; use crate::{sentry_debug, types::Scheme, ClientOptions, Envelope, Transport}; @@ -83,6 +83,7 @@ impl CurlHttpTransport { https_proxy, accept_invalid_certs, client_report_recorder, + transport_channel_capacity, .. }, client, @@ -222,6 +223,7 @@ impl CurlHttpTransport { let thread = TransportThreadOptions::new(send_fn) .with_client_report_recorder(client_report_recorder) + .with_capacity(transport_channel_capacity.unwrap_or(DEFAULT_CHANNEL_CAPACITY)) .spawn_thread(); Self { thread } } diff --git a/sentry/src/transports/mod.rs b/sentry/src/transports/mod.rs index f2bf72e25..5953cef89 100644 --- a/sentry/src/transports/mod.rs +++ b/sentry/src/transports/mod.rs @@ -6,6 +6,8 @@ use sentry_core::TransportOptions; use crate::{Transport, TransportFactory}; +#[cfg(all(sentry_any_http_transport, not(sentry_embedded_svc_http)))] +use std::num::NonZeroUsize; use std::sync::Arc; #[cfg(feature = "httpdate")] @@ -54,6 +56,10 @@ pub(crate) const HTTP_PAYLOAD_TOO_LARGE: u16 = 413; pub(crate) const HTTP_PAYLOAD_TOO_LARGE_MESSAGE: &str = "Envelope was discarded due to size limits (HTTP 413)."; +/// The default transport channel capacity. +#[cfg(all(sentry_any_http_transport, not(sentry_embedded_svc_http)))] +pub(crate) const DEFAULT_CHANNEL_CAPACITY: NonZeroUsize = NonZeroUsize::new(30).unwrap(); + #[cfg(feature = "reqwest")] type DefaultTransport = ReqwestHttpTransport; diff --git a/sentry/src/transports/reqwest.rs b/sentry/src/transports/reqwest.rs index 175904361..5ea405473 100644 --- a/sentry/src/transports/reqwest.rs +++ b/sentry/src/transports/reqwest.rs @@ -6,7 +6,7 @@ use sentry_core::TransportOptions; use super::{ tokio_thread::{TransportThread, TransportThreadOptions}, - RateLimiter, HTTP_PAYLOAD_TOO_LARGE, HTTP_PAYLOAD_TOO_LARGE_MESSAGE, + RateLimiter, DEFAULT_CHANNEL_CAPACITY, HTTP_PAYLOAD_TOO_LARGE, HTTP_PAYLOAD_TOO_LARGE_MESSAGE, }; use crate::{sentry_debug, ClientOptions, Envelope, Transport}; @@ -84,6 +84,7 @@ impl ReqwestHttpTransport { https_proxy, accept_invalid_certs, client_report_recorder, + transport_channel_capacity, .. }, client, @@ -192,6 +193,7 @@ impl ReqwestHttpTransport { let thread = TransportThreadOptions::new(send_fn) .with_client_report_recorder(client_report_recorder) + .with_capacity(transport_channel_capacity.unwrap_or(DEFAULT_CHANNEL_CAPACITY)) .spawn_thread(); Self { thread } } diff --git a/sentry/src/transports/thread.rs b/sentry/src/transports/thread.rs index c08dac483..3d4f0d1e0 100644 --- a/sentry/src/transports/thread.rs +++ b/sentry/src/transports/thread.rs @@ -1,3 +1,4 @@ +use std::num::NonZeroUsize; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::mpsc::{sync_channel, SyncSender, TrySendError}; use std::sync::Arc; @@ -7,6 +8,7 @@ use std::time::Duration; use sentry_core::client_report::{Reason as ClientReportReason, Recorder as ClientReportRecorder}; use super::ratelimit::{RateLimiter, RateLimitingCategory}; +use super::DEFAULT_CHANNEL_CAPACITY; #[cfg(doc)] use super::{StdTransportThread, StdTransportThreadOptions}; // so we can use pub re-exports in docs use crate::{sentry_debug, Envelope}; @@ -35,6 +37,8 @@ pub struct TransportThread { pub struct TransportThreadOptions { send_fn: F, client_report_recorder: ClientReportRecorder, + /// The transport channel capacity. Defaults to [`DEFAULT_CHANNEL_CAPACITY`]. + channel_capacity: NonZeroUsize, } impl TransportThreadOptions { @@ -43,6 +47,7 @@ impl TransportThreadOptions { Self { send_fn, client_report_recorder: Default::default(), + channel_capacity: DEFAULT_CHANNEL_CAPACITY, } } @@ -53,6 +58,31 @@ impl TransportThreadOptions { ..self } } + + /// Sets the transport channel capacity to the provided value. + pub(super) fn with_capacity(self, channel_capacity: NonZeroUsize) -> Self { + Self { + channel_capacity, + ..self + } + } + + /// Sets the transport channel capacity to the provided value. + /// + /// The smallest usable channel capacity is `1`; if `0` is passed to this function, we clamp + /// the capacity to `1`. + pub fn with_channel_capacity(self, channel_capacity: usize) -> Self { + let channel_capacity = channel_capacity.try_into().unwrap_or_else(|_| { + sentry_debug!( + "Cannot initialize transport with channel capacity of 0, using channel capacity + of {} instead.", + NonZeroUsize::MIN + ); + NonZeroUsize::MIN + }); + + self.with_capacity(channel_capacity) + } } impl TransportThreadOptions @@ -85,8 +115,9 @@ impl TransportThread { let TransportThreadOptions { send_fn: mut send, client_report_recorder, + channel_capacity, } = options; - let (sender, receiver) = sync_channel(30); + let (sender, receiver) = sync_channel(channel_capacity.into()); let shutdown = Arc::new(AtomicBool::new(false)); let shutdown_worker = shutdown.clone(); let handle_client_report_recorder = client_report_recorder.clone(); @@ -182,3 +213,24 @@ impl Drop for TransportThread { } } } + +#[cfg(test)] +mod tests { + use super::*; + + fn noop_options() -> TransportThreadOptions { + TransportThreadOptions::new(|_: Envelope, _: &mut RateLimiter| {}) + } + + #[test] + fn with_channel_capacity_stores_capacity() { + let options = noop_options().with_channel_capacity(42); + assert_eq!(options.channel_capacity.get(), 42); + } + + #[test] + fn with_channel_capacity_clamps_zero() { + let options = noop_options().with_channel_capacity(0); + assert_eq!(options.channel_capacity.get(), 1); + } +} diff --git a/sentry/src/transports/tokio_thread.rs b/sentry/src/transports/tokio_thread.rs index 69cd22a12..86e0e5ec1 100644 --- a/sentry/src/transports/tokio_thread.rs +++ b/sentry/src/transports/tokio_thread.rs @@ -1,3 +1,4 @@ +use std::num::NonZeroUsize; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::mpsc::{sync_channel, SyncSender, TrySendError}; use std::sync::Arc; @@ -7,6 +8,7 @@ use std::time::Duration; use sentry_core::client_report::{Reason as ClientReportReason, Recorder as ClientReportRecorder}; use super::ratelimit::{RateLimiter, RateLimitingCategory}; +use super::DEFAULT_CHANNEL_CAPACITY; #[cfg(doc)] use super::{TokioTransportThread, TokioTransportThreadOptions}; // so we can use pub re-exports in docs use crate::{sentry_debug, Envelope}; @@ -35,6 +37,8 @@ pub struct TransportThread { pub struct TransportThreadOptions { send_fn: F, client_report_recorder: ClientReportRecorder, + /// The transport channel capacity. Defaults to [`DEFAULT_CHANNEL_CAPACITY`]. + channel_capacity: NonZeroUsize, } impl TransportThreadOptions { @@ -43,6 +47,7 @@ impl TransportThreadOptions { Self { send_fn, client_report_recorder: Default::default(), + channel_capacity: DEFAULT_CHANNEL_CAPACITY, } } @@ -53,6 +58,31 @@ impl TransportThreadOptions { ..self } } + + /// Sets the transport channel capacity to the provided value. + pub(super) fn with_capacity(self, channel_capacity: NonZeroUsize) -> Self { + Self { + channel_capacity, + ..self + } + } + + /// Sets the transport channel capacity to the provided value. + /// + /// The smallest usable channel capacity is `1`; if `0` is passed to this function, we clamp + /// the capacity to `1`. + pub fn with_channel_capacity(self, channel_capacity: usize) -> Self { + let channel_capacity = channel_capacity.try_into().unwrap_or_else(|_| { + sentry_debug!( + "Cannot initialize transport with channel capacity of 0, using channel capacity + of {} instead.", + NonZeroUsize::MIN + ); + NonZeroUsize::MIN + }); + + self.with_capacity(channel_capacity) + } } impl TransportThreadOptions @@ -91,8 +121,9 @@ impl TransportThread { let TransportThreadOptions { send_fn: mut send, client_report_recorder, + channel_capacity, } = options; - let (sender, receiver) = sync_channel(30); + let (sender, receiver) = sync_channel(channel_capacity.into()); let shutdown = Arc::new(AtomicBool::new(false)); let shutdown_worker = shutdown.clone(); let handle_client_report_recorder = client_report_recorder.clone(); @@ -197,3 +228,28 @@ impl Drop for TransportThread { } } } + +#[cfg(test)] +mod tests { + use std::future::ready; + + use super::*; + + fn noop_options( + ) -> TransportThreadOptions std::future::Ready> + { + TransportThreadOptions::new(|_: Envelope, rl: RateLimiter| ready(rl)) + } + + #[test] + fn with_channel_capacity_stores_capacity() { + let options = noop_options().with_channel_capacity(42); + assert_eq!(options.channel_capacity.get(), 42); + } + + #[test] + fn with_channel_capacity_clamps_zero() { + let options = noop_options().with_channel_capacity(0); + assert_eq!(options.channel_capacity.get(), 1); + } +} diff --git a/sentry/src/transports/ureq.rs b/sentry/src/transports/ureq.rs index 521281d04..08217cb31 100644 --- a/sentry/src/transports/ureq.rs +++ b/sentry/src/transports/ureq.rs @@ -13,7 +13,7 @@ use ureq::{Agent, Proxy}; use super::{ thread::{TransportThread, TransportThreadOptions}, - RateLimiter, HTTP_PAYLOAD_TOO_LARGE, HTTP_PAYLOAD_TOO_LARGE_MESSAGE, + RateLimiter, DEFAULT_CHANNEL_CAPACITY, HTTP_PAYLOAD_TOO_LARGE, HTTP_PAYLOAD_TOO_LARGE_MESSAGE, }; use crate::{sentry_debug, types::Scheme, ClientOptions, Envelope, Transport}; @@ -94,6 +94,7 @@ impl UreqHttpTransport { ))] accept_invalid_certs, client_report_recorder, + transport_channel_capacity, .. }, agent, @@ -214,6 +215,7 @@ impl UreqHttpTransport { let thread = TransportThreadOptions::new(send_fn) .with_client_report_recorder(client_report_recorder) + .with_capacity(transport_channel_capacity.unwrap_or(DEFAULT_CHANNEL_CAPACITY)) .spawn_thread(); Self { thread } }

AltStyle によって変換されたページ (->オリジナル) /