From 8eb6963db4a31460996e78175decc8cb6c1e15d3 Mon Sep 17 00:00:00 2001 From: Julian Hofer Date: Fri, 3 Jul 2026 16:34:06 +0000 Subject: [PATCH 01/12] feat: forward backend tracing logs to pixi as JSON-RPC notifications Backend build processes previously rendered all their tracing output to stderr as plain text, so pixi could only treat it as opaque build output: no levels, no targets, no span context, and no interaction with pixi's own log filtering. This surfaces backend diagnostics as first-class tracing data in pixi. Design (supersedes the approach reviewed in #6312): - Transport: structured records travel as `log/message` JSON-RPC notifications on the existing stdin/stdout channel, instead of sentinel-tagged JSON on stderr (fragile: child processes share the inherited stderr handle) or a dedicated socket/named pipe (two platform-specific transports and a second lifecycle to manage). The channel is inherently process-scoped, so span ids stay valid across RPC calls, during setup, and between requests. - Negotiation: a new `supports_log_messages` field on `FrontendCapabilities`. Backends stay silent unless the frontend advertises it, so both mixed-version directions degrade to today's behavior without a Pixi Build API version bump. - Content split: INFO events keep flowing as rendered plain text on stderr because rattler-build uses INFO as its plaintext build-output stream, which pixi buffers and replays when a build fails. Everything else (WARN/ERROR/DEBUG/TRACE plus span lifecycles) goes over the structured channel. After activation, the stderr handler is pinned to INFO-only so records are never rendered twice. - Span fidelity: the backend mirrors span open/close over the wire (lazily: a span's open record is only sent once an event actually flows inside it, which also absorbs the pre-negotiation window), and the frontend reconstructs real `tracing` spans through runtime-interned callsites keyed by (name, target, level), so events render with their `outer:inner:` context and EnvFilter directives apply with correct metadata. Events re-emit under a `backend::`-prefixed target with fields folded into the message. - Verbosity: pixi passes its current max level to the backend via PIXI_BUILD_BACKEND_LOG_LEVEL, which the backend treats as a floor for its own filter, so `pixi -vvv` surfaces backend debug/trace without any reload machinery. The backend server loop now owns stdout through a single writer task fed by both RPC responses and notifications, replacing the deprecated jsonrpc-stdio-server dependency. --- Cargo.lock | 15 +- Cargo.toml | 1 - crates/pixi_build_backend/Cargo.toml | 2 - crates/pixi_build_backend/src/cli.rs | 85 +++- crates/pixi_build_backend/src/lib.rs | 1 + .../src/log_message_layer.rs | 295 +++++++++++++ crates/pixi_build_backend/src/server.rs | 242 ++++++++++- crates/pixi_build_frontend/Cargo.toml | 1 + .../src/backend/json_rpc.rs | 318 ++++++++++++++- .../src/backend/log_forwarder.rs | 386 ++++++++++++++++++ crates/pixi_build_frontend/src/backend/mod.rs | 1 + crates/pixi_build_frontend/src/jsonrpc/mod.rs | 2 + crates/pixi_build_types/src/capabilities.rs | 11 +- crates/pixi_build_types/src/lib.rs | 3 + .../src/procedures/log_message.rs | 167 ++++++++ crates/pixi_build_types/src/procedures/mod.rs | 1 + 16 files changed, 1486 insertions(+), 45 deletions(-) create mode 100644 crates/pixi_build_backend/src/log_message_layer.rs create mode 100644 crates/pixi_build_frontend/src/backend/log_forwarder.rs create mode 100644 crates/pixi_build_types/src/procedures/log_message.rs diff --git a/Cargo.lock b/Cargo.lock index ae950ace3c..d97917c8a3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4387,19 +4387,6 @@ dependencies = [ "unicase", ] -[[package]] -name = "jsonrpc-stdio-server" -version = "18.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6878586767497326eb3d011bd6dbb583e9f008b11528f82fd47798ec46bb6c26" -dependencies = [ - "futures", - "jsonrpc-core", - "log", - "tokio", - "tokio-util 0.6.10", -] - [[package]] name = "jsonrpsee" version = "0.26.0" @@ -6256,7 +6243,6 @@ dependencies = [ "jiff", "jsonrpc-core", "jsonrpc-http-server", - "jsonrpc-stdio-server", "miette 7.6.0", "minijinja", "ordermap", @@ -6343,6 +6329,7 @@ dependencies = [ "tokio", "tokio-util 0.7.18", "tracing", + "tracing-subscriber", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index f062a21b65..ef44409256 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -74,7 +74,6 @@ itertools = "0.15.0" jiff = "0.2.8" jsonrpc-core = "18.0.0" jsonrpc-http-server = "18.0.0" -jsonrpc-stdio-server = "18.0.0" jsonrpsee = "=0.26.0" libc = { version = "0.2.170", default-features = false } memchr = "2.8" diff --git a/crates/pixi_build_backend/Cargo.toml b/crates/pixi_build_backend/Cargo.toml index 4416671f5a..b74ee5fad0 100644 --- a/crates/pixi_build_backend/Cargo.toml +++ b/crates/pixi_build_backend/Cargo.toml @@ -46,8 +46,6 @@ pixi_build_types = { workspace = true } jsonrpc-core = { workspace = true } jsonrpc-http-server = { workspace = true } -jsonrpc-stdio-server = { workspace = true } - [dev-dependencies] insta = { version = "1.42.1", features = ["yaml", "redactions", "filters"] } diff --git a/crates/pixi_build_backend/src/cli.rs b/crates/pixi_build_backend/src/cli.rs index f7c7b61237..e65fa11dcb 100644 --- a/crates/pixi_build_backend/src/cli.rs +++ b/crates/pixi_build_backend/src/cli.rs @@ -1,14 +1,19 @@ +use std::sync::{Arc, atomic::AtomicBool}; + use clap::{Parser, Subcommand}; use clap_verbosity_flag::{InfoLevel, Verbosity}; use miette::IntoDiagnostic; use pixi_build_types::{ BackendCapabilities, FrontendCapabilities, + procedures::log_message::{LOG_LEVEL_ENV, LogMessage}, procedures::negotiate_capabilities::NegotiateCapabilitiesParams, }; use rattler_build_core::console_utils::{LoggingOutputHandler, get_default_env_filter}; -use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; +use tokio::sync::mpsc; +use tracing::Level; +use tracing_subscriber::{Layer, filter::filter_fn, layer::SubscriberExt, util::SubscriberInitExt}; -use crate::{protocol::ProtocolInstantiator, server::Server}; +use crate::{log_message_layer::LogMessageLayer, protocol::ProtocolInstantiator, server::Server}; #[allow(missing_docs)] #[derive(Parser)] @@ -33,12 +38,22 @@ pub enum Commands { Capabilities, } +/// The log channel handed to the server when it runs over stdio. +type LogMessageChannel = (mpsc::UnboundedReceiver, Arc); + /// Run the sever on the specified port or over stdin/stdout. -async fn run_server(port: Option, protocol: T) -> miette::Result<()> { - let server = Server::new(protocol); +async fn run_server( + port: Option, + protocol: T, + log_messages: Option, +) -> miette::Result<()> { + let mut server = Server::new(protocol); if let Some(port) = port { server.run_over_http(port) } else { + if let Some((receiver, enabled)) = log_messages { + server = server.with_log_messages(receiver, enabled); + } // running over stdin/stdout server.run().await } @@ -52,22 +67,57 @@ pub(crate) async fn main_impl().ok()); + let cli_level = args.verbose.log_level_filter(); + let effective_level = frontend_level.map_or(cli_level, |level| level.max(cli_level)); + // `get_default_env_filter` only enables `rattler_build` and friends, which // silently drops events from the backend crates themselves (e.g. the // "`pypi-conda-map` is set but the mapping is disabled" warning). Add a - // default directive so warnings from any target are surfaced. + // default directive so warnings from any target are surfaced — raised + // further when the frontend asked for more. + let default_directive = log_to_tracing_level_filter(effective_level) + .max(tracing_subscriber::filter::LevelFilter::WARN); let registry = tracing_subscriber::registry().with( - get_default_env_filter(args.verbose.log_level_filter()) + get_default_env_filter(effective_level) .into_diagnostic()? - .add_directive(tracing_subscriber::filter::LevelFilter::WARN.into()), + .add_directive(default_directive.into()), ); - registry.with(log_handler.clone()).init(); + // When we serve a frontend over stdio, structured log records can travel + // to it as `log/message` notifications. The layer stays dormant until + // the frontend advertises support during capability negotiation; until + // then (and for INFO events — the plaintext build-output stream — always) + // the human-readable handler keeps rendering to stderr. + let serves_stdio = args.command.is_none() && args.http_port.is_none(); + let log_messages = if serves_stdio { + let (sender, receiver) = mpsc::unbounded_channel(); + let enabled = Arc::new(AtomicBool::new(false)); + + let stderr_enabled = enabled.clone(); + let stderr_filter = filter_fn(move |metadata| { + metadata.is_span() + || *metadata.level() == Level::INFO + || !stderr_enabled.load(std::sync::atomic::Ordering::Acquire) + }); + registry + .with(log_handler.clone().with_filter(stderr_filter)) + .with(LogMessageLayer::new(sender, enabled.clone())) + .init(); + Some((receiver, enabled)) + } else { + registry.with(log_handler.clone()).init(); + None + }; let factory = factory(log_handler); match args.command { - None => run_server(args.http_port, factory).await, + None => run_server(args.http_port, factory, log_messages).await, Some(Commands::Capabilities) => { let backend_capabilities = capabilities::().await?; eprintln!( @@ -85,6 +135,21 @@ pub(crate) async fn main_impl tracing_subscriber::filter::LevelFilter { + use clap_verbosity_flag::log::LevelFilter as LogLevelFilter; + use tracing_subscriber::filter::LevelFilter; + match level { + LogLevelFilter::Off => LevelFilter::OFF, + LogLevelFilter::Error => LevelFilter::ERROR, + LogLevelFilter::Warn => LevelFilter::WARN, + LogLevelFilter::Info => LevelFilter::INFO, + LogLevelFilter::Debug => LevelFilter::DEBUG, + LogLevelFilter::Trace => LevelFilter::TRACE, + } +} + /// The entry point for the CLI which should be called from the backends implementation. pub async fn main T>( factory: F, @@ -105,7 +170,7 @@ pub async fn main_ext() -> miette::Result { let result = Factory::negotiate_capabilities(NegotiateCapabilitiesParams { - capabilities: FrontendCapabilities {}, + capabilities: FrontendCapabilities::default(), }) .await?; diff --git a/crates/pixi_build_backend/src/lib.rs b/crates/pixi_build_backend/src/lib.rs index 7fa23e5566..da89fb66a5 100644 --- a/crates/pixi_build_backend/src/lib.rs +++ b/crates/pixi_build_backend/src/lib.rs @@ -1,6 +1,7 @@ pub mod cli; pub mod generated_recipe; pub mod intermediate_backend; +pub mod log_message_layer; pub mod package_dependency; pub mod protocol; pub mod rattler_build_integration; diff --git a/crates/pixi_build_backend/src/log_message_layer.rs b/crates/pixi_build_backend/src/log_message_layer.rs new file mode 100644 index 0000000000..02e31eea78 --- /dev/null +++ b/crates/pixi_build_backend/src/log_message_layer.rs @@ -0,0 +1,295 @@ +//! A `tracing` layer that converts events and span lifecycles into +//! [`LogMessage`] records which the server forwards to the frontend as +//! `log/message` JSON-RPC notifications (see +//! [`crate::server::Server::with_log_messages`]). +//! +//! INFO-level events are skipped on purpose: by convention (inherited from +//! rattler-build) they form the plaintext build-output stream and are +//! rendered to stderr by `LoggingOutputHandler`, where the frontend captures +//! them for live display and failure replay. +//! +//! Span-open records are emitted *lazily*: when a span is created its record +//! is stashed in the span's extensions, and only when an event actually +//! flows through the channel are the unsent records of its ancestor chain +//! flushed first. This keeps spans that never produce a forwarded event off +//! the wire, and it transparently handles spans created before the frontend +//! has negotiated `supports_log_messages` (their records are still pending +//! and get flushed with the first event after activation). + +use std::sync::{ + Arc, + atomic::{AtomicBool, Ordering}, +}; + +use pixi_build_types::procedures::log_message::{ + LogEvent, LogLevel, LogMessage, LogSpanClose, LogSpanOpen, +}; +use serde_json::{Map, Value}; +use tokio::sync::mpsc; +use tracing::{ + Event, Level, Subscriber, + field::Visit, + span::{Attributes, Id}, +}; +use tracing_subscriber::{Layer, layer::Context, registry::LookupSpan}; + +/// State of a span's open record, stored in the span's extensions. +enum SpanOpenState { + /// The record has not been sent yet. + Pending(LogSpanOpen), + /// The record was sent; a close record must follow on span close. + Sent, +} + +pub struct LogMessageLayer { + sender: mpsc::UnboundedSender, + /// Flipped by the server once the frontend advertises + /// `supports_log_messages` during capability negotiation. + enabled: Arc, +} + +impl LogMessageLayer { + pub fn new(sender: mpsc::UnboundedSender, enabled: Arc) -> Self { + Self { sender, enabled } + } + + fn send(&self, record: LogMessage) { + // If the receiver is gone the server has shut down; there is nowhere + // useful to send the record to. + let _ = self.sender.send(record); + } +} + +impl Layer for LogMessageLayer +where + S: Subscriber + for<'a> LookupSpan<'a>, +{ + fn on_new_span(&self, attrs: &Attributes<'_>, id: &Id, ctx: Context<'_, S>) { + let mut visitor = FieldVisitor::for_span(); + attrs.record(&mut visitor); + let metadata = attrs.metadata(); + let parent_id = attrs + .parent() + .cloned() + .or_else(|| ctx.current_span().id().cloned()) + .map(|id| id.into_u64()); + + let record = LogSpanOpen { + id: id.into_u64(), + parent_id, + level: level_to_wire(metadata.level()), + target: metadata.target().to_string(), + name: metadata.name().to_string(), + fields: visitor.fields, + }; + if let Some(span) = ctx.span(id) { + span.extensions_mut().insert(SpanOpenState::Pending(record)); + } + } + + fn on_close(&self, id: Id, ctx: Context<'_, S>) { + let Some(span) = ctx.span(&id) else { + return; + }; + // Only spans whose open record went over the wire need a close. + if matches!(span.extensions().get(), Some(SpanOpenState::Sent)) { + self.send(LogMessage::SpanClose(LogSpanClose { id: id.into_u64() })); + } + } + + fn on_event(&self, event: &Event<'_>, ctx: Context<'_, S>) { + if !self.enabled.load(Ordering::Acquire) { + return; + } + let metadata = event.metadata(); + if *metadata.level() == Level::INFO { + return; + } + + // Flush unsent span-open records root-first so the frontend always + // sees a parent before anything that references it. + let span_id = ctx.event_span(event).map(|leaf| { + for span in leaf.scope().from_root() { + let mut extensions = span.extensions_mut(); + if let Some(state) = extensions.get_mut::() + && let SpanOpenState::Pending(record) = + std::mem::replace(state, SpanOpenState::Sent) + { + self.send(LogMessage::SpanOpen(record)); + } + } + leaf.id().into_u64() + }); + + let mut visitor = FieldVisitor::for_event(); + event.record(&mut visitor); + self.send(LogMessage::Event(LogEvent { + level: level_to_wire(metadata.level()), + target: metadata.target().to_string(), + message: visitor.message.unwrap_or_default(), + fields: visitor.fields, + span_id, + })); + } +} + +fn level_to_wire(level: &Level) -> LogLevel { + match *level { + Level::TRACE => LogLevel::Trace, + Level::DEBUG => LogLevel::Debug, + Level::INFO => LogLevel::Info, + Level::WARN => LogLevel::Warn, + Level::ERROR => LogLevel::Error, + } +} + +/// Collects the fields of an event or span into a JSON map, optionally +/// extracting the conventional `message` field. +struct FieldVisitor { + fields: Map, + message: Option, + extract_message: bool, +} + +impl FieldVisitor { + fn for_event() -> Self { + Self { + fields: Map::new(), + message: None, + extract_message: true, + } + } + + fn for_span() -> Self { + Self { + fields: Map::new(), + message: None, + extract_message: false, + } + } + + fn insert(&mut self, name: &str, value: Value) { + if self.extract_message && name == "message" { + self.message = Some(match value { + Value::String(text) => text, + other => other.to_string(), + }); + } else { + self.fields.insert(name.to_string(), value); + } + } +} + +impl Visit for FieldVisitor { + fn record_str(&mut self, field: &tracing::field::Field, value: &str) { + self.insert(field.name(), Value::String(value.to_owned())); + } + + fn record_bool(&mut self, field: &tracing::field::Field, value: bool) { + self.insert(field.name(), Value::Bool(value)); + } + + fn record_i64(&mut self, field: &tracing::field::Field, value: i64) { + self.insert(field.name(), Value::Number(value.into())); + } + + fn record_u64(&mut self, field: &tracing::field::Field, value: u64) { + self.insert(field.name(), Value::Number(value.into())); + } + + fn record_f64(&mut self, field: &tracing::field::Field, value: f64) { + match serde_json::Number::from_f64(value) { + Some(number) => self.insert(field.name(), Value::Number(number)), + None => self.insert(field.name(), Value::String(value.to_string())), + } + } + + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { + self.insert(field.name(), Value::String(format!("{:?}", value))); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; + + fn collect_records(enabled: bool, emit: impl FnOnce()) -> Vec { + let (tx, mut rx) = mpsc::unbounded_channel(); + let layer = LogMessageLayer::new(tx, Arc::new(AtomicBool::new(enabled))); + let _guard = tracing_subscriber::registry().with(layer).set_default(); + emit(); + let mut records = Vec::new(); + while let Ok(record) = rx.try_recv() { + records.push(record); + } + records + } + + #[test] + fn events_carry_their_lazily_opened_span_chain() { + let records = collect_records(true, || { + let outer = tracing::info_span!("outer"); + let _outer = outer.enter(); + let inner = tracing::debug_span!("inner", package = "foo"); + let _inner = inner.enter(); + tracing::warn!(count = 3, "something happened"); + }); + + let kinds: Vec<_> = records + .iter() + .map(|record| match record { + LogMessage::SpanOpen(open) => format!("open:{}", open.name), + LogMessage::SpanClose(_) => "close".to_string(), + LogMessage::Event(event) => format!("event:{}", event.message), + }) + .collect(); + assert_eq!( + kinds, + [ + "open:outer", + "open:inner", + "event:something happened", + "close", + "close" + ] + ); + + let LogMessage::SpanOpen(outer) = &records[0] else { + unreachable!() + }; + let LogMessage::SpanOpen(inner) = &records[1] else { + unreachable!() + }; + let LogMessage::Event(event) = &records[2] else { + unreachable!() + }; + assert_eq!(outer.parent_id, None); + assert_eq!(inner.parent_id, Some(outer.id)); + assert_eq!( + inner.fields.get("package").and_then(Value::as_str), + Some("foo") + ); + assert_eq!(event.span_id, Some(inner.id)); + assert_eq!(event.fields.get("count").and_then(Value::as_i64), Some(3)); + assert_eq!(event.level, LogLevel::Warn); + } + + #[test] + fn info_events_and_eventless_spans_stay_off_the_wire() { + let records = collect_records(true, || { + let span = tracing::info_span!("quiet"); + let _span = span.enter(); + tracing::info!("this is build output, not a log record"); + }); + assert!(records.is_empty()); + } + + #[test] + fn nothing_is_sent_before_activation() { + let records = collect_records(false, || { + tracing::error!("emitted before the frontend negotiated"); + }); + assert!(records.is_empty()); + } +} diff --git a/crates/pixi_build_backend/src/server.rs b/crates/pixi_build_backend/src/server.rs index d461cb589a..b470d8ff3e 100644 --- a/crates/pixi_build_backend/src/server.rs +++ b/crates/pixi_build_backend/src/server.rs @@ -1,4 +1,11 @@ -use std::{net::SocketAddr, path::Path, sync::Arc}; +use std::{ + net::SocketAddr, + path::Path, + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }, +}; use fs_err::tokio as tokio_fs; use jsonrpc_core::{Error, IoHandler, Params, serde_json, to_value}; @@ -10,11 +17,16 @@ use pixi_build_types::{ conda_build_v1::{CondaBuildV1Params, CondaBuildV1Result}, conda_outputs::{CondaOutputsParams, CondaOutputsResult}, initialize::InitializeParams, + log_message::LogMessage, negotiate_capabilities::NegotiateCapabilitiesParams, }, }; use serde::Serialize; -use tokio::sync::{Mutex, RwLock}; +use tokio::{ + io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, + sync::{Mutex, RwLock, mpsc}, + task::JoinSet, +}; use crate::consts::DEBUG_OUTPUT_DIR; use crate::protocol::{Protocol, ProtocolInstantiator}; @@ -22,6 +34,11 @@ use crate::protocol::{Protocol, ProtocolInstantiator}; /// A JSONRPC server that can be used to communicate with a client. pub struct Server { instatiator: T, + /// When set, `log/message` notifications received on this channel are + /// forwarded to the client over stdout, and the flag is flipped once the + /// frontend advertises `supports_log_messages` during capability + /// negotiation. + log_messages: Option<(mpsc::UnboundedReceiver, Arc)>, } enum ServerState { @@ -45,19 +62,117 @@ impl ServerState { impl Server { pub fn new(instatiator: T) -> Self { - Self { instatiator } + Self { + instatiator, + log_messages: None, + } + } + + /// Enables forwarding of structured log records to the client. + /// + /// Records received on `receiver` are sent to the client as + /// `log/message` JSON-RPC notifications, but only after the frontend has + /// advertised `supports_log_messages` during capability negotiation; + /// `enabled` is flipped at that point so the producing side (the tracing + /// layer) knows it can start emitting. Only supported when running over + /// stdio. + pub fn with_log_messages( + mut self, + receiver: mpsc::UnboundedReceiver, + enabled: Arc, + ) -> Self { + self.log_messages = Some((receiver, enabled)); + self } /// Run the server, communicating over stdin/stdout. + /// + /// Requests are read from stdin as line-delimited JSON and dispatched + /// concurrently; responses and `log/message` notifications are written + /// to stdout through a single writer task so their bytes never + /// interleave. pub async fn run(self) -> miette::Result<()> { - let io = self.setup_io(); - jsonrpc_stdio_server::ServerBuilder::new(io).build().await; + self.run_with_io(tokio::io::stdin(), tokio::io::stdout()) + .await + } + + /// The implementation of [`Self::run`], generic over the input/output + /// streams so tests can drive the server over in-process pipes. + async fn run_with_io( + mut self, + input: impl tokio::io::AsyncRead + Unpin, + mut output: impl tokio::io::AsyncWrite + Unpin + Send + 'static, + ) -> miette::Result<()> { + let (log_messages, log_enabled) = match self.log_messages.take() { + Some((receiver, enabled)) => (Some(receiver), Some(enabled)), + None => (None, None), + }; + let io = Arc::new(self.setup_io(log_enabled)); + + // All bytes destined for the output flow through this channel. + let (out_tx, mut out_rx) = mpsc::unbounded_channel::(); + let mut writer = tokio::spawn(async move { + while let Some(mut line) = out_rx.recv().await { + line.push('\n'); + if output.write_all(line.as_bytes()).await.is_err() { + break; + } + if output.flush().await.is_err() { + break; + } + } + }); + + // Forward structured log records as notifications. + let log_task = log_messages.map(|mut receiver| { + let out_tx = out_tx.clone(); + tokio::spawn(async move { + while let Some(record) = receiver.recv().await { + let notification = serde_json::json!({ + "jsonrpc": "2.0", + "method": procedures::log_message::METHOD_NAME, + "params": record, + }); + if out_tx.send(notification.to_string()).is_err() { + break; + } + } + }) + }); + + let mut requests = JoinSet::new(); + let mut stdin = BufReader::new(input).lines(); + while let Ok(Some(line)) = stdin.next_line().await { + if line.trim().is_empty() { + continue; + } + let io = io.clone(); + let out_tx = out_tx.clone(); + requests.spawn(async move { + if let Some(response) = io.handle_request(&line).await { + let _ = out_tx.send(response); + } + }); + } + + // Stdin closed: the client is gone. Finish in-flight requests, stop + // the log forwarder, and flush any queued output before returning. + while requests.join_next().await.is_some() {} + if let Some(task) = log_task { + task.abort(); + let _ = task.await; + } + drop(out_tx); + let _ = (&mut writer).await; Ok(()) } /// Run the server, communicating over HTTP. + /// + /// Notifications (and thus `log/message` forwarding) are not supported + /// in this mode. pub fn run_over_http(self, port: u16) -> miette::Result<()> { - let io = self.setup_io(); + let io = self.setup_io(None); jsonrpc_http_server::ServerBuilder::new(io) .start_http(&SocketAddr::from(([127, 0, 0, 1], port))) .into_diagnostic()? @@ -66,17 +181,27 @@ impl Server { } /// Setup the IO inner handler. - fn setup_io(self) -> IoHandler { + fn setup_io(self, log_enabled: Option>) -> IoHandler { // Construct a server let mut io = IoHandler::new(); io.add_method( procedures::negotiate_capabilities::METHOD_NAME, - move |params: Params| async move { - let params: NegotiateCapabilitiesParams = params.parse()?; - let result = T::negotiate_capabilities(params) - .await - .map_err(convert_error)?; - Ok(to_value(result).expect("failed to convert to json")) + move |params: Params| { + let log_enabled = log_enabled.clone(); + async move { + let params: NegotiateCapabilitiesParams = params.parse()?; + // Only start emitting `log/message` notifications once the + // frontend has told us it understands them. + if params.capabilities.supports_log_messages == Some(true) + && let Some(enabled) = log_enabled + { + enabled.store(true, Ordering::Release); + } + let result = T::negotiate_capabilities(params) + .await + .map_err(convert_error)?; + Ok(to_value(result).expect("failed to convert to json")) + } }, ); @@ -284,3 +409,94 @@ async fn write_json_file( .context("failed to write JSON to file")?; Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + use pixi_build_types::{ + BackendCapabilities, + procedures::{ + initialize::InitializeResult, + log_message::{LogEvent, LogLevel}, + negotiate_capabilities::NegotiateCapabilitiesResult, + }, + }; + use tokio::io::AsyncWriteExt; + + struct FakeInstantiator; + + #[async_trait::async_trait] + impl ProtocolInstantiator for FakeInstantiator { + async fn negotiate_capabilities( + _params: NegotiateCapabilitiesParams, + ) -> miette::Result { + Ok(NegotiateCapabilitiesResult { + capabilities: BackendCapabilities::default(), + }) + } + + async fn initialize( + &self, + _params: InitializeParams, + ) -> miette::Result<(Box, InitializeResult)> { + unimplemented!("not needed for this test") + } + } + + #[tokio::test] + async fn responses_and_log_notifications_share_the_output_channel() { + let (log_tx, log_rx) = mpsc::unbounded_channel(); + let enabled = Arc::new(AtomicBool::new(false)); + + let (server_read, mut client_write) = tokio::io::simplex(4096); + let (client_read, server_write) = tokio::io::simplex(4096); + + let enabled_check = enabled.clone(); + let server = tokio::spawn( + Server::new(FakeInstantiator) + .with_log_messages(log_rx, enabled.clone()) + .run_with_io(server_read, server_write), + ); + + // Records sent before negotiation must still be forwarded (the + // *layer* is what holds records back until activation; the server + // forwards whatever reaches it). + client_write + .write_all( + b"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"negotiateCapabilities\",\ + \"params\":{\"capabilities\":{\"supportsLogMessages\":true}}}\n", + ) + .await + .unwrap(); + + let mut lines = BufReader::new(client_read).lines(); + let response = lines.next_line().await.unwrap().unwrap(); + assert!(response.contains("\"id\":1")); + assert!( + enabled_check.load(Ordering::Acquire), + "negotiation must flip the activation flag" + ); + + log_tx + .send(LogMessage::Event(LogEvent { + level: LogLevel::Warn, + target: "t".to_string(), + message: "hello".to_string(), + fields: Default::default(), + span_id: None, + })) + .unwrap(); + let notification = lines.next_line().await.unwrap().unwrap(); + let parsed: serde_json::Value = serde_json::from_str(¬ification).unwrap(); + assert_eq!(parsed["method"], "log/message"); + assert_eq!(parsed["params"]["kind"], "event"); + assert_eq!(parsed["params"]["message"], "hello"); + assert!(parsed.get("id").is_none(), "notifications have no id"); + + // Closing the input shuts the server down cleanly. Note: merely + // dropping the write half would not close the underlying simplex + // stream; an explicit shutdown is required to produce an EOF. + client_write.shutdown().await.unwrap(); + server.await.unwrap().unwrap(); + } +} diff --git a/crates/pixi_build_frontend/Cargo.toml b/crates/pixi_build_frontend/Cargo.toml index 8874f1ef03..a490198f56 100644 --- a/crates/pixi_build_frontend/Cargo.toml +++ b/crates/pixi_build_frontend/Cargo.toml @@ -33,3 +33,4 @@ tokio = { workspace = true, features = [ "rt-multi-thread", ] } tokio-util = { workspace = true, features = ["io"] } +tracing-subscriber = { workspace = true, features = ["registry"] } diff --git a/crates/pixi_build_frontend/src/backend/json_rpc.rs b/crates/pixi_build_frontend/src/backend/json_rpc.rs index 2f354dc625..27783eba2c 100644 --- a/crates/pixi_build_frontend/src/backend/json_rpc.rs +++ b/crates/pixi_build_frontend/src/backend/json_rpc.rs @@ -7,7 +7,7 @@ use jsonrpsee::{ async_client::{Client, ClientBuilder}, core::{ ClientError, - client::{ClientT, Error, TransportReceiverT, TransportSenderT}, + client::{ClientT, Error, SubscriptionClientT, TransportReceiverT, TransportSenderT}, }, types::ErrorCode, }; @@ -20,6 +20,7 @@ use pixi_build_types::{ conda_build_v1::{CondaBuildV1Params, CondaBuildV1Result}, conda_outputs::{CondaOutputsParams, CondaOutputsResult}, initialize::{InitializeParams, InitializeResult}, + log_message::LogMessage, negotiate_capabilities::{NegotiateCapabilitiesParams, NegotiateCapabilitiesResult}, }, }; @@ -31,7 +32,7 @@ use tokio::{ sync::{Mutex, oneshot}, }; -use super::stderr::stream_stderr; +use super::{log_forwarder::LogForwarder, stderr::stream_stderr}; use crate::{ backend::BackendOutputStream, error::BackendError, @@ -95,6 +96,21 @@ pub enum InitializeError { Communication(#[from] Box), } +/// The name of the most verbose level enabled by the current `tracing` +/// subscriber, in the format understood by the backend's +/// [`pixi_build_types::procedures::log_message::LOG_LEVEL_ENV`]. +fn max_level_name(level: tracing::level_filters::LevelFilter) -> &'static str { + use tracing::level_filters::LevelFilter; + match level { + LevelFilter::OFF => "off", + LevelFilter::ERROR => "error", + LevelFilter::WARN => "warn", + LevelFilter::INFO => "info", + LevelFilter::DEBUG => "debug", + LevelFilter::TRACE => "trace", + } +} + impl CommunicationError { fn from_client_error( backend_identifier: String, @@ -134,6 +150,20 @@ pub struct JsonRpcBackend { manifest_path: PathBuf, /// The stderr of the backend process. stderr: Option>>>>, + /// The task that forwards `log/message` notifications from the backend + /// into this process's `tracing` subscriber. Held so it is aborted when + /// the backend is dropped. + _log_forwarder: Option, +} + +/// Aborts the wrapped task when dropped. +#[derive(Debug)] +struct AbortOnDrop(tokio::task::JoinHandle<()>); + +impl Drop for AbortOnDrop { + fn drop(&mut self) { + self.0.abort(); + } } #[allow(clippy::result_large_err)] @@ -165,6 +195,13 @@ impl JsonRpcBackend { .stdout(std::process::Stdio::piped()) .stdin(std::process::Stdio::piped()) .stderr(std::process::Stdio::piped()) + // Ask the backend to log at least as verbosely as we do; the + // records it sends back over the log channel are filtered again + // by our own subscriber. + .env( + procedures::log_message::LOG_LEVEL_ENV, + max_level_name(tracing::level_filters::LevelFilter::current()), + ) .spawn() { Ok(process) => process, @@ -233,14 +270,50 @@ impl JsonRpcBackend { let client: Client = ClientBuilder::default() // Set 24hours for request timeout because the backend may be long-running. .request_timeout(std::time::Duration::from_secs(86400)) + // Log notifications can burst faster than the forwarder drains + // them; jsonrpsee silently *unsubscribes* on overflow, so make + // the buffer generous. + .max_buffer_capacity_per_subscription(16 * 1024) .build_with_tokio(sender, receiver); + // Register interest in `log/message` notifications before the first + // request so no record can slip past unobserved. The backend only + // starts sending them after we advertise `supports_log_messages` + // below; the subscription simply stays idle for backends that never + // send any. + let log_forwarder = match client + .subscribe_to_method::(procedures::log_message::METHOD_NAME) + .await + { + Ok(mut subscription) => Some(AbortOnDrop(tokio::spawn(async move { + let mut forwarder = LogForwarder::new(); + while let Some(record) = subscription.next().await { + match record { + Ok(record) => forwarder.apply(record), + Err(err) => { + tracing::debug!( + "failed to parse a log/message notification from the build backend: {err}" + ); + } + } + } + }))), + Err(err) => { + tracing::debug!( + "failed to subscribe to log/message notifications from the build backend: {err}" + ); + None + } + }; + // Negotiate the capabilities with the backend. let negotiate_result: NegotiateCapabilitiesResult = client .request( procedures::negotiate_capabilities::METHOD_NAME, RpcParams::from(NegotiateCapabilitiesParams { - capabilities: FrontendCapabilities {}, + capabilities: FrontendCapabilities { + supports_log_messages: Some(log_forwarder.is_some()), + }, }), ) .await @@ -288,6 +361,7 @@ impl JsonRpcBackend { backend_capabilities: negotiate_result.capabilities, manifest_path, stderr: stderr.map(Mutex::new).map(Arc::new), + _log_forwarder: log_forwarder, }) } @@ -409,3 +483,241 @@ impl JsonRpcBackend { &self.backend_capabilities } } + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::{Arc as StdArc, Mutex as StdMutex}; + + use serde_json::json; + use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; + use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; + + /// Captures every event re-emitted into the `tracing` subscriber, + /// formatted as `LEVEL target [span:chain] message`. + #[derive(Clone, Default)] + struct Capture { + events: StdArc>>, + } + + impl tracing_subscriber::Layer for Capture + where + S: tracing::Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>, + { + fn on_event( + &self, + event: &tracing::Event<'_>, + ctx: tracing_subscriber::layer::Context<'_, S>, + ) { + // Only capture re-emitted backend events; the jsonrpsee client + // machinery under test emits its own events on this thread. + if !event.metadata().target().starts_with("backend::") { + return; + } + struct MessageVisitor(String); + impl tracing::field::Visit for MessageVisitor { + fn record_str(&mut self, field: &tracing::field::Field, value: &str) { + if field.name() == "message" { + self.0 = value.to_owned(); + } + } + + fn record_debug( + &mut self, + field: &tracing::field::Field, + value: &dyn std::fmt::Debug, + ) { + if field.name() == "message" { + self.0 = format!("{value:?}"); + } + } + } + let mut visitor = MessageVisitor(String::new()); + event.record(&mut visitor); + + let scope = ctx + .event_scope(event) + .into_iter() + .flat_map(|scope| scope.from_root()) + .map(|span| span.name().to_owned()) + .collect::>() + .join(":"); + self.events.lock().unwrap().push(format!( + "{} {} [{scope}] {}", + event.metadata().level(), + event.metadata().target(), + visitor.0 + )); + } + } + + type FakeBackendSender = crate::jsonrpc::Sender>; + type FakeBackendReceiver = + crate::jsonrpc::Receiver>; + + /// A fake build backend on the other end of an in-process transport. It + /// answers `negotiateCapabilities` and `initialize`, records the + /// frontend capabilities it received, and — when `log_records` is + /// non-empty — pushes each of them as a `log/message` notification + /// right after answering the negotiate request. + fn spawn_fake_backend( + log_records: Vec, + ) -> ( + FakeBackendSender, + FakeBackendReceiver, + StdArc>>, + ) { + let (frontend_end, backend_end) = tokio::io::duplex(64 * 1024); + let (frontend_read, frontend_write) = tokio::io::split(frontend_end); + let (backend_read, mut backend_write) = tokio::io::split(backend_end); + + let received_capabilities = StdArc::new(StdMutex::new(None)); + let received = received_capabilities.clone(); + tokio::spawn(async move { + let mut lines = BufReader::new(backend_read).lines(); + while let Ok(Some(line)) = lines.next_line().await { + let request: serde_json::Value = serde_json::from_str(&line).unwrap(); + let id = request["id"].clone(); + let mut responses = Vec::new(); + match request["method"].as_str().unwrap() { + "negotiateCapabilities" => { + *received.lock().unwrap() = Some(request["params"]["capabilities"].clone()); + responses.push(json!({ + "jsonrpc": "2.0", + "id": id, + "result": { + "capabilities": { + "providesCondaOutputs": true, + "providesCondaBuildV1": true, + } + } + })); + for record in &log_records { + responses.push(json!({ + "jsonrpc": "2.0", + "method": "log/message", + "params": record, + })); + } + } + "initialize" => { + responses.push(json!({"jsonrpc": "2.0", "id": id, "result": {}})); + } + other => panic!("unexpected method: {other}"), + } + for response in responses { + let mut line = response.to_string(); + line.push('\n'); + backend_write.write_all(line.as_bytes()).await.unwrap(); + } + } + }); + + ( + crate::jsonrpc::Sender::from(frontend_write), + crate::jsonrpc::Receiver::from(frontend_read), + received_capabilities, + ) + } + + async fn setup_backend_with_fake( + sender: FakeBackendSender, + receiver: FakeBackendReceiver, + ) -> JsonRpcBackend { + let dir = std::env::temp_dir(); + JsonRpcBackend::setup_with_transport( + "fake-backend".to_string(), + None, + dir.clone(), + dir.join("pixi.toml"), + dir.clone(), + None, + None, + None, + None, + None, + None, + sender, + receiver, + None, + ) + .await + .expect("setup should succeed") + } + + /// Wait until the capture contains `count` events. Notifications arrive + /// asynchronously with respect to the RPC responses. + async fn wait_for_events(capture: &Capture, count: usize) { + for _ in 0..500 { + if capture.events.lock().unwrap().len() >= count { + return; + } + tokio::time::sleep(std::time::Duration::from_millis(2)).await; + } + } + + #[tokio::test] + async fn log_messages_are_re_emitted_through_the_tracing_subscriber() { + let capture = Capture::default(); + let _guard = tracing_subscriber::registry() + .with(capture.clone()) + .set_default(); + + let (sender, receiver, received_capabilities) = spawn_fake_backend(vec![ + json!({ + "kind": "span_open", + "id": 1, + "level": "INFO", + "target": "rattler_build_core::build", + "name": "build", + }), + json!({ + "kind": "event", + "level": "WARN", + "target": "rattler_build_core::build", + "message": "watch out", + "span_id": 1, + }), + json!({"kind": "span_close", "id": 1}), + json!({ + "kind": "event", + "level": "ERROR", + "target": "pixi_build_cmake::config", + "message": "boom", + "fields": {"code": 3}, + }), + ]); + let _backend = setup_backend_with_fake(sender, receiver).await; + wait_for_events(&capture, 2).await; + + assert_eq!( + *capture.events.lock().unwrap(), + [ + "WARN backend::rattler_build_core::build [build] watch out", + "ERROR backend::pixi_build_cmake::config [] boom code=3", + ] + ); + assert_eq!( + received_capabilities + .lock() + .unwrap() + .as_ref() + .and_then(|capabilities| capabilities["supportsLogMessages"].as_bool()), + Some(true), + "the frontend must advertise supports_log_messages" + ); + } + + #[tokio::test] + async fn backends_that_never_send_log_messages_work_unchanged() { + let capture = Capture::default(); + let _guard = tracing_subscriber::registry() + .with(capture.clone()) + .set_default(); + + let (sender, receiver, _received) = spawn_fake_backend(Vec::new()); + let backend = setup_backend_with_fake(sender, receiver).await; + assert!(backend.capabilities().provides_conda_build_v1()); + assert!(capture.events.lock().unwrap().is_empty()); + } +} diff --git a/crates/pixi_build_frontend/src/backend/log_forwarder.rs b/crates/pixi_build_frontend/src/backend/log_forwarder.rs new file mode 100644 index 0000000000..0107369f51 --- /dev/null +++ b/crates/pixi_build_frontend/src/backend/log_forwarder.rs @@ -0,0 +1,386 @@ +//! Re-emits `log/message` notifications received from a build backend +//! through the frontend's own `tracing` subscriber. +//! +//! The backend mirrors its span hierarchy over the wire (see +//! [`pixi_build_types::procedures::log_message`]); this module reconstructs +//! it with real `tracing` spans so events render with their full +//! `outer:inner:` context and are subject to the frontend's own filtering. +//! Targets are prefixed with `backend::` so backend-origin records are +//! distinguishable from the frontend's, and can be filtered as a group +//! (e.g. `RUST_LOG=backend=trace`). +//! +//! `tracing`'s macros require metadata to be `&'static`, which is impossible +//! for names that only exist at runtime. The [`callsites`] module below +//! therefore interns one leaked callsite per unique `(name, target, level)` +//! tuple; the set is bounded by the number of distinct callsites in the +//! backend, which is small and stable in practice. + +use std::collections::HashMap; + +use pixi_build_types::procedures::log_message::{LogEvent, LogLevel, LogMessage, LogSpanOpen}; +use tracing::{Level, Span}; + +/// Prefix prepended to the backend's `tracing` targets on re-emission. +const TARGET_PREFIX: &str = "backend::"; + +/// Process-scoped state: maps the backend's span ids to the frontend spans +/// mirroring them. Lives for the lifetime of the backend process, so span +/// ids stay valid across RPC calls. +#[derive(Default)] +pub(crate) struct LogForwarder { + spans: HashMap, +} + +impl LogForwarder { + pub(crate) fn new() -> Self { + Self::default() + } + + pub(crate) fn apply(&mut self, record: LogMessage) { + match record { + LogMessage::SpanOpen(open) => self.open_span(open), + LogMessage::SpanClose(close) => { + self.spans.remove(&close.id); + } + LogMessage::Event(event) => { + let parent = event.span_id.and_then(|id| self.spans.get(&id)); + emit_event(&event, parent); + } + } + } + + fn open_span(&mut self, open: LogSpanOpen) { + let parent = open.parent_id.and_then(|id| self.spans.get(&id)); + let span = callsites::new_span( + &open.name, + &format!("{TARGET_PREFIX}{}", open.target), + level_from_wire(open.level), + parent.and_then(Span::id), + ); + // If the frontend's filter disabled this span, fall back to its + // parent so descendants keep the deepest *enabled* ancestor as + // context — mirroring how `tracing` treats disabled spans natively. + let span = if span.is_disabled() { + parent.cloned().unwrap_or(span) + } else { + span + }; + self.spans.insert(open.id, span); + } +} + +fn emit_event(event: &LogEvent, parent: Option<&Span>) { + let mut message = event.message.clone(); + for (key, value) in &event.fields { + use std::fmt::Write; + let _ = match value { + serde_json::Value::String(text) => write!(message, " {key}={text}"), + other => write!(message, " {key}={other}"), + }; + } + callsites::emit_event( + &format!("{TARGET_PREFIX}{}", event.target), + level_from_wire(event.level), + parent.and_then(Span::id), + &message, + ); +} + +fn level_from_wire(level: LogLevel) -> Level { + match level { + LogLevel::Trace => Level::TRACE, + LogLevel::Debug => Level::DEBUG, + LogLevel::Info => Level::INFO, + LogLevel::Warn => Level::WARN, + LogLevel::Error => Level::ERROR, + } +} + +/// Runtime-constructed `tracing` callsites. +/// +/// One callsite (and its `Metadata`) is leaked per unique +/// `(kind, name, target, level)` key. Keying on the level matters: interest +/// caching and `EnvFilter` evaluate a callsite's metadata, so reusing a +/// callsite created at a different level would make filtering decisions with +/// the wrong level. +mod callsites { + use std::{ + collections::HashMap, + sync::{Mutex, OnceLock}, + }; + + use tracing::{ + Event, Level, Metadata, + callsite::{Callsite, Identifier}, + field::FieldSet, + metadata::Kind, + span::{Id, Span}, + }; + + pub(super) struct DynCallsite { + name: &'static str, + target: &'static str, + level: Level, + is_span: bool, + fields: &'static [&'static str], + metadata: OnceLock>, + } + + impl DynCallsite { + fn metadata_ref(&'static self) -> &'static Metadata<'static> { + self.metadata.get_or_init(|| { + Metadata::new( + self.name, + self.target, + self.level, + None, + None, + None, + FieldSet::new(self.fields, Identifier(self)), + if self.is_span { + Kind::SPAN + } else { + Kind::EVENT + }, + ) + }) + } + } + + impl Callsite for DynCallsite { + fn set_interest(&self, _: tracing::subscriber::Interest) {} + fn metadata(&self) -> &Metadata<'_> { + self.metadata + .get() + .expect("metadata is initialised at interning time") + } + } + + #[derive(PartialEq, Eq, Hash)] + struct Key { + is_span: bool, + name: String, + target: String, + level: Level, + } + + fn intern(key: Key) -> &'static DynCallsite { + static INTERNED: OnceLock>> = OnceLock::new(); + let mut interned = INTERNED + .get_or_init(Default::default) + .lock() + .expect("callsite interner is poisoned"); + if let Some(&callsite) = interned.get(&key) { + return callsite; + } + let callsite: &'static DynCallsite = Box::leak(Box::new(DynCallsite { + name: Box::leak(key.name.clone().into_boxed_str()), + target: Box::leak(key.target.clone().into_boxed_str()), + level: key.level, + is_span: key.is_span, + fields: if key.is_span { &[] } else { &["message"] }, + metadata: OnceLock::new(), + })); + let _ = callsite.metadata_ref(); + tracing::callsite::register(callsite); + interned.insert(key, callsite); + callsite + } + + pub(super) fn span_callsite(name: &str, target: &str, level: Level) -> &'static DynCallsite { + intern(Key { + is_span: true, + name: name.to_owned(), + target: target.to_owned(), + level, + }) + } + + fn event_callsite(target: &str, level: Level) -> &'static DynCallsite { + intern(Key { + is_span: false, + name: "backend event".to_owned(), + target: target.to_owned(), + level, + }) + } + + /// Create a span mirroring a backend span. Returns a disabled span when + /// the frontend's subscriber is not interested in it. + pub(super) fn new_span(name: &str, target: &str, level: Level, parent: Option) -> Span { + let metadata = span_callsite(name, target, level).metadata_ref(); + let values = metadata.fields().value_set(&[]); + Span::child_of(parent, metadata, &values) + } + + /// Emit an event mirroring a backend event, if the frontend's subscriber + /// is interested in it. + pub(super) fn emit_event(target: &str, level: Level, parent: Option, message: &str) { + let metadata = event_callsite(target, level).metadata_ref(); + // Note: everything happens inside a single `get_default` closure — + // nesting (e.g. through `Event::child_of`, which calls `get_default` + // internally) would trip the dispatcher's re-entrancy protection and + // silently dispatch to the no-op subscriber. + tracing::dispatcher::get_default(move |dispatch| { + // The `tracing` macros perform this check before constructing an + // event; going through the dispatcher directly, we must do it + // ourselves or filtering would be bypassed. + if !dispatch.enabled(metadata) { + return; + } + let fields = metadata.fields(); + let message_field = fields + .field("message") + .expect("event callsites always have a message field"); + let values = [(&message_field, Some(&message as &dyn tracing::field::Value))]; + let values = fields.value_set(&values); + let event = Event::new_child_of(parent.clone(), metadata, &values); + dispatch.event(&event); + }); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::{Arc, Mutex}; + use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; + + /// Captures re-emitted events together with their full span context. + #[derive(Clone, Default)] + struct Capture { + events: Arc>>, + } + + impl tracing_subscriber::Layer for Capture + where + S: tracing::Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>, + { + fn on_event( + &self, + event: &tracing::Event<'_>, + ctx: tracing_subscriber::layer::Context<'_, S>, + ) { + struct MessageVisitor(String); + impl tracing::field::Visit for MessageVisitor { + fn record_str(&mut self, field: &tracing::field::Field, value: &str) { + if field.name() == "message" { + self.0 = value.to_owned(); + } + } + + fn record_debug( + &mut self, + field: &tracing::field::Field, + value: &dyn std::fmt::Debug, + ) { + if field.name() == "message" { + self.0 = format!("{value:?}"); + } + } + } + let mut visitor = MessageVisitor(String::new()); + event.record(&mut visitor); + + let scope = ctx + .event_scope(event) + .into_iter() + .flat_map(|scope| scope.from_root()) + .map(|span| span.name().to_owned()) + .collect::>() + .join(":"); + let level = *event.metadata().level(); + let target = event.metadata().target().to_owned(); + self.events + .lock() + .unwrap() + .push(format!("{level} {target} [{scope}] {}", visitor.0)); + } + } + + fn span_open(id: u64, parent_id: Option, name: &str) -> LogMessage { + LogMessage::SpanOpen(LogSpanOpen { + id, + parent_id, + level: LogLevel::Info, + target: "rattler_build_core::build".to_string(), + name: name.to_string(), + fields: Default::default(), + }) + } + + fn event(span_id: Option, level: LogLevel, message: &str) -> LogMessage { + LogMessage::Event(LogEvent { + level, + target: "rattler_build_core::build".to_string(), + message: message.to_string(), + fields: Default::default(), + span_id, + }) + } + + #[test] + fn events_re_emit_with_reconstructed_span_hierarchy() { + let capture = Capture::default(); + let _guard = tracing_subscriber::registry() + .with(capture.clone()) + .set_default(); + + let mut forwarder = LogForwarder::new(); + forwarder.apply(span_open(1, None, "outer")); + forwarder.apply(span_open(2, Some(1), "inner")); + forwarder.apply(event(Some(2), LogLevel::Warn, "nested")); + forwarder.apply(LogMessage::SpanClose( + pixi_build_types::procedures::log_message::LogSpanClose { id: 2 }, + )); + forwarder.apply(event(Some(1), LogLevel::Error, "after close")); + forwarder.apply(event(Some(999), LogLevel::Warn, "unknown span")); + forwarder.apply(event(None, LogLevel::Debug, "no span")); + + let events = capture.events.lock().unwrap(); + assert_eq!( + *events, + [ + "WARN backend::rattler_build_core::build [outer:inner] nested", + "ERROR backend::rattler_build_core::build [outer] after close", + "WARN backend::rattler_build_core::build [] unknown span", + "DEBUG backend::rattler_build_core::build [] no span", + ] + ); + } + + #[test] + fn fields_are_folded_into_the_message() { + let capture = Capture::default(); + let _guard = tracing_subscriber::registry() + .with(capture.clone()) + .set_default(); + + let mut fields = serde_json::Map::new(); + fields.insert("path".to_string(), serde_json::Value::from("foo/bar")); + fields.insert("count".to_string(), serde_json::Value::from(3)); + LogForwarder::new().apply(LogMessage::Event(LogEvent { + level: LogLevel::Warn, + target: "t".to_string(), + message: "base".to_string(), + fields, + span_id: None, + })); + + let events = capture.events.lock().unwrap(); + assert_eq!(*events, ["WARN backend::t [] base path=foo/bar count=3"]); + } + + #[test] + fn callsites_are_interned_by_name_target_and_level() { + let a1 = callsites::span_callsite("build", "t", Level::INFO); + let a2 = callsites::span_callsite("build", "t", Level::INFO); + let by_target = callsites::span_callsite("build", "other", Level::INFO); + let by_level = callsites::span_callsite("build", "t", Level::DEBUG); + let by_name = callsites::span_callsite("render", "t", Level::INFO); + assert!(std::ptr::eq(a1, a2)); + assert!(!std::ptr::eq(a1, by_target)); + assert!(!std::ptr::eq(a1, by_level)); + assert!(!std::ptr::eq(a1, by_name)); + } +} diff --git a/crates/pixi_build_frontend/src/backend/mod.rs b/crates/pixi_build_frontend/src/backend/mod.rs index 59e911cc11..d1311b5d08 100644 --- a/crates/pixi_build_frontend/src/backend/mod.rs +++ b/crates/pixi_build_frontend/src/backend/mod.rs @@ -13,6 +13,7 @@ use pixi_build_types::{ }, }; +mod log_forwarder; mod stderr; use crate::json_rpc::CommunicationError; diff --git a/crates/pixi_build_frontend/src/jsonrpc/mod.rs b/crates/pixi_build_frontend/src/jsonrpc/mod.rs index a10f045cfd..ffdb4d49c2 100644 --- a/crates/pixi_build_frontend/src/jsonrpc/mod.rs +++ b/crates/pixi_build_frontend/src/jsonrpc/mod.rs @@ -4,6 +4,8 @@ use serde_json::value::RawValue; mod stdio; pub(crate) use stdio::stdio_transport; +#[cfg(test)] +pub(crate) use stdio::{Receiver, Sender}; /// A helper struct to convert a serializable type into a JSON-RPC parameter. pub struct RpcParams(pub T); diff --git a/crates/pixi_build_types/src/capabilities.rs b/crates/pixi_build_types/src/capabilities.rs index 253dcd7ef4..a17eeddebb 100644 --- a/crates/pixi_build_types/src/capabilities.rs +++ b/crates/pixi_build_types/src/capabilities.rs @@ -39,6 +39,13 @@ impl BackendCapabilities { } } -#[derive(Debug, Serialize, Deserialize)] +#[derive(Default, Debug, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] /// Capabilities that the frontend provides. -pub struct FrontendCapabilities {} +pub struct FrontendCapabilities { + /// Whether the frontend understands `log/message` notifications sent by + /// the backend over the JSON-RPC channel. Backends must not send them + /// unless this is `Some(true)`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub supports_log_messages: Option, +} diff --git a/crates/pixi_build_types/src/lib.rs b/crates/pixi_build_types/src/lib.rs index 81cabd1980..3058647ca5 100644 --- a/crates/pixi_build_types/src/lib.rs +++ b/crates/pixi_build_types/src/lib.rs @@ -37,6 +37,9 @@ pub use variant::VariantValue; // structured objects instead of strings, add extra dependency groups, // and add `if()` target selectors that are passed through // to rattler-build. Older backends would silently mishandle them. +// Non-breaking addition: backends may send `log/message` notifications +// when the frontend advertises `supportsLogMessages` during capability +// negotiation (see `procedures::log_message`). /// The constraint for the pixi build api version package /// Adding this constraint when solving a pixi build backend environment ensures diff --git a/crates/pixi_build_types/src/procedures/log_message.rs b/crates/pixi_build_types/src/procedures/log_message.rs new file mode 100644 index 0000000000..202df8cb74 --- /dev/null +++ b/crates/pixi_build_types/src/procedures/log_message.rs @@ -0,0 +1,167 @@ +//! Structured log records emitted by a build backend as JSON-RPC +//! *notifications* on the same stdio channel that carries requests and +//! responses. +//! +//! The backend must only send these notifications if the frontend advertised +//! [`crate::capabilities::FrontendCapabilities::supports_log_messages`] +//! during capability negotiation. Frontends that do not understand the +//! method simply never advertise it, and backends that do not implement it +//! never send it, so the feature degrades cleanly in both directions +//! without a Pixi Build API version bump. +//! +//! Two kinds of information travel over this channel: +//! +//! * [`LogMessage::Event`]: a single `tracing` event (log line) emitted by +//! the backend. INFO-level events are deliberately *not* sent: by +//! convention (inherited from rattler-build) they form the plaintext +//! build-output stream and keep flowing over stderr where the frontend +//! captures them for progress display and failure replay. +//! * [`LogMessage::SpanOpen`] / [`LogMessage::SpanClose`]: the lifecycle of +//! the backend's `tracing` spans, so the frontend can mirror the span +//! hierarchy with real spans and events render with their full +//! `outer:inner:` context. +//! +//! Span ids are the backend's own `tracing` span ids. They are unique for +//! the lifetime of the backend process, which is safe here because the +//! notification channel lives exactly as long as the backend process. + +use serde::{Deserialize, Serialize}; +use serde_json::{Map, Value}; + +pub const METHOD_NAME: &str = "log/message"; + +/// Environment variable the frontend sets on the backend process to raise +/// the backend's log verbosity to its own. Holds a level name (`error`, +/// `warn`, `info`, `debug` or `trace`). The backend treats it as a *floor*: +/// the effective verbosity is the maximum of this value and the backend's +/// own command line flags. +pub const LOG_LEVEL_ENV: &str = "PIXI_BUILD_BACKEND_LOG_LEVEL"; + +/// A single record emitted by the backend on the log channel. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum LogMessage { + /// A `tracing` event. + Event(LogEvent), + /// A `tracing` span was opened. + SpanOpen(LogSpanOpen), + /// A previously opened span was closed. + SpanClose(LogSpanClose), +} + +/// A single `tracing` event emitted by the backend. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct LogEvent { + /// The level of the event. + pub level: LogLevel, + /// The `tracing` target of the event (usually the module path). + pub target: String, + /// The formatted message of the event. + pub message: String, + /// Additional structured fields recorded on the event. + #[serde(default, skip_serializing_if = "Map::is_empty")] + pub fields: Map, + /// Id of the span this event was emitted inside, if any. References a + /// [`LogSpanOpen::id`] previously seen on the channel. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub span_id: Option, +} + +/// A span that was opened in the backend. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct LogSpanOpen { + /// The backend-process-unique id of the span. + pub id: u64, + /// The id of the parent span, if any. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub parent_id: Option, + /// The level of the span. + pub level: LogLevel, + /// The `tracing` target of the span. + pub target: String, + /// The name of the span. + pub name: String, + /// The fields recorded on the span. + #[serde(default, skip_serializing_if = "Map::is_empty")] + pub fields: Map, +} + +/// A span that was closed in the backend. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct LogSpanClose { + /// The id of the span that was closed. + pub id: u64, +} + +/// The severity of a log record, mirroring `tracing::Level`. +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "UPPERCASE")] +pub enum LogLevel { + Trace, + Debug, + Info, + Warn, + Error, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn event_round_trips_through_json() { + let mut fields = Map::new(); + fields.insert("count".to_string(), Value::from(7)); + let record = LogMessage::Event(LogEvent { + level: LogLevel::Warn, + target: "pixi_build::recipe".to_string(), + message: "skipping unsupported variant".to_string(), + fields, + span_id: Some(42), + }); + let json = serde_json::to_string(&record).unwrap(); + assert!(json.contains("\"kind\":\"event\"")); + let LogMessage::Event(parsed) = serde_json::from_str(&json).unwrap() else { + panic!("expected Event variant"); + }; + assert_eq!(parsed.level, LogLevel::Warn); + assert_eq!(parsed.span_id, Some(42)); + assert_eq!(parsed.fields.get("count").and_then(Value::as_i64), Some(7)); + } + + #[test] + fn span_lifecycle_round_trips_through_json() { + let open = LogMessage::SpanOpen(LogSpanOpen { + id: 7, + parent_id: Some(3), + level: LogLevel::Info, + target: "pixi_build::build".to_string(), + name: "render".to_string(), + fields: Map::new(), + }); + let json = serde_json::to_string(&open).unwrap(); + assert!(json.contains("\"kind\":\"span_open\"")); + let LogMessage::SpanOpen(parsed) = serde_json::from_str(&json).unwrap() else { + panic!("expected SpanOpen variant"); + }; + assert_eq!(parsed.parent_id, Some(3)); + + let close = LogMessage::SpanClose(LogSpanClose { id: 7 }); + let json = serde_json::to_string(&close).unwrap(); + assert!(json.contains("\"kind\":\"span_close\"")); + } + + #[test] + fn empty_collections_are_omitted_from_the_wire_format() { + let record = LogMessage::Event(LogEvent { + level: LogLevel::Info, + target: "t".into(), + message: "m".into(), + fields: Map::new(), + span_id: None, + }); + let json = serde_json::to_string(&record).unwrap(); + assert!(!json.contains("fields")); + assert!(!json.contains("span_id")); + } +} diff --git a/crates/pixi_build_types/src/procedures/mod.rs b/crates/pixi_build_types/src/procedures/mod.rs index 1f307243aa..f0f2a09731 100644 --- a/crates/pixi_build_types/src/procedures/mod.rs +++ b/crates/pixi_build_types/src/procedures/mod.rs @@ -1,4 +1,5 @@ pub mod conda_build_v1; pub mod conda_outputs; pub mod initialize; +pub mod log_message; pub mod negotiate_capabilities; From 84da42c6d478c0271f82cfce3ebbbea92e29186a Mon Sep 17 00:00:00 2001 From: Julian Hofer Date: Mon, 6 Jul 2026 08:30:33 +0000 Subject: [PATCH 02/12] ci: temporary diagnostics for the linux pytest hang The Pytest linux job's 10-minute timeout is below pytest's 600s per-test timeout, so hanging tests die as an opaque job cancellation before pytest can report anything. Lower the per-test timeout for this job and raise the job timeout so a hang produces pytest's timeout diagnostics: the Python stack of the waiting test and the captured output of the stuck pixi invocation. To be reverted once the hang is understood. --- .github/workflows/ci.yml | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5c682bb09a..2ec73c6332 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -530,7 +530,10 @@ jobs: PIXI_BUILD_BACKEND_OVERRIDE: pixi-build-rust=${{ env.TARGET_RELEASE }}/pixi-build-rust test-pytest-linux-x86_64: - timeout-minutes: 10 + # TEMPORARY (revert before merge): raised from 10 so pytest's own + # per-test timeout (lowered below) fires before the job dies and we + # get stack dumps + captured pixi output for the hanging tests. + timeout-minutes: 20 name: Pytest | linux x86_64 runs-on: namespace-profile-ubuntu-24-04-x86-64-4 needs: build-binary-linux-x86_64 @@ -556,6 +559,11 @@ jobs: run: pixi run --locked test-integration-ci env: PIXI_BUILD_BACKEND_OVERRIDE: pixi-build-rust=${{ env.TARGET_RELEASE }}/pixi-build-rust + # TEMPORARY (revert before merge): per-test timeout below the job + # timeout so hangs produce pytest timeout dumps (thread stacks and + # the captured `pixi -v` output of the stuck test) instead of an + # opaque job cancellation. + PYTEST_ADDOPTS: "--color=yes --timeout=240" test-integration-windows-x86_64: timeout-minutes: 30 From b7a0b8b8e4f90bedd6833099b40de161cb0beaf5 Mon Sep 17 00:00:00 2001 From: Julian Hofer Date: Mon, 6 Jul 2026 09:25:48 +0000 Subject: [PATCH 03/12] fix: handle backend requests sequentially on the root task MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Dispatching each JSON-RPC request onto a spawned task changed the runtime geometry that build backends have always executed under, and intermittently livelocked rattler-build's script output pump: its select! loop keeps re-polling an already-EOF'd output stream (which completes immediately, forever), and with the request future running on a worker the runtime's I/O driver ended up unpolled, so the second output pipe's EOF was never delivered. The backend then spun at 100% CPU with the finished build script left as a zombie while the frontend waited indefinitely for the conda/build_v1 response — the CI hang on the Linux integration tests. Handle requests inline on the root task instead, exactly like the previous jsonrpc-stdio-server loop: read a line, await the response, queue it to the writer. The frontend serializes requests per backend, so concurrent dispatch bought nothing. Notifications are unaffected — the writer task remains the single point where responses and log/message notifications are serialized to stdout. Reproduced with 4 concurrent `pixi publish -v` loops on the rust minimal workspace (hang within <10 iterations, ~25% per iteration, confirmed with gdb/strace: Lines::poll_next_line hot loop, zero epoll activity, both pipe write-ends closed). Also reproduced with a released pixi 0.72 frontend against the new backend, proving the notification machinery (dormant in that pairing) was not the trigger. --- crates/pixi_build_backend/src/server.rs | 28 +++++++++++++------------ 1 file changed, 15 insertions(+), 13 deletions(-) diff --git a/crates/pixi_build_backend/src/server.rs b/crates/pixi_build_backend/src/server.rs index b470d8ff3e..1c467cc560 100644 --- a/crates/pixi_build_backend/src/server.rs +++ b/crates/pixi_build_backend/src/server.rs @@ -25,7 +25,6 @@ use serde::Serialize; use tokio::{ io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, sync::{Mutex, RwLock, mpsc}, - task::JoinSet, }; use crate::consts::DEBUG_OUTPUT_DIR; @@ -107,7 +106,7 @@ impl Server { Some((receiver, enabled)) => (Some(receiver), Some(enabled)), None => (None, None), }; - let io = Arc::new(self.setup_io(log_enabled)); + let io = self.setup_io(log_enabled); // All bytes destined for the output flow through this channel. let (out_tx, mut out_rx) = mpsc::unbounded_channel::(); @@ -140,24 +139,27 @@ impl Server { }) }); - let mut requests = JoinSet::new(); + // Requests are handled inline, one at a time, exactly like the + // previous `jsonrpc-stdio-server`-based loop. This is deliberate: + // the frontend serializes calls per backend anyway, and handling + // requests on the root task preserves the runtime geometry backends + // have always executed under. (Dispatching each request onto a + // spawned task was observed to livelock rattler-build's script + // output pump under load: its hot `select!` loop on a worker + // starved the runtime's I/O driver, so the second output pipe's + // EOF was never delivered.) let mut stdin = BufReader::new(input).lines(); while let Ok(Some(line)) = stdin.next_line().await { if line.trim().is_empty() { continue; } - let io = io.clone(); - let out_tx = out_tx.clone(); - requests.spawn(async move { - if let Some(response) = io.handle_request(&line).await { - let _ = out_tx.send(response); - } - }); + if let Some(response) = io.handle_request(&line).await { + let _ = out_tx.send(response); + } } - // Stdin closed: the client is gone. Finish in-flight requests, stop - // the log forwarder, and flush any queued output before returning. - while requests.join_next().await.is_some() {} + // Stdin closed: the client is gone. Stop the log forwarder and + // flush any queued output before returning. if let Some(task) = log_task { task.abort(); let _ = task.await; From b333fb57082735453f2fbd913b5d25d03e1b2699 Mon Sep 17 00:00:00 2001 From: Julian Hofer Date: Mon, 6 Jul 2026 09:25:58 +0000 Subject: [PATCH 04/12] Revert "ci: temporary diagnostics for the linux pytest hang" This reverts commit 84da42c6d478c0271f82cfce3ebbbea92e29186a. --- .github/workflows/ci.yml | 10 +--------- 1 file changed, 1 insertion(+), 9 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2ec73c6332..5c682bb09a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -530,10 +530,7 @@ jobs: PIXI_BUILD_BACKEND_OVERRIDE: pixi-build-rust=${{ env.TARGET_RELEASE }}/pixi-build-rust test-pytest-linux-x86_64: - # TEMPORARY (revert before merge): raised from 10 so pytest's own - # per-test timeout (lowered below) fires before the job dies and we - # get stack dumps + captured pixi output for the hanging tests. - timeout-minutes: 20 + timeout-minutes: 10 name: Pytest | linux x86_64 runs-on: namespace-profile-ubuntu-24-04-x86-64-4 needs: build-binary-linux-x86_64 @@ -559,11 +556,6 @@ jobs: run: pixi run --locked test-integration-ci env: PIXI_BUILD_BACKEND_OVERRIDE: pixi-build-rust=${{ env.TARGET_RELEASE }}/pixi-build-rust - # TEMPORARY (revert before merge): per-test timeout below the job - # timeout so hangs produce pytest timeout dumps (thread stacks and - # the captured `pixi -v` output of the stuck test) instead of an - # opaque job cancellation. - PYTEST_ADDOPTS: "--color=yes --timeout=240" test-integration-windows-x86_64: timeout-minutes: 30 From 40961a5cfaecf35f4574f25c435f6479585b0479 Mon Sep 17 00:00:00 2001 From: Julian Hofer Date: Mon, 6 Jul 2026 14:31:33 +0000 Subject: [PATCH 05/12] feat: surface backend tracing in pixi at warn+ with per-backend attribution Route backend log records under a backend:::: target, add a dedicated fmt layer that always renders that target so it is clear which backend produced a message, and add a backend={pixi_level} filter directive so backend logs start at warn and scale with -v like pixi's own logs. --- .../src/backend/json_rpc.rs | 14 +++- .../src/backend/log_forwarder.rs | 80 ++++++++++++------- crates/pixi_cli/src/lib.rs | 44 ++++++++-- 3 files changed, 96 insertions(+), 42 deletions(-) diff --git a/crates/pixi_build_frontend/src/backend/json_rpc.rs b/crates/pixi_build_frontend/src/backend/json_rpc.rs index 27783eba2c..ebf9b8af2d 100644 --- a/crates/pixi_build_frontend/src/backend/json_rpc.rs +++ b/crates/pixi_build_frontend/src/backend/json_rpc.rs @@ -281,12 +281,20 @@ impl JsonRpcBackend { // starts sending them after we advertise `supports_log_messages` // below; the subscription simply stays idle for backends that never // send any. + // Re-emitted records are attributed to this name (e.g. the + // `pixi-build-python` in `backend::pixi-build-python::…`); derive it + // from the executable's file stem so an overridden absolute path still + // reads cleanly. + let backend_log_name = std::path::Path::new(&backend_identifier) + .file_stem() + .map(|stem| stem.to_string_lossy().into_owned()) + .unwrap_or_else(|| backend_identifier.clone()); let log_forwarder = match client .subscribe_to_method::(procedures::log_message::METHOD_NAME) .await { Ok(mut subscription) => Some(AbortOnDrop(tokio::spawn(async move { - let mut forwarder = LogForwarder::new(); + let mut forwarder = LogForwarder::new(backend_log_name); while let Some(record) = subscription.next().await { match record { Ok(record) => forwarder.apply(record), @@ -693,8 +701,8 @@ mod tests { assert_eq!( *capture.events.lock().unwrap(), [ - "WARN backend::rattler_build_core::build [build] watch out", - "ERROR backend::pixi_build_cmake::config [] boom code=3", + "WARN backend::fake-backend::rattler_build_core::build [build] watch out", + "ERROR backend::fake-backend::pixi_build_cmake::config [] boom code=3", ] ); assert_eq!( diff --git a/crates/pixi_build_frontend/src/backend/log_forwarder.rs b/crates/pixi_build_frontend/src/backend/log_forwarder.rs index 0107369f51..931c521522 100644 --- a/crates/pixi_build_frontend/src/backend/log_forwarder.rs +++ b/crates/pixi_build_frontend/src/backend/log_forwarder.rs @@ -5,9 +5,11 @@ //! [`pixi_build_types::procedures::log_message`]); this module reconstructs //! it with real `tracing` spans so events render with their full //! `outer:inner:` context and are subject to the frontend's own filtering. -//! Targets are prefixed with `backend::` so backend-origin records are -//! distinguishable from the frontend's, and can be filtered as a group -//! (e.g. `RUST_LOG=backend=trace`). +//! Targets are prefixed with `backend::::` (e.g. +//! `backend::pixi-build-python::rattler_build_core::packaging`) so backend +//! records are distinguishable from the frontend's, name which backend +//! produced them, and can be filtered as a group (`RUST_LOG=backend=trace`) +//! or per backend (`RUST_LOG=backend::pixi-build-python=trace`). //! //! `tracing`'s macros require metadata to be `&'static`, which is impossible //! for names that only exist at runtime. The [`callsites`] module below @@ -26,14 +28,19 @@ const TARGET_PREFIX: &str = "backend::"; /// Process-scoped state: maps the backend's span ids to the frontend spans /// mirroring them. Lives for the lifetime of the backend process, so span /// ids stay valid across RPC calls. -#[derive(Default)] pub(crate) struct LogForwarder { + /// `backend::::` prepended to every re-emitted target so records are + /// attributed to the backend that produced them. + target_prefix: String, spans: HashMap, } impl LogForwarder { - pub(crate) fn new() -> Self { - Self::default() + pub(crate) fn new(backend_name: impl AsRef) -> Self { + Self { + target_prefix: format!("{TARGET_PREFIX}{}::", backend_name.as_ref()), + spans: HashMap::new(), + } } pub(crate) fn apply(&mut self, record: LogMessage) { @@ -43,8 +50,8 @@ impl LogForwarder { self.spans.remove(&close.id); } LogMessage::Event(event) => { - let parent = event.span_id.and_then(|id| self.spans.get(&id)); - emit_event(&event, parent); + let parent = event.span_id.and_then(|id| self.spans.get(&id).cloned()); + self.emit_event(&event, parent.as_ref()); } } } @@ -53,7 +60,7 @@ impl LogForwarder { let parent = open.parent_id.and_then(|id| self.spans.get(&id)); let span = callsites::new_span( &open.name, - &format!("{TARGET_PREFIX}{}", open.target), + &format!("{}{}", self.target_prefix, open.target), level_from_wire(open.level), parent.and_then(Span::id), ); @@ -67,23 +74,23 @@ impl LogForwarder { }; self.spans.insert(open.id, span); } -} -fn emit_event(event: &LogEvent, parent: Option<&Span>) { - let mut message = event.message.clone(); - for (key, value) in &event.fields { - use std::fmt::Write; - let _ = match value { - serde_json::Value::String(text) => write!(message, " {key}={text}"), - other => write!(message, " {key}={other}"), - }; + fn emit_event(&self, event: &LogEvent, parent: Option<&Span>) { + let mut message = event.message.clone(); + for (key, value) in &event.fields { + use std::fmt::Write; + let _ = match value { + serde_json::Value::String(text) => write!(message, " {key}={text}"), + other => write!(message, " {key}={other}"), + }; + } + callsites::emit_event( + &format!("{}{}", self.target_prefix, event.target), + level_from_wire(event.level), + parent.and_then(Span::id), + &message, + ); } - callsites::emit_event( - &format!("{TARGET_PREFIX}{}", event.target), - level_from_wire(event.level), - parent.and_then(Span::id), - &message, - ); } fn level_from_wire(level: LogLevel) -> Level { @@ -209,6 +216,14 @@ mod callsites { /// the frontend's subscriber is not interested in it. pub(super) fn new_span(name: &str, target: &str, level: Level, parent: Option) -> Span { let metadata = span_callsite(name, target, level).metadata_ref(); + // The `tracing` macros perform this check before constructing a + // span; going through `Span::child_of` directly, we must do it + // ourselves or filtering would be bypassed. This must be a separate + // `get_default` call: `Span::child_of` dispatches internally, and a + // nested `get_default` would silently hit the no-op subscriber. + if !tracing::dispatcher::get_default(|dispatch| dispatch.enabled(metadata)) { + return Span::none(); + } let values = metadata.fields().value_set(&[]); Span::child_of(parent, metadata, &values) } @@ -326,7 +341,7 @@ mod tests { .with(capture.clone()) .set_default(); - let mut forwarder = LogForwarder::new(); + let mut forwarder = LogForwarder::new("pixi-build-rattler-build"); forwarder.apply(span_open(1, None, "outer")); forwarder.apply(span_open(2, Some(1), "inner")); forwarder.apply(event(Some(2), LogLevel::Warn, "nested")); @@ -341,10 +356,10 @@ mod tests { assert_eq!( *events, [ - "WARN backend::rattler_build_core::build [outer:inner] nested", - "ERROR backend::rattler_build_core::build [outer] after close", - "WARN backend::rattler_build_core::build [] unknown span", - "DEBUG backend::rattler_build_core::build [] no span", + "WARN backend::pixi-build-rattler-build::rattler_build_core::build [outer:inner] nested", + "ERROR backend::pixi-build-rattler-build::rattler_build_core::build [outer] after close", + "WARN backend::pixi-build-rattler-build::rattler_build_core::build [] unknown span", + "DEBUG backend::pixi-build-rattler-build::rattler_build_core::build [] no span", ] ); } @@ -359,7 +374,7 @@ mod tests { let mut fields = serde_json::Map::new(); fields.insert("path".to_string(), serde_json::Value::from("foo/bar")); fields.insert("count".to_string(), serde_json::Value::from(3)); - LogForwarder::new().apply(LogMessage::Event(LogEvent { + LogForwarder::new("pixi-build-rattler-build").apply(LogMessage::Event(LogEvent { level: LogLevel::Warn, target: "t".to_string(), message: "base".to_string(), @@ -368,7 +383,10 @@ mod tests { })); let events = capture.events.lock().unwrap(); - assert_eq!(*events, ["WARN backend::t [] base path=foo/bar count=3"]); + assert_eq!( + *events, + ["WARN backend::pixi-build-rattler-build::t [] base path=foo/bar count=3"] + ); } #[test] diff --git a/crates/pixi_cli/src/lib.rs b/crates/pixi_cli/src/lib.rs index ce716f5e3d..11584d53ad 100644 --- a/crates/pixi_cli/src/lib.rs +++ b/crates/pixi_cli/src/lib.rs @@ -279,6 +279,19 @@ pub async fn execute() -> miette::Result<()> { execute_command(command, &global_options).await } +/// `EnvFilter` directives that scale pixi's own crates and records forwarded +/// from build backends (`backend::::…` targets) with the requested +/// verbosity, while keeping noisy dependencies one notch lower. +/// +/// The `backend` directive is load-bearing: backend records re-emit through +/// runtime-constructed callsites, and `EnvFilter` only enables those when an +/// explicit directive matches their target. +fn pixi_filter_directives(pixi_level: LevelFilter, low_level_filter: LevelFilter) -> String { + format!( + "apple_codesign=off,pixi={pixi_level},pixi_command_dispatcher={pixi_level},pixi_core={pixi_level},rattler_upload={pixi_level},uv_resolver={pixi_level},backend={pixi_level},resolvo={low_level_filter}" + ) +} + #[cfg(feature = "console-subscriber")] fn setup_logging(_args: &Args, _use_colors: bool) -> miette::Result<()> { console_subscriber::init(); @@ -289,7 +302,9 @@ fn setup_logging(_args: &Args, _use_colors: bool) -> miette::Result<()> { fn setup_logging(args: &Args, use_colors: bool) -> miette::Result<()> { use pixi_utils::indicatif::IndicatifWriter; use tracing_subscriber::{ - EnvFilter, filter::LevelFilter, prelude::__tracing_subscriber_SubscriberExt, + EnvFilter, Layer, + filter::{FilterFn, LevelFilter}, + prelude::__tracing_subscriber_SubscriberExt, util::SubscriberInitExt, }; @@ -312,17 +327,13 @@ fn setup_logging(args: &Args, use_colors: bool) -> miette::Result<()> { // CLI flags take precedence i.e. ignore RUST_LOG EnvFilter::builder() .with_default_directive(level_filter.into()) - .parse(format!( - "apple_codesign=off,pixi={pixi_level},pixi_command_dispatcher={pixi_level},pixi_core={pixi_level},rattler_upload={pixi_level},uv_resolver={pixi_level},resolvo={low_level_filter}" - )) + .parse(pixi_filter_directives(pixi_level, low_level_filter)) .into_diagnostic()? } else { // No CLI flags - use RUST_LOG if set // Parse RUST_LOG because we need to set it other our other directives let env_directives = env::var("RUST_LOG").unwrap_or_default(); - let original_directives = format!( - "apple_codesign=off,pixi={pixi_level},pixi_command_dispatcher={pixi_level},pixi_core={pixi_level},rattler_upload={pixi_level},uv_resolver={pixi_level},resolvo={low_level_filter}", - ); + let original_directives = pixi_filter_directives(pixi_level, low_level_filter); // Concatenate both directives where the LOG overrides the potential original directives let final_directives = if env_directives.is_empty() { original_directives @@ -337,15 +348,32 @@ fn setup_logging(args: &Args, use_colors: bool) -> miette::Result<()> { }; // Set up the tracing subscriber + // Records forwarded from a build backend carry a `backend::::…` + // target. They render through a dedicated layer that always shows the + // target, so the originating backend stays visible even at the default + // verbosity where the frontend's own logs hide their target. let fmt_layer = tracing_subscriber::fmt::layer() .with_ansi(use_colors) .with_target(pixi_level >= LevelFilter::INFO) .with_writer(IndicatifWriter::new(pixi_progress::global_multi_progress())) - .without_time(); + .without_time() + .with_filter(FilterFn::new(|metadata| { + !metadata.target().starts_with("backend::") + })); + + let backend_fmt_layer = tracing_subscriber::fmt::layer() + .with_ansi(use_colors) + .with_target(true) + .with_writer(IndicatifWriter::new(pixi_progress::global_multi_progress())) + .without_time() + .with_filter(FilterFn::new(|metadata| { + metadata.target().starts_with("backend::") + })); tracing_subscriber::registry() .with(env_filter) .with(fmt_layer) + .with(backend_fmt_layer) .init(); Ok(()) } From 4d3acd25b3798007b704ec928e8ad37dae86794a Mon Sep 17 00:00:00 2001 From: Julian Hofer Date: Mon, 6 Jul 2026 14:05:02 +0000 Subject: [PATCH 06/12] test: cover backend log filtering and attribution with fast unit tests --- crates/pixi_build_frontend/Cargo.toml | 2 +- .../src/backend/log_forwarder.rs | 126 ++++++++++++++++++ crates/pixi_cli/src/lib.rs | 32 +++++ 3 files changed, 159 insertions(+), 1 deletion(-) diff --git a/crates/pixi_build_frontend/Cargo.toml b/crates/pixi_build_frontend/Cargo.toml index a490198f56..4f259b3eaf 100644 --- a/crates/pixi_build_frontend/Cargo.toml +++ b/crates/pixi_build_frontend/Cargo.toml @@ -33,4 +33,4 @@ tokio = { workspace = true, features = [ "rt-multi-thread", ] } tokio-util = { workspace = true, features = ["io"] } -tracing-subscriber = { workspace = true, features = ["registry"] } +tracing-subscriber = { workspace = true, features = ["env-filter", "registry"] } diff --git a/crates/pixi_build_frontend/src/backend/log_forwarder.rs b/crates/pixi_build_frontend/src/backend/log_forwarder.rs index 931c521522..dd9c349439 100644 --- a/crates/pixi_build_frontend/src/backend/log_forwarder.rs +++ b/crates/pixi_build_frontend/src/backend/log_forwarder.rs @@ -389,6 +389,132 @@ mod tests { ); } + /// A subscriber mirroring the frontend's: an `EnvFilter` built from the + /// given directives (as pixi's CLI produces them for a given verbosity) + /// in front of the capture layer. + fn filtered_capture(directives: &str) -> (Capture, tracing::subscriber::DefaultGuard) { + let capture = Capture::default(); + let filter = tracing_subscriber::EnvFilter::builder() + .parse(directives) + .expect("directives parse"); + let guard = tracing_subscriber::registry() + .with(filter) + .with(capture.clone()) + .set_default(); + (capture, guard) + } + + fn apply_one_event_per_level(forwarder: &mut LogForwarder) { + forwarder.apply(event(None, LogLevel::Error, "error")); + forwarder.apply(event(None, LogLevel::Warn, "warn")); + forwarder.apply(event(None, LogLevel::Info, "info")); + forwarder.apply(event(None, LogLevel::Debug, "debug")); + forwarder.apply(event(None, LogLevel::Trace, "trace")); + } + + #[test] + fn default_verbosity_forwards_backend_warnings_and_errors() { + // Mirrors the `backend=warn` directive pixi installs when no `-v` + // flags are given. + let (capture, _guard) = filtered_capture("backend=warn"); + apply_one_event_per_level(&mut LogForwarder::new("pixi-build-python")); + assert_eq!( + *capture.events.lock().unwrap(), + [ + "ERROR backend::pixi-build-python::rattler_build_core::build [] error", + "WARN backend::pixi-build-python::rattler_build_core::build [] warn", + ] + ); + } + + #[test] + fn raised_verbosity_forwards_debug_and_trace() { + // Mirrors the `backend=trace` directive pixi installs at `-vvv`. + let (capture, _guard) = filtered_capture("backend=trace"); + apply_one_event_per_level(&mut LogForwarder::new("pixi-build-python")); + assert_eq!( + *capture.events.lock().unwrap(), + [ + "ERROR backend::pixi-build-python::rattler_build_core::build [] error", + "WARN backend::pixi-build-python::rattler_build_core::build [] warn", + "INFO backend::pixi-build-python::rattler_build_core::build [] info", + "DEBUG backend::pixi-build-python::rattler_build_core::build [] debug", + "TRACE backend::pixi-build-python::rattler_build_core::build [] trace", + ] + ); + } + + #[test] + fn user_directives_can_silence_backends() { + // `RUST_LOG` directives are appended after pixi's own; among equally + // specific directives the later one wins, so `RUST_LOG=backend=off` + // silences forwarded records entirely. + let (capture, _guard) = filtered_capture("backend=warn,backend=off"); + apply_one_event_per_level(&mut LogForwarder::new("pixi-build-python")); + assert_eq!(*capture.events.lock().unwrap(), [""; 0]); + } + + #[test] + fn backends_can_be_filtered_individually() { + // The attribution prefix makes per-backend directives possible, as + // promised by the module docs. + let (capture, _guard) = filtered_capture("backend=off,backend::pixi-build-python=trace"); + apply_one_event_per_level(&mut LogForwarder::new("pixi-build-python")); + apply_one_event_per_level(&mut LogForwarder::new("pixi-build-cmake")); + let events = capture.events.lock().unwrap(); + assert_eq!(events.len(), 5); + assert!( + events + .iter() + .all(|event| event.contains("backend::pixi-build-python::")) + ); + } + + #[test] + fn backend_records_need_an_explicit_backend_directive() { + // Regression guard for the directive in `pixi_cli`: `EnvFilter` + // ignores its default directive as soon as any other directive is + // present, so without an explicit `backend=` match the forwarded + // records vanish at every verbosity. + let capture = Capture::default(); + let filter = tracing_subscriber::EnvFilter::builder() + .with_default_directive(tracing::level_filters::LevelFilter::TRACE.into()) + .parse("pixi=trace") + .expect("directives parse"); + let _guard = tracing_subscriber::registry() + .with(filter) + .with(capture.clone()) + .set_default(); + apply_one_event_per_level(&mut LogForwarder::new("pixi-build-python")); + assert_eq!(*capture.events.lock().unwrap(), [""; 0]); + } + + #[test] + fn filtered_spans_degrade_to_the_deepest_enabled_ancestor() { + // The backend's spans are INFO: at default verbosity the filter + // rejects them, but events inside still render without span context. + { + let (capture, _guard) = filtered_capture("backend=warn"); + let mut forwarder = LogForwarder::new("pixi-build-python"); + forwarder.apply(span_open(1, None, "outer")); + forwarder.apply(event(Some(1), LogLevel::Warn, "inside")); + assert_eq!( + *capture.events.lock().unwrap(), + ["WARN backend::pixi-build-python::rattler_build_core::build [] inside"] + ); + } + + // Once the verbosity admits the spans, the context comes back. + let (capture, _guard) = filtered_capture("backend=info"); + let mut forwarder = LogForwarder::new("pixi-build-python"); + forwarder.apply(span_open(1, None, "outer")); + forwarder.apply(event(Some(1), LogLevel::Warn, "inside")); + assert_eq!( + *capture.events.lock().unwrap(), + ["WARN backend::pixi-build-python::rattler_build_core::build [outer] inside"] + ); + } + #[test] fn callsites_are_interned_by_name_target_and_level() { let a1 = callsites::span_callsite("build", "t", Level::INFO); diff --git a/crates/pixi_cli/src/lib.rs b/crates/pixi_cli/src/lib.rs index 11584d53ad..0b536c2f42 100644 --- a/crates/pixi_cli/src/lib.rs +++ b/crates/pixi_cli/src/lib.rs @@ -646,4 +646,36 @@ mod tests { }, ); } + + #[test] + fn verbosity_flags_map_to_log_levels() { + // `--list` satisfies `arg_required_else_help` without a subcommand. + let level = |argv: &[&str]| Args::parse_from(argv).log_level_filter(); + assert_eq!(level(&["pixi", "--list"]), LevelFilter::ERROR); + assert_eq!(level(&["pixi", "--list", "-v"]), LevelFilter::WARN); + assert_eq!(level(&["pixi", "--list", "-vv"]), LevelFilter::INFO); + assert_eq!(level(&["pixi", "--list", "-vvv"]), LevelFilter::DEBUG); + assert_eq!(level(&["pixi", "--list", "-vvvv"]), LevelFilter::TRACE); + assert_eq!(level(&["pixi", "--list", "-q"]), LevelFilter::OFF); + assert_eq!(level(&["pixi", "--list", "-q", "-vvv"]), LevelFilter::OFF); + } + + #[test] + fn filter_directives_scale_backend_records_with_verbosity() { + // Backend records only render when an explicit `backend=` directive + // matches them (see `log_forwarder`), so it must be present and track + // the pixi verbosity scale. + for pixi_level in [ + LevelFilter::WARN, + LevelFilter::INFO, + LevelFilter::DEBUG, + LevelFilter::TRACE, + ] { + let directives = pixi_filter_directives(pixi_level, LevelFilter::ERROR); + assert!( + directives.contains(&format!("backend={pixi_level}")), + "expected a backend directive at {pixi_level} in {directives}" + ); + } + } } From 73e03f51765bd891912fc3165f0f1023710baa4b Mon Sep 17 00:00:00 2001 From: Julian Hofer Date: Mon, 6 Jul 2026 15:35:34 +0000 Subject: [PATCH 07/12] fix: apply the requested default log level to unlisted crates --- crates/pixi_cli/src/lib.rs | 75 ++++++++++++++++++++++++++++++++++---- 1 file changed, 68 insertions(+), 7 deletions(-) diff --git a/crates/pixi_cli/src/lib.rs b/crates/pixi_cli/src/lib.rs index 0b536c2f42..86b00ffc17 100644 --- a/crates/pixi_cli/src/lib.rs +++ b/crates/pixi_cli/src/lib.rs @@ -286,9 +286,19 @@ pub async fn execute() -> miette::Result<()> { /// The `backend` directive is load-bearing: backend records re-emit through /// runtime-constructed callsites, and `EnvFilter` only enables those when an /// explicit directive matches their target. -fn pixi_filter_directives(pixi_level: LevelFilter, low_level_filter: LevelFilter) -> String { +/// +/// The bare `default_level` directive covers every crate not listed here. +/// It leads the string so that any bare level a user appends through +/// `RUST_LOG` overrides it (among equally specific directives the later one +/// wins). It cannot come from `EnvFilter`'s `with_default_directive`, which +/// is ignored as soon as any other directive is present. +fn pixi_filter_directives( + default_level: LevelFilter, + pixi_level: LevelFilter, + low_level_filter: LevelFilter, +) -> String { format!( - "apple_codesign=off,pixi={pixi_level},pixi_command_dispatcher={pixi_level},pixi_core={pixi_level},rattler_upload={pixi_level},uv_resolver={pixi_level},backend={pixi_level},resolvo={low_level_filter}" + "{default_level},apple_codesign=off,pixi={pixi_level},pixi_command_dispatcher={pixi_level},pixi_core={pixi_level},rattler_upload={pixi_level},uv_resolver={pixi_level},backend={pixi_level},resolvo={low_level_filter}" ) } @@ -326,14 +336,18 @@ fn setup_logging(args: &Args, use_colors: bool) -> miette::Result<()> { let env_filter = if cli_verbosity_set { // CLI flags take precedence i.e. ignore RUST_LOG EnvFilter::builder() - .with_default_directive(level_filter.into()) - .parse(pixi_filter_directives(pixi_level, low_level_filter)) + .parse(pixi_filter_directives( + level_filter, + pixi_level, + low_level_filter, + )) .into_diagnostic()? } else { // No CLI flags - use RUST_LOG if set // Parse RUST_LOG because we need to set it other our other directives let env_directives = env::var("RUST_LOG").unwrap_or_default(); - let original_directives = pixi_filter_directives(pixi_level, low_level_filter); + let original_directives = + pixi_filter_directives(level_filter, pixi_level, low_level_filter); // Concatenate both directives where the LOG overrides the potential original directives let final_directives = if env_directives.is_empty() { original_directives @@ -342,7 +356,6 @@ fn setup_logging(args: &Args, use_colors: bool) -> miette::Result<()> { }; EnvFilter::builder() - .with_default_directive(level_filter.into()) .parse(&final_directives) .into_diagnostic()? }; @@ -671,11 +684,59 @@ mod tests { LevelFilter::DEBUG, LevelFilter::TRACE, ] { - let directives = pixi_filter_directives(pixi_level, LevelFilter::ERROR); + let directives = + pixi_filter_directives(LevelFilter::ERROR, pixi_level, LevelFilter::ERROR); assert!( directives.contains(&format!("backend={pixi_level}")), "expected a backend directive at {pixi_level} in {directives}" ); } } + + /// A scoped subscriber filtering through the given directives, for + /// probing what `tracing::enabled!` sees. + fn with_filter(directives: &str, probe: impl FnOnce()) { + use tracing_subscriber::prelude::*; + let filter = tracing_subscriber::EnvFilter::builder() + .parse(directives) + .expect("directives parse"); + let subscriber = tracing_subscriber::registry().with(filter); + tracing::subscriber::with_default(subscriber, probe); + } + + #[test] + fn default_level_applies_to_unlisted_crates() { + let directives = + pixi_filter_directives(LevelFilter::ERROR, LevelFilter::WARN, LevelFilter::ERROR); + with_filter(&directives, || { + // Crates without an explicit directive surface their errors... + assert!(tracing::enabled!( + target: "rattler_networking", + tracing::Level::ERROR + )); + // ...but stay quiet below the default level... + assert!(!tracing::enabled!( + target: "rattler_networking", + tracing::Level::WARN + )); + // ...while pixi's own crates follow the more verbose pixi level. + assert!(tracing::enabled!(target: "pixi", tracing::Level::WARN)); + }); + } + + #[test] + fn user_rust_log_overrides_the_default_level() { + // A bare level in `RUST_LOG` lands after pixi's own directives and + // replaces the bare default for unlisted crates. + let directives = format!( + "{},debug", + pixi_filter_directives(LevelFilter::ERROR, LevelFilter::WARN, LevelFilter::ERROR) + ); + with_filter(&directives, || { + assert!(tracing::enabled!( + target: "rattler_networking", + tracing::Level::DEBUG + )); + }); + } } From 985a984e718ff5818cef20864c92bfc67ae8129c Mon Sep 17 00:00:00 2001 From: Julian Hofer Date: Mon, 6 Jul 2026 15:39:09 +0000 Subject: [PATCH 08/12] fix: keep explicit-root backend spans unparented on the wire --- .../src/log_message_layer.rs | 29 ++++++++++++++++++- 1 file changed, 28 insertions(+), 1 deletion(-) diff --git a/crates/pixi_build_backend/src/log_message_layer.rs b/crates/pixi_build_backend/src/log_message_layer.rs index 02e31eea78..5795815c52 100644 --- a/crates/pixi_build_backend/src/log_message_layer.rs +++ b/crates/pixi_build_backend/src/log_message_layer.rs @@ -68,10 +68,20 @@ where let mut visitor = FieldVisitor::for_span(); attrs.record(&mut visitor); let metadata = attrs.metadata(); + // An explicitly unparented span (`parent: None`) must not fall back + // to the contextual span: the registry treats it as a root, so its + // contextual "parent" would never be flushed and the wire record + // would dangle. let parent_id = attrs .parent() .cloned() - .or_else(|| ctx.current_span().id().cloned()) + .or_else(|| { + if attrs.is_contextual() { + ctx.current_span().id().cloned() + } else { + None + } + }) .map(|id| id.into_u64()); let record = LogSpanOpen { @@ -275,6 +285,23 @@ mod tests { assert_eq!(event.level, LogLevel::Warn); } + #[test] + fn explicit_root_spans_stay_unparented() { + let records = collect_records(true, || { + let outer = tracing::info_span!("outer"); + let _outer = outer.enter(); + let detached = tracing::info_span!(parent: None, "detached"); + let _detached = detached.enter(); + tracing::warn!("inside a detached span"); + }); + + let LogMessage::SpanOpen(open) = &records[0] else { + panic!("expected a span-open record, got {records:?}"); + }; + assert_eq!(open.name, "detached"); + assert_eq!(open.parent_id, None); + } + #[test] fn info_events_and_eventless_spans_stay_off_the_wire() { let records = collect_records(true, || { From dbb47b30760ffd4edf35ab19ebf4dabbff1dbae8 Mon Sep 17 00:00:00 2001 From: Julian Hofer Date: Mon, 6 Jul 2026 16:28:31 +0000 Subject: [PATCH 09/12] fix: appease rustdoc dead-code check and the typos linter `pixi_filter_directives` is only called from the non-console-subscriber `setup_logging`, so building with --all-features (as the rustdoc CI job does) flagged it as dead code; gate it on the same feature. The typos linter prefers 'parentless' over 'unparented'. --- crates/pixi_build_backend/src/log_message_layer.rs | 4 ++-- crates/pixi_cli/src/lib.rs | 1 + 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/crates/pixi_build_backend/src/log_message_layer.rs b/crates/pixi_build_backend/src/log_message_layer.rs index 5795815c52..1387218721 100644 --- a/crates/pixi_build_backend/src/log_message_layer.rs +++ b/crates/pixi_build_backend/src/log_message_layer.rs @@ -68,7 +68,7 @@ where let mut visitor = FieldVisitor::for_span(); attrs.record(&mut visitor); let metadata = attrs.metadata(); - // An explicitly unparented span (`parent: None`) must not fall back + // An explicitly parentless span (`parent: None`) must not fall back // to the contextual span: the registry treats it as a root, so its // contextual "parent" would never be flushed and the wire record // would dangle. @@ -286,7 +286,7 @@ mod tests { } #[test] - fn explicit_root_spans_stay_unparented() { + fn explicit_root_spans_stay_parentless() { let records = collect_records(true, || { let outer = tracing::info_span!("outer"); let _outer = outer.enter(); diff --git a/crates/pixi_cli/src/lib.rs b/crates/pixi_cli/src/lib.rs index 86b00ffc17..1c13dc9208 100644 --- a/crates/pixi_cli/src/lib.rs +++ b/crates/pixi_cli/src/lib.rs @@ -292,6 +292,7 @@ pub async fn execute() -> miette::Result<()> { /// `RUST_LOG` overrides it (among equally specific directives the later one /// wins). It cannot come from `EnvFilter`'s `with_default_directive`, which /// is ignored as soon as any other directive is present. +#[cfg(not(feature = "console-subscriber"))] fn pixi_filter_directives( default_level: LevelFilter, pixi_level: LevelFilter, From 70afbf77ab3979e9d2b77c5cb886740902ebef6f Mon Sep 17 00:00:00 2001 From: Julian Hofer Date: Tue, 7 Jul 2026 14:05:28 +0000 Subject: [PATCH 10/12] fix: keep stderr suppression responsive after log forwarding activates --- crates/pixi_build_backend/src/cli.rs | 21 +++--- .../src/log_message_layer.rs | 75 ++++++++++++++++++- 2 files changed, 85 insertions(+), 11 deletions(-) diff --git a/crates/pixi_build_backend/src/cli.rs b/crates/pixi_build_backend/src/cli.rs index e65fa11dcb..deadd82986 100644 --- a/crates/pixi_build_backend/src/cli.rs +++ b/crates/pixi_build_backend/src/cli.rs @@ -10,10 +10,13 @@ use pixi_build_types::{ }; use rattler_build_core::console_utils::{LoggingOutputHandler, get_default_env_filter}; use tokio::sync::mpsc; -use tracing::Level; -use tracing_subscriber::{Layer, filter::filter_fn, layer::SubscriberExt, util::SubscriberInitExt}; +use tracing_subscriber::{Layer, layer::SubscriberExt, util::SubscriberInitExt}; -use crate::{log_message_layer::LogMessageLayer, protocol::ProtocolInstantiator, server::Server}; +use crate::{ + log_message_layer::{LogMessageLayer, stderr_filter}, + protocol::ProtocolInstantiator, + server::Server, +}; #[allow(missing_docs)] #[derive(Parser)] @@ -98,14 +101,12 @@ pub(crate) async fn main_impl(enabled: Arc) -> impl Filter { + dynamic_filter_fn(move |metadata, _ctx| { + metadata.is_span() || *metadata.level() == Level::INFO || !enabled.load(Ordering::Acquire) + }) +} /// State of a span's open record, stored in the span's extensions. enum SpanOpenState { @@ -319,4 +340,56 @@ mod tests { }); assert!(records.is_empty()); } + + /// Counts the events that make it through [`stderr_filter`] to the + /// stderr-side layer. + #[derive(Clone, Default)] + struct CountEvents { + count: Arc, + } + + impl Layer for CountEvents { + fn on_event(&self, _event: &Event<'_>, _ctx: Context<'_, S>) { + self.count.fetch_add(1, Ordering::SeqCst); + } + } + + #[test] + fn stderr_suppression_applies_to_callsites_seen_before_activation() { + // A single static callsite that fires both before and after the + // frontend negotiates `supports_log_messages`. `tracing` caches + // per-callsite interest, so the filter must not let a decision made + // before activation stick around afterwards. + fn warm_callsite() { + tracing::warn!("recurring warning"); + } + + let enabled = Arc::new(AtomicBool::new(false)); + let stderr_side = CountEvents::default(); + let _guard = tracing_subscriber::registry() + .with( + stderr_side + .clone() + .with_filter(stderr_filter(enabled.clone())), + ) + .set_default(); + + warm_callsite(); + assert_eq!(stderr_side.count.load(Ordering::SeqCst), 1); + + enabled.store(true, Ordering::Release); + + // The same callsite must now be suppressed on stderr; its records + // travel over the structured channel instead. + warm_callsite(); + assert_eq!( + stderr_side.count.load(Ordering::SeqCst), + 1, + "a WARN callsite seen before activation must stop rendering to stderr afterwards" + ); + + // INFO remains the plaintext build-output stream in both phases. + tracing::info!("build output"); + assert_eq!(stderr_side.count.load(Ordering::SeqCst), 2); + } } From 9767755dfc6dddb2de6c21b6ff55309cb26807f9 Mon Sep 17 00:00:00 2001 From: Julian Hofer Date: Tue, 7 Jul 2026 14:08:19 +0000 Subject: [PATCH 11/12] fix: log when the backend log subscription ends unexpectedly --- crates/pixi_build_frontend/src/backend/json_rpc.rs | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/crates/pixi_build_frontend/src/backend/json_rpc.rs b/crates/pixi_build_frontend/src/backend/json_rpc.rs index ebf9b8af2d..5d50ec344e 100644 --- a/crates/pixi_build_frontend/src/backend/json_rpc.rs +++ b/crates/pixi_build_frontend/src/backend/json_rpc.rs @@ -294,7 +294,7 @@ impl JsonRpcBackend { .await { Ok(mut subscription) => Some(AbortOnDrop(tokio::spawn(async move { - let mut forwarder = LogForwarder::new(backend_log_name); + let mut forwarder = LogForwarder::new(&backend_log_name); while let Some(record) = subscription.next().await { match record { Ok(record) => forwarder.apply(record), @@ -305,6 +305,14 @@ impl JsonRpcBackend { } } } + // The subscription ends when the backend goes away — but + // also when jsonrpsee drops it because the notification + // buffer overflowed. Leave a trace so silently missing + // backend logs are diagnosable. + tracing::debug!( + "the log/message subscription to build backend '{backend_log_name}' ended; \ + no further backend log records will be forwarded" + ); }))), Err(err) => { tracing::debug!( From 3af990569ffd8fc662f97612094b2e748e27c75c Mon Sep 17 00:00:00 2001 From: Julian Hofer Date: Tue, 7 Jul 2026 14:09:39 +0000 Subject: [PATCH 12/12] docs: describe backend log forwarding and filtering --- docs/build/backends.md | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/docs/build/backends.md b/docs/build/backends.md index ebd84c941a..f2b73a9daa 100644 --- a/docs/build/backends.md +++ b/docs/build/backends.md @@ -130,3 +130,26 @@ We store: - Project model: `project_model.json` - Requests: `*_params.json` - Responses: `*_response.json` + +### Backend Logs + +Build backends forward their log records to Pixi, which renders them with a `backend::::` target so you can always tell which backend a message came from: + +``` +WARN backend::pixi-build-rattler-build::pixi_build_rattler_build::protocol: `debug-dir` backend configuration is deprecated and ignored +``` + +Backend warnings and errors are shown at Pixi's default verbosity. +Raising Pixi's verbosity also raises the backend's log level and forwards the additional records: `-vv` surfaces backend debug messages and `-vvv` surfaces trace messages. +The backend's regular build output is not affected by this; it keeps arriving as plain build output. + +When no verbosity flags are given, forwarded backend records can also be filtered through `RUST_LOG`, just like Pixi's own logs: + +- `RUST_LOG=backend=trace` enables all records from all backends. +- `RUST_LOG=backend::pixi-build-python=trace` enables all records from one specific backend. +- `RUST_LOG=backend=off` silences forwarded backend records entirely. + +Pixi passes its log level to the backend process through the `PIXI_BUILD_BACKEND_LOG_LEVEL` environment variable. +The backend uses whichever is more verbose: this value or its own command line flags. + +Log forwarding requires both a Pixi and a backend version that support it; with older versions on either side the backend logs only appear in the plain build output, as before.