feat(tracing): propagate log topic across spawns and label components

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 ----
     //

crates/app/src/node/wire.rs

@@ -652,26 +652,33 @@ 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<PubKey> = 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<PubKey> = 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(())
             },
         ));

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,

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.

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<M>(
     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,

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?,

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(());

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<types::Slot>,

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,

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<Duty, Vec<Event>> = HashMap::new();
 

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<Arc<AppState>>,
     Path(epoch): Path<u64>,

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,

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);
             }

crates/p2p/src/p2p.rs

@@ -630,6 +630,7 @@ impl<B: NetworkBehaviour> Node<B> {
     }
 
     /// 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<PlutoBehaviourEvent<B>>) {
         match event {
             // Identify - update peer addresses in the peer store.

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),
+        );
     }
 }
 

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,

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!(

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<Inner>,
     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`.

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;