feat: Make transport channel capacity configurable

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## Unreleased
+
+### New Features
+
+- Added configurable channel capacity for built-in background HTTP transports ([#1040](https://github.com/getsentry/sentry-rust/pull/1040)).
+
 ## 0.48.5
 
 ### Fixes

sentry/Cargo.toml

@@ -52,9 +52,9 @@ logs = ["sentry-core/logs", "sentry-tracing?/logs", "sentry-log?/logs"]
 metrics = ["sentry-core/metrics"]
 # transports
 transport = ["reqwest", "native-tls"]
-reqwest = ["dep:reqwest", "httpdate", "tokio"]
-curl = ["dep:curl", "httpdate"]
-ureq = ["dep:ureq", "httpdate"]
+reqwest = ["dep:reqwest", "dep:crossbeam-channel", "httpdate", "tokio"]
+curl = ["dep:curl", "dep:crossbeam-channel", "httpdate"]
+ureq = ["dep:ureq", "dep:crossbeam-channel", "httpdate"]
 # transport settings
 native-tls = ["dep:native-tls", "reqwest?/native-tls", "ureq?/native-tls"]
 rustls = ["dep:rustls", "reqwest?/rustls", "ureq?/rustls"]
@@ -65,6 +65,7 @@ embedded-svc-http = ["dep:embedded-svc", "dep:esp-idf-svc"]
 sentry-core = { version = "0.48.5", path = "../sentry-core", features = [
     "client",
 ] }
+crossbeam-channel = { version = "0.5.16", optional = true }
 sentry-anyhow = { version = "0.48.5", path = "../sentry-anyhow", optional = true }
 sentry-actix = { version = "0.48.5", path = "../sentry-actix", optional = true, default-features = false }
 sentry-backtrace = { version = "0.48.5", path = "../sentry-backtrace", optional = true }

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};
@@ -33,6 +33,7 @@ pub struct CurlHttpTransport {
 pub struct CurlHttpTransportOptions {
     general_options: TransportOptions,
     client: Option<CurlClient>,
+    channel_capacity: usize,
 }
 
 impl CurlHttpTransport {
@@ -86,6 +87,7 @@ impl CurlHttpTransport {
                     ..
                 },
             client,
+            channel_capacity,
         } = options;
 
         let client = client.unwrap_or_else(CurlClient::new);
@@ -224,6 +226,7 @@ impl CurlHttpTransport {
 
         let thread = TransportThreadOptions::new(send_fn)
             .with_client_report_recorder(client_report_recorder)
+            .with_channel_capacity(channel_capacity)
             .spawn_thread();
         Self { thread }
     }
@@ -238,7 +241,7 @@ impl Transport for CurlHttpTransport {
     }
 
     fn shutdown(&self, timeout: Duration) -> bool {
-        self.flush(timeout)
+        self.thread.shutdown(timeout)
     }
 }
 
@@ -248,6 +251,7 @@ impl From<TransportOptions> for CurlHttpTransportOptions {
         Self {
             general_options: value,
             client: None,
+            channel_capacity: DEFAULT_CHANNEL_CAPACITY,
         }
     }
 }
@@ -260,6 +264,21 @@ impl CurlHttpTransportOptions {
         Self { client, ..self }
     }
 
+    /// Set the capacity of the channel that queues envelopes for the background
+    /// transport thread (default: 30).
+    ///
+    /// A capacity of `0` creates a rendezvous channel: an envelope is accepted
+    /// only when the transport thread is currently waiting on the receiver,
+    /// otherwise it is dropped. A higher capacity reduces the chance of dropped
+    /// events in high-throughput scenarios at the cost of memory.
+    #[inline]
+    pub fn with_channel_capacity(self, channel_capacity: usize) -> Self {
+        Self {
+            channel_capacity,
+            ..self
+        }
+    }
+
     /// Create a [`CurlHttpTransport`] using these options.
     #[inline]
     pub fn build(self) -> CurlHttpTransport {

sentry/src/transports/mod.rs

@@ -54,6 +54,9 @@ 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).";
 
+#[cfg(sentry_any_http_transport)]
+pub(crate) const DEFAULT_CHANNEL_CAPACITY: usize = 30;
+
 #[cfg(feature = "reqwest")]
 type DefaultTransport = ReqwestHttpTransport;
 

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};
@@ -34,6 +34,7 @@ pub struct ReqwestHttpTransport {
 pub struct ReqwestHttpTransportOptions {
     general_options: TransportOptions,
     client: Option<ReqwestClient>,
+    channel_capacity: usize,
 }
 
 impl ReqwestHttpTransport {
@@ -87,6 +88,7 @@ impl ReqwestHttpTransport {
                     ..
                 },
             client,
+            channel_capacity,
         } = options;
 
         let client = client.unwrap_or_else(|| {
@@ -192,6 +194,7 @@ impl ReqwestHttpTransport {
 
         let thread = TransportThreadOptions::new(send_fn)
             .with_client_report_recorder(client_report_recorder)
+            .with_channel_capacity(channel_capacity)
             .spawn_thread();
         Self { thread }
     }
@@ -206,7 +209,7 @@ impl Transport for ReqwestHttpTransport {
     }
 
     fn shutdown(&self, timeout: Duration) -> bool {
-        self.flush(timeout)
+        self.thread.shutdown(timeout)
     }
 }
 
@@ -216,6 +219,7 @@ impl From<TransportOptions> for ReqwestHttpTransportOptions {
         Self {
             general_options: value,
             client: None,
+            channel_capacity: DEFAULT_CHANNEL_CAPACITY,
         }
     }
 }
@@ -228,6 +232,21 @@ impl ReqwestHttpTransportOptions {
         Self { client, ..self }
     }
 
+    /// Set the capacity of the channel that queues envelopes for the background
+    /// transport thread (default: 30).
+    ///
+    /// A capacity of `0` creates a rendezvous channel: an envelope is accepted
+    /// only when the transport thread is currently waiting on the receiver,
+    /// otherwise it is dropped. A higher capacity reduces the chance of dropped
+    /// events in high-throughput scenarios at the cost of memory.
+    #[inline]
+    pub fn with_channel_capacity(self, channel_capacity: usize) -> Self {
+        Self {
+            channel_capacity,
+            ..self
+        }
+    }
+
     /// Create a [`ReqwestHttpTransport`] using these options.
     #[inline]
     pub fn build(self) -> ReqwestHttpTransport {