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

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
 }

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,8 +221,32 @@ 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<F: Future>(task: F) -> tracing::instrument::Instrumented<F> {
+    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.
+///
+/// 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 ----
     //
@@ -689,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
@@ -729,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 ----
@@ -769,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
@@ -786,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).

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,26 +636,36 @@ 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) {
+                // `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,
+                            &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;
+                    }
+                    .instrument(span),
+                );
                 Ok(())
             },
         ));

crates/cli/src/commands/relay.rs

@@ -196,6 +196,7 @@ pub struct RelayP2PArgs {
     pub disable_reuseport: bool,
 }
 
+#[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

@@ -13,6 +13,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,
@@ -477,22 +478,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.