bug(tracing): register_callsite doesn't get called

hds · hds · commit e6ace819e92c · 2025-06-03T12:17:56.000+02:00
This branch contains code to reproduce an issue where a single global subscriber that takes a "long time" in `register_callsite` causes the second of 2 threads that emit the same event to not receive `register_callsite` at all, but it does receive `event`. It also contains an attempt at a fix in `DefaultCallsite::register` for when another thread is registering the same callsite, in this case we will spin-lock until the registration is complete. This fix complements the one implemented in #2938. The two fixes are really for different problems which manifest in the same way (`event` getting called without `register_callsite` having been called first). The issue for this problem is #2743.
diff --git a/tracing-core/src/callsite.rs b/tracing-core/src/callsite.rs
@@ -307,46 +307,75 @@ impl DefaultCallsite {
     // This only happens once (or if the cached interest value was corrupted).
     #[cold]
     pub fn register(&'static self) -> Interest {
-        // Attempt to advance the registration state to `REGISTERING`...
-        match self.registration.compare_exchange(
-            Self::UNREGISTERED,
-            Self::REGISTERING,
-            Ordering::AcqRel,
-            Ordering::Acquire,
-        ) {
-            Ok(_) => {
-                // Okay, we advanced the state, try to register the callsite.
-                CALLSITES.push_default(self);
-                rebuild_callsite_interest(self, &DISPATCHERS.rebuilder());
-                self.registration.store(Self::REGISTERED, Ordering::Release);
-            }
-            // Great, the callsite is already registered! Just load its
-            // previous cached interest.
-            Err(Self::REGISTERED) => {}
-            // Someone else is registering...
-            Err(_state) => {
-                debug_assert_eq!(
-                    _state,
-                    Self::REGISTERING,
-                    "weird callsite registration state"
+        let mut count = 0;
+        loop {
+            // Attempt to advance the registration state to `REGISTERING`...
+            let prev_state = self.registration.compare_exchange(
+                Self::UNREGISTERED,
+                Self::REGISTERING,
+                Ordering::AcqRel,
+                Ordering::Acquire,
+            );
+            if count == 0 {
+                println!(
+                    "{thread:?}: DefaultCallsite::register: prev_state: {prev_state:?}",
+                    thread = std::thread::current().name()
                 );
-                // Just hit `enabled` this time.
-                return Interest::sometimes();
+            }
+            count += 1;
+            match prev_state {
+                Ok(_) => {
+                    // Okay, we advanced the state, try to register the callsite.
+                    CALLSITES.push_default(self);
+                    rebuild_callsite_interest(self, &DISPATCHERS.rebuilder());
+                    self.registration.store(Self::REGISTERED, Ordering::Release);
+                    break;
+                }
+                // Great, the callsite is already registered! Just load its
+                // previous cached interest.
+                Err(Self::REGISTERED) => break,
+                // Someone else is registering...
+                Err(_state) => {
+                    debug_assert_eq!(
+                        _state,
+                        Self::REGISTERING,
+                        "weird callsite registration state: {_state}"
+                    );
+                    // The callsite is being registered. We have to wait until
+                    // registration is finished, otherwise
+                    continue;
+                }
             }
         }
 
+        println!(
+            "{thread:?}: DefaultCallsite::register: loop count: {count}",
+            thread = std::thread::current().name()
+        );
         match self.interest.load(Ordering::Relaxed) {
             Self::INTEREST_NEVER => Interest::never(),
             Self::INTEREST_ALWAYS => Interest::always(),
-            _ => Interest::sometimes(),
+            Self::INTEREST_SOMETIMES => Interest::sometimes(),
+            other_interest => {
+                println!(
+                    "{thread:?} DefaultCallsite::register: other_interest {other_interest}",
+                    thread = std::thread::current().name()
+                );
+                Interest::sometimes()
+            }
         }
     }
 
     /// Returns the callsite's cached `Interest`, or registers it for the
     /// first time if it has not yet been registered.
     #[inline]
     pub fn interest(&'static self) -> Interest {
-        match self.interest.load(Ordering::Relaxed) {
+        let interest = self.interest.load(Ordering::Relaxed);
+        println!(
+            "{thread:?} interest: {interest}",
+            thread = std::thread::current().name()
+        );
+        match interest {
             Self::INTEREST_NEVER => Interest::never(),
             Self::INTEREST_SOMETIMES => Interest::sometimes(),
             Self::INTEREST_ALWAYS => Interest::always(),
@@ -406,6 +435,10 @@ impl Callsites {
     ///
     /// This also re-computes the max level hint.
     fn rebuild_interest(&self, dispatchers: dispatchers::Rebuilder<'_>) {
+        println!(
+            "{thread:?}: rebuild_interest: dispatchers: {dispatchers:?}",
+            thread = std::thread::current().name()
+        );
         let mut max_level = LevelFilter::OFF;
         dispatchers.for_each(|dispatch| {
             // If the subscriber did not provide a max level hint, assume
@@ -493,7 +526,10 @@ fn rebuild_callsite_interest(
     dispatchers: &dispatchers::Rebuilder<'_>,
 ) {
     let meta = callsite.metadata();
-
+    println!(
+        "{thread:?}: rebuild_callsite_interest: dispatchers: {dispatchers:?}",
+        thread = std::thread::current().name()
+    );
     let mut interest = None;
     dispatchers.for_each(|dispatch| {
         let this_interest = dispatch.register_callsite(meta);
@@ -534,6 +570,22 @@ mod dispatchers {
         Write(RwLockWriteGuard<'a, Vec<dispatcher::Registrar>>),
     }
 
+    impl<'a> std::fmt::Debug for Rebuilder<'a> {
+        fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+            match self {
+                Self::JustOne => write!(f, "JustOne"),
+                Self::Read(arg0) => f
+                    .debug_tuple("Read")
+                    .field(&arg0.len() as &dyn std::fmt::Debug)
+                    .finish(),
+                Self::Write(arg0) => f
+                    .debug_tuple("Write")
+                    .field(&arg0.len() as &dyn std::fmt::Debug)
+                    .finish(),
+            }
+        }
+    }
+
     impl Dispatchers {
         pub(super) const fn new() -> Self {
             Self {
diff --git a/tracing-core/tests/missed_register_callsite_multiple_subscribers.rs b/tracing-core/tests/missed_register_callsite_multiple_subscribers.rs
diff --git a/tracing-core/tests/missed_register_callsite_single_subscriber.rs b/tracing-core/tests/missed_register_callsite_single_subscriber.rs
@@ -0,0 +1,127 @@
+use std::{
+    ptr,
+    sync::atomic::{AtomicPtr, Ordering},
+    thread,
+    time::Duration,
+};
+
+use tracing_core::{
+    callsite::{Callsite as _, DefaultCallsite},
+    dispatcher,
+    field::{FieldSet, Value},
+    span, Event, Kind, Level, Metadata, Subscriber,
+};
+
+struct TestSubscriber {
+    sleep: Duration,
+    callsite: AtomicPtr<Metadata<'static>>,
+}
+
+impl TestSubscriber {
+    fn new(sleep_micros: u64) -> Self {
+        Self {
+            sleep: Duration::from_micros(sleep_micros),
+            callsite: AtomicPtr::new(ptr::null_mut()),
+        }
+    }
+}
+
+impl Subscriber for TestSubscriber {
+    fn register_callsite(&self, metadata: &'static Metadata<'static>) -> tracing_core::Interest {
+        println!(
+            "{thread:?}: TestSubscriber::register_callsite: start",
+            thread = std::thread::current().name()
+        );
+        if !self.sleep.is_zero() {
+            thread::sleep(self.sleep);
+        }
+
+        self.callsite
+            .store(metadata as *const _ as *mut _, Ordering::SeqCst);
+
+        println!(
+            "{thread:?}: TestSubscriber::register_callsite: end",
+            thread = std::thread::current().name()
+        );
+        tracing_core::Interest::always()
+    }
+
+    fn event(&self, event: &tracing_core::Event<'_>) {
+        println!(
+            "{thread:?}: TestSubscriber::register_callsite: start",
+            thread = std::thread::current().name()
+        );
+        let stored_callsite = self.callsite.load(Ordering::SeqCst);
+        let event_callsite: *mut Metadata<'static> = event.metadata() as *const _ as *mut _;
+
+        // This assert is the actual test.
+        assert_eq!(
+            stored_callsite, event_callsite,
+            "stored callsite: {stored_callsite:#?} does not match event \
+            callsite: {event_callsite:#?}. Was `event` called before \
+            `register_callsite`?"
+        );
+    }
+
+    fn enabled(&self, _metadata: &Metadata<'_>) -> bool {
+        true
+    }
+    fn new_span(&self, _span: &span::Attributes<'_>) -> span::Id {
+        span::Id::from_u64(0)
+    }
+    fn record(&self, _span: &span::Id, _values: &span::Record<'_>) {}
+    fn record_follows_from(&self, _span: &span::Id, _follows: &span::Id) {}
+    fn enter(&self, _span: &tracing_core::span::Id) {}
+    fn exit(&self, _span: &tracing_core::span::Id) {}
+}
+
+fn emit_event() {
+    let thread = thread::current();
+    static CALLSITE: DefaultCallsite = {
+        // The values of the metadata are unimportant
+        static META: Metadata<'static> = Metadata::new(
+            "event ",
+            "module::path",
+            Level::INFO,
+            None,
+            None,
+            None,
+            FieldSet::new(&["message"], tracing_core::callsite::Identifier(&CALLSITE)),
+            Kind::EVENT,
+        );
+        DefaultCallsite::new(&META)
+    };
+    let _interest = CALLSITE.interest();
+
+    let meta = CALLSITE.metadata();
+    let field = meta.fields().field("message").unwrap();
+    let message = format!("event-from-{idx}", idx = thread.name().unwrap_or("unnamed"));
+    let values = [(&field, Some(&message as &dyn Value))];
+    let value_set = CALLSITE.metadata().fields().value_set(&values);
+
+    Event::dispatch(meta, &value_set);
+}
+
+/// Regression test for missing register_callsite call (#2743)
+///
+/// This test provokes the race condition which causes the only (global) subscriber to not receive
+/// a call to `register_callsite` before it receives a call to `event`. This occurs when the
+/// (first) call to `register_callsite` takes a long time to complete and the second thread that
+/// attempts to register the same callsite finds that some other thread is already registering and
+/// leaves `DefaultCallsite::register` before the first registration is complete.
+///
+/// Because the test depends on the interaction of multiple dispatchers in different threads,
+/// it needs to be in a test file by itself.
+#[test]
+fn event_before_register() {
+    let subscriber_register_sleep_micros = 1000;
+
+    let subscriber = TestSubscriber::new(subscriber_register_sleep_micros);
+    dispatcher::set_global_default(subscriber.into()).unwrap();
+
+    let jh1 = thread::Builder::new().name("thread-1".into()).spawn(emit_event).unwrap();
+    let jh2 = thread::Builder::new().name("thread-2".into()).spawn(emit_event).unwrap();
+
+    jh1.join().expect("failed to join thread");
+    jh2.join().expect("failed to join thread");
+}
diff --git a/tracing-subscriber/tests/missed_register_callsite_rawler.rs b/tracing-subscriber/tests/missed_register_callsite_rawler.rs
@@ -0,0 +1,45 @@
+use std::sync::{
+    Arc,
+    atomic::{AtomicBool, Ordering::SeqCst},
+};
+
+use tracing::{Event, Metadata};
+use tracing_subscriber::{
+    layer::{Context, SubscriberExt},
+    registry::LookupSpan,
+    util::SubscriberInitExt,
+};
+
+#[derive(Clone, Debug, Default)]
+struct MyLayer {
+    callsite_registered: Arc<AtomicBool>,
+}
+
+impl<S> tracing_subscriber::Layer<S> for MyLayer
+where
+    S: tracing::Subscriber + for<'a> LookupSpan<'a>,
+{
+    fn register_callsite(&self, _m: &'static Metadata<'static>) -> tracing::subscriber::Interest {
+        std::thread::sleep(std::time::Duration::from_millis(100)); // Simulate some work
+        self.callsite_registered.store(true, SeqCst);
+        tracing::subscriber::Interest::always()
+    }
+
+    fn on_event(&self, _event: &Event<'_>, _ctx: Context<'_, S>) {
+        assert!(self.callsite_registered.load(SeqCst));
+    }
+}
+
+#[test]
+fn missed_register_callsite() {
+    let my_layer = MyLayer::default();
+    tracing_subscriber::registry().with(my_layer.clone()).init();
+
+    std::thread::scope(|s| {
+        for i in 0..16 {
+            s.spawn(move || tracing::info!("Thread {} started", i));
+        }
+    });
+
+    dbg!(my_layer);
+}
diff --git a/tracing/Cargo.toml b/tracing/Cargo.toml
@@ -31,6 +31,7 @@ tracing-attributes = { path = "../tracing-attributes", version = "0.1.28", optio
 pin-project-lite = "0.2.9"
 
 [dev-dependencies]
+backtrace = "0.3.71"
 criterion = { version = "0.3.6", default-features = false }
 futures = { version = "0.3.21", default-features = false }
 log = "0.4.17"
@@ -42,17 +43,17 @@ wasm-bindgen-test = "0.3.38"
 [features]
 default = ["std", "attributes"]
 
-max_level_off   = []
+max_level_off = []
 max_level_error = []
-max_level_warn  = []
-max_level_info  = []
+max_level_warn = []
+max_level_info = []
 max_level_debug = []
 max_level_trace = []
 
-release_max_level_off   = []
+release_max_level_off = []
 release_max_level_error = []
-release_max_level_warn  = []
-release_max_level_info  = []
+release_max_level_warn = []
+release_max_level_info = []
 release_max_level_debug = []
 release_max_level_trace = []
 
diff --git a/tracing/tests/missed_register_callsite.rs b/tracing/tests/missed_register_callsite.rs
