From 9850f5e996e7689f510b21dc1bd04ed7bbc8c7c6 Mon Sep 17 00:00:00 2001 From: Raymond Khalife Date: Wed, 19 Aug 2026 19:50:58 +0000 Subject: [PATCH 1/8] refactor: introduce icp-events and port the build pipeline to it Operations now communicate with the presentation layer through typed events instead of driving progress bars directly. New icp-events crate carries the task/step/output vocabulary and the Reporter handles; the new icp-cli render module owns all wording, indicatif bars, and failure replay (absorbing MultiStepProgressBar::dump_output). The Build trait and the low-level script/wasm helpers in the core crate take a StepReporter in place of Option>; the sync path bridges into its legacy line channel until it is ported. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01QUZHfx2ng97WM2j9XoA3rr --- Cargo.lock | 10 + Cargo.toml | 1 + crates/icp-cli/Cargo.toml | 1 + crates/icp-cli/src/commands/build.rs | 18 +- crates/icp-cli/src/commands/deploy.rs | 18 +- crates/icp-cli/src/commands/project/bundle.rs | 18 +- crates/icp-cli/src/main.rs | 1 + crates/icp-cli/src/operations/build.rs | 90 ++---- crates/icp-cli/src/operations/bundle.rs | 12 +- crates/icp-cli/src/progress.rs | 19 +- crates/icp-cli/src/render/interactive.rs | 136 +++++++++ crates/icp-cli/src/render/mod.rs | 202 +++++++++++++ crates/icp-cli/src/render/plain.rs | 70 +++++ crates/icp-cli/tests/build_tests.rs | 12 +- crates/icp-events/Cargo.toml | 10 + crates/icp-events/src/lib.rs | 274 ++++++++++++++++++ crates/icp/Cargo.toml | 1 + crates/icp/src/canister/build/mod.rs | 12 +- crates/icp/src/canister/build/prebuilt.rs | 12 +- crates/icp/src/canister/build/script.rs | 18 +- crates/icp/src/canister/script.rs | 26 +- crates/icp/src/canister/sync/mod.rs | 11 + crates/icp/src/canister/sync/plugin.rs | 2 +- crates/icp/src/canister/sync/script.rs | 3 +- crates/icp/src/canister/wasm.rs | 24 +- 25 files changed, 853 insertions(+), 148 deletions(-) create mode 100644 crates/icp-cli/src/render/interactive.rs create mode 100644 crates/icp-cli/src/render/mod.rs create mode 100644 crates/icp-cli/src/render/plain.rs create mode 100644 crates/icp-events/Cargo.toml create mode 100644 crates/icp-events/src/lib.rs diff --git a/Cargo.lock b/Cargo.lock index b4b4d8099..7393ddbf9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3636,6 +3636,7 @@ dependencies = [ "ic-management-canister-types 0.8.0", "ic-utils", "icp-canister-interfaces", + "icp-events", "icp-sync-plugin", "icrc-ledger-types", "indexmap", @@ -3726,6 +3727,7 @@ dependencies = [ "ic-utils", "icp", "icp-canister-interfaces", + "icp-events", "icrc-ledger-types", "indicatif", "indoc", @@ -3769,6 +3771,14 @@ dependencies = [ "wslpath2", ] +[[package]] +name = "icp-events" +version = "1.3.0" +dependencies = [ + "serde", + "tokio", +] + [[package]] name = "icp-sync-plugin" version = "1.3.0" diff --git a/Cargo.toml b/Cargo.toml index ced350649..27b98e4f5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -60,6 +60,7 @@ ic-management-canister-types = { version = "0.8.0" } ic-utils = { version = "0.49.1" } icp = { path = "crates/icp" } icp-canister-interfaces = { path = "crates/icp-canister-interfaces" } +icp-events = { path = "crates/icp-events" } icp-sync-plugin = { path = "crates/icp-sync-plugin" } ic-identity-hsm = "0.49.1" icrc-ledger-types = "0.1.10" diff --git a/crates/icp-cli/Cargo.toml b/crates/icp-cli/Cargo.toml index 5ad3e0f0b..1cfa5180f 100644 --- a/crates/icp-cli/Cargo.toml +++ b/crates/icp-cli/Cargo.toml @@ -44,6 +44,7 @@ ic-management-canister-types.workspace = true ic-utils.workspace = true icp-canister-interfaces.workspace = true icp = { workspace = true, features = ["clap"] } +icp-events.workspace = true icrc-ledger-types.workspace = true indicatif.workspace = true indoc.workspace = true diff --git a/crates/icp-cli/src/commands/build.rs b/crates/icp-cli/src/commands/build.rs index d462f75f2..bbeeab784 100644 --- a/crates/icp-cli/src/commands/build.rs +++ b/crates/icp-cli/src/commands/build.rs @@ -6,8 +6,9 @@ use icp::context::{Context, EnvironmentSelection}; use tracing::info; use crate::{ - operations::build::build_many_with_progress_bar, + operations::build::build_many, options::{EnvironmentOpt, arg_struct_change_help}, + render::Renderer, }; /// Build canisters @@ -57,15 +58,24 @@ pub(crate) async fn exec(ctx: &Context, args: &BuildArgs) -> Result<(), anyhow:: // Build the selected canisters info!("Building canisters:"); - build_many_with_progress_bar( + let (reporter, events) = icp_events::channel(); + let render = tokio::spawn(Renderer::for_ctx(ctx.debug).run(events)); + + let result = build_many( canisters_to_build, environment_selection.name(), ctx.builder.clone(), ctx.artifacts.clone(), &ctx.dirs.package_cache()?, - ctx.debug, + &reporter, ) - .await?; + .await; + + // Close the event stream so the renderer can finish, and let it flush + // (failure dumps) before the result is acted on. + drop(reporter); + render.await?; + result?; info!("Canisters built successfully"); diff --git a/crates/icp-cli/src/commands/deploy.rs b/crates/icp-cli/src/commands/deploy.rs index b2a1fe4ca..587fe57e4 100644 --- a/crates/icp-cli/src/commands/deploy.rs +++ b/crates/icp-cli/src/commands/deploy.rs @@ -23,7 +23,7 @@ use crate::{ commands::{args::ArgsOpt, canister::create}, operations::{ binding_env_vars::set_binding_env_vars_many, - build::build_many_with_progress_bar, + build::build_many, candid_compat::check_candid_compatibility_many, create::{CreateFunding, CreateOperation, CreateTarget}, install::{install_many, resolve_install_mode_and_status}, @@ -33,6 +33,7 @@ use crate::{ }, options::{IdentityOpt, arg_struct_change_help}, progress::{ProgressManager, ProgressManagerSettings}, + render::Renderer, }; /// Deploy a project to an environment @@ -182,15 +183,24 @@ pub(crate) async fn exec(ctx: &Context, args: &DeployArgs) -> Result<(), anyhow: // Build the selected canisters info!("Building canisters:"); - build_many_with_progress_bar( + let (reporter, events) = icp_events::channel(); + let render = tokio::spawn(Renderer::for_ctx(ctx.debug).run(events)); + + let build_result = build_many( canisters_to_build, environment_selection.name(), ctx.builder.clone(), ctx.artifacts.clone(), &ctx.dirs.package_cache()?, - ctx.debug, + &reporter, ) - .await?; + .await; + + // Close the event stream so the renderer can finish, and let it flush + // (failure dumps) before the result is acted on. + drop(reporter); + render.await?; + build_result?; // Ensure the selected canisters exist, creating any that are missing. let env = ctx diff --git a/crates/icp-cli/src/commands/project/bundle.rs b/crates/icp-cli/src/commands/project/bundle.rs index b9217c0b8..f1fef3d21 100644 --- a/crates/icp-cli/src/commands/project/bundle.rs +++ b/crates/icp-cli/src/commands/project/bundle.rs @@ -3,7 +3,7 @@ use clap::{Args, ValueHint}; use icp::context::Context; use icp::prelude::*; -use crate::operations::bundle::create_bundle; +use crate::{operations::bundle::create_bundle, render::Renderer}; /// Bundle a project into a self-contained deployable archive. /// @@ -30,18 +30,26 @@ pub(crate) async fn exec(ctx: &Context, args: &BundleArgs) -> Result<(), anyhow: let canisters: Vec<_> = project.canisters.into_values().collect(); - create_bundle( + let (reporter, events) = icp_events::channel(); + let render = tokio::spawn(Renderer::for_ctx(ctx.debug).run(events)); + + let result = create_bundle( &project.dir, canisters, &args.environment, ctx.builder.clone(), ctx.artifacts.clone(), &ctx.dirs.package_cache()?, - ctx.debug, + &reporter, &args.output, ) - .await - .context("failed to create bundle")?; + .await; + + // Close the event stream so the renderer can finish, and let it flush + // (failure dumps) before the result is acted on. + drop(reporter); + render.await?; + result.context("failed to create bundle")?; Ok(()) } diff --git a/crates/icp-cli/src/main.rs b/crates/icp-cli/src/main.rs index 6fe636038..438cde234 100644 --- a/crates/icp-cli/src/main.rs +++ b/crates/icp-cli/src/main.rs @@ -21,6 +21,7 @@ mod logging; pub(crate) mod operations; mod options; mod progress; +mod render; mod telemetry; mod version; diff --git a/crates/icp-cli/src/operations/build.rs b/crates/icp-cli/src/operations/build.rs index eb85c7bb5..b0738714e 100644 --- a/crates/icp-cli/src/operations/build.rs +++ b/crates/icp-cli/src/operations/build.rs @@ -8,10 +8,8 @@ use icp::{ package::PackageCache, prelude::*, }; +use icp_events::{Reporter, StepOutcome, TaskKind, TaskOutcome, TaskReporter}; use snafu::{ResultExt, Snafu}; -use tracing::error; - -use crate::progress::{MultiStepProgressBar, ProgressManager, ProgressManagerSettings}; #[derive(Debug, Snafu)] pub enum BuildOperationError { @@ -39,18 +37,11 @@ pub struct BuildManyError { names: Vec, } -/// Holds error information from a failed canister build operation -struct BuildFailure { - canister_name: String, - error: BuildOperationError, - progress_output: Vec, -} - pub(crate) async fn build( canister_path: &Path, canister: &Canister, environment: &str, - pb: &mut MultiStepProgressBar, + task: &TaskReporter, builder: Arc, artifacts: Arc, pkg_cache: &PackageCache, @@ -60,9 +51,7 @@ pub(crate) async fn build( let step_count = canister.build.steps.len(); for (i, step) in canister.build.steps.iter().enumerate() { - let current_step = i + 1; - let pb_hdr = format!("Building: step {current_step} of {step_count} {step}"); - let tx = pb.begin_step(pb_hdr); + let reporter = task.step(i + 1, step_count, step.to_string()); let build_result = builder .build( @@ -72,12 +61,15 @@ pub(crate) async fn build( output: wasm_output_path.to_owned(), environment: environment.to_owned(), }, - Some(tx), + &reporter, pkg_cache, ) .await; - pb.end_step().await; + reporter.done(match &build_result { + Ok(()) => StepOutcome::Succeeded, + Err(_) => StepOutcome::Failed, + }); build_result?; } @@ -96,80 +88,58 @@ pub(crate) async fn build( Ok(()) } -pub(crate) async fn build_many_with_progress_bar( +pub(crate) async fn build_many( canisters: Vec<(PathBuf, Canister)>, environment: &str, builder: Arc, artifacts: Arc, pkg_cache: &PackageCache, - debug: bool, + reporter: &Reporter, ) -> Result<(), BuildManyError> { let mut futs = FuturesOrdered::new(); - let progress_manager = ProgressManager::new(ProgressManagerSettings { hidden: debug }); for (canister_path, canister) in canisters { - let mut pb = progress_manager.create_multi_step_progress_bar(&canister.name, "Build"); + let task = reporter.task(TaskKind::Build { + canister: canister.name.clone(), + }); let builder = builder.clone(); let artifacts = artifacts.clone(); + let fut = async move { - let build_result = build( + let result = build( &canister_path, &canister, environment, - &mut pb, + &task, builder, artifacts, pkg_cache, ) .await; - // Execute with progress tracking for final state - let result = ProgressManager::execute_with_progress( - &pb, - async { build_result }, - || "Built successfully".to_string(), - |err| format!("Failed to build canister: {err}"), - ) - .await; + match &result { + Ok(()) => task.finish(TaskOutcome::Succeeded), + Err(error) => task.finish(TaskOutcome::Failed { + message: error.to_string(), + }), + } - // Map error to include canister context for deferred printing - result.map_err(|error| BuildFailure { - canister_name: canister.name.clone(), - error, - progress_output: pb.dump_output(debug), - }) + result.map_err(|_| canister.name.clone()) }; futs.push_back(fut); } - // Consume the set of futures and collect errors - let mut errors: Vec = Vec::new(); + // Consume the set of futures and collect the failed canister names; the + // renderer owns displaying each failure's captured output. + let mut failed: Vec = Vec::new(); while let Some(res) = futs.next().await { - if let Err(failure) = res { - errors.push(failure); + if let Err(name) = res { + failed.push(name); } } - if !errors.is_empty() { - // Print all errors in batch - for failure in &errors { - error!( - "----- Failed to build canister '{}' -----", - failure.canister_name, - ); - error!("'{}'", failure.error); - for line in &failure.progress_output { - error!("{line}"); - } - } - - return BuildManySnafu { - names: errors - .iter() - .map(|e| e.canister_name.clone()) - .collect::>(), - } - .fail(); + if !failed.is_empty() { + return BuildManySnafu { names: failed }.fail(); } Ok(()) diff --git a/crates/icp-cli/src/operations/bundle.rs b/crates/icp-cli/src/operations/bundle.rs index 6786fc7d5..ad8305ebf 100644 --- a/crates/icp-cli/src/operations/bundle.rs +++ b/crates/icp-cli/src/operations/bundle.rs @@ -28,7 +28,9 @@ use icp::{ use snafu::{OptionExt, ResultExt, Snafu}; use tar::Builder; -use crate::operations::build::{BuildManyError, build_many_with_progress_bar}; +use icp_events::{Reporter, StepReporter}; + +use crate::operations::build::{BuildManyError, build_many}; #[derive(Debug, Snafu)] pub enum BundleError { @@ -328,7 +330,7 @@ pub(crate) async fn create_bundle( builder: Arc, artifacts: Arc, pkg_cache: &PackageCache, - debug: bool, + reporter: &Reporter, output: &Path, ) -> Result<(), BundleError> { // A bundle mirrors the workspace: the root project at the archive root and @@ -347,13 +349,13 @@ pub(crate) async fn create_bundle( validate_env_var_files(&canisters, &canonical_project_dir)?; validate_output_path(output, &canonical_sync_dirs)?; - build_many_with_progress_bar( + build_many( canisters.clone(), environment, builder, artifacts.clone(), pkg_cache, - debug, + reporter, ) .await?; @@ -727,7 +729,7 @@ async fn prepare_plugin_step( &adapter.source, canister_path, adapter.sha256.as_deref(), - None, + &StepReporter::null(), pkg_cache, ) .await diff --git a/crates/icp-cli/src/progress.rs b/crates/icp-cli/src/progress.rs index 0a0795991..74d9cc4de 100644 --- a/crates/icp-cli/src/progress.rs +++ b/crates/icp-cli/src/progress.rs @@ -13,19 +13,19 @@ pub(crate) const MAX_LINES_PER_STEP: usize = 10_000; const TICKS: &[&str] = &["✶", "✸", "✹", "✺", "✹", "✷"]; // Final tick symbols for different completion states -const TICK_EMPTY: &str = " "; -const TICK_SUCCESS: &str = "✔"; -const TICK_FAILURE: &str = "✘"; +pub(crate) const TICK_EMPTY: &str = " "; +pub(crate) const TICK_SUCCESS: &str = "✔"; +pub(crate) const TICK_FAILURE: &str = "✘"; // Color schemes for different progress states -const COLOR_REGULAR: &str = "blue"; -const COLOR_SUCCESS: &str = "green"; -const COLOR_FAILURE: &str = "red"; +pub(crate) const COLOR_REGULAR: &str = "blue"; +pub(crate) const COLOR_SUCCESS: &str = "green"; +pub(crate) const COLOR_FAILURE: &str = "red"; // Creates a progress bar style with a spinner that transitions to a final tick symbol // - end_tick: the symbol to display when the progress completes (success, failure, etc.) // - color: the color theme for the spinner and text -fn make_style(end_tick: &str, color: &str) -> ProgressStyle { +pub(crate) fn make_style(end_tick: &str, color: &str) -> ProgressStyle { // Template format: "[prefix] [spinner] [message]" let tmpl = format!("{{prefix}} {{spinner:.{color}}} {{msg}}"); @@ -63,6 +63,11 @@ impl RollingLines { self.buf.iter().map(|s| s.as_str()) } + /// Whether no lines have been pushed. + pub(crate) fn is_empty(&self) -> bool { + self.buf.is_empty() + } + /// Convert the buffer into an iterator (in order). pub(crate) fn into_iter(self) -> impl Iterator { self.buf.into_iter() diff --git a/crates/icp-cli/src/render/interactive.rs b/crates/icp-cli/src/render/interactive.rs new file mode 100644 index 000000000..d9fde82ba --- /dev/null +++ b/crates/icp-cli/src/render/interactive.rs @@ -0,0 +1,136 @@ +//! Live progress-bar renderer: one indicatif spinner per task, a rolling +//! window of the current step's output beneath it, and a ✔/✘ finish state. +//! Failed tasks replay their captured output once the stream ends. + +use std::{collections::BTreeMap, time::Duration}; + +use icp_events::{Event, EventKind, TaskId, TaskOutcome}; +use indicatif::{MultiProgress, ProgressBar}; +use itertools::Itertools; +use tracing::debug; + +use crate::progress::{ + COLOR_FAILURE, COLOR_REGULAR, COLOR_SUCCESS, RollingLines, TICK_EMPTY, TICK_FAILURE, + TICK_SUCCESS, make_style, +}; + +use super::{TaskLog, dump_failures, failure_message, step_header, success_message}; + +/// Number of output lines shown live under a task's progress bar. +const LIVE_WINDOW_LINES: usize = 4; + +pub(crate) struct InteractiveRenderer { + multi_progress: MultiProgress, + tasks: BTreeMap, +} + +struct TaskView { + log: TaskLog, + bar: ProgressBar, + /// Header of the step currently running, shown above the live window. + header: String, + /// Rolling window over the current step's most recent output lines. + window: RollingLines, +} + +impl InteractiveRenderer { + pub(crate) fn new() -> Self { + Self { + multi_progress: MultiProgress::new(), + tasks: BTreeMap::new(), + } + } + + pub(crate) fn handle(&mut self, event: Event) { + match event.kind { + EventKind::TaskStarted { task } => { + let bar = self.multi_progress.add( + ProgressBar::new_spinner().with_style(make_style(TICK_EMPTY, COLOR_REGULAR)), + ); + bar.set_prefix(format!("[{}]", task.canister())); + bar.enable_steady_tick(Duration::from_millis(120)); + + self.tasks.insert( + event.task_id, + TaskView { + log: TaskLog::new(task), + bar, + header: String::new(), + window: RollingLines::new(LIVE_WINDOW_LINES), + }, + ); + } + + EventKind::StepStarted { + number, + total, + label, + } => { + let Some(view) = self.tasks.get_mut(&event.task_id) else { + return; + }; + view.header = step_header(view.log.kind(), number, total, &label); + view.window = RollingLines::new(LIVE_WINDOW_LINES); + view.log.start_step(view.header.clone()); + view.bar.set_message(view.header.clone()); + } + + EventKind::Output { line, .. } => { + let Some(view) = self.tasks.get_mut(&event.task_id) else { + return; + }; + + debug!("{line}"); + + view.window.push(line.clone()); + view.log.push_line(line); + + // Update progress-bar with rolling terminal output + // Make the output + // │ look prettier... + // └ + let rolled = view.window.iter().map(|s| format!("│ {s}")).join("\n"); + view.bar + .set_message(format!("{}\n{rolled}\n└\n\n", view.header)); + } + + EventKind::StepCompleted { .. } => { + if let Some(view) = self.tasks.get_mut(&event.task_id) { + view.log.end_step(); + } + } + + EventKind::TaskCompleted { outcome } => { + let Some(view) = self.tasks.get_mut(&event.task_id) else { + return; + }; + + match outcome { + TaskOutcome::Succeeded => { + view.bar.set_style(make_style(TICK_SUCCESS, COLOR_SUCCESS)); + view.bar.set_message(success_message(view.log.kind())); + } + TaskOutcome::Failed { message } => { + view.bar.set_style(make_style(TICK_FAILURE, COLOR_FAILURE)); + view.bar + .set_message(failure_message(view.log.kind(), &message)); + view.log.fail(message); + } + } + + view.bar.finish(); + } + } + } + + /// Replay the captured output of failed tasks. Only the failing step is + /// shown; `--debug` runs use the plain renderer, which dumps every step. + pub(crate) fn flush(self) { + let logs = self + .tasks + .into_iter() + .map(|(id, view)| (id, view.log)) + .collect(); + dump_failures(&logs, false); + } +} diff --git a/crates/icp-cli/src/render/mod.rs b/crates/icp-cli/src/render/mod.rs new file mode 100644 index 000000000..2d2dfcfeb --- /dev/null +++ b/crates/icp-cli/src/render/mod.rs @@ -0,0 +1,202 @@ +//! Presentation layer for [`icp_events`] streams. +//! +//! Operations emit typed events through a [`icp_events::Reporter`]; a +//! [`Renderer`] consumes the stream and owns everything user-facing: wording, +//! progress bars, and the deferred failure dumps. Commands pick a renderer +//! with [`Renderer::for_ctx`] and drive it with [`Renderer::run`] alongside +//! the operation. + +use std::collections::BTreeMap; + +use icp_events::{Event, TaskId, TaskKind}; +use tokio::sync::mpsc::UnboundedReceiver; +use tracing::error; + +use crate::progress::{MAX_LINES_PER_STEP, RollingLines}; + +mod interactive; +mod plain; + +pub(crate) use interactive::InteractiveRenderer; +pub(crate) use plain::PlainRenderer; + +pub(crate) enum Renderer { + Interactive(InteractiveRenderer), + Plain(PlainRenderer), +} + +impl Renderer { + /// Pick the renderer matching how the CLI was invoked: live progress bars + /// normally, plain output under `--debug` (where indicatif bars would + /// interleave with the debug log). + pub(crate) fn for_ctx(debug: bool) -> Self { + if debug { + Renderer::Plain(PlainRenderer::new()) + } else { + Renderer::Interactive(InteractiveRenderer::new()) + } + } + + /// Drive the renderer until every reporter handle is dropped, then flush + /// deferred output (the per-task failure dumps). + pub(crate) async fn run(self, mut events: UnboundedReceiver) { + match self { + Renderer::Interactive(mut renderer) => { + while let Some(event) = events.recv().await { + renderer.handle(event); + } + renderer.flush(); + } + Renderer::Plain(mut renderer) => { + while let Some(event) = events.recv().await { + renderer.handle(event); + } + renderer.flush(); + } + } + } +} + +// Wording for each task kind. Events carry data; these helpers own the words. + +/// Live header shown while a step runs, e.g. "Building: step 1 of 3 (script)…". +/// `label` may span multiple lines. +fn step_header(kind: &TaskKind, number: usize, total: usize, label: &str) -> String { + match kind { + TaskKind::Build { .. } => format!("Building: step {number} of {total} {label}"), + } +} + +/// Label for the captured-output header, e.g. "[name] Build output:". +fn output_label(kind: &TaskKind) -> &'static str { + match kind { + TaskKind::Build { .. } => "Build", + } +} + +/// Final progress-bar message for a task that succeeded. +fn success_message(kind: &TaskKind) -> String { + match kind { + TaskKind::Build { .. } => "Built successfully".to_owned(), + } +} + +/// Final progress-bar message for a task that failed. +fn failure_message(kind: &TaskKind, message: &str) -> String { + match kind { + TaskKind::Build { .. } => format!("Failed to build canister: {message}"), + } +} + +/// First line of a task's failure dump. +fn failure_header(kind: &TaskKind) -> String { + match kind { + TaskKind::Build { canister } => { + format!("----- Failed to build canister '{canister}' -----") + } + } +} + +/// Captured output of one task, kept so a failure can be replayed after the +/// live view is gone. +pub(super) struct TaskLog { + kind: TaskKind, + finished_steps: Vec, + current_step: Option, + failure: Option, +} + +struct StepLog { + title: String, + lines: RollingLines, +} + +impl TaskLog { + fn new(kind: TaskKind) -> Self { + Self { + kind, + finished_steps: Vec::new(), + current_step: None, + failure: None, + } + } + + fn kind(&self) -> &TaskKind { + &self.kind + } + + fn start_step(&mut self, title: String) { + self.end_step(); + self.current_step = Some(StepLog { + title, + // We need _some_ limit to prevent consuming infinite memory + lines: RollingLines::new(MAX_LINES_PER_STEP), + }); + } + + fn push_line(&mut self, line: String) { + if let Some(step) = &mut self.current_step { + step.lines.push(line); + } + } + + fn end_step(&mut self) { + if let Some(step) = self.current_step.take() { + self.finished_steps.push(step); + } + } + + fn fail(&mut self, message: String) { + self.failure = Some(message); + } + + /// Render the captured output. When `all_steps` is true, output from + /// every step is included; otherwise only the last (failing) step is + /// shown. + fn dump(&self, all_steps: bool) -> Vec { + let name = self.kind.canister(); + let mut lines = Vec::new(); + + lines.push(format!("[{name}] {} output:", output_label(&self.kind))); + + let steps: &[StepLog] = if all_steps { + &self.finished_steps + } else { + self.finished_steps + .last() + .map(std::slice::from_ref) + .unwrap_or_default() + }; + + for step in steps { + for line in step.title.lines() { + if !line.is_empty() { + lines.push(format!("[{name}] {line}:")); + } + } + + if step.lines.is_empty() { + lines.push(format!("[{name}] ")); + } else { + lines.extend(step.lines.iter().map(|line| format!("[{name}] > {line}"))); + } + } + + lines + } +} + +/// Print the failure dump for every failed task, in task-creation order. +fn dump_failures(logs: &BTreeMap, all_steps: bool) { + for log in logs.values() { + let Some(message) = &log.failure else { + continue; + }; + + error!("{}", failure_header(&log.kind)); + error!("'{message}'"); + for line in log.dump(all_steps) { + error!("{line}"); + } + } +} diff --git a/crates/icp-cli/src/render/plain.rs b/crates/icp-cli/src/render/plain.rs new file mode 100644 index 000000000..6c6b5f913 --- /dev/null +++ b/crates/icp-cli/src/render/plain.rs @@ -0,0 +1,70 @@ +//! Renderer for `--debug` runs: no live progress bars (they would interleave +//! with the debug log). Output lines go to the debug log as they arrive, and +//! failed tasks dump the captured output of every step once the stream ends. + +use std::collections::BTreeMap; + +use icp_events::{Event, EventKind, TaskId, TaskOutcome}; +use tracing::debug; + +use super::{TaskLog, dump_failures, step_header}; + +pub(crate) struct PlainRenderer { + tasks: BTreeMap, +} + +impl PlainRenderer { + pub(crate) fn new() -> Self { + Self { + tasks: BTreeMap::new(), + } + } + + pub(crate) fn handle(&mut self, event: Event) { + match event.kind { + EventKind::TaskStarted { task } => { + self.tasks.insert(event.task_id, TaskLog::new(task)); + } + + EventKind::StepStarted { + number, + total, + label, + } => { + let Some(log) = self.tasks.get_mut(&event.task_id) else { + return; + }; + let header = step_header(log.kind(), number, total, &label); + log.start_step(header); + } + + EventKind::Output { line, .. } => { + let Some(log) = self.tasks.get_mut(&event.task_id) else { + return; + }; + debug!("{line}"); + log.push_line(line); + } + + EventKind::StepCompleted { .. } => { + if let Some(log) = self.tasks.get_mut(&event.task_id) { + log.end_step(); + } + } + + EventKind::TaskCompleted { outcome } => { + let Some(log) = self.tasks.get_mut(&event.task_id) else { + return; + }; + if let TaskOutcome::Failed { message } = outcome { + log.fail(message); + } + } + } + } + + /// Replay the captured output of failed tasks, including every step. + pub(crate) fn flush(self) { + dump_failures(&self.tasks, true); + } +} diff --git a/crates/icp-cli/tests/build_tests.rs b/crates/icp-cli/tests/build_tests.rs index c2e89fbf0..f60de9847 100644 --- a/crates/icp-cli/tests/build_tests.rs +++ b/crates/icp-cli/tests/build_tests.rs @@ -502,9 +502,9 @@ fn build_multiple_canisters() { .assert() .success() .stderr(contains(r#"canisters: ["canister-a", "canister-b"]"#)) - .stderr(contains("DEBUG icp::progress: building canister-a")) - .stderr(contains("DEBUG icp::progress: building canister-b")) - .stderr(contains("DEBUG icp::progress: building canister-c").not()); + .stderr(contains("DEBUG icp::render::plain: building canister-a")) + .stderr(contains("DEBUG icp::render::plain: building canister-b")) + .stderr(contains("DEBUG icp::render::plain: building canister-c").not()); } #[test] @@ -559,7 +559,7 @@ fn build_all_canisters_in_environment() { .success() .stderr(contains(r#"canisters: []"#)) .stderr(contains(r#"environment: Some("test-env")"#)) - .stderr(contains("DEBUG icp::progress: building canister-a")) - .stderr(contains("DEBUG icp::progress: building canister-b")) - .stderr(contains("DEBUG icp::progress: building canister-c").not()); // not in test-env + .stderr(contains("DEBUG icp::render::plain: building canister-a")) + .stderr(contains("DEBUG icp::render::plain: building canister-b")) + .stderr(contains("DEBUG icp::render::plain: building canister-c").not()); // not in test-env } diff --git a/crates/icp-events/Cargo.toml b/crates/icp-events/Cargo.toml new file mode 100644 index 000000000..40acc73b3 --- /dev/null +++ b/crates/icp-events/Cargo.toml @@ -0,0 +1,10 @@ +[package] +name = "icp-events" +version.workspace = true +edition = { workspace = true } +license = { workspace = true } +publish.workspace = true + +[dependencies] +serde = { workspace = true } +tokio = { workspace = true, features = ["sync", "rt"] } diff --git a/crates/icp-events/src/lib.rs b/crates/icp-events/src/lib.rs new file mode 100644 index 000000000..e22e8336e --- /dev/null +++ b/crates/icp-events/src/lib.rs @@ -0,0 +1,274 @@ +//! Typed progress events passed from operations to the presentation layer. +//! +//! Operations (and the core library underneath them) emit [`Event`]s through +//! cheap-to-clone reporter handles ([`Reporter`] → [`TaskReporter`] → +//! [`StepReporter`]); the CLI's renderers consume the event stream and decide +//! how to display it. Events carry data, not prose — wording, layout, and +//! color are the renderer's job. +//! +//! Sends never block and never fail: the channel is unbounded, and with no +//! receiver events are simply dropped, so tests and headless callers get +//! silence for free. Errors do not travel on this stream — an operation's +//! `Result` remains the source of truth; [`TaskOutcome::Failed`] exists only +//! so a renderer can paint the failure state. + +use std::sync::{ + Arc, + atomic::{AtomicU64, Ordering}, +}; + +use serde::Serialize; +use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel}; + +/// Identifies one task (one canister-level unit of work) within an event +/// stream. Ids are assigned in task-creation order, so renderers can use them +/// to present tasks in a stable order regardless of completion order. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize)] +#[serde(transparent)] +pub struct TaskId(u64); + +#[derive(Debug, Clone, Serialize)] +pub struct Event { + pub task_id: TaskId, + #[serde(flatten)] + pub kind: EventKind, +} + +#[derive(Debug, Clone, Serialize)] +#[serde(tag = "event", rename_all = "snake_case")] +pub enum EventKind { + /// The task began. Emitted once per task, before any of its steps. + TaskStarted { task: TaskKind }, + + /// A step of the task began. Steps within a task are sequential; + /// `number` is 1-based. `label` describes the step (it may span + /// multiple lines). + StepStarted { + number: usize, + total: usize, + label: String, + }, + + /// One line of output produced while the task's current step runs. + Output { stream: OutputStream, line: String }, + + /// The task's current step finished. + StepCompleted { outcome: StepOutcome }, + + /// The task finished; no further events follow for this task. + TaskCompleted { outcome: TaskOutcome }, +} + +/// What a task is doing, and to which canister. +#[derive(Debug, Clone, Serialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum TaskKind { + Build { canister: String }, +} + +impl TaskKind { + /// The canister this task operates on. + pub fn canister(&self) -> &str { + match self { + TaskKind::Build { canister } => canister, + } + } +} + +/// Where an output line came from. +#[derive(Debug, Clone, Copy, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum OutputStream { + Stdout, + Stderr, + /// A progress note from icp itself rather than a spawned tool. + Info, +} + +#[derive(Debug, Clone, Copy, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum StepOutcome { + Succeeded, + Failed, +} + +#[derive(Debug, Clone, Serialize)] +#[serde(tag = "result", rename_all = "snake_case")] +pub enum TaskOutcome { + Succeeded, + /// `message` is for display only; the typed error stays on the + /// operation's return path. + Failed { + message: String, + }, +} + +/// Create a connected reporter/receiver pair. The receiver yields `None` once +/// the reporter and every handle derived from it have been dropped. +pub fn channel() -> (Reporter, UnboundedReceiver) { + let (tx, rx) = unbounded_channel(); + let reporter = Reporter { + inner: Some(ReporterInner { + tx, + next_task_id: Arc::new(AtomicU64::new(0)), + }), + }; + (reporter, rx) +} + +/// Entry point handed to an operation; spawns [`TaskReporter`]s. +#[derive(Debug, Clone)] +pub struct Reporter { + inner: Option, +} + +#[derive(Debug, Clone)] +struct ReporterInner { + tx: UnboundedSender, + next_task_id: Arc, +} + +impl Reporter { + /// A reporter whose events go nowhere. + pub fn null() -> Self { + Self { inner: None } + } + + /// Begin a task, emitting [`EventKind::TaskStarted`]. + pub fn task(&self, task: TaskKind) -> TaskReporter { + let Some(inner) = &self.inner else { + return TaskReporter::null(); + }; + let task_id = TaskId(inner.next_task_id.fetch_add(1, Ordering::Relaxed)); + let _ = inner.tx.send(Event { + task_id, + kind: EventKind::TaskStarted { task }, + }); + TaskReporter { + tx: Some(inner.tx.clone()), + task_id, + } + } +} + +/// Reports the lifecycle of one task. +#[derive(Debug, Clone)] +pub struct TaskReporter { + tx: Option>, + task_id: TaskId, +} + +impl TaskReporter { + /// A task reporter whose events go nowhere. + pub fn null() -> Self { + Self { + tx: None, + task_id: TaskId(0), + } + } + + fn send(&self, kind: EventKind) { + if let Some(tx) = &self.tx { + let _ = tx.send(Event { + task_id: self.task_id, + kind, + }); + } + } + + /// Begin the task's next step, emitting [`EventKind::StepStarted`]. + /// `number` is 1-based. + pub fn step(&self, number: usize, total: usize, label: impl Into) -> StepReporter { + self.send(EventKind::StepStarted { + number, + total, + label: label.into(), + }); + StepReporter { + tx: self.tx.clone(), + task_id: self.task_id, + } + } + + /// Finish the task, emitting [`EventKind::TaskCompleted`]. No further + /// events should be sent for this task. + pub fn finish(&self, outcome: TaskOutcome) { + self.send(EventKind::TaskCompleted { outcome }); + } +} + +/// Reports output produced during one step of a task. +#[derive(Debug, Clone)] +pub struct StepReporter { + tx: Option>, + task_id: TaskId, +} + +impl StepReporter { + /// A step reporter whose events go nowhere. + pub fn null() -> Self { + Self { + tx: None, + task_id: TaskId(0), + } + } + + /// TEMPORARY: adapt a legacy line channel into a [`StepReporter`]. + /// + /// Output lines emitted through the returned reporter are forwarded to + /// `lines` by a background task, which drains and exits once the reporter + /// and all of its clones are dropped. Used by call sites that still + /// render through the old progress-bar channel (sync); remove once those + /// paths take a real [`Reporter`]. + pub fn bridge_lines(lines: tokio::sync::mpsc::Sender) -> Self { + let (tx, mut rx) = unbounded_channel::(); + tokio::spawn(async move { + while let Some(event) = rx.recv().await { + if let EventKind::Output { line, .. } = event.kind { + let _ = lines.send(line).await; + } + } + }); + Self { + tx: Some(tx), + task_id: TaskId(0), + } + } + + fn send(&self, kind: EventKind) { + if let Some(tx) = &self.tx { + let _ = tx.send(Event { + task_id: self.task_id, + kind, + }); + } + } + + /// Emit one line of output. + pub fn output(&self, stream: OutputStream, line: impl Into) { + self.send(EventKind::Output { + stream, + line: line.into(), + }); + } + + /// Emit one line of tool stdout. + pub fn stdout(&self, line: impl Into) { + self.output(OutputStream::Stdout, line); + } + + /// Emit one line of tool stderr. + pub fn stderr(&self, line: impl Into) { + self.output(OutputStream::Stderr, line); + } + + /// Emit a progress note from icp itself. + pub fn info(&self, line: impl Into) { + self.output(OutputStream::Info, line); + } + + /// Finish the step, emitting [`EventKind::StepCompleted`]. + pub fn done(&self, outcome: StepOutcome) { + self.send(EventKind::StepCompleted { outcome }); + } +} diff --git a/crates/icp/Cargo.toml b/crates/icp/Cargo.toml index 40690d523..d0fcb3470 100644 --- a/crates/icp/Cargo.toml +++ b/crates/icp/Cargo.toml @@ -35,6 +35,7 @@ ic-ledger-types = { workspace = true } ic-management-canister-types = { workspace = true } ic-utils = { workspace = true } icp-canister-interfaces = { workspace = true } +icp-events = { workspace = true } icp-sync-plugin = { workspace = true } icrc-ledger-types = { workspace = true } indexmap = { workspace = true } diff --git a/crates/icp/src/canister/build/mod.rs b/crates/icp/src/canister/build/mod.rs index d630d9ee4..7324e20f6 100644 --- a/crates/icp/src/canister/build/mod.rs +++ b/crates/icp/src/canister/build/mod.rs @@ -1,7 +1,7 @@ use async_trait::async_trait; +use icp_events::StepReporter; use snafu::prelude::*; -use tokio::sync::mpsc::Sender; use crate::manifest::canister::BuildStep; use crate::package::PackageCache; @@ -30,7 +30,7 @@ pub trait Build: Sync + Send { &self, step: &BuildStep, params: &Params, - stdio: Option>, + reporter: &StepReporter, pkg_cache: &PackageCache, ) -> Result<(), BuildError>; } @@ -43,14 +43,14 @@ impl Build for Builder { &self, step: &BuildStep, params: &Params, - stdio: Option>, + reporter: &StepReporter, pkg_cache: &PackageCache, ) -> Result<(), BuildError> { match step { BuildStep::Prebuilt(adapter) => { - Ok(prebuilt::build(adapter, params, stdio, pkg_cache).await?) + Ok(prebuilt::build(adapter, params, reporter, pkg_cache).await?) } - BuildStep::Script(adapter) => Ok(script::build(adapter, params, stdio).await?), + BuildStep::Script(adapter) => Ok(script::build(adapter, params, reporter).await?), } } } @@ -67,7 +67,7 @@ impl Build for UnimplementedMockBuilder { &self, _step: &BuildStep, _params: &Params, - _stdio: Option>, + _reporter: &StepReporter, _pkg_cache: &PackageCache, ) -> Result<(), BuildError> { unimplemented!("UnimplementedMockBuilder::build") diff --git a/crates/icp/src/canister/build/prebuilt.rs b/crates/icp/src/canister/build/prebuilt.rs index 774a102f9..d0482756f 100644 --- a/crates/icp/src/canister/build/prebuilt.rs +++ b/crates/icp/src/canister/build/prebuilt.rs @@ -1,5 +1,5 @@ +use icp_events::StepReporter; use snafu::prelude::*; -use tokio::sync::mpsc::Sender; use crate::{canister::wasm, fs, manifest::adapter::prebuilt::Adapter, package::PackageCache}; @@ -17,23 +17,19 @@ pub enum PrebuiltError { pub(super) async fn build( adapter: &Adapter, params: &Params, - stdio: Option>, + reporter: &StepReporter, pkg_cache: &PackageCache, ) -> Result<(), PrebuiltError> { let src = wasm::resolve( &adapter.source, ¶ms.path, adapter.sha256.as_deref(), - stdio.as_ref(), + reporter, pkg_cache, ) .await?; - if let Some(tx) = &stdio { - let _ = tx - .send(format!("Writing WASM file: {}", params.output)) - .await; - } + reporter.info(format!("Writing WASM file: {}", params.output)); fs::copy(&src, ¶ms.output).context(CopyFileSnafu)?; Ok(()) diff --git a/crates/icp/src/canister/build/script.rs b/crates/icp/src/canister/build/script.rs index 488b40077..5cbbf3021 100644 --- a/crates/icp/src/canister/build/script.rs +++ b/crates/icp/src/canister/build/script.rs @@ -1,4 +1,4 @@ -use tokio::sync::mpsc::Sender; +use icp_events::StepReporter; use crate::manifest::adapter::script::Adapter; @@ -9,7 +9,7 @@ use super::super::script::{ScriptError, execute}; pub(super) async fn build( adapter: &Adapter, params: &Params, - stdio: Option>, + reporter: &StepReporter, ) -> Result<(), ScriptError> { execute( adapter, @@ -18,7 +18,7 @@ pub(super) async fn build( ("ICP_WASM_OUTPUT_PATH", params.output.as_ref()), ("ICP_CLI_ENVIRONMENT", ¶ms.environment), ], - stdio, + reporter, ) .await } @@ -55,7 +55,7 @@ mod tests { output: "/".into(), environment: LOCAL.to_owned(), }, - None, + &StepReporter::null(), ) .await .expect("failed to build script step"); @@ -91,7 +91,7 @@ mod tests { output: "/".into(), environment: LOCAL.to_owned(), }, - None, + &StepReporter::null(), ) .await .expect("failed to build script step"); @@ -127,7 +127,7 @@ mod tests { output: out_wasm.path().to_owned(), environment: "staging".to_owned(), }, - None, + &StepReporter::null(), ) .await .expect("failed to build script step"); @@ -155,7 +155,7 @@ mod tests { output: "/".into(), environment: LOCAL.to_owned(), }, - None, + &StepReporter::null(), ) .await; @@ -179,7 +179,7 @@ mod tests { output: "/".into(), environment: LOCAL.to_owned(), }, - None, + &StepReporter::null(), ) .await; @@ -203,7 +203,7 @@ mod tests { output: "/".into(), environment: LOCAL.to_owned(), }, - None, + &StepReporter::null(), ) .await; diff --git a/crates/icp/src/canister/script.rs b/crates/icp/src/canister/script.rs index 6974a745d..fac5e640d 100644 --- a/crates/icp/src/canister/script.rs +++ b/crates/icp/src/canister/script.rs @@ -1,11 +1,11 @@ use std::process::Stdio; +use icp_events::StepReporter; use snafu::prelude::*; use tokio::{ io::{AsyncBufReadExt, BufReader}, join, process::Command, - sync::mpsc::Sender, }; use crate::manifest::adapter::script::Adapter; @@ -53,14 +53,14 @@ pub(super) async fn execute( adapter: &Adapter, cwd: &Path, envs: &[(&str, &str)], - stdio: Option>, + reporter: &StepReporter, ) -> Result<(), ScriptError> { // Normalize `command` field based on whether it's a single command or multiple. - execute_commands(&adapter.command.as_vec(), cwd, envs, stdio).await + execute_commands(&adapter.command.as_vec(), cwd, envs, reporter).await } /// Run each command in order under a shell, with `envs` set and `cwd` as the -/// working directory, streaming stdout/stderr lines to `stdio`. +/// working directory, streaming stdout/stderr lines to `reporter`. /// /// Takes already-resolved commands rather than an [`Adapter`], so the subprocess /// executor needs to know nothing about manifest types. The sync path resolves @@ -71,7 +71,7 @@ pub(super) async fn execute_commands( cmds: &[String], cwd: &Path, envs: &[(&str, &str)], - stdio: Option>, + reporter: &StepReporter, ) -> Result<(), ScriptError> { // Iterate over configured commands for input_cmd in cmds { @@ -113,14 +113,12 @@ pub(super) async fn execute_commands( // // Stdout tokio::spawn({ - // Clone the stdio sender for use in the stdout handling task - let stdio = stdio.clone(); + // Clone the reporter for use in the stdout handling task + let reporter = reporter.clone(); async move { while let Ok(Some(line)) = stdout.next_line().await { - if let Some(sender) = &stdio { - let _ = sender.send(line).await; - } + reporter.stdout(line); } Ok::<(), ScriptError>(()) } @@ -128,14 +126,12 @@ pub(super) async fn execute_commands( // // Stderr tokio::spawn({ - // Clone the stdio sender for use in the stderr handling task - let stdio = stdio.clone(); + // Clone the reporter for use in the stderr handling task + let reporter = reporter.clone(); async move { while let Ok(Some(line)) = stderr.next_line().await { - if let Some(sender) = &stdio { - let _ = sender.send(line).await; - } + reporter.stderr(line); } Ok::<(), ScriptError>(()) } diff --git a/crates/icp/src/canister/sync/mod.rs b/crates/icp/src/canister/sync/mod.rs index 519a75b71..be047b928 100644 --- a/crates/icp/src/canister/sync/mod.rs +++ b/crates/icp/src/canister/sync/mod.rs @@ -16,6 +16,17 @@ pub mod script; use script::{HostScripts, ScriptInvocation, ScriptRunError, ScriptRunner}; +/// TEMPORARY: adapt the sync path's legacy stdio line channel into a +/// [`StepReporter`](icp_events::StepReporter) for the low-level helpers that +/// now report events ([`wasm::resolve`](crate::canister::wasm::resolve), the +/// script executor). Remove once [`Synchronize`] takes a reporter directly. +pub(crate) fn stdio_reporter(stdio: Option>) -> icp_events::StepReporter { + match stdio { + Some(lines) => icp_events::StepReporter::bridge_lines(lines), + None => icp_events::StepReporter::null(), + } +} + pub struct Params { pub path: PathBuf, pub cid: Principal, diff --git a/crates/icp/src/canister/sync/plugin.rs b/crates/icp/src/canister/sync/plugin.rs index 97056d64d..12e1f796c 100644 --- a/crates/icp/src/canister/sync/plugin.rs +++ b/crates/icp/src/canister/sync/plugin.rs @@ -78,7 +78,7 @@ pub(super) async fn sync( &adapter.source, ¶ms.path, adapter.sha256.as_deref(), - stdio.as_ref(), + &super::stdio_reporter(stdio.clone()), pkg_cache, ) .await?; diff --git a/crates/icp/src/canister/sync/script.rs b/crates/icp/src/canister/sync/script.rs index 7c9d741d7..537a1c692 100644 --- a/crates/icp/src/canister/sync/script.rs +++ b/crates/icp/src/canister/sync/script.rs @@ -110,7 +110,8 @@ impl ScriptRunner for HostScripts { .iter() .map(|(k, v)| (k.as_str(), v.as_str())) .collect(); - execute_commands(&invocation.commands, &invocation.cwd, &env_refs, stdio) + let reporter = super::stdio_reporter(stdio); + execute_commands(&invocation.commands, &invocation.cwd, &env_refs, &reporter) .await .map_err(|source| ScriptRunError { source: Box::new(source), diff --git a/crates/icp/src/canister/wasm.rs b/crates/icp/src/canister/wasm.rs index 2cf2b219d..c77549dba 100644 --- a/crates/icp/src/canister/wasm.rs +++ b/crates/icp/src/canister/wasm.rs @@ -1,8 +1,8 @@ use camino::{Utf8Path, Utf8PathBuf}; +use icp_events::StepReporter; use reqwest::{Client, Method, Request}; use sha2::{Digest, Sha256}; use snafu::prelude::*; -use tokio::sync::mpsc::Sender; use url::Url; use crate::{ @@ -50,22 +50,18 @@ pub async fn resolve( source: &SourceField, base_dir: &Utf8Path, sha256: Option<&str>, - stdio: Option<&Sender>, + reporter: &StepReporter, pkg_cache: &PackageCache, ) -> Result { match source { SourceField::Local(s) => { let path = base_dir.join(&s.path); if let Some(expected) = sha256 { - if let Some(tx) = stdio { - let _ = tx.send(format!("Reading wasm: {}", s.path)).await; - } + reporter.info(format!("Reading wasm: {}", s.path)); let bytes = read(&path).context(ReadLocalSnafu { path: s.path.clone(), })?; - if let Some(tx) = stdio { - let _ = tx.send("Verifying checksum".to_string()).await; - } + reporter.info("Verifying checksum"); let actual = hex::encode(Sha256::digest(&bytes)); ensure!( actual == expected, @@ -94,17 +90,13 @@ pub async fn resolve( .await .context(LockCacheSnafu)?; if let Some(path) = cached { - if let Some(tx) = stdio { - let _ = tx.send("Using cached file".to_string()).await; - } + reporter.info("Using cached file"); return Ok(path); } } let url = Url::parse(&s.url).context(ParseUrlSnafu)?; - if let Some(tx) = stdio { - let _ = tx.send(format!("Fetching wasm: {url}")).await; - } + reporter.info(format!("Fetching wasm: {url}")); let resp = Client::new() .execute(Request::new(Method::GET, url)) .await @@ -118,9 +110,7 @@ pub async fn resolve( // Use provided sha256 as cache key (after verifying), or compute from bytes. let cache_sha = match sha256 { Some(expected) => { - if let Some(tx) = stdio { - let _ = tx.send("Verifying checksum".to_string()).await; - } + reporter.info("Verifying checksum"); let actual = hex::encode(Sha256::digest(&bytes)); ensure!( actual == expected, From e3edac0aa39d6a3c85a2cd1f37e45f67b041fcd9 Mon Sep 17 00:00:00 2001 From: Raymond Khalife Date: Wed, 19 Aug 2026 20:29:28 +0000 Subject: [PATCH 2/8] refactor: port the sync pipeline to icp-events MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Synchronize and ScriptRunner traits, and the sync-plugin runtime's run_plugin, take a StepReporter in place of Option>, which removes the temporary line-channel bridges entirely — the plugin runtime's live output forwarding is now lossless (events instead of best-effort try_send). Retained plugin stderr rides on TaskOutcome::Succeeded and failure cause chains on TaskOutcome::Failed, so the renderers own printing both; sync_many drops its ProgressManager and eprintln calls. MultiStepProgressBar and its step channel are now unused and deleted from progress.rs. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01QUZHfx2ng97WM2j9XoA3rr --- Cargo.lock | 2 + crates/icp-cli/src/commands/deploy.rs | 16 ++- crates/icp-cli/src/commands/sync.rs | 17 ++- crates/icp-cli/src/operations/build.rs | 6 +- crates/icp-cli/src/operations/sync.rs | 132 +++++++------------ crates/icp-cli/src/progress.rs | 155 ----------------------- crates/icp-cli/src/render/interactive.rs | 11 +- crates/icp-cli/src/render/mod.rs | 34 ++++- crates/icp-cli/src/render/plain.rs | 9 +- crates/icp-cli/tests/sync_tests.rs | 16 +-- crates/icp-events/Cargo.toml | 3 +- crates/icp-events/src/lib.rs | 78 ++++++++---- crates/icp-sync-plugin/Cargo.toml | 1 + crates/icp-sync-plugin/src/runtime.rs | 93 +++++++++----- crates/icp/src/canister/sync/mod.rs | 33 ++--- crates/icp/src/canister/sync/plugin.rs | 10 +- crates/icp/src/canister/sync/script.rs | 24 ++-- 17 files changed, 279 insertions(+), 361 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7393ddbf9..c58dcc1ad 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3775,6 +3775,7 @@ dependencies = [ name = "icp-events" version = "1.3.0" dependencies = [ + "candid", "serde", "tokio", ] @@ -3792,6 +3793,7 @@ dependencies = [ "hex", "ic-agent", "icp-canister-interfaces", + "icp-events", "snafu", "tokio", "wasmtime", diff --git a/crates/icp-cli/src/commands/deploy.rs b/crates/icp-cli/src/commands/deploy.rs index 587fe57e4..8cbfed936 100644 --- a/crates/icp-cli/src/commands/deploy.rs +++ b/crates/icp-cli/src/commands/deploy.rs @@ -500,7 +500,11 @@ pub(crate) async fn exec(ctx: &Context, args: &DeployArgs) -> Result<(), anyhow: .collect(); let pkg_cache = ctx.dirs.package_cache()?; - sync_many( + + let (reporter, events) = icp_events::channel(); + let render = tokio::spawn(Renderer::for_ctx(ctx.debug).run(events)); + + let sync_result = sync_many( ctx.syncer.clone(), agent.clone(), sync_canisters, @@ -508,10 +512,16 @@ pub(crate) async fn exec(ctx: &Context, args: &DeployArgs) -> Result<(), anyhow: env.network.name.clone(), canister_ids, args.proxy, - ctx.debug, &pkg_cache, + &reporter, ) - .await?; + .await; + + // Close the event stream so the renderer can finish, and let it flush + // (failure dumps) before the result is acted on. + drop(reporter); + render.await?; + sync_result?; } // Print URLs for deployed canisters diff --git a/crates/icp-cli/src/commands/sync.rs b/crates/icp-cli/src/commands/sync.rs index ac2d1e099..936e500ec 100644 --- a/crates/icp-cli/src/commands/sync.rs +++ b/crates/icp-cli/src/commands/sync.rs @@ -12,6 +12,7 @@ use tracing::info; use crate::{ operations::{proxy_management, sync::sync_many}, options::{EnvironmentOpt, IdentityOpt}, + render::Renderer, }; /// Synchronize canisters @@ -125,7 +126,11 @@ pub(crate) async fn exec(ctx: &Context, args: &SyncArgs) -> Result<(), anyhow::E .collect(); let pkg_cache = ctx.dirs.package_cache()?; - sync_many( + + let (reporter, events) = icp_events::channel(); + let render = tokio::spawn(Renderer::for_ctx(ctx.debug).run(events)); + + let result = sync_many( ctx.syncer.clone(), agent, sync_canisters, @@ -133,10 +138,16 @@ pub(crate) async fn exec(ctx: &Context, args: &SyncArgs) -> Result<(), anyhow::E env.network.name.clone(), canister_ids, args.proxy, - ctx.debug, &pkg_cache, + &reporter, ) - .await?; + .await; + + // Close the event stream so the renderer can finish, and let it flush + // (failure dumps) before the result is acted on. + drop(reporter); + render.await?; + result?; Ok(()) } diff --git a/crates/icp-cli/src/operations/build.rs b/crates/icp-cli/src/operations/build.rs index b0738714e..8df7bd142 100644 --- a/crates/icp-cli/src/operations/build.rs +++ b/crates/icp-cli/src/operations/build.rs @@ -118,10 +118,8 @@ pub(crate) async fn build_many( .await; match &result { - Ok(()) => task.finish(TaskOutcome::Succeeded), - Err(error) => task.finish(TaskOutcome::Failed { - message: error.to_string(), - }), + Ok(()) => task.finish(TaskOutcome::succeeded()), + Err(error) => task.finish(TaskOutcome::failed(error.to_string())), } result.map_err(|_| canister.name.clone()) diff --git a/crates/icp-cli/src/operations/sync.rs b/crates/icp-cli/src/operations/sync.rs index 18c77efcc..8cab97fff 100644 --- a/crates/icp-cli/src/operations/sync.rs +++ b/crates/icp-cli/src/operations/sync.rs @@ -7,12 +7,10 @@ use icp::{ package::PackageCache, prelude::PathBuf, }; +use icp_events::{Reporter, StepOutcome, TaskKind, TaskOutcome, TaskReporter}; use snafu::prelude::*; use std::collections::BTreeMap; use std::sync::Arc; -use tracing::error; - -use crate::progress::{MultiStepProgressBar, ProgressManager, ProgressManagerSettings}; #[derive(Debug, Snafu)] #[snafu(display("Canister(s) {names:?} failed to sync."))] @@ -20,15 +18,8 @@ pub struct SyncOperationError { names: Vec, } -/// Holds error information from a failed canister sync operation -struct SyncFailure { - canister_name: String, - canister_id: Principal, - error: SynchronizeError, - progress_output: Vec, -} - -/// Synchronizes a single canister using its configured sync steps +/// Synchronizes a single canister using its configured sync steps, returning +/// the stderr lines the steps retained for the persistent output channel. async fn sync_canister( syncer: &Arc, agent: &Agent, @@ -39,20 +30,15 @@ async fn sync_canister( network: &str, canister_ids: &BTreeMap, proxy: Option, - pb: &mut MultiStepProgressBar, + task: &TaskReporter, pkg_cache: &PackageCache, ) -> Result, SynchronizeError> { let step_count = canister_info.sync.steps.len(); let mut stderr_lines = Vec::new(); for (i, step) in canister_info.sync.steps.iter().enumerate() { - // Indicate to user the current step being executed - let current_step = i + 1; - let pb_hdr = format!("\nSyncing: {step} {current_step} of {step_count}"); - - let tx = pb.begin_step(pb_hdr); + let reporter = task.step(i + 1, step_count, step.to_string()); - // Execute step let sync_result = syncer .sync( step, @@ -65,13 +51,15 @@ async fn sync_canister( proxy, }, agent, - Some(tx), + &reporter, pkg_cache, ) .await; - // Ensure background receiver drains all messages - pb.end_step().await; + reporter.done(match &sync_result { + Ok(_) => StepOutcome::Succeeded, + Err(_) => StepOutcome::Failed, + }); stderr_lines.extend(sync_result?); } @@ -79,7 +67,18 @@ async fn sync_canister( Ok(stderr_lines) } -/// Orchestrates syncing multiple canisters with progress tracking +/// The rendered `source()` chain of an error, outermost cause first. +fn error_causes(error: &dyn std::error::Error) -> Vec { + let mut causes = Vec::new(); + let mut cause = error.source(); + while let Some(err) = cause { + causes.push(err.to_string()); + cause = err.source(); + } + causes +} + +/// Orchestrates syncing multiple canisters concurrently. pub(crate) async fn sync_many( syncer: Arc, agent: Agent, @@ -88,14 +87,16 @@ pub(crate) async fn sync_many( network: String, canister_ids: BTreeMap, proxy: Option, - debug: bool, pkg_cache: &PackageCache, + reporter: &Reporter, ) -> Result<(), SyncOperationError> { let mut futs = FuturesOrdered::new(); - let progress_manager = ProgressManager::new(ProgressManagerSettings { hidden: debug }); for (cid, canister_path, canister_info) in canisters { - let mut pb = progress_manager.create_multi_step_progress_bar(&canister_info.name, "Sync"); + let task = reporter.task(TaskKind::Sync { + canister: canister_info.name.clone(), + canister_id: cid, + }); let fut = { let agent = agent.clone(); @@ -105,8 +106,7 @@ pub(crate) async fn sync_many( let canister_ids = canister_ids.clone(); async move { - // Define the sync logic - let sync_result = sync_canister( + let result = sync_canister( &syncer, &agent, canister_path, @@ -116,78 +116,42 @@ pub(crate) async fn sync_many( &network, &canister_ids, proxy, - &mut pb, + &task, pkg_cache, ) .await; - // Execute with progress tracking for final state - let result = ProgressManager::execute_with_progress( - &pb, - async { sync_result }, - || format!("Synced successfully: {cid}"), - |err| format!("Failed to sync canister: {err}"), - ) - .await; - - // Print stderr lines the plugin emitted; the rolling buffer - // discards them on success, but they belong on the persistent - // output channel. - if let Ok(lines) = &result { - for line in lines { - eprintln!("[{}] {line}", canister_info.name); - } + match &result { + // The retained stderr lines ride on the outcome: the + // rolling step view discards them on success, but they + // belong on the persistent output channel. + Ok(stderr_lines) => task.finish(TaskOutcome::Succeeded { + retained_output: stderr_lines.clone(), + }), + Err(error) => task.finish(TaskOutcome::Failed { + message: error.to_string(), + causes: error_causes(error), + }), } - // Map error to include canister context for deferred printing - result.map_err(|error| SyncFailure { - canister_name: canister_info.name.clone(), - canister_id: cid, - error, - progress_output: pb.dump_output(debug), - }) + result.map(|_| ()).map_err(|_| canister_info.name.clone()) } }; futs.push_back(fut); } - // Consume the set of futures and collect errors - let mut errors: Vec = Vec::new(); + // Consume the set of futures and collect the failed canister names; the + // renderer owns displaying each failure's captured output. + let mut failed: Vec = Vec::new(); while let Some(res) = futs.next().await { - if let Err(failure) = res { - errors.push(failure); + if let Err(name) = res { + failed.push(name); } } - if !errors.is_empty() { - // Print all errors in batch - for failure in &errors { - error!( - "----- Failed to sync canister '{}': {} -----", - failure.canister_name, failure.canister_id, - ); - error!("'{}'", failure.error); - { - use std::error::Error; - let mut cause = failure.error.source(); - while let Some(err) = cause { - error!(" caused by: {err}"); - cause = err.source(); - } - } - for line in &failure.progress_output { - error!("{line}"); - } - } - - return SyncOperationSnafu { - names: errors - .iter() - .map(|e| e.canister_name.clone()) - .collect::>(), - } - .fail(); + if !failed.is_empty() { + return SyncOperationSnafu { names: failed }.fail(); } Ok(()) diff --git a/crates/icp-cli/src/progress.rs b/crates/icp-cli/src/progress.rs index 74d9cc4de..9b1a3d7e5 100644 --- a/crates/icp-cli/src/progress.rs +++ b/crates/icp-cli/src/progress.rs @@ -2,9 +2,6 @@ use std::{collections::VecDeque, time::Duration}; use futures::Future; use indicatif::{MultiProgress, ProgressBar as SimpleProgressBar, ProgressStyle}; -use itertools::Itertools; -use tokio::{sync::mpsc, task::JoinHandle}; -use tracing::debug; /// The maximum number of lines to display for a step output pub(crate) const MAX_LINES_PER_STEP: usize = 10_000; @@ -67,11 +64,6 @@ impl RollingLines { pub(crate) fn is_empty(&self) -> bool { self.buf.is_empty() } - - /// Convert the buffer into an iterator (in order). - pub(crate) fn into_iter(self) -> impl Iterator { - self.buf.into_iter() - } } /// Settings for the progress manager @@ -117,21 +109,6 @@ impl ProgressManager { pb } - /// Create a new progress bar for a multi-step operation. - pub(crate) fn create_multi_step_progress_bar( - &self, - canister_name: &str, - output_label: &str, - ) -> MultiStepProgressBar { - MultiStepProgressBar { - progress_bar: self.create_progress_bar(canister_name), - canister_name: canister_name.to_string(), - output_label: output_label.to_string(), - finished_steps: Vec::new(), - in_progress: None, - } - } - /// Execute a task with progress tracking and automatic style updates pub(crate) async fn execute_with_progress( progress_bar: &P, @@ -186,144 +163,12 @@ impl ProgressManager { } } -struct StepOutput { - title: String, - output: Vec, -} - -struct StepInProgress { - title: String, - receiver: JoinHandle>, -} - -pub(crate) struct MultiStepProgressBar { - progress_bar: SimpleProgressBar, - canister_name: String, - output_label: String, - finished_steps: Vec, - in_progress: Option, -} - -impl MultiStepProgressBar { - pub(crate) fn begin_step(&mut self, title: String) -> mpsc::Sender { - if self.in_progress.is_some() { - panic!("step already in progress"); - } - - let (tx, mut rx) = mpsc::channel::(100); - - let set_message = { - let pb = self.progress_bar.clone(); - let title = title.clone(); - - move |msg: String| { - pb.set_message(format!("{title}\n{msg}\n")); - } - }; - - // Handle logging from script commands - let handle = tokio::spawn(async move { - // Small rolling buffer to display current output while build is ongoing - let mut rolling = RollingLines::new(4); - // Total output buffer to display full build output later - let mut complete = RollingLines::new(MAX_LINES_PER_STEP); // We need _some_ limit to prevent consuming infinite memory - - while let Some(line) = rx.recv().await { - debug!("{line}"); - - // Update output buffer - rolling.push(line.clone()); - complete.push(line); - - // Update progress-bar with rolling terminal output - // Make the output - // │ look prettier... - // └ - let msg = rolling.iter().map(|s| format!("│ {s}")).join("\n"); - set_message(format!("{msg}\n└\n")); - } - - complete.into_iter().collect() - }); - - self.in_progress = Some(StepInProgress { - title, - receiver: handle, - }); - - tx - } - - pub(crate) async fn end_step(&mut self) { - let StepInProgress { title, receiver } = - self.in_progress.take().expect("no step in progress"); - let output = receiver.await.unwrap(); - - self.finished_steps.push(StepOutput { title, output }); - } - - /// Dump captured build output. When `all_steps` is true, output from every - /// step is included; otherwise only the last (failing) step is shown. - pub(crate) fn dump_output(&self, all_steps: bool) -> Vec { - let mut lines = Vec::new(); - - lines.push(format!( - "[{}] {} output:", - self.canister_name, self.output_label - )); - - let steps: &[StepOutput] = if all_steps { - &self.finished_steps - } else { - self.finished_steps - .last() - .map(std::slice::from_ref) - .unwrap_or_default() - }; - - for step_output in steps { - for line in step_output.title.lines() { - if !line.is_empty() { - lines.push(format!("[{}] {}:", self.canister_name, line)); - } - } - - if step_output.output.is_empty() { - lines.push(format!("[{}] ", self.canister_name)); - } else { - lines.extend( - step_output - .output - .iter() - .map(|s| format!("[{}] > {s}", self.canister_name)), - ); - } - } - - lines - } -} - pub(crate) trait ProgressBar { fn set_style(&self, style: ProgressStyle); fn set_message(&self, message: String); fn finish(&self); } -impl ProgressBar for MultiStepProgressBar { - fn set_style(&self, style: ProgressStyle) { - self.progress_bar.set_style(style); - } - - fn set_message(&self, message: String) { - self.progress_bar.set_message(message); - } - - fn finish(&self) { - self.progress_bar.finish(); - } -} - impl ProgressBar for SimpleProgressBar { fn set_style(&self, style: ProgressStyle) { SimpleProgressBar::set_style(self, style); diff --git a/crates/icp-cli/src/render/interactive.rs b/crates/icp-cli/src/render/interactive.rs index d9fde82ba..a180caa13 100644 --- a/crates/icp-cli/src/render/interactive.rs +++ b/crates/icp-cli/src/render/interactive.rs @@ -106,19 +106,20 @@ impl InteractiveRenderer { }; match outcome { - TaskOutcome::Succeeded => { + TaskOutcome::Succeeded { retained_output } => { view.bar.set_style(make_style(TICK_SUCCESS, COLOR_SUCCESS)); view.bar.set_message(success_message(view.log.kind())); + view.bar.finish(); + super::print_retained(view.log.kind(), &retained_output); } - TaskOutcome::Failed { message } => { + TaskOutcome::Failed { message, causes } => { view.bar.set_style(make_style(TICK_FAILURE, COLOR_FAILURE)); view.bar .set_message(failure_message(view.log.kind(), &message)); - view.log.fail(message); + view.bar.finish(); + view.log.fail(message, causes); } } - - view.bar.finish(); } } } diff --git a/crates/icp-cli/src/render/mod.rs b/crates/icp-cli/src/render/mod.rs index 2d2dfcfeb..f3ba36688 100644 --- a/crates/icp-cli/src/render/mod.rs +++ b/crates/icp-cli/src/render/mod.rs @@ -64,6 +64,7 @@ impl Renderer { fn step_header(kind: &TaskKind, number: usize, total: usize, label: &str) -> String { match kind { TaskKind::Build { .. } => format!("Building: step {number} of {total} {label}"), + TaskKind::Sync { .. } => format!("\nSyncing: {label} {number} of {total}"), } } @@ -71,6 +72,7 @@ fn step_header(kind: &TaskKind, number: usize, total: usize, label: &str) -> Str fn output_label(kind: &TaskKind) -> &'static str { match kind { TaskKind::Build { .. } => "Build", + TaskKind::Sync { .. } => "Sync", } } @@ -78,6 +80,7 @@ fn output_label(kind: &TaskKind) -> &'static str { fn success_message(kind: &TaskKind) -> String { match kind { TaskKind::Build { .. } => "Built successfully".to_owned(), + TaskKind::Sync { canister_id, .. } => format!("Synced successfully: {canister_id}"), } } @@ -85,6 +88,7 @@ fn success_message(kind: &TaskKind) -> String { fn failure_message(kind: &TaskKind, message: &str) -> String { match kind { TaskKind::Build { .. } => format!("Failed to build canister: {message}"), + TaskKind::Sync { .. } => format!("Failed to sync canister: {message}"), } } @@ -94,6 +98,18 @@ fn failure_header(kind: &TaskKind) -> String { TaskKind::Build { canister } => { format!("----- Failed to build canister '{canister}' -----") } + TaskKind::Sync { + canister, + canister_id, + } => format!("----- Failed to sync canister '{canister}': {canister_id} -----"), + } +} + +/// Print output lines a task retained past its rolling step view (e.g. +/// sync-plugin stderr), prefixed with the canister name. +fn print_retained(kind: &TaskKind, lines: &[String]) { + for line in lines { + eprintln!("[{}] {line}", kind.canister()); } } @@ -103,7 +119,7 @@ pub(super) struct TaskLog { kind: TaskKind, finished_steps: Vec, current_step: Option, - failure: Option, + failure: Option, } struct StepLog { @@ -111,6 +127,11 @@ struct StepLog { lines: RollingLines, } +struct Failure { + message: String, + causes: Vec, +} + impl TaskLog { fn new(kind: TaskKind) -> Self { Self { @@ -146,8 +167,8 @@ impl TaskLog { } } - fn fail(&mut self, message: String) { - self.failure = Some(message); + fn fail(&mut self, message: String, causes: Vec) { + self.failure = Some(Failure { message, causes }); } /// Render the captured output. When `all_steps` is true, output from @@ -189,12 +210,15 @@ impl TaskLog { /// Print the failure dump for every failed task, in task-creation order. fn dump_failures(logs: &BTreeMap, all_steps: bool) { for log in logs.values() { - let Some(message) = &log.failure else { + let Some(failure) = &log.failure else { continue; }; error!("{}", failure_header(&log.kind)); - error!("'{message}'"); + error!("'{}'", failure.message); + for cause in &failure.causes { + error!(" caused by: {cause}"); + } for line in log.dump(all_steps) { error!("{line}"); } diff --git a/crates/icp-cli/src/render/plain.rs b/crates/icp-cli/src/render/plain.rs index 6c6b5f913..31d7bf7d4 100644 --- a/crates/icp-cli/src/render/plain.rs +++ b/crates/icp-cli/src/render/plain.rs @@ -56,8 +56,13 @@ impl PlainRenderer { let Some(log) = self.tasks.get_mut(&event.task_id) else { return; }; - if let TaskOutcome::Failed { message } = outcome { - log.fail(message); + match outcome { + TaskOutcome::Succeeded { retained_output } => { + super::print_retained(log.kind(), &retained_output); + } + TaskOutcome::Failed { message, causes } => { + log.fail(message, causes); + } } } } diff --git a/crates/icp-cli/tests/sync_tests.rs b/crates/icp-cli/tests/sync_tests.rs index 31f72ef48..44fecadc8 100644 --- a/crates/icp-cli/tests/sync_tests.rs +++ b/crates/icp-cli/tests/sync_tests.rs @@ -205,7 +205,7 @@ async fn sync_aborts_when_canister_not_running() { // sync aborts early with an actionable message; the `echo "syncing"` step // never runs, so its runtime progress output must not appear. (The `--debug` // config dump echoes the step's command text, so we check for the runtime - // `DEBUG icp::progress: syncing` marker rather than the bare word "syncing".) + // `DEBUG icp::render::plain: syncing` marker rather than the bare word "syncing".) ctx.icp() .current_dir(&project_dir) .env("NO_COLOR", "1") @@ -221,7 +221,7 @@ async fn sync_aborts_when_canister_not_running() { .stderr( contains("asset sync requires it to be Running") .and(contains("icp canister start")) - .and(contains("DEBUG icp::progress: syncing").not()), + .and(contains("DEBUG icp::render::plain: syncing").not()), ); } @@ -387,9 +387,9 @@ async fn sync_multiple_canisters() { .success() .stderr(contains("Syncing canisters")) .stderr(contains(r#"canisters: ["canister-a", "canister-b"]"#)) - .stderr(contains("DEBUG icp::progress: syncing canister-a")) - .stderr(contains("DEBUG icp::progress: syncing canister-b")) - .stderr(contains("DEBUG icp::progress: syncing canister-c").not()); + .stderr(contains("DEBUG icp::render::plain: syncing canister-a")) + .stderr(contains("DEBUG icp::render::plain: syncing canister-b")) + .stderr(contains("DEBUG icp::render::plain: syncing canister-c").not()); } #[tokio::test] @@ -861,7 +861,7 @@ async fn sync_all_canisters_in_environment() { .stderr(contains("Syncing canisters")) .stderr(contains(r#"canisters: []"#)) .stderr(contains(r#"environment: Some("test-env")"#)) - .stderr(contains("DEBUG icp::progress: syncing canister-a")) - .stderr(contains("DEBUG icp::progress: syncing canister-b")) - .stderr(contains("DEBUG icp::progress: syncing canister-c").not()); // not in test-env + .stderr(contains("DEBUG icp::render::plain: syncing canister-a")) + .stderr(contains("DEBUG icp::render::plain: syncing canister-b")) + .stderr(contains("DEBUG icp::render::plain: syncing canister-c").not()); // not in test-env } diff --git a/crates/icp-events/Cargo.toml b/crates/icp-events/Cargo.toml index 40acc73b3..fb7adb41f 100644 --- a/crates/icp-events/Cargo.toml +++ b/crates/icp-events/Cargo.toml @@ -6,5 +6,6 @@ license = { workspace = true } publish.workspace = true [dependencies] +candid = { workspace = true } serde = { workspace = true } -tokio = { workspace = true, features = ["sync", "rt"] } +tokio = { workspace = true, features = ["sync"] } diff --git a/crates/icp-events/src/lib.rs b/crates/icp-events/src/lib.rs index e22e8336e..198d99caa 100644 --- a/crates/icp-events/src/lib.rs +++ b/crates/icp-events/src/lib.rs @@ -17,6 +17,7 @@ use std::sync::{ atomic::{AtomicU64, Ordering}, }; +use candid::Principal; use serde::Serialize; use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel}; @@ -63,7 +64,13 @@ pub enum EventKind { #[derive(Debug, Clone, Serialize)] #[serde(tag = "kind", rename_all = "snake_case")] pub enum TaskKind { - Build { canister: String }, + Build { + canister: String, + }, + Sync { + canister: String, + canister_id: Principal, + }, } impl TaskKind { @@ -71,6 +78,7 @@ impl TaskKind { pub fn canister(&self) -> &str { match self { TaskKind::Build { canister } => canister, + TaskKind::Sync { canister, .. } => canister, } } } @@ -95,14 +103,54 @@ pub enum StepOutcome { #[derive(Debug, Clone, Serialize)] #[serde(tag = "result", rename_all = "snake_case")] pub enum TaskOutcome { - Succeeded, - /// `message` is for display only; the typed error stays on the - /// operation's return path. + Succeeded { + /// Output lines that belong on the persistent output channel after + /// success — e.g. sync-plugin stderr, which the rolling step view + /// would otherwise discard. Most tasks retain nothing. + #[serde(skip_serializing_if = "Vec::is_empty")] + retained_output: Vec, + }, + /// Failure descriptions are for display only; the typed error stays on + /// the operation's return path. Failed { message: String, + /// The rendered `source()` chain of the failure, outermost first. + #[serde(skip_serializing_if = "Vec::is_empty")] + causes: Vec, }, } +impl TaskOutcome { + /// Success with nothing retained. + pub fn succeeded() -> Self { + TaskOutcome::Succeeded { + retained_output: Vec::new(), + } + } + + /// Failure with no cause chain. + pub fn failed(message: impl Into) -> Self { + TaskOutcome::Failed { + message: message.into(), + causes: Vec::new(), + } + } +} + +/// Create a lone [`StepReporter`] wired to its own receiver, for callers that +/// need to observe a single step's output without the task/step ceremony — +/// primarily tests. +pub fn step_channel() -> (StepReporter, UnboundedReceiver) { + let (tx, rx) = unbounded_channel(); + ( + StepReporter { + tx: Some(tx), + task_id: TaskId(0), + }, + rx, + ) +} + /// Create a connected reporter/receiver pair. The receiver yields `None` once /// the reporter and every handle derived from it have been dropped. pub fn channel() -> (Reporter, UnboundedReceiver) { @@ -213,28 +261,6 @@ impl StepReporter { } } - /// TEMPORARY: adapt a legacy line channel into a [`StepReporter`]. - /// - /// Output lines emitted through the returned reporter are forwarded to - /// `lines` by a background task, which drains and exits once the reporter - /// and all of its clones are dropped. Used by call sites that still - /// render through the old progress-bar channel (sync); remove once those - /// paths take a real [`Reporter`]. - pub fn bridge_lines(lines: tokio::sync::mpsc::Sender) -> Self { - let (tx, mut rx) = unbounded_channel::(); - tokio::spawn(async move { - while let Some(event) = rx.recv().await { - if let EventKind::Output { line, .. } = event.kind { - let _ = lines.send(line).await; - } - } - }); - Self { - tx: Some(tx), - task_id: TaskId(0), - } - } - fn send(&self, kind: EventKind) { if let Some(tx) = &self.tx { let _ = tx.send(Event { diff --git a/crates/icp-sync-plugin/Cargo.toml b/crates/icp-sync-plugin/Cargo.toml index 216e9c761..15e3c8baf 100644 --- a/crates/icp-sync-plugin/Cargo.toml +++ b/crates/icp-sync-plugin/Cargo.toml @@ -15,6 +15,7 @@ console.workspace = true hex.workspace = true ic-agent.workspace = true icp-canister-interfaces.workspace = true +icp-events.workspace = true snafu.workspace = true tokio.workspace = true wasmtime.workspace = true diff --git a/crates/icp-sync-plugin/src/runtime.rs b/crates/icp-sync-plugin/src/runtime.rs index fb284fb7f..86fe4566c 100644 --- a/crates/icp-sync-plugin/src/runtime.rs +++ b/crates/icp-sync-plugin/src/runtime.rs @@ -25,8 +25,9 @@ use camino::Utf8PathBuf; use candid::{Encode, Principal}; use ic_agent::Agent; use snafu::prelude::*; +// Aliased because wasmtime-wasi also has an `OutputStream` (imported below). +use icp_events::{OutputStream as EventStream, StepReporter}; use tokio::io::{self, AsyncWrite}; -use tokio::sync::mpsc::Sender; use wasmtime_wasi::cli::{IsTerminal, StdoutStream}; use wasmtime_wasi::p2::{OutputStream, Pollable, StreamError}; use wasmtime_wasi::{DirPerms, FilePerms}; @@ -213,7 +214,7 @@ pub fn run_plugin( identity_principal: Principal, environment: String, compute_limit_secs: u64, - stdio: Option>, + reporter: StepReporter, ) -> Result, RunPluginError> { use wasmtime::component::{Component, Linker}; use wasmtime::{Config, Engine, Store}; @@ -299,8 +300,13 @@ pub fn run_plugin( } let persistent_stderr: Arc>> = Arc::default(); - let stdout_capture = LineCapture::new("stdout", stdio.clone(), None); - let stderr_capture = LineCapture::new("stderr", stdio.clone(), Some(persistent_stderr.clone())); + let stdout_capture = LineCapture::new("stdout", EventStream::Stdout, reporter.clone(), None); + let stderr_capture = LineCapture::new( + "stderr", + EventStream::Stderr, + reporter.clone(), + Some(persistent_stderr.clone()), + ); wasi_builder .stdout(stdout_capture.clone()) .stderr(stderr_capture.clone()); @@ -376,9 +382,9 @@ pub fn run_plugin( // `LineCapture` implements both `StdoutStream` (so it can be installed on a // `WasiCtxBuilder`) and `OutputStream` / `AsyncWrite` (so the bytes written // by the guest flow through the same code path). Each write is split on -// newlines; complete lines have ANSI escapes stripped and are pushed to the -// rolling-view `Sender` via `try_send` (best-effort). For stderr, -// the same lines are also appended to `persistent`, which is drained by +// newlines; complete lines have ANSI escapes stripped and are emitted as +// output events on the step reporter (non-blocking). For stderr, the same +// lines are also appended to `persistent`, which is drained by // `run_plugin()` after `exec()` returns. Total accepted bytes are capped at // `MAX_PLUGIN_OUTPUT` per stream; further bytes are dropped and `finalize` // emits a single "… N bytes of