xmtp_common/
telemetry.rs

1//! Telemetry plumbing shared by every crate: per-task Sentry hubs that keep the
2//! process hub's client and a task's breadcrumbs in sync.
3
4/// Run a future with its own Sentry Hub so its breadcrumbs never interleave
5/// with concurrent tasks. Near-free when no Sentry client is configured.
6///
7/// Forks from `Hub::current()`, not `Hub::main()`: the fork inherits the calling
8/// scope, so the caller's transaction stays the parent of everything inside (a
9/// `main()` fork has no current span and re-roots inner spans as their own
10/// transactions).
11///
12/// A hub is a *snapshot*: a thread's hub is forked off the process hub the first
13/// time that thread touches it, so a client bound afterwards never reaches it,
14/// and captures on a client-less hub are dropped in silence. The fork therefore
15/// re-reads the process hub on every poll and rebinds whenever the client there
16/// is not the one it last took. That tracks the whole `enable` / `disable` /
17/// `enable` cycle a host can drive over FFI: the second enable installs a
18/// *different* client, and a task that had latched onto the first would keep
19/// reporting into a closed one for the rest of its (long) life.
20#[cfg(not(target_arch = "wasm32"))]
21pub fn bind_task_hub<F: core::future::Future>(
22    fut: F,
23) -> impl core::future::Future<Output = F::Output> {
24    use sentry_core::{Hub, SentryFutureExt};
25    let hub = std::sync::Arc::new(Hub::new_from_top(Hub::current()));
26    AdoptMainClient {
27        main: Hub::main(),
28        adopted: None,
29        inner: fut,
30    }
31    .bind_hub(hub)
32}
33
34/// Rebind `Hub::current()` — the task hub, since this only runs inside the
35/// `bind_hub` wrapper — to `main`'s client whenever that is not `adopted`, the
36/// client this task last took from there.
37///
38/// Identity, not presence: latching on the first client seen strands the task on
39/// a closed one after a disable/enable cycle. A cleared process hub propagates
40/// too, so a disable stops in-flight tasks. A hub whose client was inherited from
41/// the fork rather than taken from here keeps it, because a client-less process
42/// hub matches an `adopted` of `None`.
43#[cfg(not(target_arch = "wasm32"))]
44fn adopt_main_client(
45    main: &sentry_core::Hub,
46    adopted: &mut Option<std::sync::Arc<sentry_core::Client>>,
47) {
48    let current = main.client();
49    match (&current, &*adopted) {
50        (None, None) => return,
51        (Some(a), Some(b)) if std::sync::Arc::ptr_eq(a, b) => return,
52        _ => {}
53    }
54    sentry_core::Hub::current().bind_client(current.clone());
55    *adopted = current;
56}
57
58/// Runs [`adopt_main_client`] on the hub `SentryFuture` installs for the poll.
59/// Sits *inside* the `bind_hub` wrapper precisely so `Hub::current()` is the task
60/// hub while polling.
61#[cfg(not(target_arch = "wasm32"))]
62struct AdoptMainClient<F> {
63    /// The process hub, held rather than re-fetched so the steady-state check is
64    /// one read of its stack with no `Hub::main()` refcount traffic on top.
65    ///
66    /// A generation counter would make that check a single atomic load, but only
67    /// `xmtp_logging` knows when the client changes and it cannot call in here:
68    /// `xmtp_common` already depends on `xmtp_logging` for the test subscriber,
69    /// and cargo rejects the back edge as a cyclic package dependency.
70    main: std::sync::Arc<sentry_core::Hub>,
71    /// Kept alive so its address stays unique for the `ptr_eq` above.
72    adopted: Option<std::sync::Arc<sentry_core::Client>>,
73    inner: F,
74}
75
76#[cfg(not(target_arch = "wasm32"))]
77impl<F: core::future::Future> core::future::Future for AdoptMainClient<F> {
78    type Output = F::Output;
79
80    fn poll(
81        self: core::pin::Pin<&mut Self>,
82        cx: &mut core::task::Context<'_>,
83    ) -> core::task::Poll<Self::Output> {
84        // safe because we consider `inner` to be structurally pinned, and the
85        // other fields not
86        // https://doc.rust-lang.org/std/pin/#choosing-pinning-to-be-structural-for-field
87        let this = unsafe { self.get_unchecked_mut() };
88        adopt_main_client(&this.main, &mut this.adopted);
89        unsafe { core::pin::Pin::new_unchecked(&mut this.inner) }.poll(cx)
90    }
91}
92
93/// Identity passthrough: wasm is single-threaded and has no Sentry client.
94#[cfg(target_arch = "wasm32")]
95pub fn bind_task_hub<F: core::future::Future>(fut: F) -> F {
96    fut
97}
xmtp_common/telemetry.rs

xmtp_common/
telemetry.rs
