feat: gate tracing behind feature flag

Author: LinkeTh

Cargo.lock

@@ -46,20 +46,22 @@ inventory = { version = "0.3", optional = true }
 jaeb-macros = { version = "0.2", path = "jaeb-macros", optional = true }
 metrics = { version = "0.24", optional = true }
 tokio = { version = "1.51", features = ["rt", "rt-multi-thread", "sync", "macros", "time"] }
-tracing = { version = "0.1", features = ["std", "log"] }
+tracing = { version = "0.1", features = ["std", "log"], optional = true }
 
 [features]
 default = []
 macros = ["dep:jaeb-macros", "dep:inventory"]
 metrics = ["dep:metrics"]
 test-utils = []
+trace = ["dep:tracing"]
 
 [dev-dependencies]
 criterion = { version = "0.8", features = ["async_tokio"] }
 tokio = { version = "1.51", features = ["full", "test-util"] }
 async-trait = "0.1"
 eventbuzz = { version = "0.2", features = ["asynchronous"] }
 evno = "1.0.2"
+eventador = { version = "0.0.18", features = ["async"] }
 
 [[bench]]
 name = "throughput"

Cargo.toml

@@ -1,8 +1,10 @@
 use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::Duration;
 
 use async_trait::async_trait;
 use criterion::{Criterion, criterion_group, criterion_main};
+use eventador::Eventador;
 use eventbuzz::asynchronous::prelude::{ApplicationEvent, AsyncApplicationEventListener, AsyncEventbus};
 use evno::{Bus as EvnoBus, Close as EvnoClose, Emit as EvnoEmit, Guard as EvnoGuard, from_fn as evno_from_fn};
 use jaeb::{EventBus, EventHandler, HandlerResult};
@@ -99,6 +101,32 @@ fn bench_async_single_listener(c: &mut Criterion) {
         })
     });
 
+    group.bench_function("eventador_publish", |b| {
+        b.to_async(&rt).iter_custom(|iters| async move {
+            let bus = Eventador::new(1024).expect("create eventador");
+            let sub = bus.subscribe::<u64>();
+            let stop = Arc::new(AtomicBool::new(false));
+            let stop2 = Arc::clone(&stop);
+
+            let drainer = std::thread::spawn(move || {
+                while !stop2.load(Ordering::Relaxed) {
+                    let _ = sub.recv();
+                }
+            });
+
+            let start = std::time::Instant::now();
+            for i in 0..iters {
+                bus.publish(i);
+            }
+            let elapsed = start.elapsed();
+
+            stop.store(true, Ordering::Relaxed);
+            bus.publish(0u64); // sentinel to wake blocked recv
+            drainer.join().expect("eventador drainer join");
+            elapsed
+        })
+    });
+
     group.finish();
 }
 
@@ -168,6 +196,36 @@ fn bench_async_fanout_10(c: &mut Criterion) {
         })
     });
 
+    group.bench_function("eventador_publish", |b| {
+        b.to_async(&rt).iter_custom(|iters| async move {
+            let bus = Eventador::new(1024).expect("create eventador");
+            let stop = Arc::new(AtomicBool::new(false));
+            let mut drainers = Vec::with_capacity(10);
+            for _ in 0..10 {
+                let sub = bus.subscribe::<u64>();
+                let stop2 = Arc::clone(&stop);
+                drainers.push(std::thread::spawn(move || {
+                    while !stop2.load(Ordering::Relaxed) {
+                        let _ = sub.recv();
+                    }
+                }));
+            }
+
+            let start = std::time::Instant::now();
+            for i in 0..iters {
+                bus.publish(i);
+            }
+            let elapsed = start.elapsed();
+
+            stop.store(true, Ordering::Relaxed);
+            bus.publish(0u64); // sentinel to wake blocked recv
+            for d in drainers {
+                d.join().expect("eventador drainer join");
+            }
+            elapsed
+        })
+    });
+
     group.finish();
 }
 
@@ -247,6 +305,47 @@ fn bench_contention_4_publishers(c: &mut Criterion) {
     // timeouts) reduce the probability but do not eliminate it — the
     // standalone benchmark in `benches/evno_contention.rs` still triggers
     // timeouts ~2% of samples. See BENCHMARK.md for the full analysis.
+
+    group.bench_function("eventador_publish", |b| {
+        b.to_async(&rt).iter_custom(|iters| async move {
+            let workers = 4usize;
+            let per = (iters as usize) / workers;
+            let extra = (iters as usize) % workers;
+
+            let bus = Eventador::new(4096).expect("create eventador");
+            let sub = bus.subscribe::<u64>();
+            let stop = Arc::new(AtomicBool::new(false));
+            let stop2 = Arc::clone(&stop);
+
+            let drainer = std::thread::spawn(move || {
+                while !stop2.load(Ordering::Relaxed) {
+                    let _ = sub.recv();
+                }
+            });
+
+            let start = std::time::Instant::now();
+            let mut joins = Vec::with_capacity(workers);
+            for worker_idx in 0..workers {
+                let bus = bus.clone();
+                let n = per + usize::from(worker_idx < extra);
+                joins.push(tokio::spawn(async move {
+                    for i in 0..n {
+                        bus.publish(i as u64);
+                    }
+                }));
+            }
+            for join in joins {
+                join.await.expect("join");
+            }
+            let elapsed = start.elapsed();
+
+            stop.store(true, Ordering::Relaxed);
+            bus.publish(0u64); // sentinel to wake blocked recv
+            drainer.join().expect("eventador drainer join");
+            elapsed
+        })
+    });
+
     group.finish();
 }
 

benches/comparison.rs

@@ -5,7 +5,7 @@ edition = "2024"
 publish = false
 
 [dependencies]
-jaeb = { path = "../..", features = ["metrics"] }
+jaeb = { path = "../..", features = ["metrics", "trace"] }
 tokio = { version = "1", features = ["full"] }
 tracing-subscriber = { version = "0.3", features = ["fmt", "env-filter", "json"] }
 metrics-exporter-prometheus = "0.18"

examples/jaeb-demo/Cargo.toml

@@ -9,6 +9,7 @@ use std::time::Duration;
 use arc_swap::ArcSwap;
 use tokio::sync::{Mutex, Semaphore, mpsc, oneshot};
 use tokio::task::JoinHandle;
+#[cfg(feature = "trace")]
 use tracing::{error, trace, warn};
 
 use crate::error::{ConfigError, EventBusError};
@@ -349,6 +350,7 @@ impl EventBus {
             return Err(EventBusError::Stopped);
         }
 
+        #[cfg(feature = "trace")]
         trace!("event_bus.subscribe {:?}", &registered.name);
         let id = self.next_subscription_id();
         let listener = (registered.register)(id, subscription_policy, once);
@@ -357,7 +359,7 @@ impl EventBus {
         registry.add_listener(TypeId::of::<E>(), std::any::type_name::<E>(), listener);
         self.refresh_snapshot_locked(&registry).await;
 
-        Ok(Subscription::new(id, self.clone()))
+        Ok(Subscription::new(id, registered.name, self.clone()))
     }
 
     /// Register a sync handler that receives [`DeadLetter`] events.
@@ -450,7 +452,7 @@ impl EventBus {
         let mut registry = self.inner.registry.lock().await;
         registry.add_global_middleware(id, erased);
         self.refresh_snapshot_locked(&registry).await;
-        Ok(Subscription::new(id, self.clone()))
+        Ok(Subscription::new(id, None, self.clone()))
     }
 
     /// Add a global **sync** middleware that intercepts all event types.
@@ -474,7 +476,7 @@ impl EventBus {
         let mut registry = self.inner.registry.lock().await;
         registry.add_global_middleware(id, erased);
         self.refresh_snapshot_locked(&registry).await;
-        Ok(Subscription::new(id, self.clone()))
+        Ok(Subscription::new(id, None, self.clone()))
     }
 
     /// Add an async middleware scoped to a single event type `E`.
@@ -514,7 +516,7 @@ impl EventBus {
         let mut registry = self.inner.registry.lock().await;
         registry.add_typed_middleware(TypeId::of::<E>(), std::any::type_name::<E>(), slot.clone());
         self.refresh_snapshot_locked(&registry).await;
-        Ok(Subscription::new(slot.id, self.clone()))
+        Ok(Subscription::new(slot.id, None, self.clone()))
     }
 
     /// Add a **sync** middleware scoped to a single event type `E`.
@@ -549,7 +551,7 @@ impl EventBus {
         let mut registry = self.inner.registry.lock().await;
         registry.add_typed_middleware(TypeId::of::<E>(), std::any::type_name::<E>(), slot.clone());
         self.refresh_snapshot_locked(&registry).await;
-        Ok(Subscription::new(slot.id, self.clone()))
+        Ok(Subscription::new(slot.id, None, self.clone()))
     }
 
     async fn publish_erased(
@@ -645,11 +647,12 @@ impl EventBus {
         tokio::spawn(async move {
             let _keep = permit;
             let dispatch_ctx = bus.inner.full_dispatch_context();
-            if let Err(err) = bus
+            if let Err(_err) = bus
                 .publish_erased(TypeId::of::<E>(), Arc::new(event), std::any::type_name::<E>(), &dispatch_ctx)
                 .await
             {
-                error!(error = %err, "event_bus.try_publish.dispatch_failed");
+                #[cfg(feature = "trace")]
+                error!(error = %_err, "event_bus.try_publish.dispatch_failed");
             }
         });
         Ok(())
@@ -786,5 +789,6 @@ async fn control_loop(inner: std::sync::Weak<Inner>, mut notify_rx: mpsc::Unboun
             }
         }
     }
+    #[cfg(feature = "trace")]
     warn!("event_bus.control_loop.stopped");
 }

src/bus.rs

@@ -239,7 +239,7 @@ where
                 let Some(event) = event.downcast_ref::<E>() else {
                     return Err("event type mismatch".into());
                 };
-                (handler)(event)
+                handler(event)
             });
             ListenerEntry {
                 id,
@@ -275,7 +275,7 @@ where
                 Box::pin(async move {
                     let event = event.map_err(|_| "event type mismatch")?;
                     let event = (*event).clone();
-                    (handler)(event).await
+                    handler(event).await
                 })
             });
             ListenerEntry {

src/handler.rs

@@ -6,15 +6,17 @@ pub(crate) struct TimerGuard {
     start: std::time::Instant,
     name: &'static str,
     event: &'static str,
+    listener: Option<&'static str>,
 }
 
 #[cfg(feature = "metrics")]
 impl TimerGuard {
-    pub fn start(name: &'static str, event: &'static str) -> Self {
+    pub fn start(name: &'static str, event: &'static str, listener: Option<&'static str>) -> Self {
         Self {
             start: std::time::Instant::now(),
             name,
             event,
+            listener,
         }
     }
 }
@@ -23,7 +25,11 @@ impl TimerGuard {
 impl Drop for TimerGuard {
     fn drop(&mut self) {
         let dur = self.start.elapsed();
-        let histogram = histogram!(self.name, "event" => self.event);
+        let histogram = if let Some(listener) = self.listener {
+            histogram!(self.name, "event" => self.event, "listener" => listener)
+        } else {
+            histogram!(self.name, "event" => self.event)
+        };
         histogram.record(dur.as_secs_f64());
     }
 }

src/metrics.rs

@@ -11,6 +11,7 @@ use std::time::Duration;
 
 use tokio::sync::{Mutex, Notify, Semaphore, mpsc, oneshot};
 use tokio::task::AbortHandle;
+#[cfg(feature = "trace")]
 use tracing::{error, warn};
 
 #[cfg(feature = "metrics")]
@@ -506,7 +507,7 @@ async fn execute_async_listener(
         attempts += 1;
 
         #[cfg(feature = "metrics")]
-        let _timer = TimerGuard::start("eventbus.handler.duration", event_name);
+        let _timer = TimerGuard::start("eventbus.handler.duration", event_name, listener.listener_name);
 
         let handler_future = CatchUnwindFuture::new(handler(Arc::clone(&event)));
 
@@ -539,9 +540,11 @@ async fn execute_async_listener(
                 }
 
                 retries_left -= 1;
+                #[cfg(feature = "trace")]
                 warn!(
                     event = event_name,
                     listener_id = listener.subscription_id.as_u64(),
+                    listener_name = listener.listener_name,
                     attempts,
                     retries_left,
                     error = %error_message,
@@ -625,7 +628,7 @@ async fn dispatch_slot(slot: &TypeSlot, event: &EventType, event_name: &'static
         }
 
         #[cfg(feature = "metrics")]
-        let _timer = TimerGuard::start("eventbus.handler.duration", event_name);
+        let _timer = TimerGuard::start("eventbus.handler.duration", event_name, listener.name);
 
         let result = catch_unwind(AssertUnwindSafe(|| {
             // Safe because we only pass shared references to listener code.
@@ -696,16 +699,22 @@ pub(crate) async fn dispatch_with_snapshot(
 }
 
 pub(crate) fn dead_letter_from_failure(failure: &ListenerFailure) -> Option<DeadLetter> {
+    #[cfg(feature = "trace")]
     error!(
         event = failure.event_name,
         listener_id = failure.subscription_id.as_u64(),
+        listener_name = failure.listener_name,
         attempts = failure.attempts,
         error = %failure.error,
         "handler.failed"
     );
 
     #[cfg(feature = "metrics")]
-    counter!("eventbus.handler.error", "event" => failure.event_name).increment(1);
+    if let Some(name) = failure.listener_name {
+        counter!("eventbus.handler.error", "event" => failure.event_name, "listener" => name).increment(1);
+    } else {
+        counter!("eventbus.handler.error", "event" => failure.event_name).increment(1);
+    }
 
     let dead_letter_type = std::any::type_name::<DeadLetter>();
     if failure.dead_letter && failure.event_name != dead_letter_type {