diff --git a/Cargo.lock b/Cargo.lock index 2260d4dc8..cbc27a49b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3778,8 +3778,8 @@ dependencies = [ name = "icp-events" version = "1.3.0" dependencies = [ - "candid", "serde", + "serde_json", "tokio", ] diff --git a/crates/icp-cli/src/commands/canister/snapshot/download.rs b/crates/icp-cli/src/commands/canister/snapshot/download.rs index 4266303b9..1d3d2dbd9 100644 --- a/crates/icp-cli/src/commands/canister/snapshot/download.rs +++ b/crates/icp-cli/src/commands/canister/snapshot/download.rs @@ -5,8 +5,6 @@ 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; @@ -15,6 +13,8 @@ use crate::operations::snapshot_transfer::{ download_blob_to_file, download_wasm_chunk, load_download_progress, load_metadata, read_snapshot_metadata, save_metadata, }; +use icp_events::Task; + use crate::render::rendered_task; /// Download a snapshot to local disk @@ -126,12 +126,8 @@ pub(crate) async fn exec(ctx: &Context, args: &DownloadArgs) -> Result<(), anyho if !progress.wasm_module.is_complete(metadata.wasm_module_size) { rendered_task( ctx.debug, - TaskKind::SnapshotTransfer { - canister: name.to_string(), - direction: TransferDirection::Download, - blob: TransferBlob::WasmModule, - total_bytes: metadata.wasm_module_size, - }, + Task::counter("WASM module", metadata.wasm_module_size) + .on(name.to_string()), async |task| { download_blob_to_file( &agent, @@ -158,12 +154,8 @@ pub(crate) async fn exec(ctx: &Context, args: &DownloadArgs) -> Result<(), anyho if !progress.wasm_memory.is_complete(metadata.wasm_memory_size) { rendered_task( ctx.debug, - TaskKind::SnapshotTransfer { - canister: name.to_string(), - direction: TransferDirection::Download, - blob: TransferBlob::WasmMemory, - total_bytes: metadata.wasm_memory_size, - }, + Task::counter("WASM memory", metadata.wasm_memory_size) + .on(name.to_string()), async |task| { download_blob_to_file( &agent, @@ -193,12 +185,8 @@ pub(crate) async fn exec(ctx: &Context, args: &DownloadArgs) -> Result<(), anyho { rendered_task( ctx.debug, - TaskKind::SnapshotTransfer { - canister: name.to_string(), - direction: TransferDirection::Download, - blob: TransferBlob::StableMemory, - total_bytes: metadata.stable_memory_size, - }, + Task::counter("Stable memory", metadata.stable_memory_size) + .on(name.to_string()), async |task| { download_blob_to_file( &agent, diff --git a/crates/icp-cli/src/commands/canister/snapshot/upload.rs b/crates/icp-cli/src/commands/canister/snapshot/upload.rs index 3ff674e53..21c3efe72 100644 --- a/crates/icp-cli/src/commands/canister/snapshot/upload.rs +++ b/crates/icp-cli/src/commands/canister/snapshot/upload.rs @@ -8,8 +8,6 @@ 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; @@ -18,6 +16,8 @@ use crate::operations::snapshot_transfer::{ load_metadata, load_upload_progress, save_upload_progress, upload_blob_from_file, upload_snapshot_metadata, upload_wasm_chunk, }; +use icp_events::Task; + use crate::render::rendered_task; /// Upload a snapshot from local disk @@ -131,12 +131,8 @@ pub(crate) async fn exec(ctx: &Context, args: &UploadArgs) -> Result<(), anyhow: if progress.wasm_module_offset < metadata.wasm_module_size { rendered_task( ctx.debug, - TaskKind::SnapshotTransfer { - canister: name.to_string(), - direction: TransferDirection::Upload, - blob: TransferBlob::WasmModule, - total_bytes: metadata.wasm_module_size, - }, + Task::counter("WASM module", metadata.wasm_module_size) + .on(name.to_string()), async |task| { upload_blob_from_file( &agent, @@ -162,12 +158,8 @@ pub(crate) async fn exec(ctx: &Context, args: &UploadArgs) -> Result<(), anyhow: if progress.wasm_memory_offset < metadata.wasm_memory_size { rendered_task( ctx.debug, - TaskKind::SnapshotTransfer { - canister: name.to_string(), - direction: TransferDirection::Upload, - blob: TransferBlob::WasmMemory, - total_bytes: metadata.wasm_memory_size, - }, + Task::counter("WASM memory", metadata.wasm_memory_size) + .on(name.to_string()), async |task| { upload_blob_from_file( &agent, @@ -193,12 +185,8 @@ pub(crate) async fn exec(ctx: &Context, args: &UploadArgs) -> Result<(), anyhow: if progress.stable_memory_offset < metadata.stable_memory_size { rendered_task( ctx.debug, - TaskKind::SnapshotTransfer { - canister: name.to_string(), - direction: TransferDirection::Upload, - blob: TransferBlob::StableMemory, - total_bytes: metadata.stable_memory_size, - }, + Task::counter("Stable memory", metadata.stable_memory_size) + .on(name.to_string()), async |task| { upload_blob_from_file( &agent, diff --git a/crates/icp-cli/src/commands/deploy.rs b/crates/icp-cli/src/commands/deploy.rs index 5983bb7cc..0dee51a41 100644 --- a/crates/icp-cli/src/commands/deploy.rs +++ b/crates/icp-cli/src/commands/deploy.rs @@ -1,10 +1,8 @@ -use anyhow::{anyhow, bail}; +use anyhow::bail; use candid::Principal; use clap::Args; use clap_complete::ArgValueCandidates; -use futures::{StreamExt, future::try_join_all, stream::FuturesOrdered}; use ic_agent::{Agent, AgentError}; -use ic_management_canister_types::{CanisterId, CanisterIdRecord}; use icp::parsers::CyclesAmount; use icp::{ context::{CanisterSelection, Context, EnvironmentSelection}, @@ -12,26 +10,12 @@ 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}; -use std::time::Duration; -use tracing::info; use crate::options::EnvironmentOpt; use crate::{ commands::{args::ArgsOpt, canister::create}, - operations::{ - binding_env_vars::set_binding_env_vars_many, - build::build_many, - candid_compat::check_candid_compatibility_many, - create::{CreateFunding, CreateOperation, CreateTarget}, - install::{install_many, resolve_install_mode_and_status}, - proxy_management, - settings::{sync_controller_dependents, sync_settings_many}, - sync::sync_many, - }, + operations::deploy::{DeployParams, deploy, resolve_targets}, options::{IdentityOpt, arg_struct_change_help}, render::rendered, }; @@ -111,535 +95,54 @@ pub(crate) async fn exec(ctx: &Context, args: &DeployArgs) -> Result<(), anyhow: let environment_selection: EnvironmentSelection = args.environment.0.clone().into(); let identity_selection: IdentitySelection = args.identity.clone().into(); - let env = ctx.get_environment(&environment_selection).await?; - - let mut member_scoped = false; - let cnames: Vec = if args.names.is_empty() { - // No canisters specified: default to the whole environment, unless the - // command is run inside a vendored member — then scope to that member's - // own canisters. (The resolved-root notice is emitted centrally during - // project load.) - let project = ctx.project.load().await?; - let member_dir = ctx.project.member_dir(); - match icp::project::member_scoped_canisters(&project.dir, member_dir.as_deref(), &env) { - Some(scoped) => { - member_scoped = true; - scoped - } - None => env.canisters.keys().cloned().collect(), - } - } else { - // Individual canisters specified. - args.names.clone() - }; + let canisters = resolve_targets(ctx, &environment_selection, &args.names).await?; // Skip doing any work if no canisters are targeted - if cnames.is_empty() { + if canisters.is_empty() { return Ok(()); } - if args.args_opt.is_some() && cnames.len() != 1 { - anyhow::bail!("--args and --args-file can only be used when deploying a single canister"); + if args.args_opt.is_some() && canisters.len() != 1 { + bail!("--args and --args-file can only be used when deploying a single canister"); } - // A member-scoped deploy targets only the sub-project's own canisters, but - // those canisters are wired to their dependencies' ids — and the dependency - // canisters are outside the scope, so they are not (re)deployed here. If any - // are missing from the workspace store, fail fast rather than silently - // deploying an unwired canister. - if member_scoped { - let scoped: HashSet<&str> = cnames.iter().map(String::as_str).collect(); - let deployed: BTreeMap = ctx - .ids_by_environment(&environment_selection) - .await? - .into_iter() - .collect(); - let mut missing: BTreeSet = BTreeSet::new(); - for name in &cnames { - if let Some((_, canister)) = env.canisters.get(name) { - for target in canister.bindings.values() { - if !scoped.contains(target.as_str()) && !deployed.contains_key(target) { - missing.insert(target.clone()); - } - } - } - } - if !missing.is_empty() { - anyhow::bail!( - "this sub-project depends on canister(s) not yet deployed in the workspace: {}. \ - Run `icp deploy` from the workspace root first (or deploy them explicitly by name).", - missing.into_iter().collect::>().join(", ") - ); - } - } - - let canisters_to_build = try_join_all( - cnames - .iter() - .map(|name| ctx.get_canister_and_path_for_env(name, &environment_selection)), - ) - .await?; + let params = DeployParams { + environment: environment_selection.clone(), + identity: identity_selection.clone(), + canisters: canisters.clone(), + mode: args.mode.clone(), + subnet: args.subnet, + proxy: args.proxy, + cycles: args.cycles.get(), + no_create: args.no_create, + yes: args.yes, + init_args: args.args_opt.resolve_bytes()?, + }; - // Build the selected canisters - info!("Building canisters:"); - - 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 + // One reporter for the whole deploy: every phase and every canister lands + // on the same stream, so the renderer can order the run without the + // command having to await it phase by phase. + let report = rendered(ctx.debug, async |reporter| { + deploy(ctx, ¶ms, reporter).await }) .await?; - // Ensure the selected canisters exist, creating any that are missing. - let env = ctx - .get_environment(&environment_selection) - .await - .map_err(|e| anyhow!(e))?; - let agent = ctx - .get_agent_for_env(&identity_selection, &environment_selection) - .await - .map_err(|e| anyhow!(e))?; - let existing_canisters = ctx - .ids_by_environment(&environment_selection) - .await - .map_err(|e| anyhow!(e))?; - let canisters_to_create = cnames - .iter() - .filter(|name| !existing_canisters.contains_key(*name)) - .collect::>(); - - if canisters_to_create.is_empty() { - info!("All canisters already exist"); - } else if args.no_create { - bail!( - "`--no-create` was specified but the following canisters do not exist: {}", - canisters_to_create.iter().format(", ") - ); - } else { - info!("Creating canisters:"); - let target = match (args.subnet, args.proxy) { - (Some(subnet), _) => CreateTarget::Subnet(subnet), - (_, Some(proxy)) => CreateTarget::Proxy(proxy), - _ => CreateTarget::None, - }; - let create_operation = CreateOperation::new( - agent.clone(), - target, - CreateFunding::Cycles(args.cycles.get()), - existing_canisters.into_values().collect(), - ); - 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())), - } - - 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))?; - } - Err(err) => { - error = Some(err.into()); - } - } - idx += 1; - } - if let Some(err) = error { - return Err(err); - } - Ok(()) - }) - .await?; - } - - ctx.update_custom_domains(&environment_selection).await; - - info!("Setting environment variables:"); - let env = ctx - .get_environment(&environment_selection) - .await - .map_err(|e| anyhow!(e))?; - - let env_canisters = &env.canisters; - let target_canisters = try_join_all(cnames.iter().map(|name| { - let environment_selection = environment_selection.clone(); - async move { - let cid = ctx - .get_canister_id_for_env( - &CanisterSelection::Named(name.clone()), - &environment_selection, - ) - .await - .map_err(|e| anyhow!(e))?; - let (_, info) = env_canisters - .get(name) - .ok_or_else(|| anyhow!("Canister id exists but no canister info"))?; - Ok::<_, anyhow::Error>((cid, info.clone())) + // Terminal output is the command's job. The operation reports what it did; + // this decides how to say it. + if !args.json { + for (name, id) in &report.created { + println!("Created canister {name} with ID {id}"); } - })) - .await?; - - let canister_list = ctx - .ids_by_environment(&environment_selection) - .await - .map_err(|e| anyhow!(e))?; - - 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))?; - - rendered(ctx.debug, async |reporter| { - sync_settings_many( - agent.clone(), - args.proxy, - target_canisters, - canister_list, - reporter, - ) - .await - }) - .await - .map_err(|e| anyhow!(e))?; - - // Install the selected canisters - - let canisters = try_join_all(cnames.iter().map(|name| { - let environment_selection = environment_selection.clone(); - let agent = agent.clone(); - async move { - let cid = ctx - .get_canister_id_for_env( - &CanisterSelection::Named(name.clone()), - &environment_selection, - ) - .await - .map_err(|e| anyhow!(e))?; - - let (mode, status) = - resolve_install_mode_and_status(&agent, args.proxy, name, &cid, &args.mode).await?; - - let env = ctx.get_environment(&environment_selection).await?; - let (_canister_path, canister_info) = - env.get_canister_info(name).map_err(|e| anyhow!(e))?; - - // CLI --args/--args-file take priority over manifest init_args - let init_args_bytes = if args.args_opt.is_some() { - args.args_opt.resolve_bytes()? - } else { - canister_info - .init_args - .as_ref() - .map(|ia| ia.to_bytes()) - .transpose()? - }; - - Ok::<_, anyhow::Error>((name.clone(), cid, mode, status, init_args_bytes)) - } - })) - .await?; - - if !args.yes { - info!("Checking compatibility:"); - 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:"); - - rendered(ctx.debug, async |reporter| { - install_many( - agent.clone(), - args.proxy, - canisters, - ctx.artifacts.clone(), - reporter, - ) - .await - }) - .await?; - - // Sync the selected canisters - - // Prepare list of canisters with their info for syncing - let env = ctx - .get_environment(&environment_selection) - .await - .map_err(|e| anyhow!(e))?; - - let env_canisters = &env.canisters; - let sync_canisters = try_join_all(cnames.iter().map(|name| { - let environment_selection = environment_selection.clone(); - async move { - let cid = ctx - .get_canister_id_for_env( - &CanisterSelection::Named(name.clone()), - &environment_selection, - ) - .await - .map_err(|e| anyhow!(e))?; - let (canister_path, info) = env_canisters - .get(name) - .ok_or_else(|| anyhow!("Canister id exists but no canister info"))?; - Ok::<_, anyhow::Error>((cid, canister_path.clone(), info.clone())) - } - })) - .await?; - - // Filter out canisters with no sync steps - let sync_canisters: Vec<_> = sync_canisters - .into_iter() - .filter(|(_, _, info)| !info.sync.steps.is_empty()) - .collect(); - - if sync_canisters.is_empty() { - info!("No canisters have sync steps configured"); - } else { - // Asset sync requires the canister to be Running. install_code is status- - // preserving, so a canister that entered deploy Stopped/Stopping (handed out - // Stopped from a pool, or left so by an earlier interrupted deploy) is still - // not Running here. Start each canister we're about to sync. Per the IC spec - // start_canister is synchronous — its Ok reply means the canister is already - // Running, so no status poll is needed — and idempotent (no-op if Running). - let proxy = args.proxy; - try_join_all(sync_canisters.iter().map(|(cid, _, _)| { - let agent = agent.clone(); - let cid = *cid; - async move { - proxy_management::start_canister( - &agent, - proxy, - CanisterIdRecord { - canister_id: CanisterId::from(cid), - }, - ) - .await - .map_err(|e| anyhow!(e)) - } - })) - .await?; - - // start_canister is synchronous, so each canister is now Running in the - // subnet's *certified* state — but IC query calls are eventually-consistent - // reads, answered by a single replica that may still lag the height at which - // the restart committed and would then observe the just-vacated Stopped state. - // The sync plugin's first calls are queries, so without this wait sync can fail - // with a transient IC0508 right after a restart. Wait until the query path - // consistently sees the canister Running before handing off. - try_join_all(sync_canisters.iter().map(|(cid, _, _)| { - let agent = agent.clone(); - let cid = *cid; - async move { wait_until_serving_queries(&agent, cid).await } - })) - .await?; - - // TODO: When `--proxy` is used and the canister was newly created, the proxy - // canister is its only controller. Sync steps (e.g. asset uploads to a frontend - // canister) will fail because the user's identity lacks the required permissions. - // The fix is to make a proxy call to the frontend canister's `grant_permission` - // method to permit the user identity to upload assets directly before syncing. - info!("Syncing canisters:"); - - let canister_ids: BTreeMap = ctx - .ids_by_environment(&environment_selection) - .await? - .into_iter() - .collect(); - - let pkg_cache = ctx.dirs.package_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 - }) + let agent = ctx + .get_agent_for_env(&identity_selection, &environment_selection) .await?; - } - - // Print URLs for deployed canisters - print_canister_urls( - ctx, - &environment_selection, - agent.clone(), - &cnames, - args.json, - ) - .await?; + print_canister_urls(ctx, &environment_selection, agent, &canisters, args.json).await?; Ok(()) } -/// A method name no real canister exports — used purely as a liveness probe. -/// Querying it is side-effect-free: the replica rejects an unknown method before -/// any canister code runs (no cycles, no logs, no state change), and the reject -/// reason tells us whether the canister is serving queries yet. -const READINESS_PROBE_METHOD: &str = ""; - -/// Wait until the canister's *query* path consistently observes it as Running. -/// -/// After `start_canister` the canister is Running in the subnet's certified -/// state, but query calls are eventually-consistent reads: each is answered by a -/// single replica that may still lag the restart's commit height and would then -/// see the just-vacated Stopped state. The sync plugin's first calls are queries, -/// so without this wait sync can fail with a transient IC0508 right after a -/// restart. -/// -/// We probe with a query for a method no canister exports and classify the result: -/// -/// - a reject of "is stopped"/"is stopping" (IC0508/IC0509) means the replica is -/// still lagging behind the restart. -/// - any other reject (e.g. "no query method"), or a reply, means the replica got -/// far enough to answer for a non-status reason, so it sees the canister Running. -/// - a transport or timeout error is inconclusive. -/// -/// We require a few consecutive ready observations, spaced out so they may land on -/// different replicas, to raise confidence the lagging set has drained. This is not -/// a hard guarantee — query reads are per-node and boundary nodes load-balance -/// across replicas — but it makes the post-restart race rare. -async fn wait_until_serving_queries( - agent: &Agent, - canister_id: Principal, -) -> Result<(), anyhow::Error> { - const REQUIRED_CONSECUTIVE: u32 = 2; - // Total wall-clock budget for the whole wait — the hard cap on the failure - // path. PROBE_TIMEOUT below only bounds a single hung probe (so retries keep - // flowing); this outer budget is what guarantees we give up promptly, rather - // than attempts * (probe timeout + interval). - const READINESS_BUDGET: Duration = Duration::from_secs(30); - const POLL_INTERVAL: Duration = Duration::from_millis(500); - const PROBE_TIMEOUT: Duration = Duration::from_secs(2); - - let poll = async { - let mut consecutive_ready: u32 = 0; - loop { - let probe = agent - .query(&canister_id, READINESS_PROBE_METHOD) - .with_arg(Vec::::new()) - .call(); - let ready = match tokio::time::timeout(PROBE_TIMEOUT, probe).await { - Ok(Ok(_)) => true, // replied -> Running - Ok(Err(err)) => is_serving_reject(&err), // non-stopped reject -> Running - Err(_elapsed) => false, // probe timed out -> inconclusive - }; - - if ready { - consecutive_ready += 1; - if consecutive_ready >= REQUIRED_CONSECUTIVE { - return; - } - } else { - consecutive_ready = 0; - } - tokio::time::sleep(POLL_INTERVAL).await; - } - }; - - match tokio::time::timeout(READINESS_BUDGET, poll).await { - Ok(()) => Ok(()), - Err(_elapsed) => bail!( - "canister {canister_id} did not start serving queries within {}s after being \ - started; the asset sync plugin's first call would fail. Re-run the deploy.", - READINESS_BUDGET.as_secs() - ), - } -} - -/// True when a query error is a *reject from the replica* that indicates the -/// canister is Running and serving — i.e. a positive readiness signal. -/// -/// A reject means the replica processed the request to a verdict (e.g. "no such -/// query method"), so the canister is up — unless the reject says it is -/// stopped/stopping (IC0508/IC0509, with a message-substring fallback), which is -/// a replica still lagging behind the restart. Every other `AgentError` -/// (transport, HTTP, timeout, …) is inconclusive — not evidence the canister is -/// serving — and returns false so the caller retries rather than proceeding. -fn is_serving_reject(err: &AgentError) -> bool { - let reject = match err { - AgentError::CertifiedReject { reject, .. } - | AgentError::UncertifiedReject { reject, .. } => reject, - _ => return false, - }; - let stopped = matches!( - reject.error_code.as_deref(), - Some("IC0508") | Some("IC0509") - ) || reject.reject_message.contains("is stopped") - || reject.reject_message.contains("is stopping"); - !stopped -} - /// Checks whether a canister speaks the HTTP gateway protocol — i.e. exposes an /// `http_request` query method — so we print its gateway (frontend) URL instead /// of a Candid UI URL. diff --git a/crates/icp-cli/src/operations/binding_env_vars.rs b/crates/icp-cli/src/operations/binding_env_vars.rs index 5e1296ef3..4ff9d01a8 100644 --- a/crates/icp-cli/src/operations/binding_env_vars.rs +++ b/crates/icp-cli/src/operations/binding_env_vars.rs @@ -4,7 +4,7 @@ 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 icp_events::{Reporter, Task, TaskOutcome}; use snafu::Snafu; use tracing::error; @@ -111,10 +111,7 @@ pub(crate) async fn set_binding_env_vars_many( let mut futs = FuturesOrdered::new(); for (cid, info) in target_canisters { - let task = reporter.task(TaskKind::UpdateEnvironmentVariables { - canister: info.name.clone(), - canister_id: cid, - }); + let task = reporter.task(Task::new("Updating environment variables").on(&info.name)); // Each canister receives only the ids it is wired to (its own project's // canisters by their local names, plus any declared dependencies under @@ -139,7 +136,9 @@ pub(crate) async fn set_binding_env_vars_many( let result = set_env_vars_for_canister(&agent, proxy, &cid, &info, &binding_vars).await; match &result { - Ok(()) => task.finish(TaskOutcome::succeeded()), + Ok(()) => task.finish(TaskOutcome::succeeded_with( + "Environment variables updated successfully", + )), Err(error) => task.finish(TaskOutcome::failed(error.to_string())), } diff --git a/crates/icp-cli/src/operations/build.rs b/crates/icp-cli/src/operations/build.rs index 8df7bd142..0521ea5d7 100644 --- a/crates/icp-cli/src/operations/build.rs +++ b/crates/icp-cli/src/operations/build.rs @@ -8,7 +8,7 @@ use icp::{ package::PackageCache, prelude::*, }; -use icp_events::{Reporter, StepOutcome, TaskKind, TaskOutcome, TaskReporter}; +use icp_events::{Reporter, StepOutcome, Task, TaskOutcome, TaskReporter}; use snafu::{ResultExt, Snafu}; #[derive(Debug, Snafu)] @@ -99,9 +99,7 @@ pub(crate) async fn build_many( let mut futs = FuturesOrdered::new(); for (canister_path, canister) in canisters { - let task = reporter.task(TaskKind::Build { - canister: canister.name.clone(), - }); + let task = reporter.task(Task::new("Building").on(&canister.name)); let builder = builder.clone(); let artifacts = artifacts.clone(); @@ -118,7 +116,7 @@ pub(crate) async fn build_many( .await; match &result { - Ok(()) => task.finish(TaskOutcome::succeeded()), + Ok(()) => task.finish(TaskOutcome::succeeded_with("Built successfully")), Err(error) => task.finish(TaskOutcome::failed(error.to_string())), } diff --git a/crates/icp-cli/src/operations/candid_compat.rs b/crates/icp-cli/src/operations/candid_compat.rs index b2d7108db..566e903ec 100644 --- a/crates/icp-cli/src/operations/candid_compat.rs +++ b/crates/icp-cli/src/operations/candid_compat.rs @@ -8,7 +8,7 @@ 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 icp_events::{Failure, Reporter, Task, TaskOutcome}; use snafu::Snafu; use tracing::debug; @@ -25,10 +25,7 @@ pub(crate) async fn check_candid_compatibility_many( let mut check_futs = FuturesOrdered::new(); for (name, cid, mode) in canisters { - let task = reporter.task(TaskKind::CandidCheck { - canister: name.to_owned(), - canister_id: cid, - }); + let task = reporter.task(Task::new("Checking Candid compatibility").on(name)); let is_upgrade = matches!(mode, CanisterInstallMode::Upgrade(_)); let agent = agent.clone(); let artifacts = artifacts.clone(); @@ -44,10 +41,20 @@ pub(crate) async fn check_candid_compatibility_many( let result = check_canister_candid_compat(&agent, &cid, name, &*artifacts).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())), + Ok(()) => task.finish(TaskOutcome::succeeded_with("Compatible")), + // The incompatibility report is far too long for a progress + // bar, so the bar gets the verdict and the report is what the + // deferred dump prints. Both are this check's own words: only + // it knows a breaking interface change is what went wrong. + Err(failure) => task.finish(TaskOutcome::Failed( + Failure::new("incompatible interface") + .with_detail(vec![format!( + "You are making a BREAKING change. Other canisters or frontend \ + clients relying on your canister may stop working.\n\n{}", + failure.details, + )]) + .with_epilogue("Use --yes to bypass this check."), + )), } result.map_err(|failure| failure.canister_name) diff --git a/crates/icp-cli/src/operations/deploy.rs b/crates/icp-cli/src/operations/deploy.rs new file mode 100644 index 000000000..8d4e20998 --- /dev/null +++ b/crates/icp-cli/src/operations/deploy.rs @@ -0,0 +1,664 @@ +//! Deploy as one operation: build, create, wire, check, install, sync. +//! +//! Deploy used to be orchestration written into the command, with each phase +//! opening its own renderer and the phase headings printed directly between +//! them — the sequencing was what kept those headings from tearing through a +//! live progress view. Here the whole run is a single task tree on a single +//! event stream: each phase is a group task, and the per-canister tasks the +//! sub-operations start nest under it, because the reporter they are handed +//! is scoped to the phase. They do not know they are being composed. +//! +//! Nothing in here writes to the terminal. Progress goes out as events; +//! results come back on [`DeployReport`] for the command to print. That split +//! is what lets the same orchestration run somewhere without a terminal at +//! all. + +use std::collections::{BTreeMap, BTreeSet, HashSet}; +use std::time::Duration; + +use anyhow::{anyhow, bail}; +use candid::Principal; +use futures::{StreamExt, future::try_join_all, stream::FuturesOrdered}; +use ic_agent::{Agent, AgentError}; +use ic_management_canister_types::{CanisterId, CanisterIdRecord}; +use icp::context::{CanisterSelection, Context, EnvironmentSelection}; +use icp::identity::IdentitySelection; +use icp_events::{Reporter, Task, TaskOutcome, TaskReporter}; +use itertools::Itertools; + +use crate::operations::{ + binding_env_vars::set_binding_env_vars_many, + build::build_many, + candid_compat::check_candid_compatibility_many, + create::{CreateFunding, CreateOperation, CreateTarget}, + install::{install_many, resolve_install_mode_and_status}, + proxy_management, + settings::{sync_controller_dependents, sync_settings_many}, + sync::sync_many, +}; + +/// Everything a deploy needs that the command line supplies. Resolved by the +/// command so this layer never touches clap. +pub(crate) struct DeployParams { + pub(crate) environment: EnvironmentSelection, + pub(crate) identity: IdentitySelection, + /// Canisters to deploy, already resolved from the command line (or from + /// the environment when none were named). + pub(crate) canisters: Vec, + pub(crate) mode: String, + pub(crate) subnet: Option, + pub(crate) proxy: Option, + pub(crate) cycles: u128, + pub(crate) no_create: bool, + /// Skip the Candid interface compatibility check. + pub(crate) yes: bool, + /// Install arguments, already resolved to bytes. Only ever set when a + /// single canister is being deployed. + pub(crate) init_args: Option>, +} + +/// What the deploy did, for the command to report. +pub(crate) struct DeployReport { + /// Canisters created during this deploy, in creation order. + pub(crate) created: Vec<(String, Principal)>, +} + +/// Run a full deploy, reporting progress as one task tree. +pub(crate) async fn deploy( + ctx: &Context, + params: &DeployParams, + reporter: &Reporter, +) -> Result { + let environment_selection = ¶ms.environment; + let cnames = ¶ms.canisters; + + let canisters_to_build = try_join_all( + cnames + .iter() + .map(|name| ctx.get_canister_and_path_for_env(name, environment_selection)), + ) + .await?; + + // Build + let pkg_cache = ctx.dirs.package_cache()?; + let phase = reporter.task(Task::group("Building canisters:")); + let result = build_many( + canisters_to_build, + environment_selection.name(), + ctx.builder.clone(), + ctx.artifacts.clone(), + &pkg_cache, + &phase.reporter(), + ) + .await; + finish(&phase, result)?; + + // Create any canisters that do not exist yet + let env = ctx + .get_environment(environment_selection) + .await + .map_err(|e| anyhow!(e))?; + let agent = ctx + .get_agent_for_env(¶ms.identity, environment_selection) + .await + .map_err(|e| anyhow!(e))?; + let existing_canisters = ctx + .ids_by_environment(environment_selection) + .await + .map_err(|e| anyhow!(e))?; + let canisters_to_create = cnames + .iter() + .filter(|name| !existing_canisters.contains_key(*name)) + .collect::>(); + + let created = if canisters_to_create.is_empty() { + reporter.notice("All canisters already exist"); + Vec::new() + } else if params.no_create { + bail!( + "`--no-create` was specified but the following canisters do not exist: {}", + canisters_to_create.iter().format(", ") + ); + } else { + let phase = reporter.task(Task::group("Creating canisters:")); + let result = create_canisters( + ctx, + params, + &agent, + &env, + &canisters_to_create, + existing_canisters.into_values().collect(), + &phase.reporter(), + ) + .await; + finish(&phase, result)? + }; + + ctx.update_custom_domains(environment_selection).await; + + // Wire canister ids into each other's environment variables, then apply + // manifest settings. + let env = ctx + .get_environment(environment_selection) + .await + .map_err(|e| anyhow!(e))?; + let env_canisters = &env.canisters; + let target_canisters = try_join_all(cnames.iter().map(|name| async move { + let cid = ctx + .get_canister_id_for_env( + &CanisterSelection::Named(name.clone()), + environment_selection, + ) + .await + .map_err(|e| anyhow!(e))?; + let (_, info) = env_canisters + .get(name) + .ok_or_else(|| anyhow!("Canister id exists but no canister info"))?; + Ok::<_, anyhow::Error>((cid, info.clone())) + })) + .await?; + + let canister_list = ctx + .ids_by_environment(environment_selection) + .await + .map_err(|e| anyhow!(e))?; + + let phase = reporter.task(Task::group("Setting environment variables:")); + let result = set_binding_env_vars_many( + agent.clone(), + params.proxy, + &env.name, + target_canisters.clone(), + canister_list.clone(), + &phase.reporter(), + ) + .await + .map_err(|e| anyhow!(e)); + finish(&phase, result)?; + + let phase = reporter.task(Task::group("Applying canister settings:")); + let result = sync_settings_many( + agent.clone(), + params.proxy, + target_canisters, + canister_list, + &phase.reporter(), + ) + .await + .map_err(|e| anyhow!(e)); + finish(&phase, result)?; + + // Resolve install plans + let canisters = try_join_all(cnames.iter().map(|name| { + let agent = agent.clone(); + async move { + let cid = ctx + .get_canister_id_for_env( + &CanisterSelection::Named(name.clone()), + environment_selection, + ) + .await + .map_err(|e| anyhow!(e))?; + + let (mode, status) = + resolve_install_mode_and_status(&agent, params.proxy, name, &cid, ¶ms.mode) + .await?; + + let env = ctx.get_environment(environment_selection).await?; + let (_canister_path, canister_info) = + env.get_canister_info(name).map_err(|e| anyhow!(e))?; + + // Command-line arguments take priority over manifest init_args. + let init_args_bytes = match ¶ms.init_args { + Some(bytes) => Some(bytes.clone()), + None => canister_info + .init_args + .as_ref() + .map(|ia| ia.to_bytes()) + .transpose()?, + }; + + Ok::<_, anyhow::Error>((name.clone(), cid, mode, status, init_args_bytes)) + } + })) + .await?; + + if !params.yes { + let phase = reporter.task(Task::group("Checking Candid compatibility:")); + let result = check_candid_compatibility_many( + agent.clone(), + canisters + .iter() + .map(|(name, cid, mode, _, _)| (&**name, *cid, *mode)), + ctx.artifacts.clone(), + &phase.reporter(), + ) + .await + .map_err(|e| anyhow!(e)); + finish(&phase, result)?; + } + + // Install + let phase = reporter.task(Task::group("Installing canisters:")); + let result = install_many( + agent.clone(), + params.proxy, + canisters, + ctx.artifacts.clone(), + &phase.reporter(), + ) + .await + .map_err(|e| anyhow!(e)); + finish(&phase, result)?; + + sync(ctx, params, &agent, reporter).await?; + + Ok(DeployReport { created }) +} + +/// Create the missing canisters, recording each id as it lands. +async fn create_canisters( + ctx: &Context, + params: &DeployParams, + agent: &Agent, + env: &icp::Environment, + canisters_to_create: &[&String], + existing_ids: Vec, + reporter: &Reporter, +) -> Result, anyhow::Error> { + let target = match (params.subnet, params.proxy) { + (Some(subnet), _) => CreateTarget::Subnet(subnet), + (_, Some(proxy)) => CreateTarget::Proxy(proxy), + _ => CreateTarget::None, + }; + let create_operation = CreateOperation::new( + agent.clone(), + target, + CreateFunding::Cycles(params.cycles), + existing_ids, + ); + + let mut futs = FuturesOrdered::new(); + for name in canisters_to_create.iter() { + let task = reporter.task(Task::new("Creating").on(*name)); + 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_with("Created successfully")), + // The command's returned error reports this; a deferred dump + // would only say it twice. + Err(err) => task.finish(TaskOutcome::failed_silently(err.to_string())), + } + + result + }); + } + + // Cache errors until all futures are processed. Otherwise we risk dropping a canister id. + let mut created = Vec::new(); + 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"); + ctx.set_canister_id_for_env(canister_name, id, ¶ms.environment) + .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, + params.proxy, + canister_name, + ¶ms.environment, + ) + .await + .map_err(|e| anyhow!(e))?; + created.push(((*canister_name).clone(), id)); + } + Err(err) => { + error = Some(err.into()); + } + } + idx += 1; + } + match error { + Some(err) => Err(err), + None => Ok(created), + } +} + +/// Run the sync steps of every canister that has any. +async fn sync( + ctx: &Context, + params: &DeployParams, + agent: &Agent, + reporter: &Reporter, +) -> Result<(), anyhow::Error> { + let environment_selection = ¶ms.environment; + let env = ctx + .get_environment(environment_selection) + .await + .map_err(|e| anyhow!(e))?; + + let env_canisters = &env.canisters; + let sync_canisters = try_join_all(params.canisters.iter().map(|name| async move { + let cid = ctx + .get_canister_id_for_env( + &CanisterSelection::Named(name.clone()), + environment_selection, + ) + .await + .map_err(|e| anyhow!(e))?; + let (canister_path, info) = env_canisters + .get(name) + .ok_or_else(|| anyhow!("Canister id exists but no canister info"))?; + Ok::<_, anyhow::Error>((cid, canister_path.clone(), info.clone())) + })) + .await?; + + // Filter out canisters with no sync steps + let sync_canisters: Vec<_> = sync_canisters + .into_iter() + .filter(|(_, _, info)| !info.sync.steps.is_empty()) + .collect(); + + if sync_canisters.is_empty() { + reporter.notice("No canisters have sync steps configured"); + return Ok(()); + } + + // Asset sync requires the canister to be Running. install_code is status- + // preserving, so a canister that entered deploy Stopped/Stopping (handed out + // Stopped from a pool, or left so by an earlier interrupted deploy) is still + // not Running here. Start each canister we're about to sync. Per the IC spec + // start_canister is synchronous — its Ok reply means the canister is already + // Running, so no status poll is needed — and idempotent (no-op if Running). + let proxy = params.proxy; + try_join_all(sync_canisters.iter().map(|(cid, _, _)| { + let agent = agent.clone(); + let cid = *cid; + async move { + proxy_management::start_canister( + &agent, + proxy, + CanisterIdRecord { + canister_id: CanisterId::from(cid), + }, + ) + .await + .map_err(|e| anyhow!(e)) + } + })) + .await?; + + // start_canister is synchronous, so each canister is now Running in the + // subnet's *certified* state — but IC query calls are eventually-consistent + // reads, answered by a single replica that may still lag the height at which + // the restart committed and would then observe the just-vacated Stopped state. + // The sync plugin's first calls are queries, so without this wait sync can fail + // with a transient IC0508 right after a restart. Wait until the query path + // consistently sees the canister Running before handing off. + try_join_all(sync_canisters.iter().map(|(cid, _, _)| { + let agent = agent.clone(); + let cid = *cid; + async move { wait_until_serving_queries(&agent, cid).await } + })) + .await?; + + // TODO: When `--proxy` is used and the canister was newly created, the proxy + // canister is its only controller. Sync steps (e.g. asset uploads to a frontend + // canister) will fail because the user's identity lacks the required permissions. + // The fix is to make a proxy call to the frontend canister's `grant_permission` + // method to permit the user identity to upload assets directly before syncing. + let canister_ids: BTreeMap = ctx + .ids_by_environment(environment_selection) + .await? + .into_iter() + .collect(); + + let pkg_cache = ctx.dirs.package_cache()?; + + let phase = reporter.task(Task::group("Syncing canisters:")); + let result = sync_many( + ctx.syncer.clone(), + agent.clone(), + sync_canisters, + environment_selection.name().to_owned(), + env.network.name.clone(), + canister_ids, + proxy, + &pkg_cache, + &phase.reporter(), + ) + .await + .map_err(|e| anyhow!(e)); + finish(&phase, result)?; + + Ok(()) +} + +/// Close a phase's group task from its result, and hand the result back. +/// +/// The phase heading carries no failure text of its own: whichever child +/// failed already said what went wrong, and the error itself is on the return +/// path. +fn finish(phase: &TaskReporter, result: Result) -> Result { + match &result { + Ok(_) => phase.finish(TaskOutcome::succeeded()), + Err(error) => phase.finish(TaskOutcome::failed_silently(error.to_string())), + } + result +} + +/// Resolve the canisters a deploy targets, and check that a member-scoped +/// deploy is not about to wire canisters to dependencies that do not exist. +pub(crate) async fn resolve_targets( + ctx: &Context, + environment_selection: &EnvironmentSelection, + named: &[String], +) -> Result, anyhow::Error> { + let env = ctx.get_environment(environment_selection).await?; + + let mut member_scoped = false; + let cnames: Vec = if named.is_empty() { + // No canisters specified: default to the whole environment, unless the + // command is run inside a vendored member — then scope to that member's + // own canisters. (The resolved-root notice is emitted centrally during + // project load.) + let project = ctx.project.load().await?; + let member_dir = ctx.project.member_dir(); + match icp::project::member_scoped_canisters(&project.dir, member_dir.as_deref(), &env) { + Some(scoped) => { + member_scoped = true; + scoped + } + None => env.canisters.keys().cloned().collect(), + } + } else { + named.to_vec() + }; + + // A member-scoped deploy targets only the sub-project's own canisters, but + // those canisters are wired to their dependencies' ids — and the dependency + // canisters are outside the scope, so they are not (re)deployed here. If any + // are missing from the workspace store, fail fast rather than silently + // deploying an unwired canister. + if member_scoped { + let scoped: HashSet<&str> = cnames.iter().map(String::as_str).collect(); + let deployed: BTreeMap = ctx + .ids_by_environment(environment_selection) + .await? + .into_iter() + .collect(); + let mut missing: BTreeSet = BTreeSet::new(); + for name in &cnames { + if let Some((_, canister)) = env.canisters.get(name) { + for target in canister.bindings.values() { + if !scoped.contains(target.as_str()) && !deployed.contains_key(target) { + missing.insert(target.clone()); + } + } + } + } + if !missing.is_empty() { + anyhow::bail!( + "this sub-project depends on canister(s) not yet deployed in the workspace: {}. \ + Run `icp deploy` from the workspace root first (or deploy them explicitly by name).", + missing.into_iter().collect::>().join(", ") + ); + } + } + + Ok(cnames) +} + +/// A method name no real canister exports — used purely as a liveness probe. +/// Querying it is side-effect-free: the replica rejects an unknown method before +/// any canister code runs (no cycles, no logs, no state change), and the reject +/// reason tells us whether the canister is serving queries yet. +const READINESS_PROBE_METHOD: &str = ""; + +/// Wait until the canister's *query* path consistently observes it as Running. +/// +/// After `start_canister` the canister is Running in the subnet's certified +/// state, but query calls are eventually-consistent reads: each is answered by a +/// single replica that may still lag the restart's commit height and would then +/// see the just-vacated Stopped state. The sync plugin's first calls are queries, +/// so without this wait sync can fail with a transient IC0508 right after a +/// restart. +/// +/// We probe with a query for a method no canister exports and classify the result: +/// +/// - a reject of "is stopped"/"is stopping" (IC0508/IC0509) means the replica is +/// still lagging behind the restart. +/// - any other reject (e.g. "no query method"), or a reply, means the replica got +/// far enough to answer for a non-status reason, so it sees the canister Running. +/// - a transport or timeout error is inconclusive. +/// +/// We require a few consecutive ready observations, spaced out so they may land on +/// different replicas, to raise confidence the lagging set has drained. This is not +/// a hard guarantee — query reads are per-node and boundary nodes load-balance +/// across replicas — but it makes the post-restart race rare. +async fn wait_until_serving_queries( + agent: &Agent, + canister_id: Principal, +) -> Result<(), anyhow::Error> { + const REQUIRED_CONSECUTIVE: u32 = 2; + // Total wall-clock budget for the whole wait — the hard cap on the failure + // path. PROBE_TIMEOUT below only bounds a single hung probe (so retries keep + // flowing); this outer budget is what guarantees we give up promptly, rather + // than attempts * (probe timeout + interval). + const READINESS_BUDGET: Duration = Duration::from_secs(30); + const POLL_INTERVAL: Duration = Duration::from_millis(500); + const PROBE_TIMEOUT: Duration = Duration::from_secs(2); + + let poll = async { + let mut consecutive_ready: u32 = 0; + loop { + let probe = agent + .query(&canister_id, READINESS_PROBE_METHOD) + .with_arg(Vec::::new()) + .call(); + let ready = match tokio::time::timeout(PROBE_TIMEOUT, probe).await { + Ok(Ok(_)) => true, // replied -> Running + Ok(Err(err)) => is_serving_reject(&err), // non-stopped reject -> Running + Err(_elapsed) => false, // probe timed out -> inconclusive + }; + + if ready { + consecutive_ready += 1; + if consecutive_ready >= REQUIRED_CONSECUTIVE { + return; + } + } else { + consecutive_ready = 0; + } + tokio::time::sleep(POLL_INTERVAL).await; + } + }; + + match tokio::time::timeout(READINESS_BUDGET, poll).await { + Ok(()) => Ok(()), + Err(_elapsed) => bail!( + "canister {canister_id} did not start serving queries within {}s after being \ + started; the asset sync plugin's first call would fail. Re-run the deploy.", + READINESS_BUDGET.as_secs() + ), + } +} + +/// True when a query error is a *reject from the replica* that indicates the +/// canister is Running and serving — i.e. a positive readiness signal. +/// +/// A reject means the replica processed the request to a verdict (e.g. "no such +/// query method"), so the canister is up — unless the reject says it is +/// stopped/stopping (IC0508/IC0509, with a message-substring fallback), which is +/// a replica still lagging behind the restart. Every other `AgentError` +/// (transport, HTTP, timeout, …) is inconclusive — not evidence the canister is +/// serving — and returns false so the caller retries rather than proceeding. +fn is_serving_reject(err: &AgentError) -> bool { + let reject = match err { + AgentError::CertifiedReject { reject, .. } + | AgentError::UncertifiedReject { reject, .. } => reject, + _ => return false, + }; + let stopped = matches!( + reject.error_code.as_deref(), + Some("IC0508") | Some("IC0509") + ) || reject.reject_message.contains("is stopped") + || reject.reject_message.contains("is stopping"); + !stopped +} + +#[cfg(test)] +mod tests { + use super::*; + use ic_agent::agent::{RejectCode, RejectResponse}; + + fn reject(error_code: Option<&str>, reject_message: &str) -> AgentError { + AgentError::UncertifiedReject { + reject: RejectResponse { + reject_code: RejectCode::CanisterError, + reject_message: reject_message.to_string(), + error_code: error_code.map(String::from), + }, + operation: None, + } + } + + #[test] + fn stopped_rejects_are_not_a_readiness_signal() { + // A replica still lagging behind the restart. + assert!(!is_serving_reject(&reject( + Some("IC0508"), + "Canister abc is stopped" + ))); + assert!(!is_serving_reject(&reject( + None, + "Canister abc is stopping" + ))); + } + + #[test] + fn any_other_reject_means_the_canister_is_serving() { + // The replica got far enough to answer for a non-status reason. + assert!(is_serving_reject(&reject( + Some("IC0536"), + "Canister abc has no query method ''" + ))); + } + + #[test] + fn a_transport_error_is_inconclusive() { + // Not evidence of anything; the caller must retry. + assert!(!is_serving_reject(&AgentError::InvalidReplicaStatus)); + } +} diff --git a/crates/icp-cli/src/operations/install.rs b/crates/icp-cli/src/operations/install.rs index cc79c9c8d..950b603a2 100644 --- a/crates/icp-cli/src/operations/install.rs +++ b/crates/icp-cli/src/operations/install.rs @@ -6,7 +6,7 @@ use ic_management_canister_types::{ ClearChunkStoreArgs, InstallChunkedCodeArgs, InstallCodeArgs, UpgradeFlags, UploadChunkArgs, WasmMemoryPersistence, }; -use icp_events::{Reporter, TaskKind, TaskOutcome}; +use icp_events::{Reporter, Task, TaskOutcome}; use sha2::{Digest, Sha256}; use snafu::{ResultExt, Snafu}; use std::sync::Arc; @@ -360,10 +360,7 @@ pub(crate) async fn install_many( let mut futs = FuturesOrdered::new(); for (name, cid, mode, status, init_args) in canisters { - let task = reporter.task(TaskKind::Install { - canister: name.clone(), - canister_id: cid, - }); + let task = reporter.task(Task::new("Installing").on(&name)); let agent = agent.clone(); let artifacts = artifacts.clone(); @@ -391,7 +388,7 @@ pub(crate) async fn install_many( .await; match &result { - Ok(()) => task.finish(TaskOutcome::succeeded()), + Ok(()) => task.finish(TaskOutcome::succeeded_with("Installed successfully")), Err(error) => task.finish(TaskOutcome::failed(error.to_string())), } diff --git a/crates/icp-cli/src/operations/mod.rs b/crates/icp-cli/src/operations/mod.rs index 918b80613..55e3f8f81 100644 --- a/crates/icp-cli/src/operations/mod.rs +++ b/crates/icp-cli/src/operations/mod.rs @@ -5,6 +5,7 @@ pub(crate) mod call_output; pub(crate) mod candid_compat; pub(crate) mod canister_migration; pub(crate) mod create; +pub(crate) mod deploy; pub(crate) mod install; pub(crate) mod proxy; pub(crate) mod proxy_management; diff --git a/crates/icp-cli/src/operations/settings.rs b/crates/icp-cli/src/operations/settings.rs index 083fb31e8..2d6ec9261 100644 --- a/crates/icp-cli/src/operations/settings.rs +++ b/crates/icp-cli/src/operations/settings.rs @@ -15,7 +15,7 @@ use icp::{ context::{Context, EnvironmentSelection}, store_id::IdMapping, }; -use icp_events::{Reporter, TaskKind, TaskOutcome}; +use icp_events::{Reporter, Task, TaskOutcome}; use itertools::Itertools; use num_traits::ToPrimitive; use snafu::{ResultExt, Snafu}; @@ -224,10 +224,7 @@ pub(crate) async fn sync_settings_many( let ids = Arc::new(ids); for (cid, info) in target_canisters { - let task = reporter.task(TaskKind::UpdateSettings { - canister: info.name.clone(), - canister_id: cid, - }); + let task = reporter.task(Task::new("Updating canister settings").on(&info.name)); let agent = agent.clone(); let ids = ids.clone(); @@ -246,7 +243,9 @@ pub(crate) async fn sync_settings_many( .await; match &result { - Ok(()) => task.finish(TaskOutcome::succeeded()), + Ok(()) => task.finish(TaskOutcome::succeeded_with( + "Canister settings updated successfully", + )), Err(error) => task.finish(TaskOutcome::failed(error.to_string())), } diff --git a/crates/icp-cli/src/operations/sync.rs b/crates/icp-cli/src/operations/sync.rs index 8cab97fff..c8e76df90 100644 --- a/crates/icp-cli/src/operations/sync.rs +++ b/crates/icp-cli/src/operations/sync.rs @@ -7,7 +7,7 @@ use icp::{ package::PackageCache, prelude::PathBuf, }; -use icp_events::{Reporter, StepOutcome, TaskKind, TaskOutcome, TaskReporter}; +use icp_events::{Failure, Reporter, StepOutcome, Task, TaskOutcome, TaskReporter}; use snafu::prelude::*; use std::collections::BTreeMap; use std::sync::Arc; @@ -93,10 +93,7 @@ pub(crate) async fn sync_many( let mut futs = FuturesOrdered::new(); for (cid, canister_path, canister_info) in canisters { - let task = reporter.task(TaskKind::Sync { - canister: canister_info.name.clone(), - canister_id: cid, - }); + let task = reporter.task(Task::new("Syncing").on(&canister_info.name)); let fut = { let agent = agent.clone(); @@ -126,12 +123,12 @@ pub(crate) async fn sync_many( // rolling step view discards them on success, but they // belong on the persistent output channel. Ok(stderr_lines) => task.finish(TaskOutcome::Succeeded { + message: Some(format!("Synced successfully: {cid}")), retained_output: stderr_lines.clone(), }), - Err(error) => task.finish(TaskOutcome::Failed { - message: error.to_string(), - causes: error_causes(error), - }), + Err(error) => task.finish(TaskOutcome::Failed( + Failure::new(error.to_string()).with_causes(error_causes(error)), + )), } result.map(|_| ()).map_err(|_| canister_info.name.clone()) diff --git a/crates/icp-cli/src/render/interactive.rs b/crates/icp-cli/src/render/interactive.rs index 389962bcd..f43bb6916 100644 --- a/crates/icp-cli/src/render/interactive.rs +++ b/crates/icp-cli/src/render/interactive.rs @@ -1,21 +1,20 @@ -//! Live progress-bar renderer: one indicatif spinner per task, a rolling +//! Live progress-bar renderer: one indicatif widget 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. +//! Nested tasks indent under their parent, and group tasks print as plain +//! headings above the bars. 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 icp_events::{Event, EventKind, Shape, TaskId, TaskOutcome}; use indicatif::{MultiProgress, ProgressBar, ProgressStyle}; use itertools::Itertools; -use tracing::debug; +use tracing::{debug, info}; 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, -}; +use super::{INDENT, RollingLines, TaskInfo, TaskLog, dump_failures, retained_lines}; /// Number of output lines shown live under a task's progress bar. const LIVE_WINDOW_LINES: usize = 4; @@ -27,7 +26,9 @@ pub(crate) struct InteractiveRenderer { struct TaskView { log: TaskLog, - bar: ProgressBar, + /// The live widget, or `None` for a group — a heading has no live state, + /// so it is printed once above the bars rather than animated. + bar: Option, /// 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. @@ -47,39 +48,86 @@ impl InteractiveRenderer { } } + /// How deeply a task nests, from its parent's depth. A task whose parent + /// is unknown is treated as top-level. + fn depth_of(&self, parent: Option) -> usize { + parent + .and_then(|id| self.tasks.get(&id)) + .map(|view| view.log.info().depth + 1) + .unwrap_or(0) + } + + /// Announce a heading. + /// + /// A top-level heading closes the live view first: dropping the + /// `MultiProgress` leaves its last frame on the terminal, so the bars + /// above become scrollback and the heading lands beneath them, with the + /// next phase's bars drawing below that. Without this the finished bars + /// would be redrawn *under* each new heading and the run would read out + /// of order. Nested headings have live siblings to preserve, so they are + /// written through the live view instead. + /// + /// The heading goes out over tracing rather than as a bar so it survives + /// a hidden draw target — under a pipe or in CI there are no bars, but + /// the phase headings still belong in the log. + fn heading(&mut self, depth: usize, title: &str) { + let line = format!("{}{title}", INDENT.repeat(depth)); + if depth == 0 { + self.multi_progress = MultiProgress::new(); + info!("{line}"); + } else { + self.multi_progress.suspend(|| info!("{line}")); + } + } + pub(crate) fn handle(&mut self, event: Event) { match event.kind { - EventKind::TaskStarted { task } => { + EventKind::TaskStarted { + parent, + subject, + title, + shape, + } => { + let info = TaskInfo { + subject, + title, + shape, + depth: self.depth_of(parent), + }; + // 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)), + // MultiProgress switches the draw target, and `set_style` + // adapts a style to stderr only while the bar is still + // detached. + let bar = match info.shape { + // A heading has no live state; it just titles whatever + // nests beneath it. + Shape::Group => { + self.heading(info.depth, &info.title); + None + } + Shape::Counter { total } => Some( + self.multi_progress.add( + ProgressBar::new(total) + .with_style(counter_style()) + .with_prefix(info.widget_prefix()), + ), ), - _ => { - let mut bar = ProgressBar::new_spinner() + Shape::Spinner => { + let 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); - } + .with_prefix(info.widget_prefix()) + .with_message(info.running_message()); let bar = self.multi_progress.add(bar); bar.enable_steady_tick(Duration::from_millis(120)); - bar + Some(bar) } }; self.tasks.insert( event.task_id, TaskView { - log: TaskLog::new(task), + log: TaskLog::new(info), bar, header: String::new(), headline: String::new(), @@ -89,8 +137,10 @@ impl InteractiveRenderer { } EventKind::Progress { position } => { - if let Some(view) = self.tasks.get(&event.task_id) { - view.bar.set_position(position); + if let Some(view) = self.tasks.get(&event.task_id) + && let Some(bar) = &view.bar + { + bar.set_position(position); } } @@ -102,7 +152,7 @@ impl InteractiveRenderer { let Some(view) = self.tasks.get_mut(&event.task_id) else { return; }; - view.header = step_header(view.log.kind(), number, total, &label); + view.header = view.log.info().step_header(number, total, &label); view.headline = view .header .lines() @@ -125,7 +175,9 @@ impl InteractiveRenderer { // 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()); + if let Some(bar) = &view.bar { + bar.set_message(view.header.clone()); + } } EventKind::Output { line, .. } => { @@ -133,7 +185,15 @@ impl InteractiveRenderer { return; }; - debug!("[{}] {line}", view.log.kind().canister()); + // A group has no live window to roll; its output is a notice, + // so it goes out as a line of its own. + let Some(bar) = &view.bar else { + let line = view.log.info().line(&line); + self.multi_progress.suspend(|| info!("{line}")); + return; + }; + + debug!("{}", view.log.info().line(&line)); view.window.push(line.clone()); view.log.push_line(line); @@ -143,8 +203,7 @@ impl InteractiveRenderer { // │ 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)); + bar.set_message(format!("{}\n{rolled}\n└\n\n", view.header)); } EventKind::StepCompleted { .. } => { @@ -158,34 +217,42 @@ impl InteractiveRenderer { 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::Succeeded { + message, + retained_output, + } => { + if let Some(bar) = &view.bar { + // A counter's template has no message slot, so a + // task driving one leaves the message unset and + // the bar simply freezes at its final position. + if let Some(message) = message { + bar.set_style(make_style(TICK_SUCCESS, COLOR_SUCCESS)); + bar.set_message(message); + } + bar.finish(); + } + let lines = retained_lines(view.log.info(), &retained_output); + self.multi_progress.suspend(|| { + for line in lines { + eprintln!("{line}"); + } + }); } - 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); + TaskOutcome::Failed(failure) => { + if let Some(bar) = &view.bar { + bar.set_style(make_style(TICK_FAILURE, COLOR_FAILURE)); + bar.set_message(view.log.info().failure_message(&failure.message)); + bar.finish(); + } + view.log.fail(failure); } // Skipped keeps the neutral style — nothing succeeded or // failed. TaskOutcome::Skipped { reason } => { - view.bar.finish_with_message(format!("Skipped ({reason})")); + if let Some(bar) = &view.bar { + bar.finish_with_message(format!("Skipped ({reason})")); + } } } } @@ -204,8 +271,8 @@ impl InteractiveRenderer { } } -/// Style for a byte-transfer bar. -fn transfer_style() -> ProgressStyle { +/// Style for a determinate byte counter. +fn counter_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") diff --git a/crates/icp-cli/src/render/mod.rs b/crates/icp-cli/src/render/mod.rs index 84df894b4..884d54396 100644 --- a/crates/icp-cli/src/render/mod.rs +++ b/crates/icp-cli/src/render/mod.rs @@ -1,14 +1,22 @@ //! 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. +//! Operations emit events through a [`Reporter`]; a [`Renderer`] consumes the +//! stream and owns everything user-facing. Commands pick a renderer with +//! [`Renderer::for_ctx`] and drive it with [`Renderer::run`] alongside the +//! operation. +//! +//! The event stream carries no operation vocabulary — a task announces itself +//! with a title, an optional subject, and the shape of its widget. Everything +//! below composes display text from those three facts and nothing else, so a +//! new kind of work needs no change here at all. The chrome (brackets, +//! dashes, indentation, tick marks) is the renderer's; the nouns are the +//! operation's. use std::collections::{BTreeMap, VecDeque}; -use icp_events::{Event, Reporter, TaskId, TaskKind, TaskOutcome, TaskReporter, TransferBlob}; +use icp_events::{ + Event, Failure, FailureReport, Reporter, Shape, Task, TaskId, TaskOutcome, TaskReporter, +}; use tokio::sync::mpsc::UnboundedReceiver; use tracing::error; @@ -24,6 +32,9 @@ 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; +/// Indentation applied per level of task nesting. +const INDENT: &str = " "; + pub(crate) enum Renderer { Interactive(InteractiveRenderer), Plain(PlainRenderer), @@ -61,10 +72,10 @@ impl Renderer { } } -/// 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. +/// Run an operation 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)); @@ -77,21 +88,23 @@ pub(crate) async fn rendered(debug: bool, op: impl AsyncFnOnce(&Reporter) -> 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 +/// Run a single task under its own renderer: starts `task`, 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, + task: Task, op: impl AsyncFnOnce(&TaskReporter) -> Result, ) -> Result { rendered(debug, async |reporter| { - let task = reporter.task(kind); + let task = reporter.task(task); let result = op(&task).await; match &result { Ok(_) => task.finish(TaskOutcome::succeeded()), - Err(error) => task.finish(TaskOutcome::failed(error.to_string())), + // The command's returned error reports this; a dump would only + // say it twice. + Err(error) => task.finish(TaskOutcome::failed_silently(error.to_string())), } result @@ -99,143 +112,65 @@ pub(crate) async fn rendered_task( .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..."), - } +/// What a task said about itself, plus where it sits in the task tree. Every +/// piece of display text is composed from these. +pub(super) struct TaskInfo { + subject: Option, + title: String, + shape: Shape, + /// Levels of nesting below the root, for indentation. + depth: usize, } -/// 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", +impl TaskInfo { + /// Prefix each line of captured or retained output so concurrent tasks + /// stay attributable. Tasks with no subject contribute no prefix. + fn line(&self, text: &str) -> String { + match &self.subject { + Some(subject) => format!("[{subject}] {text}"), + None => text.to_owned(), + } } -} -/// 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 task's live widget. A counter's bar is too narrow to + /// carry both, so it is labelled by what it is transferring; everything + /// else is labelled by what it is working on. + fn widget_prefix(&self) -> String { + let indent = INDENT.repeat(self.depth); + match (&self.shape, &self.subject) { + (Shape::Counter { .. }, _) | (_, None) => format!("{indent}{}", self.title), + (_, Some(subject)) => format!("{indent}[{subject}]"), + } } -} -/// 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", + /// Message shown while the task runs, before any step reports in. + fn running_message(&self) -> String { + format!("{}...", self.title) } -} -/// 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(), + /// Live header shown while a step runs. `label` may span multiple lines. + fn step_header(&self, number: usize, total: usize, label: &str) -> String { + format!("{}: step {number} of {total} {label}", self.title) } -} -/// 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(), + /// Final widget message on failure. + fn failure_message(&self, message: &str) -> String { + format!("{} failed: {message}", self.title) } -} -/// 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}' -----")) + /// First line of the task's deferred failure dump. + fn failure_header(&self) -> String { + match &self.subject { + Some(subject) => format!("----- {} failed: '{subject}' -----", self.title), + None => format!("----- {} failed -----", self.title), } - 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, + info: TaskInfo, finished_steps: Vec, current_step: Option, failure: Option, @@ -246,11 +181,6 @@ struct StepLog { lines: RollingLines, } -struct Failure { - message: String, - causes: Vec, -} - /// A fixed-capacity rolling buffer that always holds the last `capacity` items. #[derive(Debug)] struct RollingLines { @@ -286,17 +216,17 @@ impl RollingLines { } impl TaskLog { - fn new(kind: TaskKind) -> Self { + fn new(info: TaskInfo) -> Self { Self { - kind, + info, finished_steps: Vec::new(), current_step: None, failure: None, } } - fn kind(&self) -> &TaskKind { - &self.kind + fn info(&self) -> &TaskInfo { + &self.info } fn start_step(&mut self, title: String) { @@ -320,23 +250,20 @@ impl TaskLog { } } - fn fail(&mut self, message: String, causes: Vec) { - self.failure = Some(Failure { message, causes }); + fn fail(&mut self, failure: Failure) { + self.failure = Some(failure); } /// 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. + /// shown. Tasks that never reported a step 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 info = &self.info; + let mut lines = vec![info.line(&format!("{} output:", info.title))]; let steps: &[StepLog] = if all_steps { &self.finished_steps @@ -350,14 +277,18 @@ impl TaskLog { for step in steps { for line in step.title.lines() { if !line.is_empty() { - lines.push(format!("[{name}] {line}:")); + lines.push(info.line(&format!("{line}:"))); } } if step.lines.is_empty() { - lines.push(format!("[{name}] ")); + lines.push(info.line("")); } else { - lines.extend(step.lines.iter().map(|line| format!("[{name}] > {line}"))); + lines.extend( + step.lines + .iter() + .map(|line| info.line(&format!("> {line}"))), + ); } } @@ -365,41 +296,187 @@ impl TaskLog { } } -/// Print the failure dump for every failed task, in task-creation order. +/// Print output lines a task retained past its rolling step view (e.g. +/// sync-plugin stderr), attributed to the task that produced them. +fn retained_lines(info: &TaskInfo, lines: &[String]) -> Vec { + lines.iter().map(|line| info.line(line)).collect() +} + +/// Print the failure dump for every failed task, in task-creation order, +/// followed by any epilogues the failures asked for (each printed once). fn dump_failures(logs: &BTreeMap, all_steps: bool) { - let mut candid_failures = false; + let mut epilogues: Vec<&str> = Vec::new(); 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, + let body = match &failure.report { + // The command's returned error already carries this one. + FailureReport::Silent => continue, + FailureReport::Summary => { + let mut body = vec![format!("'{}'", failure.message)]; + body.extend( + failure + .causes + .iter() + .map(|cause| format!(" caused by: {cause}")), ); + body } - _ => { - error!("'{}'", failure.message); - for cause in &failure.causes { - error!(" caused by: {cause}"); - } - } + FailureReport::Detail { lines } => lines.clone(), + }; + + error!("{}", log.info().failure_header()); + for line in body { + error!("{line}"); } for line in log.dump(all_steps) { error!("{line}"); } + + if let Some(epilogue) = failure.epilogue.as_deref() + && !epilogues.contains(&epilogue) + { + epilogues.push(epilogue); + } + } + + for epilogue in epilogues { + error!("{epilogue}"); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn info(title: &str, subject: Option<&str>, depth: usize) -> TaskInfo { + TaskInfo { + subject: subject.map(str::to_owned), + title: title.to_owned(), + shape: Shape::Spinner, + depth, + } + } + + /// Every line of display text is composed from title, subject and shape. + /// These are the compositions, spelled out so a change to any of them is + /// a deliberate one. + #[test] + fn display_text_is_composed_from_title_and_subject() { + let build = info("Building", Some("frontend"), 0); + + assert_eq!(build.widget_prefix(), "[frontend]"); + assert_eq!(build.running_message(), "Building..."); + assert_eq!( + build.step_header(1, 3, "(script)"), + "Building: step 1 of 3 (script)" + ); + assert_eq!(build.line("> hello"), "[frontend] > hello"); + assert_eq!(build.failure_message("exit 1"), "Building failed: exit 1"); + assert_eq!( + build.failure_header(), + "----- Building failed: 'frontend' -----" + ); + } + + /// A task with no subject contributes no `[name]` prefix, and is labelled + /// by what it is doing instead. + #[test] + fn a_subjectless_task_is_labelled_by_its_title() { + let bundle = info("Bundling", None, 0); + + assert_eq!(bundle.widget_prefix(), "Bundling"); + assert_eq!(bundle.line("> hello"), "> hello"); + assert_eq!(bundle.failure_header(), "----- Bundling failed -----"); + } + + /// A counter's bar has no room for both, so it is labelled by what is + /// moving rather than by the canister it belongs to. + #[test] + fn a_counter_is_labelled_by_its_title_even_with_a_subject() { + let transfer = TaskInfo { + shape: Shape::Counter { total: 4096 }, + ..info("WASM module", Some("frontend"), 0) + }; + + assert_eq!(transfer.widget_prefix(), "WASM module"); + // Output still attributes to the canister. + assert_eq!(transfer.line("note"), "[frontend] note"); + } + + #[test] + fn nesting_indents_the_widget_label() { + assert_eq!( + info("Installing", Some("frontend"), 1).widget_prefix(), + " [frontend]" + ); + assert_eq!(info("Deploying", None, 2).widget_prefix(), " Deploying"); + } + + #[test] + fn a_dump_reproduces_the_captured_step_output() { + let mut log = TaskLog::new(info("Building", Some("my-canister"), 0)); + log.start_step("Building: step 1 of 2 (script)".to_owned()); + log.push_line("hidden".to_owned()); + log.end_step(); + log.start_step("Building: step 2 of 2 (script)".to_owned()); + log.push_line("boom".to_owned()); + log.end_step(); + + assert_eq!( + log.dump(false), + vec![ + "[my-canister] Building output:", + "[my-canister] Building: step 2 of 2 (script):", + "[my-canister] > boom", + ] + ); + assert_eq!( + log.dump(true), + vec![ + "[my-canister] Building output:", + "[my-canister] Building: step 1 of 2 (script):", + "[my-canister] > hidden", + "[my-canister] Building: step 2 of 2 (script):", + "[my-canister] > boom", + ], + "every step is replayed under --debug" + ); + } + + /// A step that produced nothing still says so, rather than rendering as a + /// bare header. + #[test] + fn a_silent_step_is_reported_as_such() { + let mut log = TaskLog::new(info("Syncing", Some("frontend"), 0)); + log.start_step("Syncing: step 1 of 1 (assets)".to_owned()); + log.end_step(); + + assert_eq!( + log.dump(false), + vec![ + "[frontend] Syncing output:", + "[frontend] Syncing: step 1 of 1 (assets):", + "[frontend] ", + ] + ); + } + + /// Tasks that never reported a step have nothing to replay. + #[test] + fn a_stepless_task_dumps_nothing() { + let log = TaskLog::new(info("Installing", Some("frontend"), 0)); + assert!(log.dump(true).is_empty()); } - if candid_failures { - error!("Use --yes to bypass this check."); + #[test] + fn retained_output_is_attributed_to_its_task() { + let info = info("Syncing", Some("frontend"), 0); + assert_eq!( + retained_lines(&info, &["one".to_owned(), "two".to_owned()]), + vec!["[frontend] one", "[frontend] two"] + ); } } diff --git a/crates/icp-cli/src/render/plain.rs b/crates/icp-cli/src/render/plain.rs index ffc0faac7..73efddf95 100644 --- a/crates/icp-cli/src/render/plain.rs +++ b/crates/icp-cli/src/render/plain.rs @@ -4,10 +4,10 @@ use std::collections::BTreeMap; -use icp_events::{Event, EventKind, TaskId, TaskOutcome}; -use tracing::debug; +use icp_events::{Event, EventKind, Shape, TaskId, TaskOutcome}; +use tracing::{debug, info}; -use super::{TaskLog, dump_failures, step_header}; +use super::{INDENT, TaskInfo, TaskLog, dump_failures, retained_lines}; pub(crate) struct PlainRenderer { tasks: BTreeMap, @@ -20,10 +20,35 @@ impl PlainRenderer { } } + /// How deeply a task nests, from its parent's depth. A task whose parent + /// is unknown is treated as top-level. + fn depth_of(&self, parent: Option) -> usize { + parent + .and_then(|id| self.tasks.get(&id)) + .map(|log| log.info().depth + 1) + .unwrap_or(0) + } + pub(crate) fn handle(&mut self, event: Event) { match event.kind { - EventKind::TaskStarted { task } => { - self.tasks.insert(event.task_id, TaskLog::new(task)); + EventKind::TaskStarted { + parent, + subject, + title, + shape, + } => { + let info = TaskInfo { + subject, + title, + shape, + depth: self.depth_of(parent), + }; + // A heading has no live state to animate, so with the bars + // gone it is simply a line. + if matches!(info.shape, Shape::Group) { + info!("{}{}", INDENT.repeat(info.depth), info.title); + } + self.tasks.insert(event.task_id, TaskLog::new(info)); } EventKind::StepStarted { @@ -34,7 +59,7 @@ impl PlainRenderer { let Some(log) = self.tasks.get_mut(&event.task_id) else { return; }; - let header = step_header(log.kind(), number, total, &label); + let header = log.info().step_header(number, total, &label); log.start_step(header); } @@ -44,16 +69,22 @@ impl PlainRenderer { }; // Mark command boundaries so interleaved output stays // attributable to the command producing it. - debug!("[{}] $ {command}", log.kind().canister()); + debug!("{}", log.info().line(&format!("$ {command}"))); } EventKind::Output { line, .. } => { let Some(log) = self.tasks.get_mut(&event.task_id) else { return; }; - // Prefix with the canister so interleaved concurrent tasks + // A group's output is a notice rather than tool output, so it + // stays on the user-facing channel. + if matches!(log.info().shape, Shape::Group) { + info!("{}", log.info().line(&line)); + return; + } + // Prefix with the subject so interleaved concurrent tasks // stay attributable. - debug!("[{}] {line}", log.kind().canister()); + debug!("{}", log.info().line(&line)); log.push_line(line); } @@ -71,11 +102,15 @@ impl PlainRenderer { return; }; match outcome { - TaskOutcome::Succeeded { retained_output } => { - super::print_retained(log.kind(), &retained_output); + TaskOutcome::Succeeded { + retained_output, .. + } => { + for line in retained_lines(log.info(), &retained_output) { + eprintln!("{line}"); + } } - TaskOutcome::Failed { message, causes } => { - log.fail(message, causes); + TaskOutcome::Failed(failure) => { + log.fail(failure); } TaskOutcome::Skipped { .. } => {} } diff --git a/crates/icp-cli/tests/assets/__pycache__/limit_transfer.cpython-312.pyc b/crates/icp-cli/tests/assets/__pycache__/limit_transfer.cpython-312.pyc new file mode 100644 index 000000000..29f27c431 Binary files /dev/null and b/crates/icp-cli/tests/assets/__pycache__/limit_transfer.cpython-312.pyc differ diff --git a/crates/icp-cli/tests/build_tests.rs b/crates/icp-cli/tests/build_tests.rs index 2e5554d0e..12fe9aafb 100644 --- a/crates/icp-cli/tests/build_tests.rs +++ b/crates/icp-cli/tests/build_tests.rs @@ -175,9 +175,9 @@ fn build_adapter_display_failing_build_output() { // Invoke build let expected_output = indoc! {r#" - ERR ----- Failed to build canister 'my-canister' ----- + ERR ----- Building failed: 'my-canister' ----- ERR 'command 'for i in $(seq 1 5); do echo "failing build step $i"; done; exit 1' failed with status code 1' - ERR [my-canister] Build output: + ERR [my-canister] Building output: ERR [my-canister] Building: step 3 of 3 (script): ERR [my-canister] for i in $(seq 1 5); do echo "failing build step $i"; done; exit 1: ERR [my-canister] > failing build step 1 @@ -228,9 +228,9 @@ fn build_adapter_display_failing_middle_step_output() { // Only step 2 output should be shown, not step 1 or step 3 let expected_output = indoc! {r#" - ERR ----- Failed to build canister 'my-canister' ----- + ERR ----- Building failed: 'my-canister' ----- ERR 'command 'echo "step 2 failing"; exit 1' failed with status code 1' - ERR [my-canister] Build output: + ERR [my-canister] Building output: ERR [my-canister] Building: step 2 of 3 (script): ERR [my-canister] echo "step 2 failing"; exit 1: ERR [my-canister] > step 2 failing @@ -275,9 +275,9 @@ fn build_adapter_display_failing_prebuilt_output() { // Invoke build let expected_output = indoc! {r#" - ERR ----- Failed to build canister 'my-canister' ----- + ERR ----- Building failed: 'my-canister' ----- ERR 'failed to read wasm file at '/nonexistent/path/to/wasm.wasm'' - ERR [my-canister] Build output: + ERR [my-canister] Building output: ERR [my-canister] Building: step 2 of 2 (pre-built): ERR [my-canister] path: /nonexistent/path/to/wasm.wasm, sha: invalid: ERR [my-canister] > Reading wasm: /nonexistent/path/to/wasm.wasm @@ -319,9 +319,9 @@ fn build_adapter_display_failing_build_output_no_output() { // Invoke build let expected_output = indoc! {r#" - ERR ----- Failed to build canister 'my-canister' ----- + ERR ----- Building failed: 'my-canister' ----- ERR 'command 'exit 1' failed with status code 1' - ERR [my-canister] Build output: + ERR [my-canister] Building output: ERR [my-canister] Building: step 2 of 2 (script): ERR [my-canister] exit 1: ERR [my-canister] @@ -405,9 +405,9 @@ fn build_adapter_display_script_multiple_commands_output() { // Invoke build let expected_output = indoc! {r#" - ERR ----- Failed to build canister 'my-canister' ----- + ERR ----- Building failed: 'my-canister' ----- ERR 'build did not produce a wasm output file' - ERR [my-canister] Build output: + ERR [my-canister] Building output: ERR [my-canister] Building: step 1 of 1 (script): ERR [my-canister] echo "command 1": ERR [my-canister] echo "command 2": diff --git a/crates/icp-cli/tests/deploy_tests.rs b/crates/icp-cli/tests/deploy_tests.rs index 36241f796..6d446fe6a 100644 --- a/crates/icp-cli/tests/deploy_tests.rs +++ b/crates/icp-cli/tests/deploy_tests.rs @@ -818,7 +818,7 @@ async fn deploy_upgrade_rejects_incompatible_candid() { ]) .assert() .failure() - .stderr(contains("Candid interface compatibility check failed")); + .stderr(contains("Checking Candid compatibility failed")); // Deploy with --yes should succeed ctx.icp() diff --git a/crates/icp-events/Cargo.toml b/crates/icp-events/Cargo.toml index fb7adb41f..a215bd09a 100644 --- a/crates/icp-events/Cargo.toml +++ b/crates/icp-events/Cargo.toml @@ -6,6 +6,9 @@ license = { workspace = true } publish.workspace = true [dependencies] -candid = { workspace = true } serde = { workspace = true } tokio = { workspace = true, features = ["sync"] } + +[dev-dependencies] +serde_json = { workspace = true } +tokio = { workspace = true, features = ["macros", "rt"] } diff --git a/crates/icp-events/src/lib.rs b/crates/icp-events/src/lib.rs index 7753fbe66..e1b025d42 100644 --- a/crates/icp-events/src/lib.rs +++ b/crates/icp-events/src/lib.rs @@ -1,29 +1,38 @@ -//! Typed progress events passed from operations to the presentation layer. +//! Progress events passed from operations to a 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. +//! [`StepReporter`]); a consumer reads the stream and decides how to display +//! it. +//! +//! The crate is deliberately ignorant of *what* the work is. A task announces +//! itself with a [`Task`] descriptor — a title, an optional subject, and the +//! [`Shape`] of the widget it drives — and nothing here enumerates builds, +//! syncs or installs. That keeps the operation vocabulary out of every crate +//! that merely reports progress, and lets a host that has no terminal at all +//! (a deploy running inside a canister, say) consume the same stream. +//! +//! Chrome is the consumer's job: brackets, dashes, tick marks, indentation +//! and colour are all applied by whoever renders the stream. What travels on +//! the wire is the operation's own nouns. //! //! 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. +//! so a consumer 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. +/// Identifies one task within an event stream. Ids are assigned in +/// task-creation order, so consumers can 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); @@ -39,7 +48,23 @@ pub struct Event { #[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 }, + /// + /// `parent` is set when this task was spawned from another task's + /// reporter, which is how a composite operation (deploy calling build, + /// install, sync, …) forwards its children's progress onto one stream. + TaskStarted { + #[serde(skip_serializing_if = "Option::is_none")] + parent: Option, + /// What the work is being done *to* — a canister name, typically. + /// Consumers use it to attribute output lines. + #[serde(skip_serializing_if = "Option::is_none")] + subject: Option, + /// What the work *is*, in the operation's own words, phrased so it + /// reads as work in progress: "Building", "Checking compatibility". + title: String, + #[serde(flatten)] + shape: Shape, + }, /// A step of the task began. Steps within a task are sequential; /// `number` is 1-based. `label` describes the step (it may span @@ -51,16 +76,15 @@ pub enum EventKind { }, /// A shell command within the task's current step began executing. - /// Script steps run their commands in order; renderers can use this to + /// Script steps run their commands in order; consumers 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`). + /// How far a quantifiable task has come, against the `total` its + /// [`Shape::Counter`] declared. Progress { position: u64 }, /// The task's current step finished. @@ -70,79 +94,69 @@ pub enum EventKind { 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, - }, +/// The kind of widget a task drives — the one presentation fact an operation +/// legitimately knows, because it is a property of the work rather than of +/// the display. +#[derive(Debug, Clone, Copy, Serialize)] +#[serde(tag = "shape", rename_all = "snake_case")] +pub enum Shape { + /// An announcement with no live state of its own. It titles whatever + /// nests beneath it, and a childless one is simply a notice. + Group, + /// Work of unknown duration. + Spinner, + /// Quantifiable work, measured against `total` (bytes, today). + Counter { total: 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, +/// How a task announces itself. Built by the operation and handed to +/// [`Reporter::task`] or [`TaskReporter::subtask`]. +#[derive(Debug, Clone)] +pub struct Task { + subject: Option, + title: String, + shape: Shape, +} + +impl Task { + /// Work of unknown duration, e.g. an install. + pub fn new(title: impl Into) -> Self { + Self { + subject: None, + title: title.into(), + shape: Shape::Spinner, } } -} -/// 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, -} + /// An announcement that titles the tasks nested under it — a deploy + /// phase, say. With no children it is just a notice, so its title is + /// rendered verbatim rather than being decorated. + pub fn group(title: impl Into) -> Self { + Self { + subject: None, + title: title.into(), + shape: Shape::Group, + } + } -/// The snapshot blob being transferred. -#[derive(Debug, Clone, Copy, Serialize)] -#[serde(rename_all = "snake_case")] -pub enum TransferBlob { - WasmModule, - WasmMemory, - StableMemory, + /// Quantifiable work, reported through [`TaskReporter::progress`]. + pub fn counter(title: impl Into, total: u64) -> Self { + Self { + subject: None, + title: title.into(), + shape: Shape::Counter { total }, + } + } + + /// Name what the work is being done to, so output can be attributed to it. + pub fn on(mut self, subject: impl Into) -> Self { + self.subject = Some(subject.into()); + self + } } /// Where an output line came from. -#[derive(Debug, Clone, Copy, Serialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] #[serde(rename_all = "snake_case")] pub enum OutputStream { Stdout, @@ -151,7 +165,7 @@ pub enum OutputStream { Info, } -#[derive(Debug, Clone, Copy, Serialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] #[serde(rename_all = "snake_case")] pub enum StepOutcome { Succeeded, @@ -162,54 +176,117 @@ pub enum StepOutcome { #[serde(tag = "result", rename_all = "snake_case")] pub enum TaskOutcome { Succeeded { + /// Closing line for the task's widget. Left `None` when the operation + /// has nothing to add beyond "it worked", or when the widget has no + /// message slot to put it in. + #[serde(skip_serializing_if = "Option::is_none")] + message: Option, /// 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. + /// success — e.g. sync-plugin stderr, which a 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, - }, + Failed(Failure), /// 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 }, } +/// Everything a consumer needs to paint a failure. None of it is load-bearing +/// — the operation's `Err` is. +#[derive(Debug, Clone, Serialize)] +pub struct Failure { + /// Short description, kept terse enough for a progress bar. + pub message: String, + /// The rendered `source()` chain of the failure, outermost first. + #[serde(skip_serializing_if = "Vec::is_empty")] + pub causes: Vec, + /// Whether, and how, this failure is replayed once the live view is gone. + #[serde(flatten)] + pub report: FailureReport, + /// A note printed once after every dump, when at least one task failed + /// asking for it. Deduplicated across tasks. + #[serde(skip_serializing_if = "Option::is_none")] + pub epilogue: Option, +} + +/// What a failed task leaves behind after the live view is torn down. +#[derive(Debug, Clone, Serialize)] +#[serde(tag = "report", rename_all = "snake_case")] +pub enum FailureReport { + /// Nothing: the failure reaches the user on the command's returned error, + /// and a dump would only say it twice. + Silent, + /// The message, then its cause chain. + Summary, + /// These lines in place of the message and causes — for a failure whose + /// real content is a report rather than a sentence. + Detail { lines: Vec }, +} + impl TaskOutcome { - /// Success with nothing retained. + /// Success with nothing to say and nothing retained. pub fn succeeded() -> Self { TaskOutcome::Succeeded { + message: None, + retained_output: Vec::new(), + } + } + + /// Success with a closing line for the task's widget. + pub fn succeeded_with(message: impl Into) -> Self { + TaskOutcome::Succeeded { + message: Some(message.into()), retained_output: Vec::new(), } } - /// Failure with no cause chain. + /// Failure summarised by `message`, with no cause chain. pub fn failed(message: impl Into) -> Self { - TaskOutcome::Failed { + TaskOutcome::Failed(Failure::new(message)) + } + + /// Failure that the command's returned error already reports, so it needs + /// no deferred dump of its own. + pub fn failed_silently(message: impl Into) -> Self { + TaskOutcome::Failed(Failure { + report: FailureReport::Silent, + ..Failure::new(message) + }) + } +} + +impl Failure { + /// A failure summarised by `message` alone. + pub fn new(message: impl Into) -> Self { + Self { message: message.into(), causes: Vec::new(), + report: FailureReport::Summary, + epilogue: None, } } -} -/// 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, - ) + /// Attach the rendered `source()` chain, outermost first. + pub fn with_causes(mut self, causes: Vec) -> Self { + self.causes = causes; + self + } + + /// Replace the dumped summary with `lines`. + pub fn with_detail(mut self, lines: Vec) -> Self { + self.report = FailureReport::Detail { lines }; + self + } + + /// Add a note printed once after all dumps. + pub fn with_epilogue(mut self, epilogue: impl Into) -> Self { + self.epilogue = Some(epilogue.into()); + self + } } /// Create a connected reporter/receiver pair. The receiver yields `None` once @@ -221,14 +298,36 @@ pub fn channel() -> (Reporter, UnboundedReceiver) { tx, next_task_id: Arc::new(AtomicU64::new(0)), }), + parent: None, }; (reporter, rx) } +/// 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, + ) +} + /// Entry point handed to an operation; spawns [`TaskReporter`]s. +/// +/// A reporter carries the scope it was made in. An operation cannot tell the +/// difference between the reporter a command handed it and one scoped to a +/// parent task by a composite operation — which is what lets `deploy` run +/// `build_many` unmodified and have its tasks nest under the build phase. #[derive(Debug, Clone)] pub struct Reporter { inner: Option, + /// Task the reporter's tasks nest under, if any. + parent: Option, } #[derive(Debug, Clone)] @@ -237,33 +336,55 @@ struct ReporterInner { next_task_id: Arc, } +impl ReporterInner { + fn start(&self, parent: Option, task: Task) -> TaskReporter { + let task_id = TaskId(self.next_task_id.fetch_add(1, Ordering::Relaxed)); + let _ = self.tx.send(Event { + task_id, + kind: EventKind::TaskStarted { + parent, + subject: task.subject, + title: task.title, + shape: task.shape, + }, + }); + TaskReporter { + inner: Some(self.clone()), + task_id, + } + } +} + impl Reporter { /// A reporter whose events go nowhere. pub fn null() -> Self { - Self { inner: None } + Self { + inner: None, + parent: 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, + /// Begin a task, emitting [`EventKind::TaskStarted`]. It nests under + /// whatever scope this reporter carries. + pub fn task(&self, task: Task) -> TaskReporter { + match &self.inner { + Some(inner) => inner.start(self.parent, task), + None => TaskReporter::null(), } } + + /// Announce something in passing: a task with no work under it, finished + /// as soon as it is started. + pub fn notice(&self, text: impl Into) { + self.task(Task::group(text)) + .finish(TaskOutcome::succeeded()); + } } -/// Reports the lifecycle of one task. +/// Reports the lifecycle of one task, and spawns the tasks nested under it. #[derive(Debug, Clone)] pub struct TaskReporter { - tx: Option>, + inner: Option, task_id: TaskId, } @@ -271,14 +392,32 @@ impl TaskReporter { /// A task reporter whose events go nowhere. pub fn null() -> Self { Self { - tx: None, + inner: None, task_id: TaskId(0), } } + /// Begin a task nested under this one. A composite operation uses this to + /// forward the progress of the operations it calls onto one stream. + pub fn subtask(&self, task: Task) -> TaskReporter { + match &self.inner { + Some(inner) => inner.start(Some(self.task_id), task), + None => TaskReporter::null(), + } + } + + /// A reporter scoped to this task. Hand it to an operation that does not + /// know it is being composed: whatever tasks it starts nest here. + pub fn reporter(&self) -> Reporter { + Reporter { + inner: self.inner.clone(), + parent: Some(self.task_id), + } + } + fn send(&self, kind: EventKind) { - if let Some(tx) = &self.tx { - let _ = tx.send(Event { + if let Some(inner) = &self.inner { + let _ = inner.tx.send(Event { task_id: self.task_id, kind, }); @@ -294,17 +433,30 @@ impl TaskReporter { label: label.into(), }); StepReporter { - tx: self.tx.clone(), + tx: self.inner.as_ref().map(|inner| inner.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. + /// Report how far the task has come, against the total its + /// [`Shape::Counter`] declared. pub fn progress(&self, position: u64) { self.send(EventKind::Progress { position }); } + /// Emit a line of output attributed to this task rather than to a step. + pub fn output(&self, stream: OutputStream, line: impl Into) { + self.send(EventKind::Output { + stream, + line: line.into(), + }); + } + + /// Emit a progress note from icp itself. + pub fn info(&self, line: impl Into) { + self.output(OutputStream::Info, line); + } + /// Finish the task, emitting [`EventKind::TaskCompleted`]. No further /// events should be sent for this task. pub fn finish(&self, outcome: TaskOutcome) { @@ -313,6 +465,9 @@ impl TaskReporter { } /// Reports output produced during one step of a task. +/// +/// This handle carries no task vocabulary at all, which is what lets the core +/// library's ports (`Arc`, `Arc`) accept one. #[derive(Debug, Clone)] pub struct StepReporter { tx: Option>, @@ -373,3 +528,347 @@ impl StepReporter { self.send(EventKind::StepCompleted { outcome }); } } + +#[cfg(test)] +mod tests { + use super::*; + + fn drain(rx: &mut UnboundedReceiver) -> Vec { + let mut events = Vec::new(); + while let Ok(event) = rx.try_recv() { + events.push(event); + } + events + } + + #[tokio::test] + async fn null_reporters_emit_nothing_and_never_panic() { + let reporter = Reporter::null(); + let task = reporter.task(Task::new("Working").on("thing")); + let child = task.subtask(Task::new("Nested")); + let step = task.step(1, 1, "only"); + + // Every method must be safe to call with no receiver attached. + step.command("echo hi"); + step.stdout("out"); + step.stderr("err"); + step.info("note"); + step.done(StepOutcome::Succeeded); + task.progress(42); + task.info("note"); + child.finish(TaskOutcome::succeeded()); + task.finish(TaskOutcome::succeeded()); + + // Handles derived from a null reporter are themselves null. + assert!(TaskReporter::null().inner.is_none()); + assert!(StepReporter::null().tx.is_none()); + assert!(task.reporter().inner.is_none()); + } + + #[tokio::test] + async fn task_ids_follow_creation_order_across_nesting() { + let (reporter, mut rx) = channel(); + let parent = reporter.task(Task::group("Phase")); + let first = parent.subtask(Task::new("A")); + let second = parent.subtask(Task::new("B")); + + // Finishing out of order must not disturb the ids. + second.finish(TaskOutcome::succeeded()); + first.finish(TaskOutcome::succeeded()); + parent.finish(TaskOutcome::succeeded()); + + let started: Vec<(TaskId, Option)> = drain(&mut rx) + .into_iter() + .filter_map(|e| match e.kind { + EventKind::TaskStarted { parent, .. } => Some((e.task_id, parent)), + _ => None, + }) + .collect(); + assert_eq!( + started, + vec![ + (TaskId(0), None), + (TaskId(1), Some(TaskId(0))), + (TaskId(2), Some(TaskId(0))), + ] + ); + } + + /// A composite operation hands plain `Reporter`s to the operations it + /// calls; those must still land in the same stream, nested under it. + #[tokio::test] + async fn a_task_can_hand_out_a_reporter_for_uncomposed_callees() { + let (reporter, mut rx) = channel(); + let phase = reporter.task(Task::group("Phase")); + let inner = phase.reporter(); + inner + .task(Task::new("Callee")) + .finish(TaskOutcome::succeeded()); + + let started: Vec<(String, Option)> = drain(&mut rx) + .into_iter() + .filter_map(|e| match e.kind { + EventKind::TaskStarted { title, parent, .. } => Some((title, parent)), + _ => None, + }) + .collect(); + assert_eq!( + started, + vec![ + ("Phase".to_owned(), None), + // The callee asked for a top-level task and got a child of + // the phase, without knowing the difference. + ("Callee".to_owned(), Some(TaskId(0))), + ] + ); + } + + #[tokio::test] + async fn events_arrive_in_emission_order_and_carry_their_task_id() { + let (reporter, mut rx) = channel(); + let task = reporter.task(Task::new("Building").on("frontend")); + let step = task.step(1, 2, "compile"); + step.command("make"); + step.stdout("line one"); + step.stderr("line two"); + step.done(StepOutcome::Succeeded); + task.finish(TaskOutcome::succeeded()); + + let events = drain(&mut rx); + assert!(events.iter().all(|e| e.task_id == TaskId(0))); + + let shape: Vec = events + .iter() + .map(|e| match &e.kind { + EventKind::TaskStarted { title, .. } => format!("started {title}"), + EventKind::StepStarted { + number, + total, + label, + } => format!("step {number}/{total} {label}"), + EventKind::CommandStarted { command } => format!("$ {command}"), + EventKind::Output { stream, line } => format!("{stream:?} {line}"), + EventKind::Progress { position } => format!("progress {position}"), + EventKind::StepCompleted { .. } => "step done".to_owned(), + EventKind::TaskCompleted { .. } => "task done".to_owned(), + }) + .collect(); + assert_eq!( + shape, + vec![ + "started Building", + "step 1/2 compile", + "$ make", + "Stdout line one", + "Stderr line two", + "step done", + "task done", + ] + ); + } + + /// A step reporter keeps its own channel handle, so output emitted after + /// the task reporter is gone still arrives. This is what makes the + /// spawned stdout/stderr readers in the core library safe. + #[tokio::test] + async fn step_reporter_outlives_its_task_reporter() { + let (reporter, mut rx) = channel(); + let step = { + let task = reporter.task(Task::new("Working")); + task.step(1, 1, "run") + }; + step.stdout("after the task handle dropped"); + + let lines: Vec = drain(&mut rx) + .into_iter() + .filter_map(|e| match e.kind { + EventKind::Output { line, .. } => Some(line), + _ => None, + }) + .collect(); + assert_eq!(lines, vec!["after the task handle dropped".to_owned()]); + } + + /// The receiver must close once every handle is gone, otherwise a + /// consumer driving the stream to completion would hang. + #[tokio::test] + async fn stream_closes_when_all_handles_drop() { + let (reporter, mut rx) = channel(); + let task = reporter.task(Task::new("Working")); + let step = task.step(1, 1, "run"); + + drop(reporter); + drop(task); + assert!(rx.recv().await.is_some(), "buffered events still deliver"); + + drop(step); + while rx.recv().await.is_some() {} + // Reaching here means recv() returned None rather than hanging. + } + + /// The stream is the CLI's structured account of a run, so its tags and + /// field names are pinned here rather than left to chance. + #[test] + fn wire_format_is_stable() { + let event = |kind| { + serde_json::to_value(Event { + task_id: TaskId(7), + kind, + }) + .expect("event should serialize") + }; + + assert_eq!( + event(EventKind::TaskStarted { + parent: Some(TaskId(2)), + subject: Some("frontend".to_owned()), + title: "Building".to_owned(), + shape: Shape::Spinner, + }), + serde_json::json!({ + "task_id": 7, "event": "task_started", "parent": 2, + "subject": "frontend", "title": "Building", "shape": "spinner", + }) + ); + // A top-level task with no subject omits both fields rather than + // sending nulls. + assert_eq!( + event(EventKind::TaskStarted { + parent: None, + subject: None, + title: "Building canisters:".to_owned(), + shape: Shape::Group, + }), + serde_json::json!({ + "task_id": 7, "event": "task_started", + "title": "Building canisters:", "shape": "group", + }) + ); + assert_eq!( + event(EventKind::TaskStarted { + parent: None, + subject: None, + title: "WASM module".to_owned(), + shape: Shape::Counter { total: 4096 }, + }), + serde_json::json!({ + "task_id": 7, "event": "task_started", + "title": "WASM module", "shape": "counter", "total": 4096, + }) + ); + assert_eq!( + event(EventKind::StepStarted { + number: 1, + total: 3, + label: "compile".to_owned(), + }), + serde_json::json!({ + "task_id": 7, "event": "step_started", + "number": 1, "total": 3, "label": "compile", + }) + ); + assert_eq!( + event(EventKind::CommandStarted { + command: "make build".to_owned(), + }), + serde_json::json!({ + "task_id": 7, "event": "command_started", "command": "make build", + }) + ); + assert_eq!( + event(EventKind::Output { + stream: OutputStream::Stderr, + line: "boom".to_owned(), + }), + serde_json::json!({ + "task_id": 7, "event": "output", "stream": "stderr", "line": "boom", + }) + ); + assert_eq!( + event(EventKind::Progress { position: 1024 }), + serde_json::json!({ "task_id": 7, "event": "progress", "position": 1024 }) + ); + assert_eq!( + event(EventKind::StepCompleted { + outcome: StepOutcome::Failed, + }), + serde_json::json!({ "task_id": 7, "event": "step_completed", "outcome": "failed" }) + ); + + // Task outcomes nest under `outcome`, internally tagged by `result`. + assert_eq!( + event(EventKind::TaskCompleted { + outcome: TaskOutcome::succeeded(), + }), + serde_json::json!({ + "task_id": 7, "event": "task_completed", + "outcome": { "result": "succeeded" }, + }) + ); + assert_eq!( + event(EventKind::TaskCompleted { + outcome: TaskOutcome::Succeeded { + message: Some("Built successfully".to_owned()), + retained_output: vec!["kept".to_owned()], + }, + }), + serde_json::json!({ + "task_id": 7, "event": "task_completed", + "outcome": { + "result": "succeeded", "message": "Built successfully", + "retained_output": ["kept"], + }, + }) + ); + assert_eq!( + event(EventKind::TaskCompleted { + outcome: TaskOutcome::Failed( + Failure::new("no") + .with_causes(vec!["because".to_owned()]) + .with_epilogue("try harder"), + ), + }), + serde_json::json!({ + "task_id": 7, "event": "task_completed", + "outcome": { + "result": "failed", "message": "no", "causes": ["because"], + "report": "summary", "epilogue": "try harder", + }, + }) + ); + assert_eq!( + event(EventKind::TaskCompleted { + outcome: TaskOutcome::Failed( + Failure::new("no").with_detail(vec!["why".to_owned()]) + ), + }), + serde_json::json!({ + "task_id": 7, "event": "task_completed", + "outcome": { + "result": "failed", "message": "no", + "report": "detail", "lines": ["why"], + }, + }) + ); + assert_eq!( + event(EventKind::TaskCompleted { + outcome: TaskOutcome::failed_silently("no"), + }), + serde_json::json!({ + "task_id": 7, "event": "task_completed", + "outcome": { "result": "failed", "message": "no", "report": "silent" }, + }) + ); + assert_eq!( + event(EventKind::TaskCompleted { + outcome: TaskOutcome::Skipped { + reason: "not an upgrade".to_owned(), + }, + }), + serde_json::json!({ + "task_id": 7, "event": "task_completed", + "outcome": { "result": "skipped", "reason": "not an upgrade" }, + }) + ); + } +}