chore: always enable stderr printing on verbose mode in the backends (#4814)

remimimimimi · web-flow · commit d1e724f8e373 · 2025-10-28T16:20:39.000+02:00
diff --git a/crates/pixi_build_backend_passthrough/src/lib.rs b/crates/pixi_build_backend_passthrough/src/lib.rs
@@ -66,6 +66,7 @@ impl InMemoryBackend for PassthroughBackend {
     fn conda_outputs(
         &self,
         params: CondaOutputsParams,
+        _output_stream: &(dyn BackendOutputStream + Send + 'static),
     ) -> Result<CondaOutputsResult, Box<CommunicationError>> {
         Ok(CondaOutputsResult {
             outputs: vec![CondaOutput {
diff --git a/crates/pixi_build_frontend/src/backend/in_memory/mod.rs b/crates/pixi_build_frontend/src/backend/in_memory/mod.rs
@@ -55,6 +55,7 @@ pub trait InMemoryBackend: Send {
     fn conda_outputs(
         &self,
         params: CondaOutputsParams,
+        output_stream: &(dyn BackendOutputStream + Send + 'static),
     ) -> Result<CondaOutputsResult, Box<CommunicationError>> {
         unimplemented!()
     }
diff --git a/crates/pixi_build_frontend/src/backend/json_rpc.rs b/crates/pixi_build_frontend/src/backend/json_rpc.rs
@@ -32,7 +32,7 @@ use tokio::{
     sync::{Mutex, oneshot},
 };
 
-use super::stderr::{stderr_buffer, stream_stderr};
+use super::stderr::stream_stderr;
 use crate::{
     backend::BackendOutputStream,
     error::BackendError,
@@ -287,7 +287,7 @@ impl JsonRpcBackend {
         request: CondaBuildV1Params,
         output_stream: W,
     ) -> Result<CondaBuildV1Result, CommunicationError> {
-        // Capture all of stderr and discard it
+        // Capture all of stderr and stream it
         let stderr = self.stderr.as_ref().map(|stderr| {
             // Cancellation signal
             let (cancel_tx, cancel_rx) = oneshot::channel();
@@ -334,16 +334,17 @@ impl JsonRpcBackend {
     }
 
     /// Call the `conda/outputs` method on the backend.
-    pub async fn conda_outputs(
+    pub async fn conda_outputs<W: BackendOutputStream + Send + 'static>(
         &self,
         request: CondaOutputsParams,
+        output_stream: W,
     ) -> Result<CondaOutputsResult, CommunicationError> {
-        // Capture all of stderr and discard it
+        // Capture all of stderr and stream it
         let stderr = self.stderr.as_ref().map(|stderr| {
             // Cancellation signal
             let (cancel_tx, cancel_rx) = oneshot::channel();
             // Spawn the stderr forwarding task
-            let handle = tokio::spawn(stderr_buffer(stderr.clone(), cancel_rx));
+            let handle = tokio::spawn(stream_stderr(stderr.clone(), cancel_rx, output_stream));
             (cancel_tx, handle)
         });
 
diff --git a/crates/pixi_build_frontend/src/backend/mod.rs b/crates/pixi_build_frontend/src/backend/mod.rs
@@ -152,19 +152,22 @@ impl Backend {
     }
 
     /// Returns the outputs that this backend can produce.
-    pub async fn conda_outputs(
+    pub async fn conda_outputs<W: BackendOutputStream + Send + 'static>(
         &self,
         params: CondaOutputsParams,
+        output_stream: W,
     ) -> Result<CondaOutputsResult, CommunicationError> {
         assert!(
             self.inner.capabilities().provides_conda_outputs(),
             "This backend does not support the conda outputs procedure"
         );
         match &self.inner {
-            BackendImplementation::JsonRpc(json_rpc) => json_rpc.conda_outputs(params).await,
-            BackendImplementation::InMemory(in_memory) => {
-                in_memory.conda_outputs(params).map_err(|e| *e)
+            BackendImplementation::JsonRpc(json_rpc) => {
+                json_rpc.conda_outputs(params, output_stream).await
             }
+            BackendImplementation::InMemory(in_memory) => in_memory
+                .conda_outputs(params, &output_stream)
+                .map_err(|e| *e),
         }
     }
 }
diff --git a/crates/pixi_build_frontend/src/backend/stderr.rs b/crates/pixi_build_frontend/src/backend/stderr.rs
@@ -7,35 +7,6 @@ use tokio::{
     sync::{Mutex, oneshot},
 };
 
-/// Stderr stream that captures the stderr output of the backend and stores it
-/// in a buffer for later use.
-pub(crate) async fn stderr_buffer(
-    buffer: Arc<Mutex<Lines<BufReader<ChildStderr>>>>,
-    cancel: oneshot::Receiver<()>,
-) -> Result<String, std::io::Error> {
-    // Create a future that continuously read from the buffer and stores the lines
-    // until all data is received.
-    let mut lines = Vec::new();
-    let read_and_buffer = async {
-        let mut buffer = buffer.lock().await;
-        while let Some(line) = buffer.next_line().await? {
-            lines.push(line);
-        }
-        Ok(lines.join("\n"))
-    };
-
-    // Either wait until the cancel signal is received or the `read_and_buffer`
-    // finishes which means there is no more data to read.
-    tokio::select! {
-        _ = cancel => {
-            Ok(lines.join("\n"))
-        }
-        result = read_and_buffer => {
-            result
-        }
-    }
-}
-
 /// Stderr stream that captures the stderr output of the backend and stores it
 /// in a buffer for later use.
 pub(crate) async fn stream_stderr<W: BackendOutputStream>(
diff --git a/crates/pixi_command_dispatcher/src/build_backend_metadata/mod.rs b/crates/pixi_command_dispatcher/src/build_backend_metadata/mod.rs
@@ -1,3 +1,11 @@
+use std::{
+    collections::{BTreeMap, BTreeSet, HashSet},
+    hash::{Hash, Hasher},
+    path::PathBuf,
+    sync::Mutex,
+};
+
+use futures::{SinkExt, channel::mpsc::UnboundedSender};
 use miette::Diagnostic;
 use once_cell::sync::Lazy;
 use pathdiff::diff_paths;
@@ -9,12 +17,6 @@ use pixi_record::{InputHash, PinnedSourceSpec};
 use pixi_spec::{SourceAnchor, SourceSpec};
 use rand::random;
 use rattler_conda_types::{ChannelConfig, ChannelUrl};
-use std::{
-    collections::{BTreeMap, BTreeSet, HashSet},
-    hash::{Hash, Hasher},
-    path::PathBuf,
-    sync::Mutex,
-};
 use thiserror::Error;
 use tracing::instrument;
 use xxhash_rust::xxh3::Xxh3;
@@ -100,6 +102,7 @@ impl BuildBackendMetadataSpec {
     pub(crate) async fn request(
         self,
         command_dispatcher: CommandDispatcher,
+        log_sink: UnboundedSender<String>,
     ) -> Result<BuildBackendMetadata, CommandDispatcherError<BuildBackendMetadataError>> {
         // Ensure that the source is checked out before proceeding.
         let source_checkout = command_dispatcher
@@ -195,6 +198,7 @@ impl BuildBackendMetadataSpec {
                 source_checkout,
                 backend,
                 additional_glob_hash,
+                log_sink,
             )
             .await?;
 
@@ -318,6 +322,7 @@ impl BuildBackendMetadataSpec {
         source_checkout: SourceCheckout,
         backend: Backend,
         additional_glob_hash: Vec<u8>,
+        mut log_sink: UnboundedSender<String>,
     ) -> Result<CachedCondaMetadata, CommandDispatcherError<BuildBackendMetadataError>> {
         let backend_identifier = backend.identifier().to_string();
         let params = CondaOutputsParams {
@@ -338,7 +343,9 @@ impl BuildBackendMetadataSpec {
             ),
         };
         let outputs = backend
-            .conda_outputs(params)
+            .conda_outputs(params, move |line| {
+                let _err = futures::executor::block_on(log_sink.send(line));
+            })
             .await
             .map_err(BuildBackendMetadataError::Communication)
             .map_err(CommandDispatcherError::Failed)?;
diff --git a/crates/pixi_command_dispatcher/src/command_dispatcher_processor/build_backend_metadata.rs b/crates/pixi_command_dispatcher/src/command_dispatcher_processor/build_backend_metadata.rs
@@ -1,6 +1,6 @@
 use std::{collections::hash_map::Entry, sync::Arc};
 
-use futures::FutureExt;
+use futures::{FutureExt, channel::mpsc::UnboundedSender};
 use tokio_util::sync::CancellationToken;
 
 use super::{CommandDispatcherProcessor, PendingDeduplicatingTask, TaskResult};
@@ -68,19 +68,23 @@ impl CommandDispatcherProcessor {
                         .insert(source_metadata_id, reporter_id);
                 }
 
+                // Open a channel to receive build output.
+                let (log_sink, rx) = futures::channel::mpsc::unbounded();
+
                 if let Some((reporter, reporter_id)) = self
                     .reporter
                     .as_deref_mut()
                     .and_then(Reporter::as_build_backend_metadata_reporter)
                     .zip(reporter_id)
                 {
-                    reporter.on_started(reporter_id)
+                    reporter.on_started(reporter_id, Box::new(rx))
                 }
 
                 self.queue_build_backend_metadata_task(
                     source_metadata_id,
                     task.spec,
                     task.cancellation_token,
+                    log_sink,
                 );
             }
         }
@@ -92,13 +96,16 @@ impl CommandDispatcherProcessor {
         build_backend_metadata_id: BuildBackendMetadataId,
         spec: BuildBackendMetadataSpec,
         cancellation_token: CancellationToken,
+        log_sink: UnboundedSender<String>,
     ) {
         let dispatcher = self.create_task_command_dispatcher(
             CommandDispatcherContext::BuildBackendMetadata(build_backend_metadata_id),
         );
+
+        // Open a channel to receive build output.
         self.pending_futures.push(
             cancellation_token
-                .run_until_cancelled_owned(spec.request(dispatcher))
+                .run_until_cancelled_owned(spec.request(dispatcher, log_sink))
                 .map(move |result| {
                     TaskResult::BuildBackendMetadata(
                         build_backend_metadata_id,
@@ -133,7 +140,8 @@ impl CommandDispatcherProcessor {
             .and_then(Reporter::as_build_backend_metadata_reporter)
             .zip(self.build_backend_metadata_reporters.remove(&id))
         {
-            reporter.on_finished(reporter_id);
+            let failed = result.is_err();
+            reporter.on_finished(reporter_id, failed);
         }
 
         self.build_backend_metadata
diff --git a/crates/pixi_command_dispatcher/src/command_dispatcher_processor/source_build.rs b/crates/pixi_command_dispatcher/src/command_dispatcher_processor/source_build.rs
@@ -1,4 +1,4 @@
-use std::collections::hash_map::Entry;
+use std::{collections::hash_map::Entry, sync::Arc};
 
 use futures::FutureExt;
 use tokio_util::sync::CancellationToken;
@@ -8,6 +8,7 @@ use crate::{
     CommandDispatcherError, Reporter, SourceBuildError, SourceBuildResult, SourceBuildSpec,
     command_dispatcher::{CommandDispatcherContext, SourceBuildId, SourceBuildTask},
 };
+use rattler_repodata_gateway::RunExportsReporter;
 
 impl CommandDispatcherProcessor {
     /// Constructs a new [`SourceBuildId`] for the given `task`.
@@ -61,15 +62,6 @@ impl CommandDispatcherProcessor {
                         .insert(source_build_id, reporter_id);
                 }
 
-                if let Some((reporter, reporter_id)) = self
-                    .reporter
-                    .as_deref_mut()
-                    .and_then(Reporter::as_source_build_reporter)
-                    .zip(reporter_id)
-                {
-                    reporter.on_started(reporter_id)
-                }
-
                 self.queue_source_build_task(source_build_id, task.spec, task.cancellation_token);
             }
         }
@@ -82,19 +74,27 @@ impl CommandDispatcherProcessor {
         spec: SourceBuildSpec,
         cancellation_token: CancellationToken,
     ) {
-        let dispatcher = self
-            .create_task_command_dispatcher(CommandDispatcherContext::SourceBuild(source_build_id));
-
         let dispatcher_context = CommandDispatcherContext::SourceBuild(source_build_id);
+        let dispatcher = self.create_task_command_dispatcher(dispatcher_context);
+
         let reporter_context = self.reporter_context(dispatcher_context);
-        let run_exports_reporter = self
-            .reporter
-            .as_mut()
-            .and_then(|reporter| reporter.create_run_exports_reporter(reporter_context));
+        let (tx, rx) = futures::channel::mpsc::unbounded::<String>();
+
+        let mut run_exports_reporter: Option<Arc<dyn RunExportsReporter>> = None;
+        if let Some(reporter) = self.reporter.as_mut() {
+            let created = reporter.create_run_exports_reporter(reporter_context);
+            if let Some((source_reporter, reporter_id)) = reporter
+                .as_source_build_reporter()
+                .zip(self.source_build_reporters.get(&source_build_id).copied())
+            {
+                source_reporter.on_started(reporter_id, Box::new(rx));
+            }
+            run_exports_reporter = created;
+        }
 
         self.pending_futures.push(
             cancellation_token
-                .run_until_cancelled_owned(spec.build(dispatcher, run_exports_reporter))
+                .run_until_cancelled_owned(spec.build(dispatcher, run_exports_reporter.clone(), tx))
                 .map(move |result| {
                     TaskResult::SourceBuild(
                         source_build_id,
@@ -122,7 +122,8 @@ impl CommandDispatcherProcessor {
             .and_then(Reporter::as_source_build_reporter)
             .zip(self.source_build_reporters.remove(&id))
         {
-            reporter.on_finished(reporter_id);
+            let failed = result.as_ref().is_err();
+            reporter.on_finished(reporter_id, failed);
         }
 
         self.source_build
diff --git a/crates/pixi_command_dispatcher/src/reporter.rs b/crates/pixi_command_dispatcher/src/reporter.rs
@@ -143,10 +143,14 @@ pub trait BuildBackendMetadataReporter {
     ) -> BuildBackendMetadataId;
 
     /// Called when the operation has started.
-    fn on_started(&mut self, id: BuildBackendMetadataId);
+    fn on_started(
+        &mut self,
+        id: BuildBackendMetadataId,
+        backend_output_stream: Box<dyn Stream<Item = String> + Unpin + Send>,
+    );
 
     /// Called when the operation has finished.
-    fn on_finished(&mut self, id: BuildBackendMetadataId);
+    fn on_finished(&mut self, id: BuildBackendMetadataId, failed: bool);
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize)]
@@ -183,10 +187,14 @@ pub trait SourceBuildReporter {
     ) -> SourceBuildId;
 
     /// Called when the operation has started.
-    fn on_started(&mut self, id: SourceBuildId);
+    fn on_started(
+        &mut self,
+        id: SourceBuildId,
+        backend_output_stream: Box<dyn Stream<Item = String> + Unpin + Send>,
+    );
 
     /// Called when the operation has finished.
-    fn on_finished(&mut self, id: SourceBuildId);
+    fn on_finished(&mut self, id: SourceBuildId, failed: bool);
 }
 
 /// A trait that is used to report the progress of a source build performed by
diff --git a/crates/pixi_command_dispatcher/src/source_build/mod.rs b/crates/pixi_command_dispatcher/src/source_build/mod.rs
diff --git a/crates/pixi_command_dispatcher/tests/integration/event_reporter.rs b/crates/pixi_command_dispatcher/tests/integration/event_reporter.rs
diff --git a/crates/pixi_reporters/src/sync_reporter.rs b/crates/pixi_reporters/src/sync_reporter.rs

Original file line number	Diff line number	Diff line change
`@@ -55,6 +55,7 @@ pub trait InMemoryBackend: Send {`
`55`	`55`	`fn conda_outputs(`
`56`	`56`	`&self,`
`57`	`57`	`params: CondaOutputsParams,`
	`58`	`+ output_stream: &(dyn BackendOutputStream + Send + 'static),`
`58`	`59`	`) -> Result<CondaOutputsResult, Box<CommunicationError>> {`
`59`	`60`	`unimplemented!()`
`60`	`61`	`}`
Original file line number	Diff line number	Diff line change
`@@ -152,19 +152,22 @@ impl Backend {`
`152`	`152`	`}`
`153`	`153`
`154`	`154`	`/// Returns the outputs that this backend can produce.`
`155`		`- pub async fn conda_outputs(`
	`155`	`+ pub async fn conda_outputs<W: BackendOutputStream + Send + 'static>(`
`156`	`156`	`&self,`
`157`	`157`	`params: CondaOutputsParams,`
	`158`	`+ output_stream: W,`
`158`	`159`	`) -> Result<CondaOutputsResult, CommunicationError> {`
`159`	`160`	`assert!(`
`160`	`161`	`self.inner.capabilities().provides_conda_outputs(),`
`161`	`162`	`"This backend does not support the conda outputs procedure"`
`162`	`163`	`);`
`163`	`164`	`match &self.inner {`
`164`		`- BackendImplementation::JsonRpc(json_rpc) => json_rpc.conda_outputs(params).await,`
`165`		`- BackendImplementation::InMemory(in_memory) => {`
`166`		`- in_memory.conda_outputs(params).map_err(\|e\| *e)`
	`165`	`+ BackendImplementation::JsonRpc(json_rpc) => {`
	`166`	`+ json_rpc.conda_outputs(params, output_stream).await`
`167`	`167`	`}`
	`168`	`+ BackendImplementation::InMemory(in_memory) => in_memory`
	`169`	`+ .conda_outputs(params, &output_stream)`
	`170`	`+ .map_err(\|e\| *e),`
`168`	`171`	`}`
`169`	`172`	`}`
`170`	`173`	`}`