From 7836e72ffbed0f3531d56dccd84973912bc56c94 Mon Sep 17 00:00:00 2001 From: Bohdan Ohorodnii <273991985+varex83agent@users.noreply.github.com> Date: Tue, 25 Aug 2026 17:06:12 +0200 Subject: [PATCH 1/4] feat(tracing): propagate log topic across spawns; label components (#588) `MetricsLayer` labels `app_log_{warn,error}_total` with the `topic` field from the nearest enclosing span. Span context is not carried across `tokio::spawn`, and pluto set the `topic` field in only two places, so almost every warn/error was counted under `topic=""`. Add a span-propagating spawn helper (`pluto_tracing::spawn`) that attaches `Span::current()` to the spawned future, restoring context-like topic propagation, and set a `&'static str` `topic` root span on each long-running component, reusing charon's topic names: - sched, tracker, sigagg, bcast (+recast), parsigex, vapi, qbft, p2p, peerinfo, dkg, relay, vmock, and app-start as the catch-all. Adds tests asserting the helper propagates the topic across the task boundary while a bare `tokio::spawn` does not. Co-Authored-By: Bohdan Ohorodnii <35969035+varex83@users.noreply.github.com> --- crates/app/src/node/mod.rs | 9 ++ crates/app/src/node/wire.rs | 34 ++++--- crates/cli/src/commands/relay.rs | 1 + crates/consensus/src/qbft/component.rs | 33 ++++--- crates/consensus/src/qbft/runner.rs | 3 + crates/core/src/bcast/mod.rs | 1 + crates/core/src/bcast/recast.rs | 1 + crates/core/src/scheduler.rs | 1 + crates/core/src/sigagg.rs | 1 + crates/core/src/tracker/mod.rs | 1 + crates/core/src/validatorapi/router.rs | 12 +++ crates/dkg/src/dkg.rs | 3 +- crates/p2p/src/bootnode.rs | 12 ++- crates/p2p/src/p2p.rs | 1 + crates/parsigex/src/behaviour.rs | 21 +++- crates/peerinfo/src/protocol.rs | 12 +++ crates/relay-server/src/web.rs | 2 +- .../testutil/src/validatormock/component.rs | 6 +- crates/tracing/src/lib.rs | 4 + crates/tracing/src/spawn.rs | 95 +++++++++++++++++++ 20 files changed, 212 insertions(+), 41 deletions(-) create mode 100644 crates/tracing/src/spawn.rs diff --git a/crates/app/src/node/mod.rs b/crates/app/src/node/mod.rs index 9c134c847..06ee5e537 100644 --- a/crates/app/src/node/mod.rs +++ b/crates/app/src/node/mod.rs @@ -226,6 +226,15 @@ impl App { /// Loads the cluster lock + key, builds the consensus component and P2P /// behaviours, wires the core workflow, and drives the node. +/// +/// Carries the `app-start` topic as the catch-all for log metrics not +/// attributed to a more specific component (mirrors charon's `app.Run`). +#[tracing::instrument( + name = "app-start", + level = "debug", + skip_all, + fields(topic = "app-start") +)] async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> { // ---- (1) Load cluster lock + key, derive peers and this node's index ---- // diff --git a/crates/app/src/node/wire.rs b/crates/app/src/node/wire.rs index 40dab2ff6..7bbe6db67 100644 --- a/crates/app/src/node/wire.rs +++ b/crates/app/src/node/wire.rs @@ -652,26 +652,32 @@ pub async fn wire_core_workflow( move |duty: Duty, value: pbcore::UnsignedDataSet| { let dutydb = Arc::clone(&dutydb); let tracker = Arc::clone(&tracker); - tokio::spawn(async move { - let core_set = - match unsigneddata::unsigned_data_set_from_proto(&duty.duty_type, &value) { + let span = tracing::debug_span!("app-start", topic = "app-start"); + tokio::spawn(tracing::Instrument::instrument( + async move { + let core_set = match unsigneddata::unsigned_data_set_from_proto( + &duty.duty_type, + &value, + ) { Ok(set) => set, Err(err) => { tracing::warn!(?err, "dutydb: decode unsigned data set"); return; } }; - let pubkeys: Vec = core_set.keys().copied().collect(); - // Logged before the error moves into the tracker's `Arc`. - let step_err = match dutydb.store(duty.clone(), core_set).await { - Ok(()) => None, - Err(err) => { - tracing::warn!(?err, "dutydb: store"); - Some(owned_step_err(err)) - } - }; - tracker.duty_db_stored(duty, &pubkeys, step_err).await; - }); + let pubkeys: Vec = core_set.keys().copied().collect(); + // Logged before the error moves into the tracker's `Arc`. + let step_err = match dutydb.store(duty.clone(), core_set).await { + Ok(()) => None, + Err(err) => { + tracing::warn!(?err, "dutydb: store"); + Some(owned_step_err(err)) + } + }; + tracker.duty_db_stored(duty, &pubkeys, step_err).await; + }, + span, + )); Ok(()) }, )); diff --git a/crates/cli/src/commands/relay.rs b/crates/cli/src/commands/relay.rs index 44e9d7950..0572bdbd8 100644 --- a/crates/cli/src/commands/relay.rs +++ b/crates/cli/src/commands/relay.rs @@ -293,6 +293,7 @@ pub struct RelayLokiArgs { pub loki_service: String, } +#[tracing::instrument(name = "relay", level = "debug", skip_all, fields(topic = "relay"))] pub async fn run( config: pluto_relay_server::config::Config, ct: CancellationToken, diff --git a/crates/consensus/src/qbft/component.rs b/crates/consensus/src/qbft/component.rs index 820ac25ef..1db7cc82d 100644 --- a/crates/consensus/src/qbft/component.rs +++ b/crates/consensus/src/qbft/component.rs @@ -12,6 +12,7 @@ use prost::{Message, Name}; use prost_types::Any; use tokio::{sync::mpsc, task::JoinHandle}; use tokio_util::sync::CancellationToken; +use tracing::Instrument as _; use crate::{ instance::InstanceIo, @@ -474,22 +475,26 @@ impl Consensus { .expect("start must be called exactly once"); let instances = Arc::clone(&self.instances); - tokio::spawn(async move { - loop { - tokio::select! { - () = ct.cancelled() => return, - duty = expired_rx.recv() => match duty { - Some(duty) => { - instances - .lock() - .unwrap_or_else(PoisonError::into_inner) - .remove(&duty); - } - None => return, - }, + let span = tracing::debug_span!("qbft", topic = "qbft"); + tokio::spawn( + async move { + loop { + tokio::select! { + () = ct.cancelled() => return, + duty = expired_rx.recv() => match duty { + Some(duty) => { + instances + .lock() + .unwrap_or_else(PoisonError::into_inner) + .remove(&duty); + } + None => return, + }, + } } } - }) + .instrument(span), + ) } /// Returns existing instance I/O for `duty`, or creates an empty one. diff --git a/crates/consensus/src/qbft/runner.rs b/crates/consensus/src/qbft/runner.rs index bd05764e8..bef2b4c0d 100644 --- a/crates/consensus/src/qbft/runner.rs +++ b/crates/consensus/src/qbft/runner.rs @@ -123,6 +123,7 @@ pub(crate) async fn propose_priority( } /// Hashes and packs the local value, then starts or joins the duty runner. +#[tracing::instrument(name = "qbft", level = "debug", skip_all, fields(topic = "qbft"))] async fn propose( consensus: &Consensus, duty: Duty, @@ -166,6 +167,7 @@ where } /// Starts participating in a duty without a local proposal value. +#[tracing::instrument(name = "qbft", level = "debug", skip_all, fields(topic = "qbft"))] pub(crate) async fn participate( consensus: &Consensus, duty: Duty, @@ -194,6 +196,7 @@ pub(crate) async fn participate( } /// Runs one consensus instance and publishes its completion result. +#[tracing::instrument(name = "qbft", level = "debug", skip_all, fields(topic = "qbft"))] pub(crate) async fn run_instance( consensus: &Consensus, duty: Duty, diff --git a/crates/core/src/bcast/mod.rs b/crates/core/src/bcast/mod.rs index 99a3f53ee..91738365b 100644 --- a/crates/core/src/bcast/mod.rs +++ b/crates/core/src/bcast/mod.rs @@ -268,6 +268,7 @@ impl Broadcaster { /// success record the broadcast count and submission delay. Internal-only /// duties (randao, prepare-aggregator, prepare-sync-contribution) are /// no-ops; deprecated and unknown duty types return an error. + #[tracing::instrument(name = "bcast", level = "debug", skip_all, fields(topic = "bcast"))] pub async fn broadcast(&self, mut duty: Duty, set: SignedDataSet) -> Result<()> { match duty.duty_type { DutyType::Attester => self.broadcast_attester(&duty, &set).await?, diff --git a/crates/core/src/bcast/recast.rs b/crates/core/src/bcast/recast.rs index 4514349a6..07117d85b 100644 --- a/crates/core/src/bcast/recast.rs +++ b/crates/core/src/bcast/recast.rs @@ -92,6 +92,7 @@ impl Recaster { } /// Called when new slots tick. + #[tracing::instrument(name = "bcast", level = "debug", skip_all, fields(topic = "bcast"))] pub async fn slot_ticked(&self, slot: Slot) -> Result<()> { if !slot.first_in_epoch() { return Ok(()); diff --git a/crates/core/src/scheduler.rs b/crates/core/src/scheduler.rs index d29b5abe9..5437d788b 100644 --- a/crates/core/src/scheduler.rs +++ b/crates/core/src/scheduler.rs @@ -313,6 +313,7 @@ struct SchedulerActor { } impl SchedulerActor { + #[tracing::instrument(name = "sched", level = "debug", skip_all, fields(topic = "sched"))] async fn run( mut self, mut slot_rx: sync::mpsc::Receiver, diff --git a/crates/core/src/sigagg.rs b/crates/core/src/sigagg.rs index de1dfcdd1..50b26ca51 100644 --- a/crates/core/src/sigagg.rs +++ b/crates/core/src/sigagg.rs @@ -143,6 +143,7 @@ impl Aggregator { /// /// If aggregation fails for any validator the entire call returns that /// error immediately — no partial results are emitted. + #[tracing::instrument(name = "sigagg", level = "debug", skip_all, fields(topic = "sigagg"))] pub async fn aggregate( &self, duty: &Duty, diff --git a/crates/core/src/tracker/mod.rs b/crates/core/src/tracker/mod.rs index fb0272138..af8a96c8d 100644 --- a/crates/core/src/tracker/mod.rs +++ b/crates/core/src/tracker/mod.rs @@ -455,6 +455,7 @@ impl TrackerService { ); } + #[tracing::instrument(name = "tracker", level = "debug", skip_all, fields(topic = "tracker"))] async fn run(mut self) { let mut events: HashMap> = HashMap::new(); diff --git a/crates/core/src/validatorapi/router.rs b/crates/core/src/validatorapi/router.rs index b675fcba3..39d720c4b 100644 --- a/crates/core/src/validatorapi/router.rs +++ b/crates/core/src/validatorapi/router.rs @@ -250,9 +250,21 @@ pub fn new_router( ) .route("/eth/v1/node/version", get(node_version)) .fallback(proxy_handler) + // Attach the `vapi` topic to every request so warn/error logs emitted + // while handling it are counted under `app_log_{warn,error}_total{topic="vapi"}`. + .layer(middleware::from_fn(with_vapi_topic)) .with_state(state) } +/// Middleware that runs each request handler inside a `vapi` topic span so log +/// metrics are attributed to the validator API component. +async fn with_vapi_topic(req: Request, next: Next) -> Response { + use tracing::Instrument as _; + + let span = tracing::debug_span!("vapi", topic = "vapi"); + next.run(req).instrument(span).await +} + async fn attester_duties( State(state): State>, Path(epoch): Path, diff --git a/crates/dkg/src/dkg.rs b/crates/dkg/src/dkg.rs index d5bc9d911..5c07cd765 100644 --- a/crates/dkg/src/dkg.rs +++ b/crates/dkg/src/dkg.rs @@ -377,6 +377,7 @@ fn default_tracing_config() -> TracingConfig { } /// Runs the DKG entrypoint. +#[tracing::instrument(name = "dkg", level = "debug", skip_all, fields(topic = "dkg"))] pub async fn run(conf: Config, ct: CancellationToken) -> Result<(), DkgError> { if ct.is_cancelled() { return Err(DkgError::ShutdownRequestedBeforeStartup); @@ -594,7 +595,7 @@ async fn run_inner(conf: Config, ct: CancellationToken) -> Result<(), DkgError> let sync_clients = handlers.sync.clone(); let sync_server = handlers.sync_server.clone(); let network_ct = ct.child_token(); - let network_task = tokio::spawn(drive_dkg_network(node, network_ct.clone())); + let network_task = pluto_tracing::spawn(drive_dkg_network(node, network_ct.clone())); let result = run_ceremony( &conf, diff --git a/crates/p2p/src/bootnode.rs b/crates/p2p/src/bootnode.rs index ea2d2ec39..2d0ec1bab 100644 --- a/crates/p2p/src/bootnode.rs +++ b/crates/p2p/src/bootnode.rs @@ -6,7 +6,7 @@ use backon::Retryable; use libp2p::Multiaddr; use pluto_eth2util::enr::Record; use tokio_util::sync::CancellationToken; -use tracing::{info, warn}; +use tracing::{Instrument as _, info, warn}; use url::Url; use crate::{ @@ -127,9 +127,13 @@ pub async fn new_relays( let mutable_clone = mutable.clone(); let cancel_clone = cancel.child_token(); - tokio::spawn(async move { - resolve_relay(cancel_clone, url, hash, mutable_clone).await; - }); + let span = tracing::debug_span!("relay", topic = "relay"); + tokio::spawn( + async move { + resolve_relay(cancel_clone, url, hash, mutable_clone).await; + } + .instrument(span), + ); resp.push(mutable); } diff --git a/crates/p2p/src/p2p.rs b/crates/p2p/src/p2p.rs index ceb4b5806..5aebae30f 100644 --- a/crates/p2p/src/p2p.rs +++ b/crates/p2p/src/p2p.rs @@ -630,6 +630,7 @@ impl Node { } /// Handles a swarm event to update metrics and logging. + #[tracing::instrument(name = "p2p", level = "debug", skip_all, fields(topic = "p2p"))] fn handle_event(&mut self, event: &SwarmEvent>) { match event { // Identify - update peer addresses in the peer store. diff --git a/crates/parsigex/src/behaviour.rs b/crates/parsigex/src/behaviour.rs index 4505150e4..0d55ae04a 100644 --- a/crates/parsigex/src/behaviour.rs +++ b/crates/parsigex/src/behaviour.rs @@ -21,6 +21,7 @@ use libp2p::{ }, }; use tokio::sync::{RwLock, mpsc, oneshot}; +use tracing::Instrument as _; use pluto_core::{ eth2signeddata, @@ -203,6 +204,12 @@ impl Handle { result_rx.await.map_err(|_| Error::Closed)? } + #[tracing::instrument( + name = "parsigex", + level = "debug", + skip_all, + fields(topic = "parsigex") + )] async fn enqueue( &self, duty: Duty, @@ -498,12 +505,16 @@ impl Behaviour { /// subscribers async). fn notify_subscribers(&self, duty: Duty, data_set: ParSignedDataSet) { let shared_subs = self.shared_subs.clone(); - tokio::spawn(async move { - let subs = shared_subs.subs.read().await.clone(); - for sub in &subs { - sub(duty.clone(), data_set.clone()).await; + let span = tracing::debug_span!("parsigex", topic = "parsigex"); + tokio::spawn( + async move { + let subs = shared_subs.subs.read().await.clone(); + for sub in &subs { + sub(duty.clone(), data_set.clone()).await; + } } - }); + .instrument(span), + ); } } diff --git a/crates/peerinfo/src/protocol.rs b/crates/peerinfo/src/protocol.rs index fe485f09a..6bd94570c 100644 --- a/crates/peerinfo/src/protocol.rs +++ b/crates/peerinfo/src/protocol.rs @@ -281,6 +281,12 @@ impl ProtocolState { /// Sends a peer info request and waits for a response. /// /// Returns the response `PeerInfo` on success. + #[tracing::instrument( + name = "peerinfo", + level = "debug", + skip_all, + fields(topic = "peerinfo") + )] pub async fn send_peer_info( &self, mut stream: Stream, @@ -301,6 +307,12 @@ impl ProtocolState { /// Receives a peer info request and sends a response. /// /// Returns the stream for potential reuse after successfully responding. + #[tracing::instrument( + name = "peerinfo", + level = "debug", + skip_all, + fields(topic = "peerinfo") + )] pub async fn recv_peer_info( &self, mut stream: Stream, diff --git a/crates/relay-server/src/web.rs b/crates/relay-server/src/web.rs index dcbdd6a83..d51d17307 100644 --- a/crates/relay-server/src/web.rs +++ b/crates/relay-server/src/web.rs @@ -127,7 +127,7 @@ pub async fn enr_server( let resolver_handle = state.p2p_config.external_host.clone().map(|external_host| { let state = state.clone(); let ct = ct.child_token(); - tokio::spawn(resolve_external_host_periodically(state, external_host, ct)) + pluto_tracing::spawn(resolve_external_host_periodically(state, external_host, ct)) }); info!( diff --git a/crates/testutil/src/validatormock/component.rs b/crates/testutil/src/validatormock/component.rs index c7ea50c81..432062d2c 100644 --- a/crates/testutil/src/validatormock/component.rs +++ b/crates/testutil/src/validatormock/component.rs @@ -138,6 +138,7 @@ impl Component { } /// Called externally each slot. Mirrors Go's `Component.SlotTicked`. + #[tracing::instrument(name = "vmock", level = "debug", skip_all, fields(topic = "vmock"))] pub async fn slot_ticked(&self, slot: u64) -> Result<()> { if self.delay_on_startup().await { return Ok(()); @@ -270,6 +271,7 @@ impl Drop for Component { } } +#[tracing::instrument(name = "vmock", level = "debug", skip_all, fields(topic = "vmock"))] async fn run_scheduler( inner: Arc, cancel: CancellationToken, @@ -287,7 +289,7 @@ async fn run_scheduler( let Some(scheduled) = maybe else { break }; let inner_for_task = Arc::clone(&inner); let cancel_for_task = cancel.clone(); - duties.spawn(async move { + duties.spawn(tracing::Instrument::instrument(async move { let start_time = scheduled.start_time; let slot = scheduled.slot; let duty_label = scheduled.duty_type.clone(); @@ -312,7 +314,7 @@ async fn run_scheduler( } } } - }); + }, tracing::Span::current())); } // Reap finished duties to keep the JoinSet bounded. Disabled when // empty — `Some(_)` does not match `None`. diff --git a/crates/tracing/src/lib.rs b/crates/tracing/src/lib.rs index 7a4d407d1..7632d6b58 100644 --- a/crates/tracing/src/lib.rs +++ b/crates/tracing/src/lib.rs @@ -16,6 +16,10 @@ pub mod layers; /// Metrics for the tracing. pub mod metrics; +/// Span-propagating task spawning. +pub mod spawn; + pub use config::{ConsoleConfig, LokiConfig, TracingConfig, TracingConfigBuilder}; pub use init::{LokiInit, init}; +pub use spawn::spawn; diff --git a/crates/tracing/src/spawn.rs b/crates/tracing/src/spawn.rs new file mode 100644 index 000000000..02ebce97d --- /dev/null +++ b/crates/tracing/src/spawn.rs @@ -0,0 +1,95 @@ +//! Span-propagating task spawning. +//! +//! [`tokio::spawn`] does **not** carry the current [`tracing`] span into the +//! spawned future: the new task starts with an empty span stack. That breaks +//! the `topic` label used by [`crate::layers::metrics::MetricsLayer`], because +//! a `warn!`/`error!` emitted from a bare spawn lands on `topic=""` even when +//! the spawning code is inside a component's `topic` span. +//! +//! Charon derives the same label from `context.Context`, which *is* propagated +//! into goroutines. To restore context-like propagation here, wrap the spawned +//! future with [`tracing::Instrument`] and attach [`tracing::Span::current`]. +//! +//! Prefer [`spawn`] over [`tokio::spawn`] in long-running components so that +//! the component's root `topic` span is inherited by its subtasks by default. + +use std::future::Future; + +use tokio::task::JoinHandle; +use tracing::Instrument as _; + +/// Like [`tokio::spawn`], but attaches the current [`tracing::Span`] to the +/// spawned future so span context (and therefore the metrics `topic` label) is +/// propagated across the task boundary. +/// +/// Use this instead of [`tokio::spawn`] when the calling code runs inside a +/// component's `topic` span and the spawned work should be attributed to the +/// same topic. +pub fn spawn(future: F) -> JoinHandle +where + F: Future + Send + 'static, + F::Output: Send + 'static, +{ + tokio::spawn(future.instrument(tracing::Span::current())) +} + +#[cfg(test)] +mod tests { + use tracing_subscriber::layer::SubscriberExt as _; + + use crate::{layers::metrics::MetricsLayer, metrics::TRACING_METRICS}; + + #[tokio::test] + async fn spawn_propagates_topic_across_task_boundary() { + let topic = "spawn_helper_propagation_test"; + let subscriber = tracing_subscriber::registry().with(MetricsLayer); + + let before = TRACING_METRICS.error_total[&topic.to_owned()].get(); + + // `Instrument` captures both the span and the dispatcher at spawn time, + // so the default subscriber set below applies to the spawned task. + let guard = tracing::subscriber::set_default(subscriber); + + let span = tracing::info_span!("component", topic); + let handle = { + let _enter = span.enter(); + super::spawn(async { + tracing::error!("boom from spawned task"); + }) + }; + handle.await.unwrap(); + + drop(guard); + + let after = TRACING_METRICS.error_total[&topic.to_owned()].get(); + assert_eq!( + after, + before.saturating_add(1), + "spawned task should inherit topic" + ); + } + + #[tokio::test] + async fn bare_tokio_spawn_loses_topic() { + // Documents the behaviour the helper fixes: a bare spawn drops the + // topic and is counted under the empty label. + let topic = "spawn_helper_bare_test"; + let subscriber = tracing_subscriber::registry().with(MetricsLayer); + + let before = TRACING_METRICS.error_total[&topic.to_owned()].get(); + + let guard = tracing::subscriber::set_default(subscriber); + let span = tracing::info_span!("component", topic); + let handle = { + let _enter = span.enter(); + tokio::spawn(async { + tracing::error!("boom from bare spawned task"); + }) + }; + handle.await.unwrap(); + drop(guard); + + let after = TRACING_METRICS.error_total[&topic.to_owned()].get(); + assert_eq!(after, before, "bare spawn must not inherit topic"); + } +} From d1ef1ff449d865c1288d01727c2d2df8a592071d Mon Sep 17 00:00:00 2001 From: Bohdan Ohorodnii <273991985+varex83agent@users.noreply.github.com> Date: Mon, 31 Aug 2026 12:16:06 +0200 Subject: [PATCH 2/4] docs(tracing): disambiguate spawn intra-doc link The rustdoc build failed with -D warnings because [`spawn`] is ambiguous between the `spawn` module and the `spawn` function. Add parentheses to link to the function. Co-Authored-By: Bohdan Ohorodnii --- crates/tracing/src/spawn.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/tracing/src/spawn.rs b/crates/tracing/src/spawn.rs index 02ebce97d..f3342d08a 100644 --- a/crates/tracing/src/spawn.rs +++ b/crates/tracing/src/spawn.rs @@ -10,7 +10,7 @@ //! into goroutines. To restore context-like propagation here, wrap the spawned //! future with [`tracing::Instrument`] and attach [`tracing::Span::current`]. //! -//! Prefer [`spawn`] over [`tokio::spawn`] in long-running components so that +//! Prefer [`spawn()`] over [`tokio::spawn`] in long-running components so that //! the component's root `topic` span is inherited by its subtasks by default. use std::future::Future; From 5af9a332af7e062e1f53ff65032c43846ff45d78 Mon Sep 17 00:00:00 2001 From: Bohdan Ohorodnii <273991985+varex83agent@users.noreply.github.com> Date: Mon, 31 Aug 2026 12:19:50 +0200 Subject: [PATCH 3/4] style(app): wrap comment to satisfy rustfmt comment_width Co-Authored-By: Bohdan Ohorodnii --- crates/app/src/node/wire.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/crates/app/src/node/wire.rs b/crates/app/src/node/wire.rs index 7bbe6db67..813538b4a 100644 --- a/crates/app/src/node/wire.rs +++ b/crates/app/src/node/wire.rs @@ -666,7 +666,8 @@ pub async fn wire_core_workflow( } }; let pubkeys: Vec = core_set.keys().copied().collect(); - // Logged before the error moves into the tracker's `Arc`. + // Logged before the error moves into the tracker's + // `Arc`. let step_err = match dutydb.store(duty.clone(), core_set).await { Ok(()) => None, Err(err) => { From 3586d2e528addeb67470406c6b832a75aadbde76 Mon Sep 17 00:00:00 2001 From: Bohdan Ohorodnii <273991985+varex83agent@users.noreply.github.com> Date: Fri, 18 Sep 2026 00:02:31 +0200 Subject: [PATCH 4/4] refactor(tracing): inline span propagation and close subtask topic gaps Addresses review feedback on #669. Drop the `pluto_tracing::spawn` wrapper: a helper that must be remembered at every spawn site is the wrong shape for this. Both callers now inline `tokio::spawn(fut.instrument(Span::current()))` with a comment, and the propagation tests move next to `MetricsLayer`, which is what reads the span. Close the subtask gaps the wrapper's two call sites were hiding. `JoinSet::spawn` and `spawn_blocking` drop the span stack exactly like `tokio::spawn`, so the node's lifecycle JoinSet, the QBFT instance JoinSet and its blocking core, the scheduler's subscriber loops and slot ticker, the readiness checker and the priority consensus spawn all lost their topic. The node's JoinSet now opens `app-start` per task, mirroring charon's `lifecycle.Manager` handing each background hook a `log.WithTopic(context.Background(), "app-start")` context; the rest re-attach the caller's span. Two of these are charon-parity bugs rather than plumbing: - `InclusionChecker::run` carried no topic at all; charon sets `tracker` (core/tracker/inclusion.go). - The scheduler's subscriber loops were left unlabelled on the theory that callback errors belong to other components, but charon logs "Emit scheduled slot event" under `sched` (core/scheduler). Also fix the formatting rustfmt had silently given up on: wrapping a long `async move` block in `Instrument::instrument(fut, span)` left the body at its old indentation while `--check` still passed. Hoisting the future into a named binding first keeps rustfmt in charge. Co-Authored-By: Bohdan Ohorodnii <35969035+varex83@users.noreply.github.com> --- crates/app/src/monitoringapi/checker.rs | 31 +++--- crates/app/src/node/mod.rs | 49 +++++++--- crates/app/src/node/wire.rs | 14 ++- crates/consensus/src/qbft/runner.rs | 65 ++++++++----- crates/core/src/scheduler.rs | 71 +++++++++----- crates/core/src/tracker/inclusion.rs | 4 + crates/dkg/src/dkg.rs | 10 +- crates/priority/src/prioritiser.rs | 21 ++-- crates/relay-server/src/web.rs | 15 ++- .../testutil/src/validatormock/component.rs | 9 +- crates/tracing/src/layers/metrics.rs | 46 +++++++++ crates/tracing/src/lib.rs | 4 - crates/tracing/src/spawn.rs | 95 ------------------- 13 files changed, 238 insertions(+), 196 deletions(-) delete mode 100644 crates/tracing/src/spawn.rs diff --git a/crates/app/src/monitoringapi/checker.rs b/crates/app/src/monitoringapi/checker.rs index ad6f2d20a..0cb5dc015 100644 --- a/crates/app/src/monitoringapi/checker.rs +++ b/crates/app/src/monitoringapi/checker.rs @@ -8,7 +8,7 @@ use pluto_eth2api::EthBeaconNodeApiClient; use pluto_p2p::p2p_context::P2PContext; use tokio::{sync::mpsc, time::MissedTickBehavior}; use tokio_util::sync::CancellationToken; -use tracing::{error, warn}; +use tracing::{Instrument as _, Span, error, warn}; use super::{ metrics::MONITORING_METRICS, @@ -52,17 +52,24 @@ pub fn start_ready_checker( let readiness = ReadyState::new(); // Both background tasks are detached; their lifecycle is bound to `ct` and // they stop when the token is cancelled. - let _version_task = tokio::spawn(run_beacon_node_version_metric( - beacon_node.clone(), - ct.clone(), - )); - let _task = tokio::spawn(run_ready_checker( - p2p_context, - beacon_node, - validator_api_calls, - ct, - readiness.clone(), - )); + // + // `tokio::spawn` starts a task with an empty span stack, so both futures + // are re-attached to the caller's span; charon's monitoring API sets no + // topic of its own and runs as a lifecycle hook, which puts its logs on + // `app-start`. + let _version_task = tokio::spawn( + run_beacon_node_version_metric(beacon_node.clone(), ct.clone()).instrument(Span::current()), + ); + let _task = tokio::spawn( + run_ready_checker( + p2p_context, + beacon_node, + validator_api_calls, + ct, + readiness.clone(), + ) + .instrument(Span::current()), + ); readiness } diff --git a/crates/app/src/node/mod.rs b/crates/app/src/node/mod.rs index b27c72516..8f64b2302 100644 --- a/crates/app/src/node/mod.rs +++ b/crates/app/src/node/mod.rs @@ -23,6 +23,7 @@ pub use config::AppConfig; use std::{ collections::HashMap, + future::Future, sync::Arc, time::{Duration, SystemTime}, }; @@ -37,6 +38,7 @@ use pluto_testutil::{ }; use tokio::task::JoinSet; use tokio_util::sync::CancellationToken; +use tracing::Instrument as _; use behaviour::{CoreBehaviour, CoreHandles}; use wire::{ParSigExSeam, SlotTickFn, ValidatorInfo, WireInputs, WiredComponents}; @@ -219,6 +221,21 @@ impl App { } } +/// Puts a long-lived background task under the `app-start` topic. +/// +/// Charon's lifecycle manager hands every background hook a +/// `log.WithTopic(context.Background(), "app-start")` context +/// (`app/lifecycle/hook.go`), so a task that never sets a topic of its own +/// still reports under `app-start`. `tokio::spawn` has no such inheritance — +/// it starts the task with an empty span stack — so the span is attached here +/// instead, at the one place background tasks are started. +/// +/// Tasks that open their own topic span (`sched`, `tracker`, `health`, …) +/// shadow this one, exactly as a nested `log.WithTopic` does in Go. +fn background(task: F) -> tracing::instrument::Instrumented { + task.instrument(tracing::debug_span!("app-start", topic = "app-start")) +} + /// Loads the cluster lock + key, builds the consensus component and P2P /// behaviours, wires the core workflow, and drives the node. /// @@ -698,37 +715,37 @@ async fn run_lifecycle( // Supervise the scheduler actor alongside the other long-lived tasks so // its exit triggers node shutdown (it only exits on cancellation). - tasks.extend([async move { + tasks.extend([background(async move { let _ = scheduler_task.await; Ok::<(), AppError>(()) - }]); + })]); // Swarm drive loop (push-based routing inside behaviours). { let ct = ct.clone(); - tasks.spawn(async move { + tasks.spawn(background(async move { drive_network(node, ct).await; Ok(()) - }); + })); } // ParSigDB trim task. { let parsigdb = Arc::clone(&parsigdb); - tasks.spawn(async move { + tasks.spawn(background(async move { parsigdb.trim(parsigdb_deadliner_rx).await; Ok(()) - }); + })); } // Networked inclusion checker: polls the beacon node once per due slot and // resolves each tracked duty's on-chain inclusion step. { let ct = ct.clone(); - tasks.spawn(async move { + tasks.spawn(background(async move { inclusion_checker.run(ct).await; Ok(()) - }); + })); } // Private-key lock maintenance loop. Only spawn `run` when locking is @@ -738,7 +755,9 @@ async fn run_lifecycle( let svc = Arc::clone(svc); // A lock-maintenance failure fails the run (Charon parity); a graceful // `close()` returns `Ok`. - tasks.spawn(async move { svc.run().await.map_err(AppError::PrivKeyLock) }); + tasks.spawn(background(async move { + svc.run().await.map_err(AppError::PrivKeyLock) + })); } // ---- Monitoring API ---- @@ -778,10 +797,10 @@ async fn run_lifecycle( Box::new(health::ViseGatherer), num_validators, ); - tasks.spawn(async move { + tasks.spawn(background(async move { checker.run(ct).await; Ok(()) - }); + })); } // Validator API axum server. Each request bumps the readiness "vc @@ -795,20 +814,20 @@ async fn run_lifecycle( } }, )); - tasks.spawn(serve_validator_api( + tasks.spawn(background(serve_validator_api( validator_api_addr, validator_api_router, ct.clone(), - )); + ))); // Monitoring HTTP server (metrics + livez + readyz). - tasks.spawn(serve_monitoring_api( + tasks.spawn(background(serve_monitoring_api( monitoring_addr, monitoringapi::router_with_state( monitoringapi::MonitoringState::new(readiness).with_labels(monitoring_labels), ), ct.clone(), - )); + ))); // Supervise: stop on cancellation or first task completion. A failed task // fails the whole run (Charon parity). diff --git a/crates/app/src/node/wire.rs b/crates/app/src/node/wire.rs index 38bead90e..e1a692a42 100644 --- a/crates/app/src/node/wire.rs +++ b/crates/app/src/node/wire.rs @@ -48,6 +48,7 @@ use pluto_eth2api::{ }; use pluto_featureset::{Feature, FeatureSet, Status}; use tokio_util::sync::CancellationToken; +use tracing::Instrument as _; use crate::node::AppError; @@ -635,8 +636,11 @@ pub async fn wire_core_workflow( move |duty: Duty, value: pbcore::UnsignedDataSet| { let dutydb = Arc::clone(&dutydb); let tracker = Arc::clone(&tracker); - let span = tracing::debug_span!("app-start", topic = "app-start"); - tokio::spawn(tracing::Instrument::instrument( + // `tokio::spawn` starts the task with an empty span stack, so + // re-attach the caller's span: this callback runs during core + // wiring, under the node's `app-start` topic. + let span = tracing::Span::current(); + tokio::spawn( async move { let core_set = match unsigneddata::unsigned_data_set_from_proto( &duty.duty_type, @@ -659,9 +663,9 @@ pub async fn wire_core_workflow( } }; tracker.duty_db_stored(duty, &pubkeys, step_err).await; - }, - span, - )); + } + .instrument(span), + ); Ok(()) }, )); diff --git a/crates/consensus/src/qbft/runner.rs b/crates/consensus/src/qbft/runner.rs index 5b67f991e..7e1045ed9 100644 --- a/crates/consensus/src/qbft/runner.rs +++ b/crates/consensus/src/qbft/runner.rs @@ -18,6 +18,7 @@ use tokio::{ time::Duration, }; use tokio_util::sync::CancellationToken; +use tracing::Instrument as _; use crate::{ instance::{self, InstanceIo, RunnerError, RunnerResult}, @@ -265,43 +266,53 @@ async fn run_instance_inner( Sniffer::new(i64::try_from(nodes).expect("node count fits i64"), peer_idx), )); + // `JoinSet::spawn` starts each task with an empty span stack, just like + // `tokio::spawn`, so the `qbft` topic opened by this function would not + // reach any of the instance's subtasks. Charon gets this for free by + // passing the instance `ctx` into every goroutine; here the span is + // re-attached to each spawned future by hand. + let qbft_span = tracing::Span::current(); + let mut tasks = JoinSet::new(); - tasks.spawn(bridge_mpsc_to_crossbeam( - instance_ct.clone(), - inner_recv_rx, - core_recv_tx, - )); - tasks.spawn(bridge_mpsc_to_crossbeam( - instance_ct.clone(), - hash_rx, - core_hash_tx, - )); - tasks.spawn(bridge_mpsc_to_crossbeam( - instance_ct.clone(), - verify_rx, - core_verify_tx, - )); + tasks.spawn( + bridge_mpsc_to_crossbeam(instance_ct.clone(), inner_recv_rx, core_recv_tx) + .instrument(qbft_span.clone()), + ); + tasks.spawn( + bridge_mpsc_to_crossbeam(instance_ct.clone(), hash_rx, core_hash_tx) + .instrument(qbft_span.clone()), + ); + tasks.spawn( + bridge_mpsc_to_crossbeam(instance_ct.clone(), verify_rx, core_verify_tx) + .instrument(qbft_span.clone()), + ); { let transport = Arc::clone(&transport); let instance_ct = instance_ct.clone(); let transport_error = Arc::clone(&transport_error); - tasks.spawn(async move { - if let Err(err) = transport.process_receives(instance_ct, outer_rx).await { - *transport_error - .lock() - .unwrap_or_else(PoisonError::into_inner) = Some(err.to_string()); + tasks.spawn( + async move { + if let Err(err) = transport.process_receives(instance_ct, outer_rx).await { + *transport_error + .lock() + .unwrap_or_else(PoisonError::into_inner) = Some(err.to_string()); + } } - }); + .instrument(qbft_span.clone()), + ); } { let instance_ct = instance_ct.clone(); let core_cts = Arc::clone(&core_cts); - tasks.spawn(async move { - instance_ct.cancelled().await; - core_cts.cancel(); - }); + tasks.spawn( + async move { + instance_ct.cancelled().await; + core_cts.cancel(); + } + .instrument(qbft_span.clone()), + ); } let decide_callback: DecideCallback = { @@ -379,7 +390,11 @@ async fn run_instance_inner( let core_ct_for_run = core_ct.clone(); let core_duty = duty.clone(); + // The blocking core runs the `Definition` callbacks, which log their own + // warnings; entering the span keeps those on `topic="qbft"`. + let core_span = qbft_span.clone(); let core_result = tokio::task::spawn_blocking(move || { + let _entered = core_span.enter(); qbft::run( &core_ct_for_run, &def, diff --git a/crates/core/src/scheduler.rs b/crates/core/src/scheduler.rs index 772a1bd69..8bf1992b7 100644 --- a/crates/core/src/scheduler.rs +++ b/crates/core/src/scheduler.rs @@ -6,6 +6,7 @@ use std::{ use backon::{BackoffBuilder, Retryable}; use tokio::{sync, task::JoinHandle}; use tokio_util::{future::FutureExt, sync::CancellationToken}; +use tracing::Instrument as _; use crate::{scheduler::metrics::SCHEDULER_METRICS, types}; use pluto_eth2api::{v1, valcache}; @@ -121,7 +122,7 @@ impl SchedulerBuilder { // TODO: We might want to return a handle so clients can `.abort()` them // to drop the subscription let label: Arc = Arc::from(label.as_ref()); - tokio::spawn(async move { + let pump = async move { loop { match rx.recv().await { Ok(slot) => { @@ -129,11 +130,12 @@ impl SchedulerBuilder { // delay later slots for this subscriber. let fut = f(&slot); let label = Arc::clone(&label); - tokio::spawn(async move { + let emit = async move { if let Err(err) = fut.await { tracing::error!(err = ?err, slot = %slot.slot, label = &*label, "Emit scheduled slot event"); } - }); + }; + tokio::spawn(emit.instrument(sched_span())); } // NOTE: Handlers are spawned per event above, so the // receive loop drains immediately. Lag therefore no longer @@ -157,7 +159,8 @@ impl SchedulerBuilder { Err(sync::broadcast::error::RecvError::Closed) => break, } } - }); + }; + tokio::spawn(pump.instrument(sched_span())); } /// Subscribes a callback function for triggered duties. @@ -170,7 +173,7 @@ impl SchedulerBuilder { let mut rx = self.duty_broadcast.subscribe(); let label: Arc = Arc::from(label.as_ref()); - tokio::spawn(async move { + let pump = async move { loop { match rx.recv().await { Ok((duty, set)) => { @@ -181,11 +184,12 @@ impl SchedulerBuilder { // subscriber. let fut = f(&duty, &set); let label = Arc::clone(&label); - tokio::spawn(async move { + let trigger = async move { if let Err(err) = fut.await { tracing::error!(err = ?err, %duty, label = &*label, "Trigger duty subscriber error"); } - }); + }; + tokio::spawn(trigger.instrument(sched_span())); } // NOTE: Same as in `subscribe_slot` Err(sync::broadcast::error::RecvError::Lagged(skipped)) => { @@ -195,7 +199,8 @@ impl SchedulerBuilder { Err(sync::broadcast::error::RecvError::Closed) => break, } } - }); + }; + tokio::spawn(pump.instrument(sched_span())); } /// Add a source of chain reorgs to the scheduler. @@ -297,6 +302,18 @@ impl SchedulerHandle { } } +/// The scheduler's `sched` topic span. +/// +/// Charon opens it once in `Scheduler.Run` and every goroutine started from +/// there inherits it through `context.Context`, so subscriber and slot-ticker +/// errors are all reported under `sched`. `tokio::spawn` starts a task with an +/// empty span stack instead, and several of these tasks are started from +/// wiring code rather than from the actor loop, so each opens the span itself +/// rather than inheriting one. +fn sched_span() -> tracing::Span { + tracing::debug_span!("sched", topic = "sched") +} + struct SchedulerActor { client: pluto_eth2api::EthBeaconNodeApiClient, validator_cache: valcache::ValidatorCache, @@ -429,23 +446,26 @@ impl SchedulerActor { let ct = ct.clone(); let slot = slot.clone(); let broadcast = self.duty_broadcast.clone(); - tokio::spawn(async move { - if delay_slot_offset(&slot, &duty) - .with_cancellation_token_owned(ct) - .await - .is_none() - { - // Cancelled early - return; - } + tokio::spawn( + async move { + if delay_slot_offset(&slot, &duty) + .with_cancellation_token_owned(ct) + .await + .is_none() + { + // Cancelled early + return; + } - SCHEDULER_METRICS.duty_total[&duty.duty_type.to_string()] - .inc_by(def_set.len() as u64); + SCHEDULER_METRICS.duty_total[&duty.duty_type.to_string()] + .inc_by(def_set.len() as u64); - // NOTE: Ignore send errors, it means that there are no - // subscribers. - let _ = broadcast.send((duty.clone(), def_set.clone())); - }); + // NOTE: Ignore send errors, it means that there are no + // subscribers. + let _ = broadcast.send((duty.clone(), def_set.clone())); + } + .instrument(sched_span()), + ); } if slot.last_in_epoch() @@ -657,7 +677,7 @@ async fn new_slot_ticker( }; let (tx, rx) = sync::mpsc::channel(CHANNEL_BUFFER_SIZE); - tokio::spawn(async move { + let ticker = async move { let mut slot = current_slot(); loop { @@ -695,7 +715,8 @@ async fn new_slot_ticker( slot = next_slot; } - }); + }; + tokio::spawn(ticker.instrument(sched_span())); Ok(rx) } diff --git a/crates/core/src/tracker/inclusion.rs b/crates/core/src/tracker/inclusion.rs index 4f614b72c..900b0f108 100644 --- a/crates/core/src/tracker/inclusion.rs +++ b/crates/core/src/tracker/inclusion.rs @@ -693,6 +693,10 @@ impl InclusionChecker { /// Drives inclusion checking until `cancel` fires: once per due slot, ask /// the beacon node whether that slot produced a block, feed the verdict to /// the core, then trim submissions old enough to count as missed. + /// + /// Runs under the `tracker` topic, matching charon's + /// `InclusionChecker.Run`. + #[tracing::instrument(name = "tracker", level = "debug", skip_all, fields(topic = "tracker"))] pub async fn run(self: Arc, cancel: CancellationToken) { let mut ticker = tokio::time::interval(Duration::from_secs(1)); let mut checked_slot: Option = None; diff --git a/crates/dkg/src/dkg.rs b/crates/dkg/src/dkg.rs index 3f58b81ea..89fe8e040 100644 --- a/crates/dkg/src/dkg.rs +++ b/crates/dkg/src/dkg.rs @@ -7,7 +7,7 @@ use pluto_app::{privkeylock, utils::UtilsError}; use pluto_core::version; use tokio::select; use tokio_util::sync::CancellationToken; -use tracing::{debug, error, info, warn}; +use tracing::{Instrument as _, Span, debug, error, info, warn}; pub use crate::{ aggregate::{AggregateError, agg_deposit_data, agg_lock_hash_sig, agg_validator_registrations}, @@ -581,7 +581,13 @@ async fn run_inner(conf: Config, ct: CancellationToken) -> Result<(), DkgError> let sync_clients = handlers.sync.clone(); let sync_server = handlers.sync_server.clone(); let network_ct = ct.child_token(); - let network_task = pluto_tracing::spawn(drive_dkg_network(node, network_ct.clone())); + // A bare `tokio::spawn` starts the driver with an empty span stack, which + // would drop the `dkg` topic set by `run` and count the driver's warnings + // on `app_log_warn_total{topic=""}`. Re-attach the current span so the + // subtask keeps it, mirroring charon passing `context.Context` into the + // goroutine. + let network_task = + tokio::spawn(drive_dkg_network(node, network_ct.clone()).instrument(Span::current())); let result = run_ceremony() .conf(&conf) diff --git a/crates/priority/src/prioritiser.rs b/crates/priority/src/prioritiser.rs index 39c754efa..fd03783c2 100644 --- a/crates/priority/src/prioritiser.rs +++ b/crates/priority/src/prioritiser.rs @@ -29,6 +29,7 @@ use pluto_core::{ use pluto_p2p::p2p_context::P2PContext; use tokio::sync::{mpsc, oneshot}; use tokio_util::sync::CancellationToken; +use tracing::Instrument as _; use crate::{ calculate, @@ -518,14 +519,20 @@ fn start_consensus( let consensus = inner.consensus.clone(); let duty = duty.clone(); let ct = ct.clone(); - tokio::spawn(async move { - // Fire-and-forget so the instance keeps servicing peer requests while - // consensus runs. The instance token reaches consensus, so cancellation - // tears the proposal down; a propose failure is unexpected. - if let Err(err) = consensus.propose_priority(duty, result, &ct).await { - tracing::warn!(%err, "Priority protocol consensus"); + tokio::spawn( + async move { + // Fire-and-forget so the instance keeps servicing peer requests + // while consensus runs. The instance token reaches consensus, so + // cancellation tears the proposal down; a propose failure is + // unexpected. + if let Err(err) = consensus.propose_priority(duty, result, &ct).await { + tracing::warn!(%err, "Priority protocol consensus"); + } } - }); + // `tokio::spawn` starts the task with an empty span stack; re-attach + // the caller's span so the warning above keeps its topic. + .instrument(tracing::Span::current()), + ); Ok(()) } diff --git a/crates/relay-server/src/web.rs b/crates/relay-server/src/web.rs index 96d5be8a0..8721c46f0 100644 --- a/crates/relay-server/src/web.rs +++ b/crates/relay-server/src/web.rs @@ -17,7 +17,7 @@ use libp2p::{Multiaddr, PeerId, multiaddr}; use pluto_eth2util::enr::{EnrEntry, Record}; use tokio::{net::TcpListener, sync::RwLock}; use tokio_util::sync::CancellationToken; -use tracing::{debug, info, instrument, warn}; +use tracing::{Instrument as _, Span, debug, info, instrument, warn}; use vise_exporter::{MetricsExporter, MetricsServer}; use crate::{ @@ -123,11 +123,20 @@ pub async fn enr_server( ) -> Result<()> { info!("Starting ENR server"); - // Start external host resolver task if configured + // Start external host resolver task if configured. + // + // `tokio::spawn` gives the new task an empty span stack, so the resolver + // would lose the `relay` topic this server runs under and its warnings + // would land on `app_log_warn_total{topic=""}`. Re-attaching the current + // span restores what charon gets for free by handing the goroutine its + // `context.Context`. let resolver_handle = state.p2p_config.external_host.clone().map(|external_host| { let state = state.clone(); let ct = ct.child_token(); - pluto_tracing::spawn(resolve_external_host_periodically(state, external_host, ct)) + tokio::spawn( + resolve_external_host_periodically(state, external_host, ct) + .instrument(Span::current()), + ) }); info!( diff --git a/crates/testutil/src/validatormock/component.rs b/crates/testutil/src/validatormock/component.rs index 0e8de91a6..2d0c5f430 100644 --- a/crates/testutil/src/validatormock/component.rs +++ b/crates/testutil/src/validatormock/component.rs @@ -24,7 +24,7 @@ use tokio::{ task::JoinHandle, }; use tokio_util::sync::CancellationToken; -use tracing::warn; +use tracing::{Instrument as _, warn}; use super::{ SignFunc, @@ -290,7 +290,7 @@ async fn run_scheduler( let Some(scheduled) = maybe else { break }; let inner_for_task = Arc::clone(&inner); let cancel_for_task = cancel.clone(); - duties.spawn(tracing::Instrument::instrument(async move { + let duty_task = async move { let start_time = scheduled.start_time; let slot = scheduled.slot; let duty_label = scheduled.duty_type.clone(); @@ -315,7 +315,10 @@ async fn run_scheduler( } } } - }, tracing::Span::current())); + }; + // `JoinSet::spawn` starts the task with an empty span stack, + // so re-attach the `vmock` span opened by this function. + duties.spawn(duty_task.instrument(tracing::Span::current())); } // Reap finished duties to keep the JoinSet bounded. Disabled when // empty — `Some(_)` does not match `None`. diff --git a/crates/tracing/src/layers/metrics.rs b/crates/tracing/src/layers/metrics.rs index 2137ca4e9..5fa3834b3 100644 --- a/crates/tracing/src/layers/metrics.rs +++ b/crates/tracing/src/layers/metrics.rs @@ -82,6 +82,7 @@ where #[cfg(test)] mod tests { use super::*; + use tracing::Instrument as _; use tracing_subscriber::layer::SubscriberExt as _; #[test] @@ -102,6 +103,51 @@ mod tests { assert_eq!(TRACING_METRICS.warn_total[&topic.to_owned()].get(), 1); } + #[tokio::test] + async fn instrumented_spawn_keeps_topic_across_task_boundary() { + // `tokio::spawn` starts a task with an empty span stack, so a subtask + // only keeps its parent's topic when the future is explicitly + // re-attached to the current span. Callers that spawn from inside a + // topic span must do this by hand; the two assertions below pin both + // halves of that contract. + let topic = "metrics_layer_spawn_topic"; + let subscriber = tracing_subscriber::registry().with(MetricsLayer); + + let before = TRACING_METRICS.error_total[&topic.to_owned()].get(); + + // `Instrument` captures the dispatcher as well as the span, so the + // default subscriber set here applies inside the spawned task. + let guard = tracing::subscriber::set_default(subscriber); + let span = tracing::info_span!("component", topic); + + let instrumented = { + let _enter = span.enter(); + tokio::spawn( + async { + tracing::error!("boom from instrumented task"); + } + .instrument(tracing::Span::current()), + ) + }; + instrumented.await.unwrap(); + + let bare = { + let _enter = span.enter(); + tokio::spawn(async { + tracing::error!("boom from bare task"); + }) + }; + bare.await.unwrap(); + + drop(guard); + + assert_eq!( + TRACING_METRICS.error_total[&topic.to_owned()].get(), + before.saturating_add(1), + "only the instrumented spawn should be counted under the topic" + ); + } + #[test] fn events_without_topic_use_empty_label() { let subscriber = tracing_subscriber::registry().with(MetricsLayer); diff --git a/crates/tracing/src/lib.rs b/crates/tracing/src/lib.rs index 5ef52f9b3..7e7da3c2b 100644 --- a/crates/tracing/src/lib.rs +++ b/crates/tracing/src/lib.rs @@ -16,10 +16,6 @@ pub mod layers; /// Metrics for the tracing. pub mod metrics; -/// Span-propagating task spawning. -pub mod spawn; - pub use config::{ConsoleConfig, LokiConfig, TracingConfig}; pub use init::{LokiWorker, init}; -pub use spawn::spawn; diff --git a/crates/tracing/src/spawn.rs b/crates/tracing/src/spawn.rs deleted file mode 100644 index f3342d08a..000000000 --- a/crates/tracing/src/spawn.rs +++ /dev/null @@ -1,95 +0,0 @@ -//! Span-propagating task spawning. -//! -//! [`tokio::spawn`] does **not** carry the current [`tracing`] span into the -//! spawned future: the new task starts with an empty span stack. That breaks -//! the `topic` label used by [`crate::layers::metrics::MetricsLayer`], because -//! a `warn!`/`error!` emitted from a bare spawn lands on `topic=""` even when -//! the spawning code is inside a component's `topic` span. -//! -//! Charon derives the same label from `context.Context`, which *is* propagated -//! into goroutines. To restore context-like propagation here, wrap the spawned -//! future with [`tracing::Instrument`] and attach [`tracing::Span::current`]. -//! -//! Prefer [`spawn()`] over [`tokio::spawn`] in long-running components so that -//! the component's root `topic` span is inherited by its subtasks by default. - -use std::future::Future; - -use tokio::task::JoinHandle; -use tracing::Instrument as _; - -/// Like [`tokio::spawn`], but attaches the current [`tracing::Span`] to the -/// spawned future so span context (and therefore the metrics `topic` label) is -/// propagated across the task boundary. -/// -/// Use this instead of [`tokio::spawn`] when the calling code runs inside a -/// component's `topic` span and the spawned work should be attributed to the -/// same topic. -pub fn spawn(future: F) -> JoinHandle -where - F: Future + Send + 'static, - F::Output: Send + 'static, -{ - tokio::spawn(future.instrument(tracing::Span::current())) -} - -#[cfg(test)] -mod tests { - use tracing_subscriber::layer::SubscriberExt as _; - - use crate::{layers::metrics::MetricsLayer, metrics::TRACING_METRICS}; - - #[tokio::test] - async fn spawn_propagates_topic_across_task_boundary() { - let topic = "spawn_helper_propagation_test"; - let subscriber = tracing_subscriber::registry().with(MetricsLayer); - - let before = TRACING_METRICS.error_total[&topic.to_owned()].get(); - - // `Instrument` captures both the span and the dispatcher at spawn time, - // so the default subscriber set below applies to the spawned task. - let guard = tracing::subscriber::set_default(subscriber); - - let span = tracing::info_span!("component", topic); - let handle = { - let _enter = span.enter(); - super::spawn(async { - tracing::error!("boom from spawned task"); - }) - }; - handle.await.unwrap(); - - drop(guard); - - let after = TRACING_METRICS.error_total[&topic.to_owned()].get(); - assert_eq!( - after, - before.saturating_add(1), - "spawned task should inherit topic" - ); - } - - #[tokio::test] - async fn bare_tokio_spawn_loses_topic() { - // Documents the behaviour the helper fixes: a bare spawn drops the - // topic and is counted under the empty label. - let topic = "spawn_helper_bare_test"; - let subscriber = tracing_subscriber::registry().with(MetricsLayer); - - let before = TRACING_METRICS.error_total[&topic.to_owned()].get(); - - let guard = tracing::subscriber::set_default(subscriber); - let span = tracing::info_span!("component", topic); - let handle = { - let _enter = span.enter(); - tokio::spawn(async { - tracing::error!("boom from bare spawned task"); - }) - }; - handle.await.unwrap(); - drop(guard); - - let after = TRACING_METRICS.error_total[&topic.to_owned()].get(); - assert_eq!(after, before, "bare spawn must not inherit topic"); - } -}