diff --git a/Cargo.lock b/Cargo.lock index 67ca726a3..2260d4dc8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3637,6 +3637,7 @@ dependencies = [ "ic-management-canister-types 0.8.0", "ic-utils", "icp-canister-interfaces", + "icp-events", "icp-sync-plugin", "icrc-ledger-types", "indexmap", @@ -3728,6 +3729,7 @@ dependencies = [ "ic-utils", "icp", "icp-canister-interfaces", + "icp-events", "icrc-ledger-types", "indicatif", "indoc", @@ -3772,6 +3774,15 @@ dependencies = [ "wslpath2", ] +[[package]] +name = "icp-events" +version = "1.3.0" +dependencies = [ + "candid", + "serde", + "tokio", +] + [[package]] name = "icp-sync-plugin" version = "1.3.0" @@ -3785,6 +3796,7 @@ dependencies = [ "hex", "ic-agent", "icp-canister-interfaces", + "icp-events", "snafu", "tokio", "wasmtime", diff --git a/Cargo.toml b/Cargo.toml index 30710a835..019341003 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 3b85f287b..ed0fcf1ed 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..6517ef437 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::rendered, }; /// Build canisters @@ -57,14 +58,18 @@ pub(crate) async fn exec(ctx: &Context, args: &BuildArgs) -> Result<(), anyhow:: // Build the selected canisters info!("Building canisters:"); - build_many_with_progress_bar( - canisters_to_build, - environment_selection.name(), - ctx.builder.clone(), - ctx.artifacts.clone(), - &ctx.dirs.package_cache()?, - ctx.debug, - ) + let pkg_cache = ctx.dirs.package_cache()?; + rendered(ctx.debug, async |reporter| { + build_many( + canisters_to_build, + environment_selection.name(), + ctx.builder.clone(), + ctx.artifacts.clone(), + &pkg_cache, + reporter, + ) + .await + }) .await?; info!("Canisters built successfully"); diff --git a/crates/icp-cli/src/commands/canister/snapshot/download.rs b/crates/icp-cli/src/commands/canister/snapshot/download.rs index d8b3aa661..4266303b9 100644 --- a/crates/icp-cli/src/commands/canister/snapshot/download.rs +++ b/crates/icp-cli/src/commands/canister/snapshot/download.rs @@ -5,14 +5,17 @@ use icp::context::Context; use icp::prelude::*; use tracing::info; +use icp_events::{TaskKind, TransferBlob, TransferDirection}; + use super::SnapshotId; use crate::commands::args; use crate::operations::misc::format_timestamp; use crate::operations::snapshot_transfer::{ - BlobType, SnapshotPaths, SnapshotTransferError, create_transfer_progress_bar, - delete_download_progress, download_blob_to_file, download_wasm_chunk, load_download_progress, - load_metadata, read_snapshot_metadata, save_metadata, + BlobType, SnapshotPaths, SnapshotTransferError, delete_download_progress, + download_blob_to_file, download_wasm_chunk, load_download_progress, load_metadata, + read_snapshot_metadata, save_metadata, }; +use crate::render::rendered_task; /// Download a snapshot to local disk #[derive(Debug, Args)] @@ -121,20 +124,30 @@ pub(crate) async fn exec(ctx: &Context, args: &DownloadArgs) -> Result<(), anyho // Download WASM module if metadata.wasm_module_size > 0 { if !progress.wasm_module.is_complete(metadata.wasm_module_size) { - let pb = create_transfer_progress_bar(metadata.wasm_module_size, "WASM module"); - download_blob_to_file( - &agent, - args.proxy, - cid, - snapshot_id, - BlobType::WasmModule, - metadata.wasm_module_size, - paths, - &mut progress, - &pb, + rendered_task( + ctx.debug, + TaskKind::SnapshotTransfer { + canister: name.to_string(), + direction: TransferDirection::Download, + blob: TransferBlob::WasmModule, + total_bytes: metadata.wasm_module_size, + }, + async |task| { + download_blob_to_file( + &agent, + args.proxy, + cid, + snapshot_id, + BlobType::WasmModule, + metadata.wasm_module_size, + paths, + &mut progress, + task, + ) + .await + }, ) .await?; - pb.finish_with_message("done"); } else { info!("WASM module: already complete"); } @@ -143,20 +156,30 @@ pub(crate) async fn exec(ctx: &Context, args: &DownloadArgs) -> Result<(), anyho // Download WASM memory if metadata.wasm_memory_size > 0 { if !progress.wasm_memory.is_complete(metadata.wasm_memory_size) { - let pb = create_transfer_progress_bar(metadata.wasm_memory_size, "WASM memory"); - download_blob_to_file( - &agent, - args.proxy, - cid, - snapshot_id, - BlobType::WasmMemory, - metadata.wasm_memory_size, - paths, - &mut progress, - &pb, + rendered_task( + ctx.debug, + TaskKind::SnapshotTransfer { + canister: name.to_string(), + direction: TransferDirection::Download, + blob: TransferBlob::WasmMemory, + total_bytes: metadata.wasm_memory_size, + }, + async |task| { + download_blob_to_file( + &agent, + args.proxy, + cid, + snapshot_id, + BlobType::WasmMemory, + metadata.wasm_memory_size, + paths, + &mut progress, + task, + ) + .await + }, ) .await?; - pb.finish_with_message("done"); } else { info!("WASM memory: already complete"); } @@ -168,21 +191,30 @@ pub(crate) async fn exec(ctx: &Context, args: &DownloadArgs) -> Result<(), anyho .stable_memory .is_complete(metadata.stable_memory_size) { - let pb = - create_transfer_progress_bar(metadata.stable_memory_size, "Stable memory"); - download_blob_to_file( - &agent, - args.proxy, - cid, - snapshot_id, - BlobType::StableMemory, - metadata.stable_memory_size, - paths, - &mut progress, - &pb, + rendered_task( + ctx.debug, + TaskKind::SnapshotTransfer { + canister: name.to_string(), + direction: TransferDirection::Download, + blob: TransferBlob::StableMemory, + total_bytes: metadata.stable_memory_size, + }, + async |task| { + download_blob_to_file( + &agent, + args.proxy, + cid, + snapshot_id, + BlobType::StableMemory, + metadata.stable_memory_size, + paths, + &mut progress, + task, + ) + .await + }, ) .await?; - pb.finish_with_message("done"); } else { info!("Stable memory: already complete"); } diff --git a/crates/icp-cli/src/commands/canister/snapshot/upload.rs b/crates/icp-cli/src/commands/canister/snapshot/upload.rs index a8d97f2c4..3ff674e53 100644 --- a/crates/icp-cli/src/commands/canister/snapshot/upload.rs +++ b/crates/icp-cli/src/commands/canister/snapshot/upload.rs @@ -8,14 +8,17 @@ use icp::prelude::*; use serde::Serialize; use tracing::info; +use icp_events::{TaskKind, TransferBlob, TransferDirection}; + use super::SnapshotId; use crate::commands::args; use crate::operations::misc::format_timestamp; use crate::operations::snapshot_transfer::{ - BlobType, SnapshotPaths, SnapshotTransferError, UploadProgress, create_transfer_progress_bar, - delete_upload_progress, load_metadata, load_upload_progress, save_upload_progress, - upload_blob_from_file, upload_snapshot_metadata, upload_wasm_chunk, + BlobType, SnapshotPaths, SnapshotTransferError, UploadProgress, delete_upload_progress, + load_metadata, load_upload_progress, save_upload_progress, upload_blob_from_file, + upload_snapshot_metadata, upload_wasm_chunk, }; +use crate::render::rendered_task; /// Upload a snapshot from local disk #[derive(Debug, Args)] @@ -126,19 +129,29 @@ pub(crate) async fn exec(ctx: &Context, args: &UploadArgs) -> Result<(), anyhow: // Upload WASM module if metadata.wasm_module_size > 0 { if progress.wasm_module_offset < metadata.wasm_module_size { - let pb = create_transfer_progress_bar(metadata.wasm_module_size, "WASM module"); - upload_blob_from_file( - &agent, - args.proxy, - cid, - &snapshot_id_bytes, - BlobType::WasmModule, - paths, - &mut progress, - &pb, + rendered_task( + ctx.debug, + TaskKind::SnapshotTransfer { + canister: name.to_string(), + direction: TransferDirection::Upload, + blob: TransferBlob::WasmModule, + total_bytes: metadata.wasm_module_size, + }, + async |task| { + upload_blob_from_file( + &agent, + args.proxy, + cid, + &snapshot_id_bytes, + BlobType::WasmModule, + paths, + &mut progress, + task, + ) + .await + }, ) .await?; - pb.finish_with_message("done"); } else { info!("WASM module: already complete"); } @@ -147,19 +160,29 @@ pub(crate) async fn exec(ctx: &Context, args: &UploadArgs) -> Result<(), anyhow: // Upload WASM memory if metadata.wasm_memory_size > 0 { if progress.wasm_memory_offset < metadata.wasm_memory_size { - let pb = create_transfer_progress_bar(metadata.wasm_memory_size, "WASM memory"); - upload_blob_from_file( - &agent, - args.proxy, - cid, - &snapshot_id_bytes, - BlobType::WasmMemory, - paths, - &mut progress, - &pb, + rendered_task( + ctx.debug, + TaskKind::SnapshotTransfer { + canister: name.to_string(), + direction: TransferDirection::Upload, + blob: TransferBlob::WasmMemory, + total_bytes: metadata.wasm_memory_size, + }, + async |task| { + upload_blob_from_file( + &agent, + args.proxy, + cid, + &snapshot_id_bytes, + BlobType::WasmMemory, + paths, + &mut progress, + task, + ) + .await + }, ) .await?; - pb.finish_with_message("done"); } else { info!("WASM memory: already complete"); } @@ -168,20 +191,29 @@ pub(crate) async fn exec(ctx: &Context, args: &UploadArgs) -> Result<(), anyhow: // Upload stable memory if metadata.stable_memory_size > 0 { if progress.stable_memory_offset < metadata.stable_memory_size { - let pb = - create_transfer_progress_bar(metadata.stable_memory_size, "Stable memory"); - upload_blob_from_file( - &agent, - args.proxy, - cid, - &snapshot_id_bytes, - BlobType::StableMemory, - paths, - &mut progress, - &pb, + rendered_task( + ctx.debug, + TaskKind::SnapshotTransfer { + canister: name.to_string(), + direction: TransferDirection::Upload, + blob: TransferBlob::StableMemory, + total_bytes: metadata.stable_memory_size, + }, + async |task| { + upload_blob_from_file( + &agent, + args.proxy, + cid, + &snapshot_id_bytes, + BlobType::StableMemory, + paths, + &mut progress, + task, + ) + .await + }, ) .await?; - pb.finish_with_message("done"); } else { info!("Stable memory: already complete"); } diff --git a/crates/icp-cli/src/commands/deploy.rs b/crates/icp-cli/src/commands/deploy.rs index b2a1fe4ca..5983bb7cc 100644 --- a/crates/icp-cli/src/commands/deploy.rs +++ b/crates/icp-cli/src/commands/deploy.rs @@ -12,6 +12,7 @@ use icp::{ network::Configuration as NetworkConfiguration, }; use icp_canister_interfaces::candid_ui::MAINNET_CANDID_UI_CID; +use icp_events::{TaskKind, TaskOutcome}; use itertools::Itertools; use serde::Serialize; use std::collections::{BTreeMap, BTreeSet, HashSet}; @@ -23,7 +24,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}, @@ -32,7 +33,7 @@ use crate::{ sync::sync_many, }, options::{IdentityOpt, arg_struct_change_help}, - progress::{ProgressManager, ProgressManagerSettings}, + render::rendered, }; /// Deploy a project to an environment @@ -182,14 +183,18 @@ pub(crate) async fn exec(ctx: &Context, args: &DeployArgs) -> Result<(), anyhow: // Build the selected canisters info!("Building canisters:"); - build_many_with_progress_bar( - canisters_to_build, - environment_selection.name(), - ctx.builder.clone(), - ctx.artifacts.clone(), - &ctx.dirs.package_cache()?, - ctx.debug, - ) + let pkg_cache = ctx.dirs.package_cache()?; + rendered(ctx.debug, async |reporter| { + build_many( + canisters_to_build, + environment_selection.name(), + ctx.builder.clone(), + ctx.artifacts.clone(), + &pkg_cache, + reporter, + ) + .await + }) .await?; // Ensure the selected canisters exist, creating any that are missing. @@ -230,61 +235,65 @@ pub(crate) async fn exec(ctx: &Context, args: &DeployArgs) -> Result<(), anyhow: CreateFunding::Cycles(args.cycles.get()), existing_canisters.into_values().collect(), ); - let mut futs = FuturesOrdered::new(); - let progress_manager = ProgressManager::new(ProgressManagerSettings { hidden: ctx.debug }); - for name in canisters_to_create.iter() { - let pb = progress_manager.create_progress_bar(name); - pb.set_message("Creating..."); - let create_op = create_operation.clone(); - let (_, canister_info) = env.get_canister_info(name).map_err(|e| anyhow!(e))?; - futs.push_back(async move { - ProgressManager::execute_with_custom_progress( - &pb, - create_op.create(&canister_info.settings.into()), - || "Created successfully".to_string(), - |err: &_| err.to_string(), - |_| false, - ) - .await - }); - } - - // Cache errors until all futures are processed. Otherwise we risk dropping a canister id. - let mut error: Option = None; - let mut idx = 0; - while let Some(res) = futs.next().await { - match res { - Ok(id) => { - let canister_name = canisters_to_create - .get(idx) - .expect("should have tried to create every canister"); - if !args.json { - println!("Created canister {canister_name} with ID {id}"); + rendered(ctx.debug, async |reporter| { + let mut futs = FuturesOrdered::new(); + for name in canisters_to_create.iter() { + let task = reporter.task(TaskKind::Create { + canister: (*name).clone(), + }); + let create_op = create_operation.clone(); + let (_, canister_info) = env.get_canister_info(name).map_err(|e| anyhow!(e))?; + futs.push_back(async move { + let result = create_op.create(&canister_info.settings.into()).await; + + match &result { + Ok(_) => task.finish(TaskOutcome::succeeded()), + Err(err) => task.finish(TaskOutcome::failed(err.to_string())), } - ctx.set_canister_id_for_env(canister_name, id, &environment_selection) + + result + }); + } + + // Cache errors until all futures are processed. Otherwise we risk dropping a canister id. + let mut error: Option = None; + let mut idx = 0; + while let Some(res) = futs.next().await { + match res { + Ok(id) => { + let canister_name = canisters_to_create + .get(idx) + .expect("should have tried to create every canister"); + if !args.json { + println!("Created canister {canister_name} with ID {id}"); + } + ctx.set_canister_id_for_env(canister_name, id, &environment_selection) + .await + .map_err(|e| anyhow!(e))?; + // Apply controller settings for any already-created canister that was + // waiting for this one to exist (e.g. created via `icp canister create`). + sync_controller_dependents( + ctx, + &agent, + args.proxy, + canister_name, + &environment_selection, + ) .await .map_err(|e| anyhow!(e))?; - // Apply controller settings for any already-created canister that was - // waiting for this one to exist (e.g. created via `icp canister create`). - sync_controller_dependents( - ctx, - &agent, - args.proxy, - canister_name, - &environment_selection, - ) - .await - .map_err(|e| anyhow!(e))?; - } - Err(err) => { - error = Some(err.into()); + } + Err(err) => { + error = Some(err.into()); + } } + idx += 1; } - idx += 1; - } - if let Some(err) = error { - return Err(err); - } + if let Some(err) = error { + return Err(err); + } + Ok(()) + }) + .await?; } ctx.update_custom_domains(&environment_selection).await; @@ -319,24 +328,30 @@ pub(crate) async fn exec(ctx: &Context, args: &DeployArgs) -> Result<(), anyhow: .await .map_err(|e| anyhow!(e))?; - set_binding_env_vars_many( - agent.clone(), - args.proxy, - &env.name, - target_canisters.clone(), - canister_list.clone(), - ctx.debug, - ) + rendered(ctx.debug, async |reporter| { + set_binding_env_vars_many( + agent.clone(), + args.proxy, + &env.name, + target_canisters.clone(), + canister_list.clone(), + reporter, + ) + .await + }) .await .map_err(|e| anyhow!(e))?; - sync_settings_many( - agent.clone(), - args.proxy, - target_canisters, - canister_list, - ctx.debug, - ) + rendered(ctx.debug, async |reporter| { + sync_settings_many( + agent.clone(), + args.proxy, + target_canisters, + canister_list, + reporter, + ) + .await + }) .await .map_err(|e| anyhow!(e))?; @@ -379,27 +394,33 @@ pub(crate) async fn exec(ctx: &Context, args: &DeployArgs) -> Result<(), anyhow: if !args.yes { info!("Checking compatibility:"); - check_candid_compatibility_many( - agent.clone(), - canisters - .iter() - .map(|(name, cid, mode, _, _)| (&**name, *cid, *mode)), - ctx.artifacts.clone(), - ctx.debug, - ) + rendered(ctx.debug, async |reporter| { + check_candid_compatibility_many( + agent.clone(), + canisters + .iter() + .map(|(name, cid, mode, _, _)| (&**name, *cid, *mode)), + ctx.artifacts.clone(), + reporter, + ) + .await + }) .await .map_err(|e| anyhow!(e))?; } info!("Installing canisters:"); - install_many( - agent.clone(), - args.proxy, - canisters, - ctx.artifacts.clone(), - ctx.debug, - ) + rendered(ctx.debug, async |reporter| { + install_many( + agent.clone(), + args.proxy, + canisters, + ctx.artifacts.clone(), + reporter, + ) + .await + }) .await?; // Sync the selected canisters @@ -490,17 +511,21 @@ pub(crate) async fn exec(ctx: &Context, args: &DeployArgs) -> Result<(), anyhow: .collect(); let pkg_cache = ctx.dirs.package_cache()?; - sync_many( - ctx.syncer.clone(), - agent.clone(), - sync_canisters, - environment_selection.name().to_owned(), - env.network.name.clone(), - canister_ids, - args.proxy, - ctx.debug, - &pkg_cache, - ) + + rendered(ctx.debug, async |reporter| { + sync_many( + ctx.syncer.clone(), + agent.clone(), + sync_canisters, + environment_selection.name().to_owned(), + env.network.name.clone(), + canister_ids, + args.proxy, + &pkg_cache, + reporter, + ) + .await + }) .await?; } diff --git a/crates/icp-cli/src/commands/network/start.rs b/crates/icp-cli/src/commands/network/start.rs index f06942f22..8c7e6764f 100644 --- a/crates/icp-cli/src/commands/network/start.rs +++ b/crates/icp-cli/src/commands/network/start.rs @@ -22,7 +22,7 @@ use icp::{ }; use tracing::{debug, info, warn}; -use crate::progress::{ProgressManager, ProgressManagerSettings}; +use crate::render::{ProgressManager, ProgressManagerSettings}; use super::args::NetworkOrEnvironmentArgs; use icp::context::Context; diff --git a/crates/icp-cli/src/commands/network/update.rs b/crates/icp-cli/src/commands/network/update.rs index 7ff1c0807..f5e960843 100644 --- a/crates/icp-cli/src/commands/network/update.rs +++ b/crates/icp-cli/src/commands/network/update.rs @@ -3,7 +3,7 @@ use std::sync::{Arc, OnceLock}; use clap::Parser; use icp::{context::Context, network::managed::cache::download_launcher_version}; -use crate::progress::{ProgressManager, ProgressManagerSettings}; +use crate::render::{ProgressManager, ProgressManagerSettings}; /// Update icp-cli-network-launcher to the latest version. #[derive(Parser, Debug)] diff --git a/crates/icp-cli/src/commands/project/bundle.rs b/crates/icp-cli/src/commands/project/bundle.rs index b9217c0b8..0c17d81df 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::rendered}; /// Bundle a project into a self-contained deployable archive. /// @@ -30,16 +30,20 @@ pub(crate) async fn exec(ctx: &Context, args: &BundleArgs) -> Result<(), anyhow: let canisters: Vec<_> = project.canisters.into_values().collect(); - create_bundle( - &project.dir, - canisters, - &args.environment, - ctx.builder.clone(), - ctx.artifacts.clone(), - &ctx.dirs.package_cache()?, - ctx.debug, - &args.output, - ) + let pkg_cache = ctx.dirs.package_cache()?; + rendered(ctx.debug, async |reporter| { + create_bundle( + &project.dir, + canisters, + &args.environment, + ctx.builder.clone(), + ctx.artifacts.clone(), + &pkg_cache, + reporter, + &args.output, + ) + .await + }) .await .context("failed to create bundle")?; diff --git a/crates/icp-cli/src/commands/sync.rs b/crates/icp-cli/src/commands/sync.rs index ac2d1e099..6421535b3 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::rendered, }; /// Synchronize canisters @@ -125,17 +126,21 @@ pub(crate) async fn exec(ctx: &Context, args: &SyncArgs) -> Result<(), anyhow::E .collect(); let pkg_cache = ctx.dirs.package_cache()?; - sync_many( - ctx.syncer.clone(), - agent, - sync_canisters, - environment_selection.name().to_owned(), - env.network.name.clone(), - canister_ids, - args.proxy, - ctx.debug, - &pkg_cache, - ) + + rendered(ctx.debug, async |reporter| { + sync_many( + ctx.syncer.clone(), + agent, + sync_canisters, + environment_selection.name().to_owned(), + env.network.name.clone(), + canister_ids, + args.proxy, + &pkg_cache, + reporter, + ) + .await + }) .await?; Ok(()) diff --git a/crates/icp-cli/src/main.rs b/crates/icp-cli/src/main.rs index 66047c977..1173a4b3b 100644 --- a/crates/icp-cli/src/main.rs +++ b/crates/icp-cli/src/main.rs @@ -20,7 +20,7 @@ mod dist; mod logging; pub(crate) mod operations; mod options; -mod progress; +mod render; mod telemetry; mod version; diff --git a/crates/icp-cli/src/operations/binding_env_vars.rs b/crates/icp-cli/src/operations/binding_env_vars.rs index a2118f193..5e1296ef3 100644 --- a/crates/icp-cli/src/operations/binding_env_vars.rs +++ b/crates/icp-cli/src/operations/binding_env_vars.rs @@ -4,11 +4,10 @@ use futures::{StreamExt, stream::FuturesOrdered}; use ic_agent::{Agent, export::Principal}; use ic_management_canister_types::{CanisterSettings, EnvironmentVariable, UpdateSettingsArgs}; use icp::Canister; +use icp_events::{Reporter, TaskKind, TaskOutcome}; use snafu::Snafu; use tracing::error; -use crate::progress::{ProgressManager, ProgressManagerSettings}; - use super::proxy::UpdateOrProxyError; use super::proxy_management; @@ -30,13 +29,6 @@ pub struct SetBindingEnvVarsManyError { names: Vec, } -/// Holds error information from a failed environment variable update operation -struct BindingEnvVarsFailure { - canister_name: String, - canister_id: Principal, - error: BindingEnvVarsOperationError, -} - pub(crate) async fn set_env_vars_for_canister( agent: &Agent, proxy: Option, @@ -77,14 +69,14 @@ pub(crate) async fn set_env_vars_for_canister( Ok(()) } -/// Orchestrates setting environment variables for multiple canisters with progress tracking +/// Orchestrates setting environment variables for multiple canisters concurrently. pub(crate) async fn set_binding_env_vars_many( agent: Agent, proxy: Option, environment_name: &str, target_canisters: Vec<(Principal, Canister)>, canister_list: BTreeMap, - debug: bool, + reporter: &Reporter, ) -> Result<(), SetBindingEnvVarsManyError> { // Check that all the canisters in this environment have an id // We need to have all the ids to generate environment variables @@ -117,11 +109,12 @@ pub(crate) async fn set_binding_env_vars_many( } let mut futs = FuturesOrdered::new(); - let progress_manager = ProgressManager::new(ProgressManagerSettings { hidden: debug }); for (cid, info) in target_canisters { - let pb = progress_manager.create_progress_bar(&info.name); - let canister_name = info.name.clone(); + let task = reporter.task(TaskKind::UpdateEnvironmentVariables { + canister: info.name.clone(), + canister_id: cid, + }); // Each canister receives only the ids it is wired to (its own project's // canisters by their local names, plus any declared dependencies under @@ -141,59 +134,30 @@ pub(crate) async fn set_binding_env_vars_many( }) .collect(); - let settings_fn = { - let agent = agent.clone(); - let pb = pb.clone(); + let agent = agent.clone(); + futs.push_back(async move { + let result = set_env_vars_for_canister(&agent, proxy, &cid, &info, &binding_vars).await; - async move { - pb.set_message("Updating environment variables..."); - set_env_vars_for_canister(&agent, proxy, &cid, &info, &binding_vars).await + match &result { + Ok(()) => task.finish(TaskOutcome::succeeded()), + Err(error) => task.finish(TaskOutcome::failed(error.to_string())), } - }; - futs.push_back(async move { - let result = ProgressManager::execute_with_progress( - &pb, - settings_fn, - || "Environment variables updated successfully".to_string(), - |err| format!("Failed to update environment variables: {err}"), - ) - .await; - - // Map error to include canister context for deferred printing - result.map_err(|error| BindingEnvVarsFailure { - canister_name, - canister_id: cid, - error, - }) + result.map_err(|_| info.name.clone()) }); } - // Consume the set of futures and collect errors - let mut errors: Vec = Vec::new(); + // Collect the failed canister names; the renderer owns displaying each + // failure. + 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 update environment variables for canister '{}': {} -----", - failure.canister_name, failure.canister_id, - ); - error!("'{}'", failure.error); - } - - return SetBindingEnvVarsManySnafu { - names: errors - .iter() - .map(|e| e.canister_name.clone()) - .collect::>(), - } - .fail(); + if !failed.is_empty() { + return SetBindingEnvVarsManySnafu { names: failed }.fail(); } Ok(()) diff --git a/crates/icp-cli/src/operations/build.rs b/crates/icp-cli/src/operations/build.rs index eb85c7bb5..8df7bd142 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,56 @@ 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(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/operations/candid_compat.rs b/crates/icp-cli/src/operations/candid_compat.rs index 07355a012..b2d7108db 100644 --- a/crates/icp-cli/src/operations/candid_compat.rs +++ b/crates/icp-cli/src/operations/candid_compat.rs @@ -8,13 +8,11 @@ use candid_parser::utils::CandidSource; use futures::{StreamExt, stream::FuturesOrdered}; use ic_agent::Agent; use ic_management_canister_types::CanisterInstallMode; +use icp_events::{Reporter, TaskKind, TaskOutcome}; use snafu::Snafu; -use tracing::{debug, error}; +use tracing::debug; -use crate::{ - operations::{misc::fetch_canister_metadata, wasm::extract_candid_service}, - progress::{ProgressManager, ProgressManagerSettings}, -}; +use crate::operations::{misc::fetch_canister_metadata, wasm::extract_candid_service}; /// Checks Candid interface compatibility for all canisters that would be /// upgraded. Aborts if any canister has an incompatible interface. @@ -22,63 +20,51 @@ pub(crate) async fn check_candid_compatibility_many( agent: Agent, canisters: impl IntoIterator, artifacts: Arc, - debug: bool, + reporter: &Reporter, ) -> Result<(), CandidCheckManyError> { let mut check_futs = FuturesOrdered::new(); - let check_progress = ProgressManager::new(ProgressManagerSettings { hidden: debug }); for (name, cid, mode) in canisters { - let pb = check_progress.create_progress_bar(name); + let task = reporter.task(TaskKind::CandidCheck { + canister: name.to_owned(), + canister_id: cid, + }); let is_upgrade = matches!(mode, CanisterInstallMode::Upgrade(_)); let agent = agent.clone(); let artifacts = artifacts.clone(); check_futs.push_back(async move { if !is_upgrade { - pb.finish_with_message("Skipped (not an upgrade)"); - return Ok::<_, CandidCheckFailure>(()); + task.finish(TaskOutcome::Skipped { + reason: "not an upgrade".to_owned(), + }); + return Ok::<_, String>(()); } - pb.set_message("Checking compatibility..."); + let result = check_canister_candid_compat(&agent, &cid, name, &*artifacts).await; - ProgressManager::execute_with_progress( - &pb, - check_canister_candid_compat(&agent, &cid, name, &*artifacts), - || "Compatible".to_string(), - |_| "Incompatible".to_string(), - ) - .await + match &result { + Ok(()) => task.finish(TaskOutcome::succeeded()), + // The renderer words the breaking-change dump; the incompatibility + // details ride as the failure message. + Err(failure) => task.finish(TaskOutcome::failed(failure.details.clone())), + } + + result.map_err(|failure| failure.canister_name) }); } - let mut check_failures: Vec = Vec::new(); + // Collect the failed canister names; the renderer owns displaying each + // failure. + let mut failed: Vec = Vec::new(); while let Some(res) = check_futs.next().await { - if let Err(failure) = res { - check_failures.push(failure); + if let Err(name) = res { + failed.push(name); } } - if !check_failures.is_empty() { - for failure in &check_failures { - error!( - " ----- Candid interface compatibility check failed: '{}' ({}) -----", - failure.canister_name, failure.canister_id, - ); - error!( - "You are making a BREAKING change. Other canisters or frontend clients \ - relying on your canister may stop working.\n\n{}", - failure.details, - ); - } - error!("Use --yes to bypass this check."); - - return CandidCheckManySnafu { - names: check_failures - .iter() - .map(|f| f.canister_name.clone()) - .collect::>(), - } - .fail(); + if !failed.is_empty() { + return CandidCheckManySnafu { names: failed }.fail(); } Ok(()) @@ -108,7 +94,6 @@ async fn check_canister_candid_compat( } CandidCompatibility::Incompatible(details) => Err(CandidCheckFailure { canister_name: canister_name.to_owned(), - canister_id: *canister_id, details, }), } @@ -187,7 +172,6 @@ pub(crate) async fn check_candid_compatibility( /// Holds error information from a failed candid compatibility check struct CandidCheckFailure { canister_name: String, - canister_id: Principal, details: String, } diff --git a/crates/icp-cli/src/operations/install.rs b/crates/icp-cli/src/operations/install.rs index 24f90fd4e..cc79c9c8d 100644 --- a/crates/icp-cli/src/operations/install.rs +++ b/crates/icp-cli/src/operations/install.rs @@ -6,12 +6,11 @@ use ic_management_canister_types::{ ClearChunkStoreArgs, InstallChunkedCodeArgs, InstallCodeArgs, UpgradeFlags, UploadChunkArgs, WasmMemoryPersistence, }; +use icp_events::{Reporter, TaskKind, TaskOutcome}; use sha2::{Digest, Sha256}; use snafu::{ResultExt, Snafu}; use std::sync::Arc; -use tracing::{debug, error, warn}; - -use crate::progress::{ProgressManager, ProgressManagerSettings}; +use tracing::{debug, warn}; use super::misc::fetch_canister_metadata; use super::proxy::UpdateOrProxyError; @@ -71,13 +70,6 @@ pub struct InstallManyError { names: Vec, } -/// Holds error information from a failed canister install operation -struct InstallFailure { - canister_name: String, - canister_id: Principal, - error: InstallOperationError, -} - /// Resolve a mode string ("auto", "install", "reinstall", "upgrade") into /// a [`CanisterInstallMode`]. For "auto", queries `canister_status` to /// determine whether the canister already has code installed. @@ -349,7 +341,7 @@ async fn stop_and_start_if_upgrade( install_result } -/// Installs code to multiple canisters and displays progress bars. +/// Installs code to multiple canisters concurrently. pub(crate) async fn install_many( agent: Agent, proxy: Option, @@ -363,22 +355,20 @@ pub(crate) async fn install_many( ), >, artifacts: Arc, - debug: bool, + reporter: &Reporter, ) -> Result<(), InstallManyError> { let mut futs = FuturesOrdered::new(); - let progress_manager = ProgressManager::new(ProgressManagerSettings { hidden: debug }); for (name, cid, mode, status, init_args) in canisters { - let pb = progress_manager.create_progress_bar(&name); + let task = reporter.task(TaskKind::Install { + canister: name.clone(), + canister_id: cid, + }); let agent = agent.clone(); - let install_fn = { - let pb = pb.clone(); - let artifacts = artifacts.clone(); - let name = name.clone(); - - async move { - pb.set_message("Installing..."); + let artifacts = artifacts.clone(); + futs.push_back(async move { + let result = async { let wasm = artifacts.lookup(&name).await.map_err(|_| { InstallOperationError::ArtifactNotFound { canister_name: name.clone(), @@ -398,48 +388,28 @@ pub(crate) async fn install_many( ) .await } - }; - - futs.push_back(async move { - let result = ProgressManager::execute_with_progress( - &pb, - install_fn, - || "Installed successfully".to_string(), - |err| format!("Failed to install canister: {err}"), - ) .await; - result.map_err(|error| InstallFailure { - canister_name: name.clone(), - canister_id: cid, - error, - }) + match &result { + Ok(()) => task.finish(TaskOutcome::succeeded()), + Err(error) => task.finish(TaskOutcome::failed(error.to_string())), + } + + result.map_err(|_| name) }); } - let mut errors: Vec = Vec::new(); + // Collect the failed canister names; the renderer owns displaying each + // failure. + 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() { - for failure in &errors { - error!( - "----- Failed to install canister '{}': {} -----", - failure.canister_name, failure.canister_id, - ); - error!("'{}'", failure.error); - } - - return InstallManySnafu { - names: errors - .iter() - .map(|e| e.canister_name.clone()) - .collect::>(), - } - .fail(); + if !failed.is_empty() { + return InstallManySnafu { names: failed }.fail(); } Ok(()) diff --git a/crates/icp-cli/src/operations/settings.rs b/crates/icp-cli/src/operations/settings.rs index c8f1d0b8a..083fb31e8 100644 --- a/crates/icp-cli/src/operations/settings.rs +++ b/crates/icp-cli/src/operations/settings.rs @@ -15,12 +15,11 @@ use icp::{ context::{Context, EnvironmentSelection}, store_id::IdMapping, }; +use icp_events::{Reporter, TaskKind, TaskOutcome}; use itertools::Itertools; use num_traits::ToPrimitive; use snafu::{ResultExt, Snafu}; -use tracing::{error, warn}; - -use crate::progress::{ProgressManager, ProgressManagerSettings}; +use tracing::warn; use super::proxy::UpdateOrProxyError; use super::proxy_management; @@ -46,13 +45,6 @@ pub struct SyncSettingsManyError { names: Vec, } -/// Holds error information from a failed canister settings update operation -struct SettingsFailure { - canister_name: String, - canister_id: Principal, - error: SyncSettingsOperationError, -} - /// Compare two LogVisibility values in an order-insensitive manner. /// For AllowedViewers, the principal lists are compared as sets. fn log_visibility_eq(a: &LogVisibility, b: &LogVisibility) -> bool { @@ -226,23 +218,21 @@ pub(crate) async fn sync_settings_many( proxy: Option, target_canisters: Vec<(Principal, Canister)>, ids: IdMapping, - debug: bool, + reporter: &Reporter, ) -> Result<(), SyncSettingsManyError> { let mut futs = FuturesOrdered::new(); - let progress_manager = ProgressManager::new(ProgressManagerSettings { hidden: debug }); let ids = Arc::new(ids); for (cid, info) in target_canisters { - let pb = progress_manager.create_progress_bar(&info.name); - let canister_name = info.name.clone(); + let task = reporter.task(TaskKind::UpdateSettings { + canister: info.name.clone(), + canister_id: cid, + }); + let agent = agent.clone(); let ids = ids.clone(); - let settings_fn = { - let agent = agent.clone(); - let pb = pb.clone(); - - async move { - pb.set_message("Updating canister settings..."); + futs.push_back(async move { + let result = async { let unresolved = sync_settings(&agent, proxy, &cid, &info, &ids).await?; for name in &unresolved { warn!( @@ -253,51 +243,28 @@ pub(crate) async fn sync_settings_many( } Ok::<_, SyncSettingsOperationError>(()) } - }; - - futs.push_back(async move { - let result = ProgressManager::execute_with_progress( - &pb, - settings_fn, - || "Canister settings updated successfully".to_string(), - |err| format!("Failed to update canister settings: {err}"), - ) .await; - // Map error to include canister context for deferred printing - result.map_err(|error| SettingsFailure { - canister_name, - canister_id: cid, - error, - }) + match &result { + Ok(()) => task.finish(TaskOutcome::succeeded()), + Err(error) => task.finish(TaskOutcome::failed(error.to_string())), + } + + result.map_err(|_| info.name.clone()) }); } - // Consume the set of futures and collect errors - let mut errors: Vec = Vec::new(); + // Collect the failed canister names; the renderer owns displaying each + // failure. + 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 update settings for canister '{}': {} -----", - failure.canister_name, failure.canister_id, - ); - error!("'{}'", failure.error); - } - - return SyncSettingsManySnafu { - names: errors - .iter() - .map(|e| e.canister_name.clone()) - .collect::>(), - } - .fail(); + if !failed.is_empty() { + return SyncSettingsManySnafu { names: failed }.fail(); } Ok(()) diff --git a/crates/icp-cli/src/operations/snapshot_transfer.rs b/crates/icp-cli/src/operations/snapshot_transfer.rs index 24a6388de..4a4de887c 100644 --- a/crates/icp-cli/src/operations/snapshot_transfer.rs +++ b/crates/icp-cli/src/operations/snapshot_transfer.rs @@ -19,7 +19,7 @@ use icp::{ fs::lock::{DirectoryStructureLock, LWrite, LockError, PathsAccess}, prelude::*, }; -use indicatif::{ProgressBar, ProgressStyle}; +use icp_events::TaskReporter; use serde::{Deserialize, Serialize}; use snafu::{ResultExt, Snafu}; use tokio::{ @@ -424,19 +424,6 @@ where } } -/// Create a progress bar for byte transfers. -pub fn create_transfer_progress_bar(total_bytes: u64, label: &str) -> ProgressBar { - let pb = ProgressBar::new(total_bytes); - pb.set_style( - ProgressStyle::default_bar() - .template("{prefix} [{elapsed_precise}] [{wide_bar:.cyan/blue}] {bytes}/{total_bytes} ({bytes_per_sec}, {eta})") - .expect("invalid progress bar template") - .progress_chars("#>-"), - ); - pb.set_prefix(label.to_string()); - pb -} - /// Read snapshot metadata from a canister. pub async fn read_snapshot_metadata( agent: &Agent, @@ -513,7 +500,7 @@ pub async fn download_blob_to_file( total_size: u64, paths: LWrite<&SnapshotPaths>, progress: &mut DownloadProgress, - progress_bar: &ProgressBar, + task: &TaskReporter, ) -> Result<(), SnapshotTransferError> { let output_path = paths.blob_path(blob_type); @@ -547,7 +534,7 @@ pub async fn download_blob_to_file( // Set initial progress based on frontier let initial_bytes = progress.blob_progress(blob_type).frontier; - progress_bar.set_position(initial_bytes); + task.progress(initial_bytes); // Determine which chunks need downloading let snapshot_id_vec = snapshot_id.to_vec(); @@ -612,7 +599,7 @@ pub async fn download_blob_to_file( save_download_progress(progress, paths)?; // Update progress bar to show frontier position - progress_bar.set_position(progress.blob_progress(blob_type).frontier); + task.progress(progress.blob_progress(blob_type).frontier); } Ok(()) @@ -663,7 +650,7 @@ pub async fn upload_blob_from_file( blob_type: BlobType, paths: LWrite<&SnapshotPaths>, progress: &mut UploadProgress, - progress_bar: &ProgressBar, + task: &TaskReporter, ) -> Result { let input_path = paths.blob_path(blob_type); let file_size = std::fs::metadata(&input_path) @@ -690,7 +677,7 @@ pub async fn upload_blob_from_file( .context(SeekBlobFileSnafu { path: &input_path })?; } - progress_bar.set_position(start_offset); + task.progress(start_offset); // Read all chunks and launch uploads concurrently let snapshot_id_vec = snapshot_id.to_vec(); @@ -742,7 +729,7 @@ pub async fn upload_blob_from_file( while let Some(&size) = completed.get(&next_report_offset) { completed.remove(&next_report_offset); next_report_offset += size; - progress_bar.set_position(next_report_offset); + task.progress(next_report_offset); // Update and save progress match blob_type { 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 deleted file mode 100644 index 0a0795991..000000000 --- a/crates/icp-cli/src/progress.rs +++ /dev/null @@ -1,334 +0,0 @@ -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; - -// Animation frames for the spinner - creates a rotating star effect -const TICKS: &[&str] = &["✶", "✸", "✹", "✺", "✹", "✷"]; - -// Final tick symbols for different completion states -const TICK_EMPTY: &str = " "; -const TICK_SUCCESS: &str = "✔"; -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"; - -// 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 { - // Template format: "[prefix] [spinner] [message]" - let tmpl = format!("{{prefix}} {{spinner:.{color}}} {{msg}}"); - - ProgressStyle::with_template(&tmpl) - .expect("invalid style template") - // Combine animation frames with the final completion symbol - .tick_strings(&[TICKS, &[end_tick]].concat()) -} - -/// A fixed-capacity rolling buffer that always holds the last `capacity` items. -#[derive(Debug)] -pub(crate) struct RollingLines { - buf: VecDeque, - capacity: usize, -} - -impl RollingLines { - /// Create a new buffer with a fixed capacity. - pub(crate) fn new(capacity: usize) -> Self { - let buf = VecDeque::with_capacity(capacity); - Self { buf, capacity } - } - - /// Push a new line, evicting the oldest if full. - pub(crate) fn push(&mut self, line: String) { - if self.buf.len() == self.capacity { - self.buf.pop_front(); - } - - self.buf.push_back(line); - } - - /// Get an iterator over the current contents (in order). - pub(crate) fn iter(&self) -> impl Iterator { - self.buf.iter().map(|s| s.as_str()) - } - - /// 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 -pub(crate) struct ProgressManagerSettings { - /// Whether to hide the progress bars - pub(crate) hidden: bool, -} - -/// Shared progress bar utilities for build and sync commands -pub(crate) struct ProgressManager { - pub(crate) multi_progress: MultiProgress, -} - -impl ProgressManager { - pub(crate) fn new(settings: ProgressManagerSettings) -> Self { - let multi_progress = MultiProgress::new(); - - if settings.hidden { - multi_progress.set_draw_target(indicatif::ProgressDrawTarget::hidden()); - } - - Self { multi_progress } - } - - /// Create a new progress bar with standard configuration - pub(crate) fn create_progress_bar(&self, canister_name: &str) -> SimpleProgressBar { - let pb = self.create_independent_progress_bar(); - pb.set_prefix(format!("[{canister_name}]")); - pb - } - - pub(crate) fn create_independent_progress_bar(&self) -> SimpleProgressBar { - let pb = self - .multi_progress - .add(SimpleProgressBar::new_spinner().with_style(make_style( - TICK_EMPTY, // end_tick - COLOR_REGULAR, // color - ))); - - // Auto-tick spinner - pb.enable_steady_tick(Duration::from_millis(120)); - - 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, - task: F, - success_message: impl Fn() -> String, - error_message: impl Fn(&E) -> String, - ) -> Result - where - F: Future>, - P: ProgressBar, - { - // Delegate to execute_with_custom_progress with no special error handling - Self::execute_with_custom_progress( - progress_bar, - task, - success_message, - error_message, - |_| false, // No errors are treated as success - ) - .await - } - - /// Execute a task with custom progress handling for errors that should display as success - pub(crate) async fn execute_with_custom_progress( - progress_bar: &P, - task: F, - success_message: impl Fn() -> String, - error_message: impl Fn(&E) -> String, - is_success_error: impl Fn(&E) -> bool, - ) -> Result - where - F: Future>, - P: ProgressBar, - { - // Execute the task and capture the result - let result = task.await; - - // Update the progress bar style and message based on result - let (style, message) = match &result { - Ok(_) => (make_style(TICK_SUCCESS, COLOR_SUCCESS), success_message()), - Err(err) if is_success_error(err) => { - (make_style(TICK_SUCCESS, COLOR_SUCCESS), error_message(err)) - } - Err(err) => (make_style(TICK_FAILURE, COLOR_FAILURE), error_message(err)), - }; - - progress_bar.set_style(style); - progress_bar.set_message(message); - progress_bar.finish(); - - result - } -} - -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); - } - - fn set_message(&self, message: String) { - SimpleProgressBar::set_message(self, message); - } - - fn finish(&self) { - SimpleProgressBar::finish(self); - } -} diff --git a/crates/icp-cli/src/render/interactive.rs b/crates/icp-cli/src/render/interactive.rs new file mode 100644 index 000000000..389962bcd --- /dev/null +++ b/crates/icp-cli/src/render/interactive.rs @@ -0,0 +1,213 @@ +//! 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, TaskKind, TaskOutcome}; +use indicatif::{MultiProgress, ProgressBar, ProgressStyle}; +use itertools::Itertools; +use tracing::debug; + +use super::style::{ + COLOR_FAILURE, COLOR_REGULAR, COLOR_SUCCESS, TICK_EMPTY, TICK_FAILURE, TICK_SUCCESS, make_style, +}; +use super::{ + RollingLines, TaskLog, dump_failures, failure_message, step_header, success_message, + transfer_label, +}; + +/// 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. + /// While a script command runs this is the step headline plus that + /// command, so output is attributed to the command producing it. + header: String, + /// First line of the current step's full header, used to rebuild the + /// live header when a command starts. + headline: 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 } => { + // Bars are configured fully before insertion: adding to the + // MultiProgress can draw the initial frame, and it must not + // appear unstyled or unlabeled. + let bar = match &task { + // Quantifiable transfers get a byte bar instead of a + // spinner, labeled by the blob rather than the canister. + TaskKind::SnapshotTransfer { + blob, total_bytes, .. + } => self.multi_progress.add( + ProgressBar::new(*total_bytes) + .with_style(transfer_style()) + .with_prefix(transfer_label(blob)), + ), + _ => { + let mut bar = ProgressBar::new_spinner() + .with_style(make_style(TICK_EMPTY, COLOR_REGULAR)) + .with_prefix(format!("[{}]", task.canister())); + if let Some(message) = super::running_message(&task) { + bar = bar.with_message(message); + } + let bar = self.multi_progress.add(bar); + bar.enable_steady_tick(Duration::from_millis(120)); + bar + } + }; + + self.tasks.insert( + event.task_id, + TaskView { + log: TaskLog::new(task), + bar, + header: String::new(), + headline: String::new(), + window: RollingLines::new(LIVE_WINDOW_LINES), + }, + ); + } + + EventKind::Progress { position } => { + if let Some(view) = self.tasks.get(&event.task_id) { + view.bar.set_position(position); + } + } + + 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.headline = view + .header + .lines() + .find(|line| !line.is_empty()) + .unwrap_or_default() + .to_owned(); + view.window = RollingLines::new(LIVE_WINDOW_LINES); + view.log.start_step(view.header.clone()); + // The bar is deliberately not updated here: the header shows + // once the step's first output line arrives (the Output + // branch), so silent steps draw nothing. + } + + EventKind::CommandStarted { command } => { + let Some(view) = self.tasks.get_mut(&event.task_id) else { + return; + }; + // Show only the running command under the step headline, and + // reset the live window so a previous command's output isn't + // attributed to this one. The captured log is unaffected. + view.header = format!("{}\n$ {command}", view.headline); + view.window = RollingLines::new(LIVE_WINDOW_LINES); + 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.log.kind().canister()); + + 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; + }; + + // A transfer's byte bar has no message or tick slot; it just + // freezes at its final position. + if matches!(view.log.kind(), TaskKind::SnapshotTransfer { .. }) { + view.bar.finish(); + if let TaskOutcome::Failed { message, causes } = outcome { + view.log.fail(message, causes); + } + return; + } + + match outcome { + 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, causes } => { + view.bar.set_style(make_style(TICK_FAILURE, COLOR_FAILURE)); + view.bar + .set_message(failure_message(view.log.kind(), &message)); + view.bar.finish(); + view.log.fail(message, causes); + } + // Skipped keeps the neutral style — nothing succeeded or + // failed. + TaskOutcome::Skipped { reason } => { + view.bar.finish_with_message(format!("Skipped ({reason})")); + } + } + } + } + } + + /// 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); + } +} + +/// Style for a byte-transfer bar. +fn transfer_style() -> ProgressStyle { + ProgressStyle::default_bar() + .template("{prefix} [{elapsed_precise}] [{wide_bar:.cyan/blue}] {bytes}/{total_bytes} ({bytes_per_sec}, {eta})") + .expect("invalid progress bar template") + .progress_chars("#>-") +} diff --git a/crates/icp-cli/src/render/mod.rs b/crates/icp-cli/src/render/mod.rs new file mode 100644 index 000000000..84df894b4 --- /dev/null +++ b/crates/icp-cli/src/render/mod.rs @@ -0,0 +1,405 @@ +//! 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, VecDeque}; + +use icp_events::{Event, Reporter, TaskId, TaskKind, TaskOutcome, TaskReporter, TransferBlob}; +use tokio::sync::mpsc::UnboundedReceiver; +use tracing::error; + +mod interactive; +mod plain; +mod spinner; +mod style; + +pub(crate) use interactive::InteractiveRenderer; +pub(crate) use plain::PlainRenderer; +pub(crate) use spinner::{ProgressManager, ProgressManagerSettings}; + +/// The maximum number of lines to display for a step output +const MAX_LINES_PER_STEP: usize = 10_000; + +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(); + } + } + } +} + +/// Run one operation phase with a fresh event channel and a renderer driving +/// its display: the reporter is handed to `op`, and once `op` finishes the +/// stream is closed and the renderer flushes (failure dumps) before the +/// operation's result is returned. +pub(crate) async fn rendered(debug: bool, op: impl AsyncFnOnce(&Reporter) -> T) -> T { + let (reporter, events) = icp_events::channel(); + let render = tokio::spawn(Renderer::for_ctx(debug).run(events)); + + let result = op(&reporter).await; + + drop(reporter); + render.await.expect("renderer task panicked"); + + result +} + +/// Run a single task under its own renderer: starts a task of `kind`, hands +/// its reporter to `op`, and finishes the task from the result before the +/// renderer flushes. +pub(crate) async fn rendered_task( + debug: bool, + kind: TaskKind, + op: impl AsyncFnOnce(&TaskReporter) -> Result, +) -> Result { + rendered(debug, async |reporter| { + let task = reporter.task(kind); + let result = op(&task).await; + + match &result { + Ok(_) => task.finish(TaskOutcome::succeeded()), + Err(error) => task.finish(TaskOutcome::failed(error.to_string())), + } + + result + }) + .await +} + +// Wording for each task kind. Events carry data; these helpers own the words. + +/// Message shown while a task runs, before any step reports in. Multi-step +/// tasks (build, sync) have none — their step headers take over. +fn running_message(kind: &TaskKind) -> Option<&'static str> { + match kind { + // Build and sync step headers take over; a transfer's byte bar has no + // message slot at all. + TaskKind::Build { .. } | TaskKind::Sync { .. } | TaskKind::SnapshotTransfer { .. } => None, + TaskKind::Create { .. } => Some("Creating..."), + TaskKind::Install { .. } => Some("Installing..."), + TaskKind::UpdateSettings { .. } => Some("Updating canister settings..."), + TaskKind::UpdateEnvironmentVariables { .. } => Some("Updating environment variables..."), + TaskKind::CandidCheck { .. } => Some("Checking compatibility..."), + } +} + +/// Prefix label for a snapshot-transfer byte bar. +fn transfer_label(blob: &TransferBlob) -> &'static str { + match blob { + TransferBlob::WasmModule => "WASM module", + TransferBlob::WasmMemory => "WASM memory", + TransferBlob::StableMemory => "Stable memory", + } +} + +/// 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::Sync { .. } => format!("\nSyncing: {label} {number} of {total}"), + // Only build and sync report steps; a generic header for the rest. + _ => 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::Sync { .. } => "Sync", + // Only build and sync capture step output; a generic label for the rest. + _ => "Build", + } +} + +/// Final progress-bar message for a task that succeeded. +fn success_message(kind: &TaskKind) -> String { + match kind { + TaskKind::Build { .. } => "Built successfully".to_owned(), + TaskKind::Sync { canister_id, .. } => format!("Synced successfully: {canister_id}"), + TaskKind::Create { .. } => "Created successfully".to_owned(), + TaskKind::Install { .. } => "Installed successfully".to_owned(), + TaskKind::UpdateSettings { .. } => "Canister settings updated successfully".to_owned(), + TaskKind::UpdateEnvironmentVariables { .. } => { + "Environment variables updated successfully".to_owned() + } + TaskKind::CandidCheck { .. } => "Compatible".to_owned(), + // A transfer's byte bar has no message slot; nothing to show. + TaskKind::SnapshotTransfer { .. } => "done".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}"), + TaskKind::Sync { .. } => format!("Failed to sync canister: {message}"), + // Create failures surface through the command's returned error; the + // bar shows the bare message. + TaskKind::Create { .. } => message.to_owned(), + TaskKind::Install { .. } => format!("Failed to install canister: {message}"), + TaskKind::UpdateSettings { .. } => { + format!("Failed to update canister settings: {message}") + } + TaskKind::UpdateEnvironmentVariables { .. } => { + format!("Failed to update environment variables: {message}") + } + TaskKind::CandidCheck { .. } => "Incompatible".to_owned(), + // Transfer failures surface through the command's returned error. + TaskKind::SnapshotTransfer { .. } => message.to_owned(), + } +} + +/// First line of a task's failure dump, or `None` for kinds that don't get a +/// deferred dump (their failure travels on the command's returned error). +fn failure_header(kind: &TaskKind) -> Option { + match kind { + TaskKind::Build { canister } => { + Some(format!("----- Failed to build canister '{canister}' -----")) + } + TaskKind::Sync { + canister, + canister_id, + } => Some(format!( + "----- Failed to sync canister '{canister}': {canister_id} -----" + )), + TaskKind::Create { .. } => None, + TaskKind::Install { + canister, + canister_id, + } => Some(format!( + "----- Failed to install canister '{canister}': {canister_id} -----" + )), + TaskKind::UpdateSettings { + canister, + canister_id, + } => Some(format!( + "----- Failed to update settings for canister '{canister}': {canister_id} -----" + )), + TaskKind::UpdateEnvironmentVariables { + canister, + canister_id, + } => Some(format!( + "----- Failed to update environment variables for canister '{canister}': {canister_id} -----" + )), + TaskKind::CandidCheck { + canister, + canister_id, + } => Some(format!( + " ----- Candid interface compatibility check failed: '{canister}' ({canister_id}) -----" + )), + TaskKind::SnapshotTransfer { .. } => None, + } +} + +/// 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()); + } +} + +/// 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, +} + +struct Failure { + message: String, + causes: Vec, +} + +/// A fixed-capacity rolling buffer that always holds the last `capacity` items. +#[derive(Debug)] +struct RollingLines { + buf: VecDeque, + capacity: usize, +} + +impl RollingLines { + /// Create a new buffer with a fixed capacity. + fn new(capacity: usize) -> Self { + let buf = VecDeque::with_capacity(capacity); + Self { buf, capacity } + } + + /// Push a new line, evicting the oldest if full. + fn push(&mut self, line: String) { + if self.buf.len() == self.capacity { + self.buf.pop_front(); + } + + self.buf.push_back(line); + } + + /// Get an iterator over the current contents (in order). + fn iter(&self) -> impl Iterator { + self.buf.iter().map(|s| s.as_str()) + } + + /// Whether no lines have been pushed. + fn is_empty(&self) -> bool { + self.buf.is_empty() + } +} + +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, causes: Vec) { + self.failure = Some(Failure { message, causes }); + } + + /// Render the captured output. When `all_steps` is true, output from + /// every step is included; otherwise only the last (failing) step is + /// shown. Tasks that never reported a step (the single-action kinds) + /// have nothing to replay. + fn dump(&self, all_steps: bool) -> Vec { + if self.finished_steps.is_empty() && self.current_step.is_none() { + return Vec::new(); + } + + 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) { + let mut candid_failures = false; + + for log in logs.values() { + let Some(failure) = &log.failure else { + continue; + }; + let Some(header) = failure_header(&log.kind) else { + continue; + }; + + error!("{header}"); + match &log.kind { + TaskKind::CandidCheck { .. } => { + candid_failures = true; + error!( + "You are making a BREAKING change. Other canisters or frontend clients \ + relying on your canister may stop working.\n\n{}", + failure.message, + ); + } + _ => { + error!("'{}'", failure.message); + for cause in &failure.causes { + error!(" caused by: {cause}"); + } + } + } + for line in log.dump(all_steps) { + error!("{line}"); + } + } + + if candid_failures { + error!("Use --yes to bypass this check."); + } +} diff --git a/crates/icp-cli/src/render/plain.rs b/crates/icp-cli/src/render/plain.rs new file mode 100644 index 000000000..ffc0faac7 --- /dev/null +++ b/crates/icp-cli/src/render/plain.rs @@ -0,0 +1,90 @@ +//! 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::CommandStarted { command } => { + let Some(log) = self.tasks.get_mut(&event.task_id) else { + return; + }; + // Mark command boundaries so interleaved output stays + // attributable to the command producing it. + debug!("[{}] $ {command}", log.kind().canister()); + } + + EventKind::Output { line, .. } => { + let Some(log) = self.tasks.get_mut(&event.task_id) else { + return; + }; + // Prefix with the canister so interleaved concurrent tasks + // stay attributable. + debug!("[{}] {line}", log.kind().canister()); + log.push_line(line); + } + + EventKind::StepCompleted { .. } => { + if let Some(log) = self.tasks.get_mut(&event.task_id) { + log.end_step(); + } + } + + // No live display to advance. + EventKind::Progress { .. } => {} + + EventKind::TaskCompleted { outcome } => { + let Some(log) = self.tasks.get_mut(&event.task_id) else { + return; + }; + match outcome { + TaskOutcome::Succeeded { retained_output } => { + super::print_retained(log.kind(), &retained_output); + } + TaskOutcome::Failed { message, causes } => { + log.fail(message, causes); + } + TaskOutcome::Skipped { .. } => {} + } + } + } + } + + /// 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/src/render/spinner.rs b/crates/icp-cli/src/render/spinner.rs new file mode 100644 index 000000000..fc4dcf010 --- /dev/null +++ b/crates/icp-cli/src/render/spinner.rs @@ -0,0 +1,75 @@ +//! A direct-use spinner widget for commands that render their own UX (e.g. +//! `icp network start`) rather than consuming an event stream through a +//! [`Renderer`](super::Renderer). + +use std::time::Duration; + +use futures::Future; +use indicatif::{MultiProgress, ProgressBar}; + +use super::style::{ + COLOR_FAILURE, COLOR_REGULAR, COLOR_SUCCESS, TICK_EMPTY, TICK_FAILURE, TICK_SUCCESS, make_style, +}; + +/// Settings for the progress manager +pub(crate) struct ProgressManagerSettings { + /// Whether to hide the progress bars + pub(crate) hidden: bool, +} + +/// Spinner utilities for commands that drive their own progress display. +pub(crate) struct ProgressManager { + multi_progress: MultiProgress, +} + +impl ProgressManager { + pub(crate) fn new(settings: ProgressManagerSettings) -> Self { + let multi_progress = MultiProgress::new(); + + if settings.hidden { + multi_progress.set_draw_target(indicatif::ProgressDrawTarget::hidden()); + } + + Self { multi_progress } + } + + pub(crate) fn create_independent_progress_bar(&self) -> ProgressBar { + let pb = self + .multi_progress + .add(ProgressBar::new_spinner().with_style(make_style( + TICK_EMPTY, // end_tick + COLOR_REGULAR, // color + ))); + + // Auto-tick spinner + pb.enable_steady_tick(Duration::from_millis(120)); + + pb + } + + /// Execute a task with progress tracking and automatic style updates + pub(crate) async fn execute_with_progress( + progress_bar: &ProgressBar, + task: F, + success_message: impl Fn() -> String, + error_message: impl Fn(&E) -> String, + ) -> Result + where + F: Future>, + { + // Execute the task and capture the result + let result = task.await; + + // Update the progress bar style and message based on result + let (style, message) = match &result { + Ok(_) => (make_style(TICK_SUCCESS, COLOR_SUCCESS), success_message()), + Err(err) => (make_style(TICK_FAILURE, COLOR_FAILURE), error_message(err)), + }; + + progress_bar.set_style(style); + progress_bar.set_message(message); + progress_bar.finish(); + + result + } +} diff --git a/crates/icp-cli/src/render/style.rs b/crates/icp-cli/src/render/style.rs new file mode 100644 index 000000000..5fa5e2b9f --- /dev/null +++ b/crates/icp-cli/src/render/style.rs @@ -0,0 +1,30 @@ +//! Spinner styling shared by the interactive renderer and the direct-use +//! spinner widget. + +use indicatif::ProgressStyle; + +// Animation frames for the spinner - creates a rotating star effect +const TICKS: &[&str] = &["✶", "✸", "✹", "✺", "✹", "✷"]; + +// Final tick symbols for different completion states +pub(super) const TICK_EMPTY: &str = " "; +pub(super) const TICK_SUCCESS: &str = "✔"; +pub(super) const TICK_FAILURE: &str = "✘"; + +// Color schemes for different progress states +pub(super) const COLOR_REGULAR: &str = "blue"; +pub(super) const COLOR_SUCCESS: &str = "green"; +pub(super) 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 +pub(super) fn make_style(end_tick: &str, color: &str) -> ProgressStyle { + // Template format: "[prefix] [spinner] [message]" + let tmpl = format!("{{prefix}} {{spinner:.{color}}} {{msg}}"); + + ProgressStyle::with_template(&tmpl) + .expect("invalid style template") + // Combine animation frames with the final completion symbol + .tick_strings(&[TICKS, &[end_tick]].concat()) +} diff --git a/crates/icp-cli/tests/build_tests.rs b/crates/icp-cli/tests/build_tests.rs index c2e89fbf0..2e5554d0e 100644 --- a/crates/icp-cli/tests/build_tests.rs +++ b/crates/icp-cli/tests/build_tests.rs @@ -502,9 +502,13 @@ 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: [canister-a] building canister-a", + )) + .stderr(contains( + "DEBUG icp::render::plain: [canister-b] building canister-b", + )) + .stderr(contains("DEBUG icp::render::plain: [canister-c] building canister-c").not()); } #[test] @@ -559,7 +563,11 @@ 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: [canister-a] building canister-a", + )) + .stderr(contains( + "DEBUG icp::render::plain: [canister-b] building canister-b", + )) + .stderr(contains("DEBUG icp::render::plain: [canister-c] building canister-c").not()); // not in test-env } diff --git a/crates/icp-cli/tests/sync_tests.rs b/crates/icp-cli/tests/sync_tests.rs index 31f72ef48..741772b96 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: [my-canister] 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: [my-canister] syncing").not()), ); } @@ -387,9 +387,13 @@ 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: [canister-a] syncing canister-a", + )) + .stderr(contains( + "DEBUG icp::render::plain: [canister-b] syncing canister-b", + )) + .stderr(contains("DEBUG icp::render::plain: [canister-c] syncing canister-c").not()); } #[tokio::test] @@ -861,7 +865,11 @@ 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: [canister-a] syncing canister-a", + )) + .stderr(contains( + "DEBUG icp::render::plain: [canister-b] syncing canister-b", + )) + .stderr(contains("DEBUG icp::render::plain: [canister-c] syncing 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..fb7adb41f --- /dev/null +++ b/crates/icp-events/Cargo.toml @@ -0,0 +1,11 @@ +[package] +name = "icp-events" +version.workspace = true +edition = { workspace = true } +license = { workspace = true } +publish.workspace = true + +[dependencies] +candid = { workspace = true } +serde = { workspace = true } +tokio = { workspace = true, features = ["sync"] } diff --git a/crates/icp-events/src/lib.rs b/crates/icp-events/src/lib.rs new file mode 100644 index 000000000..7753fbe66 --- /dev/null +++ b/crates/icp-events/src/lib.rs @@ -0,0 +1,375 @@ +//! 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 candid::Principal; +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, + }, + + /// A shell command within the task's current step began executing. + /// Script steps run their commands in order; renderers can use this to + /// attribute the output that follows to the command producing it. + CommandStarted { command: String }, + + /// One line of output produced while the task's current step runs. + Output { stream: OutputStream, line: String }, + + /// How far a quantifiable task has come, in the unit its [`TaskKind`] + /// declares (e.g. bytes out of [`TaskKind::SnapshotTransfer`]'s + /// `total_bytes`). + Progress { position: u64 }, + + /// 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, + }, + Sync { + canister: String, + canister_id: Principal, + }, + Create { + canister: String, + }, + Install { + canister: String, + canister_id: Principal, + }, + UpdateSettings { + canister: String, + canister_id: Principal, + }, + UpdateEnvironmentVariables { + canister: String, + canister_id: Principal, + }, + CandidCheck { + canister: String, + canister_id: Principal, + }, + SnapshotTransfer { + canister: String, + direction: TransferDirection, + blob: TransferBlob, + total_bytes: u64, + }, +} + +impl TaskKind { + /// The canister this task operates on. + pub fn canister(&self) -> &str { + match self { + TaskKind::Build { canister } + | TaskKind::Sync { canister, .. } + | TaskKind::Create { canister } + | TaskKind::Install { canister, .. } + | TaskKind::UpdateSettings { canister, .. } + | TaskKind::UpdateEnvironmentVariables { canister, .. } + | TaskKind::CandidCheck { canister, .. } + | TaskKind::SnapshotTransfer { canister, .. } => canister, + } + } +} + +/// Which way a snapshot blob is moving relative to the local machine. +#[derive(Debug, Clone, Copy, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum TransferDirection { + Upload, + Download, +} + +/// The snapshot blob being transferred. +#[derive(Debug, Clone, Copy, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum TransferBlob { + WasmModule, + WasmMemory, + StableMemory, +} + +/// 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 { + /// 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, + }, + /// The task did not apply and no work was done (e.g. a Candid + /// compatibility check on an install that is not an upgrade). + Skipped { reason: String }, +} + +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) { + 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, + } + } + + /// Report how far the task has come, emitting [`EventKind::Progress`]. + /// The unit is whatever the task's [`TaskKind`] declares. + pub fn progress(&self, position: u64) { + self.send(EventKind::Progress { position }); + } + + /// 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), + } + } + + fn send(&self, kind: EventKind) { + if let Some(tx) = &self.tx { + let _ = tx.send(Event { + task_id: self.task_id, + kind, + }); + } + } + + /// Report that a shell command within the step began executing, emitting + /// [`EventKind::CommandStarted`]. + pub fn command(&self, command: impl Into) { + self.send(EventKind::CommandStarted { + command: command.into(), + }); + } + + /// 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-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