From abe6c9c281f3d94c75466707277cdb5e103a9890 Mon Sep 17 00:00:00 2001 From: lcian Date: Mon, 1 Sep 2025 08:43:54 +0200 Subject: [PATCH 1/4] feat: futures::Stream extension --- sentry-core/Cargo.toml | 2 +- sentry-core/src/futures.rs | 63 ++++++++++++++++++++++++++++++++++++++ sentry-core/src/lib.rs | 1 + 3 files changed, 65 insertions(+), 1 deletion(-) diff --git a/sentry-core/Cargo.toml b/sentry-core/Cargo.toml index c97f9d004..e04631294 100644 --- a/sentry-core/Cargo.toml +++ b/sentry-core/Cargo.toml @@ -34,6 +34,7 @@ serde = { version = "1.0.104", features = ["derive"] } serde_json = { version = "1.0.46" } url = { version = "2.1.1" } uuid = { version = "1.0.0", features = ["v4", "serde"], optional = true } +futures = "0.3.24" [dev-dependencies] # Because we re-export all the public API in `sentry`, we actually run all the @@ -45,7 +46,6 @@ sentry = { path = "../sentry", default-features = false, features = [ ] } anyhow = "1.0.30" criterion = "0.5" -futures = "0.3.24" rayon = "1.5.3" thiserror = "2.0.12" tokio = { version = "1.44", features = ["rt", "rt-multi-thread", "macros"] } diff --git a/sentry-core/src/futures.rs b/sentry-core/src/futures.rs index 5d8e7f05e..600962449 100644 --- a/sentry-core/src/futures.rs +++ b/sentry-core/src/futures.rs @@ -3,6 +3,8 @@ use std::pin::Pin; use std::sync::Arc; use std::task::{Context, Poll}; +use futures::Stream; + use crate::Hub; /// A future that binds a `Hub` to its execution. @@ -66,6 +68,67 @@ pub trait SentryFutureExt: Sized { impl SentryFutureExt for F where F: Future {} +/// A stream that binds a `Hub` to its execution. +/// +/// This activates the given hub for the duration of the inner streams `poll_next` +/// method. Users usually do not need to construct this type manually, but +/// rather use the [`StreamExt::bind_hub`] method instead. +/// +/// [`StreamExt::bind_hub`]: trait.StreamExt.html#method.bind_hub +#[derive(Debug)] +pub struct SentryStream { + hub: Arc, + stream: S, +} + +impl SentryStream { + /// Creates a new bound stream with a `Hub`. + pub fn new(hub: Arc, stream: S) -> Self { + Self { hub, stream } + } +} + +impl Stream for SentryStream +where + S: Stream, +{ + type Item = S::Item; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let hub = self.hub.clone(); + // https://doc.rust-lang.org/std/pin/index.html#pinning-is-structural-for-field + let stream = unsafe { self.map_unchecked_mut(|s| &mut s.stream) }; + #[cfg(feature = "client")] + { + let _guard = crate::hub_impl::SwitchGuard::new(hub); + stream.poll_next(cx) + } + #[cfg(not(feature = "client"))] + { + let _ = hub; + stream.poll_next(cx) + } + } +} + +/// Stream extensions for Sentry. +pub trait SentryStreamExt: Sized { + /// Binds a hub to the execution of this stream. + /// + /// This ensures that the stream is polled within the given hub. + fn bind_hub(self, hub: H) -> SentryStream + where + H: Into>, + { + SentryStream { + stream: self, + hub: hub.into(), + } + } +} + +impl SentryStreamExt for S where S: Stream {} + #[cfg(all(test, feature = "test"))] mod tests { use crate::test::with_captured_events; diff --git a/sentry-core/src/lib.rs b/sentry-core/src/lib.rs index a39e63ca5..4e4a3bde8 100644 --- a/sentry-core/src/lib.rs +++ b/sentry-core/src/lib.rs @@ -126,6 +126,7 @@ pub use crate::breadcrumbs::IntoBreadcrumbs; pub use crate::clientoptions::{BeforeCallback, ClientOptions, SessionMode}; pub use crate::error::{capture_error, event_from_error, parse_type_from_debug}; pub use crate::futures::{SentryFuture, SentryFutureExt}; +pub use crate::futures::{SentryStream, SentryStreamExt}; pub use crate::hub::Hub; pub use crate::integration::Integration; pub use crate::intodsn::IntoDsn; From 6c52b029c3eb788f3a2ebeda5fcb7d0e7c84f004 Mon Sep 17 00:00:00 2001 From: lcian <17258265+lcian@users.noreply.github.com> Date: Tue, 30 Jun 2026 16:29:51 +0200 Subject: [PATCH 2/4] improve --- sentry-core/src/futures.rs | 63 ------------------ sentry-core/src/lib.rs | 3 +- sentry-core/src/stream.rs | 131 +++++++++++++++++++++++++++++++++++++ 3 files changed, 133 insertions(+), 64 deletions(-) create mode 100644 sentry-core/src/stream.rs diff --git a/sentry-core/src/futures.rs b/sentry-core/src/futures.rs index 600962449..5d8e7f05e 100644 --- a/sentry-core/src/futures.rs +++ b/sentry-core/src/futures.rs @@ -3,8 +3,6 @@ use std::pin::Pin; use std::sync::Arc; use std::task::{Context, Poll}; -use futures::Stream; - use crate::Hub; /// A future that binds a `Hub` to its execution. @@ -68,67 +66,6 @@ pub trait SentryFutureExt: Sized { impl SentryFutureExt for F where F: Future {} -/// A stream that binds a `Hub` to its execution. -/// -/// This activates the given hub for the duration of the inner streams `poll_next` -/// method. Users usually do not need to construct this type manually, but -/// rather use the [`StreamExt::bind_hub`] method instead. -/// -/// [`StreamExt::bind_hub`]: trait.StreamExt.html#method.bind_hub -#[derive(Debug)] -pub struct SentryStream { - hub: Arc, - stream: S, -} - -impl SentryStream { - /// Creates a new bound stream with a `Hub`. - pub fn new(hub: Arc, stream: S) -> Self { - Self { hub, stream } - } -} - -impl Stream for SentryStream -where - S: Stream, -{ - type Item = S::Item; - - fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - let hub = self.hub.clone(); - // https://doc.rust-lang.org/std/pin/index.html#pinning-is-structural-for-field - let stream = unsafe { self.map_unchecked_mut(|s| &mut s.stream) }; - #[cfg(feature = "client")] - { - let _guard = crate::hub_impl::SwitchGuard::new(hub); - stream.poll_next(cx) - } - #[cfg(not(feature = "client"))] - { - let _ = hub; - stream.poll_next(cx) - } - } -} - -/// Stream extensions for Sentry. -pub trait SentryStreamExt: Sized { - /// Binds a hub to the execution of this stream. - /// - /// This ensures that the stream is polled within the given hub. - fn bind_hub(self, hub: H) -> SentryStream - where - H: Into>, - { - SentryStream { - stream: self, - hub: hub.into(), - } - } -} - -impl SentryStreamExt for S where S: Stream {} - #[cfg(all(test, feature = "test"))] mod tests { use crate::test::with_captured_events; diff --git a/sentry-core/src/lib.rs b/sentry-core/src/lib.rs index 4ed213f40..6fbf038af 100644 --- a/sentry-core/src/lib.rs +++ b/sentry-core/src/lib.rs @@ -118,6 +118,7 @@ mod integration; mod intodsn; mod performance; mod scope; +mod stream; mod transport; // public api or exports from this crate @@ -126,12 +127,12 @@ pub use crate::breadcrumbs::IntoBreadcrumbs; pub use crate::clientoptions::{BeforeCallback, ClientOptions, SessionMode}; pub use crate::error::{capture_error, event_from_error, parse_type_from_debug}; pub use crate::futures::{SentryFuture, SentryFutureExt}; -pub use crate::futures::{SentryStream, SentryStreamExt}; pub use crate::hub::Hub; pub use crate::integration::Integration; pub use crate::intodsn::IntoDsn; pub use crate::performance::*; pub use crate::scope::{Scope, ScopeGuard}; +pub use crate::stream::{SentryStream, SentryStreamExt}; pub use crate::transport::{Transport, TransportFactory, TransportOptions}; #[cfg(feature = "logs")] mod logger; // structured logging macros exported with `#[macro_export]` diff --git a/sentry-core/src/stream.rs b/sentry-core/src/stream.rs new file mode 100644 index 000000000..37ebc00ce --- /dev/null +++ b/sentry-core/src/stream.rs @@ -0,0 +1,131 @@ +use std::pin::Pin; +use std::sync::Arc; +use std::task::{Context, Poll}; + +use futures::Stream; + +use crate::Hub; + +/// A stream that binds a `Hub` to its execution. +/// +/// This activates the given hub for the duration of the inner streams `poll_next` +/// method. Users usually do not need to construct this type manually, but +/// rather use the [`StreamExt::bind_hub`] method instead. +/// +/// [`StreamExt::bind_hub`]: trait.StreamExt.html#method.bind_hub +#[derive(Debug)] +pub struct SentryStream { + hub: Arc, + stream: S, +} + +impl SentryStream { + /// Creates a new bound stream with a `Hub`. + pub fn new(hub: Arc, stream: S) -> Self { + Self { hub, stream } + } +} + +impl Stream for SentryStream +where + S: Stream, +{ + type Item = S::Item; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let hub = self.hub.clone(); + // https://doc.rust-lang.org/std/pin/index.html#pinning-is-structural-for-field + let stream = unsafe { self.map_unchecked_mut(|s| &mut s.stream) }; + #[cfg(feature = "client")] + { + let _guard = crate::hub_impl::SwitchGuard::new(hub); + stream.poll_next(cx) + } + #[cfg(not(feature = "client"))] + { + let _ = hub; + stream.poll_next(cx) + } + } +} + +/// Stream extensions for Sentry. +pub trait SentryStreamExt: Sized { + /// Binds a hub to the execution of this stream. + /// + /// This ensures that the stream is polled within the given hub. + fn bind_hub(self, hub: H) -> SentryStream + where + H: Into>, + { + SentryStream { + stream: self, + hub: hub.into(), + } + } +} + +impl SentryStreamExt for S where S: Stream {} + +#[cfg(all(test, feature = "test"))] +mod tests { + use crate::test::with_captured_events; + use crate::{capture_error, capture_message, configure_scope, Hub, Level, SentryStreamExt}; + use futures::StreamExt; + use tokio::runtime::Runtime; + + #[derive(Debug)] + struct TestError(&'static str); + + impl std::fmt::Display for TestError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.0) + } + } + + impl std::error::Error for TestError {} + + #[test] + fn test_streams() { + let mut events = with_captured_events(|| { + let runtime = Runtime::new().unwrap(); + + // Two real streams, each bound to its own hub. The work inside each + // stream runs during `poll_next`, so the captured errors must end up + // tagged with the scope of the hub the stream was bound to. + runtime.block_on(async { + let stream1 = futures::stream::once(async { + configure_scope(|scope| scope.set_transaction(Some("transaction1"))); + capture_error(&TestError("oh no from 1")); + }) + .bind_hub(Hub::new_from_top(Hub::current())); + + let stream2 = futures::stream::once(async { + configure_scope(|scope| scope.set_transaction(Some("transaction2"))); + capture_error(&TestError("oh no from 2")); + }) + .bind_hub(Hub::new_from_top(Hub::current())); + + stream1.collect::>().await; + stream2.collect::>().await; + }); + + capture_message("oh hai from outside", Level::Info); + }); + + events.sort_by(|a, b| a.transaction.cmp(&b.transaction)); + assert_eq!(events.len(), 3); + + // The message captured outside any bound stream has no transaction and no + // exception, and sorts first. + assert_eq!(events[0].transaction, None); + assert!(events[0].exception.is_empty()); + + // The errors captured inside `poll_next` carry the scope of their bound + // hub and the expected exception payload. + assert_eq!(events[1].transaction, Some("transaction1".into())); + assert_eq!(events[1].exception[0].value, Some("oh no from 1".into())); + assert_eq!(events[2].transaction, Some("transaction2".into())); + assert_eq!(events[2].exception[0].value, Some("oh no from 2".into())); + } +} From dc753c99c7b1db2ef9a652ceedca9f0733f9dbdd Mon Sep 17 00:00:00 2001 From: lcian <17258265+lcian@users.noreply.github.com> Date: Tue, 30 Jun 2026 16:33:30 +0200 Subject: [PATCH 3/4] improve --- sentry-core/src/stream.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/sentry-core/src/stream.rs b/sentry-core/src/stream.rs index 37ebc00ce..96380d610 100644 --- a/sentry-core/src/stream.rs +++ b/sentry-core/src/stream.rs @@ -6,9 +6,9 @@ use futures::Stream; use crate::Hub; -/// A stream that binds a `Hub` to its execution. +/// A stream that binds a `Hub` to its polling. /// -/// This activates the given hub for the duration of the inner streams `poll_next` +/// This activates the given hub for the duration of the inner stream's `poll_next` /// method. Users usually do not need to construct this type manually, but /// rather use the [`StreamExt::bind_hub`] method instead. /// @@ -51,7 +51,7 @@ where /// Stream extensions for Sentry. pub trait SentryStreamExt: Sized { - /// Binds a hub to the execution of this stream. + /// Binds a hub to this stream. /// /// This ensures that the stream is polled within the given hub. fn bind_hub(self, hub: H) -> SentryStream From 52a238e661c3fb71fc77b660393d48d6131c3e4e Mon Sep 17 00:00:00 2001 From: lcian <17258265+lcian@users.noreply.github.com> Date: Tue, 30 Jun 2026 16:40:50 +0200 Subject: [PATCH 4/4] improve --- CHANGELOG.md | 6 ++++++ Cargo.lock | 1 + sentry-core/Cargo.toml | 3 ++- sentry-core/src/stream.rs | 2 +- 4 files changed, 10 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 742947eb4..c363fcd61 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog +## Unreleased + +### New Features + +- Added `SentryStream` and `SentryStreamExt` to `sentry-core`, which bind a `Hub` to a `Stream` so that it is polled within the given hub, mirroring the existing `SentryFuture` and `SentryFutureExt`. Use by bringing `SentryStreamExt` in scope and calling `bind_hub` on a stream ([#1214](https://github.com/getsentry/sentry-rust/pull/1214)). + ## 0.48.3 The Sentry Rust SDK now reports data discarded by the SDK to Sentry’s [Stats](https://docs.sentry.io/product/stats/) page. The SDK reports approximate counts for drops from transports, queues, rate-limit backoff, sampling, event processors, and `before_send*` callbacks, including span counts for dropped transactions and byte counts for dropped logs and metrics. diff --git a/Cargo.lock b/Cargo.lock index f5c1754c6..603273d55 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3640,6 +3640,7 @@ dependencies = [ "anyhow", "criterion", "futures", + "futures-core", "log", "rand 0.9.3", "rayon", diff --git a/sentry-core/Cargo.toml b/sentry-core/Cargo.toml index fe3d935b7..c1c5d894e 100644 --- a/sentry-core/Cargo.toml +++ b/sentry-core/Cargo.toml @@ -31,6 +31,7 @@ logs = [] metrics = [] [dependencies] +futures-core = "0.3.24" log = { version = "0.4.8", optional = true, features = ["std"] } rand = { version = "0.9.3", optional = true } sentry-types = { version = "0.48.3", path = "../sentry-types" } @@ -38,9 +39,9 @@ serde = { version = "1.0.104", features = ["derive"] } serde_json = { version = "1.0.46" } url = { version = "2.1.1" } uuid = { version = "1.0.0", features = ["v4", "serde"], optional = true } -futures = "0.3.24" [dev-dependencies] +futures = "0.3.24" # Because we re-export all the public API in `sentry`, we actually run all the # doctests using the `sentry` crate. This also takes care of the doctest # limitation documented in https://github.com/rust-lang/rust/issues/45599. diff --git a/sentry-core/src/stream.rs b/sentry-core/src/stream.rs index 96380d610..3c7e99121 100644 --- a/sentry-core/src/stream.rs +++ b/sentry-core/src/stream.rs @@ -2,7 +2,7 @@ use std::pin::Pin; use std::sync::Arc; use std::task::{Context, Poll}; -use futures::Stream; +use futures_core::Stream; use crate::Hub;